feat(effects): exact-new-set guard closes the over-write false-pass gap

docs/EFFECT_KIT.md

@@ -22,7 +22,8 @@ from the reference apps.
 
 1. **The bundle declares WHAT must be true** — typed `Effect` contracts on
    each consequential step (`record_written` for mutations; `field_equals`
-   for a unique persisted field or independently read business outcome),
+   for a unique persisted field or independently read business outcome;
+   `exact_new_set` for the full set of records an action may add),
    at-most-once counts, idempotency keys, and `{param: ...}` references that
    bind to the run's governed parameters. Contracts are substrate-neutral.
 2. **The deployment declares WHERE truth lives** — the `effects:` section of
@@ -58,8 +59,8 @@ from the reference apps.
 
 All substrates share one judge (`runtime/effects/_common.py`), so
 at-most-once counting, idempotency-key de-duplication, field read-back,
-collateral-loss detection, and the duplicate-write guard below behave
-identically everywhere.
+collateral-loss detection, the duplicate-write guard, and the `exact_new_set`
+over-write guard below behave identically everywhere.
 
 ### The duplicate-write / idempotency guard (`count_new_only`)
 
@@ -70,6 +71,63 @@ legitimately matches pre-existing rows (e.g. "an encounter for this patient").
 It requires a readable pre-state — an unreachable baseline is INDETERMINATE →
 HALT, never a guess. Available on every substrate.
 
+### The over-write guard (`exact_new_set`) — opt-in, and why you want it
+
+Every other kind answers **"is my record there?"**. None of them answers
+**"and nothing else?"**. A contract set that declares one `record_written`
+per intended new record is silent about the records it never named, so an
+actuation that writes the 6 intended rows **and 31 unintended ones** satisfies
+every declared contract: the runtime CONFIRMS while the system of record holds
+writes nobody asked for. That is a **false pass** — the one error direction
+this design must never take. (The 6-vs-37 case is not hypothetical: it was
+measured in a 150-trial benchmark study of an agent asked to download 6
+records.)
+
+`Effect(kind=exact_new_set, ...)` closes it. One **table-scoped** effect
+declares the FULL set of records the action may add:
+
+```yaml
+effects:
+  - kind: exact_new_set
+    # `match` is the SCOPE, not a target selector. Empty = the whole read set.
+    match: {user_id: "32"}
+    # One selector per intended record. Repeat a selector to declare that many
+    # identical additions. Values may be literals or {param: ...} references.
+    new_records:
+      - {user_id: "32", song_id: "199"}
+      - {user_id: "32", song_id: "9"}
+    # Must equal len(new_records). Stated explicitly so an edit that drops a
+    # member fails loud instead of silently weakening the contract.
+    expected_count: 2
+    # How a record ADDED by this action is told apart from one already there.
+    # A surrogate key is the RIGHT choice here even though it is the wrong
+    # thing to pin in a selector.
+    identity_field: id
+```
+
+It REFUTES: an addition no member names (the guard), a missing or duplicated
+member, a wrong cardinality, and collateral loss inside the same scope. It
+requires a **real pre-action baseline**: with an unreachable baseline, or a
+record on either side carrying no `identity_field` value, the added set cannot
+be enumerated and the verdict is INDETERMINATE → HALT. Newness is never
+guessed. `new_records: []` with `expected_count: 0` is the meaningful
+assertion *"this action adds NOTHING to this read set."*
+
+Available on every substrate (it runs in the shared judge). It is judged
+against the pre-action snapshot, so the current-state read-back paths (durable
+resume, attended qualified read-back) refuse it rather than judge it against a
+synthesized empty baseline.
+
+**Backward compatibility and the honest boundary.** This kind is **additive
+and opt-in**. Flow contracts are operator-authored — there is no derivation
+step that could turn the guard on for you — so every contract written before
+this option judges **exactly** as it did before, and its `contract_hash` is
+byte-identical (the new fields enter the digest only on the new kind).
+The boundary follows directly: **an existing contract does not detect an
+over-write unless the operator declares an `exact_new_set` effect for that
+read set.** Declare one on any step where an unintended extra write would
+matter.
+
 ### The SQL table-delta audit
 
 `capture_table_counts(connect, tables)` + `audit_table_deltas(before, after,

docs/design/EFFECT_VERIFIER.md

@@ -35,8 +35,10 @@ class EffectVerifier(Protocol):
 ```
 
 - **`Effect`** is the RFC's typed effect. Kinds: `record_written` (a record
-  matching a selector exists *exactly* `expected_count` times — at-most-once)
-  and `field_equals` (a read-back of one field). Substrate-neutral: the SAME
+  matching a selector exists *exactly* `expected_count` times — at-most-once),
+  `field_equals` (a read-back of one field), and `exact_new_set` (the records
+  ADDED inside a scope are EXACTLY the declared set — the over-write guard;
+  see [`docs/EFFECT_KIT.md`](../EFFECT_KIT.md)). Substrate-neutral: the SAME
   `Effect` is checked by every verifier.
 - **`capture_pre_state`** snapshots the system of record *before* the action —
   a baseline for delta/at-most-once counting and for collateral-loss

openadapt_flow/bundle_validation.py

@@ -50,6 +50,7 @@
 from openadapt_flow.ir import (
     BundleManifest,
     BundleProvenance,
+    Effect,
     Interstitial,
     ParamKind,
     ParamSpec,
@@ -183,6 +184,26 @@ class _SealedCanonicalOmission:
                 "field; a populated band remains sealed content"
             ),
         ),
+        _SealedCanonicalOmission(
+            owner=Effect,
+            field_name="new_records",
+            omitted_values=([],),
+            reason=(
+                "the empty declared-addition set predates the v2 field; only "
+                "an exact_new_set effect populates it, and a populated set is "
+                "the whole over-write contract and remains sealed content"
+            ),
+        ),
+        _SealedCanonicalOmission(
+            owner=Effect,
+            field_name="identity_field",
+            omitted_values=("id",),
+            reason=(
+                "the 'id' default predates the v2 field and is inert on every "
+                "kind but exact_new_set; a contract naming ANY other identity "
+                "column remains sealed content"
+            ),
+        ),
     ),
 }
 

openadapt_flow/runtime/effects/_common.py

@@ -3,9 +3,10 @@
 
 Every substrate (REST, FHIR, filesystem) normalizes its system of record into
 a list of plain dicts and calls :func:`judge_records`, so the *decision* logic
--- at-most-once counting, idempotency-key de-duplication, field read-back, and
-collateral-loss detection -- lives in exactly ONE place (the same
-single-source-of-truth discipline the fault-model study uses for ``classify``).
+-- at-most-once counting, idempotency-key de-duplication, field read-back,
+collateral-loss detection, and the ``exact_new_set`` over-write guard -- lives
+in exactly ONE place (the same single-source-of-truth discipline the
+fault-model study uses for ``classify``).
 """
 
 from __future__ import annotations
@@ -78,6 +79,9 @@ def judge_records(
             unavailable=True,
         )
 
+    if effect.kind is EffectKind.EXACT_NEW_SET:
+        return _judge_exact_new_set(effect, before, current, substrate)
+
     matched = [r for r in current if record_matches(r, effect.match)]
     if effect.idempotency_key is not None:
         matched = [
@@ -106,6 +110,146 @@ def judge_records(
     return _judge_record_written(effect, before, current, matched, substrate)
 
 
+#: How many offending records one refutation reason names before it stops
+#: listing them. The COUNT is always exact; the sample keeps an audit line
+#: readable when an agent added dozens of unintended rows.
+_SAMPLE_LIMIT = 5
+
+
+def _selector_text(selector: dict[str, Any]) -> str:
+    return "{" + ", ".join(f"{k}={v}" for k, v in sorted(selector.items())) + "}"
+
+
+def _judge_exact_new_set(
+    effect: Effect,
+    before: EffectState,
+    current: list[dict[str, Any]],
+    substrate: str,
+) -> EffectVerdict:
+    """Judge the ``exact_new_set`` over-write guard.
+
+    The claim: the records ADDED to the scoped read set by this action are
+    EXACTLY ``effect.new_records`` -- each declared member added exactly as
+    many times as it is declared, and NO record added that no member names.
+    ``effect.match`` is the SCOPE (empty = the whole read set), not a target
+    selector.
+
+    Refuses (INDETERMINATE) rather than guesses when the added set cannot be
+    enumerated: an unreachable baseline, or any record on either side missing
+    ``effect.identity_field``.
+    """
+    if not before.reachable:
+        return _indeterminate(
+            effect,
+            substrate,
+            "exact_new_set requires a readable pre-state baseline, but the "
+            "system of record was unreachable before the action -- the set of "
+            "records this action ADDED cannot be enumerated; HALT",
+        )
+
+    identity = effect.identity_field
+    for label, records in (("pre-state", before.records), ("current", current)):
+        missing = sum(1 for r in records if r.get(identity, None) is None)
+        if missing:
+            return _indeterminate(
+                effect,
+                substrate,
+                f"exact_new_set cannot enumerate the added set: {missing} "
+                f"{label} record(s) carry no {identity!r} value. Supply an "
+                "identity_field that every record of the read set carries "
+                "(widen the query to return a stable identity column), or "
+                "remove this guard -- newness is never guessed",
+            )
+
+    before_identities = {str(r[identity]) for r in before.records}
+    in_scope = [r for r in current if record_matches(r, effect.match)]
+    added = [r for r in in_scope if str(r[identity]) not in before_identities]
+
+    reasons: list[str] = []
+
+    # (1) The headline: how many records the action added versus how many it
+    # was allowed to add.
+    if len(added) != effect.expected_count:
+        reasons.append(
+            f"the action added {len(added)} record(s) to the scoped read set "
+            f"but the contract declares exactly {effect.expected_count}"
+        )
+
+    # (2) Per declared member: present exactly as many times as declared.
+    declared: dict[str, tuple[dict[str, Any], int]] = {}
+    for selector in effect.new_records:
+        plain = {k: str(v) for k, v in selector.items()}
+        key = _selector_text(plain)
+        found, multiplicity = declared.get(key, (plain, 0))
+        declared[key] = (found, multiplicity + 1)
+    for key, (selector, multiplicity) in sorted(declared.items()):
+        observed = sum(1 for r in added if record_matches(r, selector))
+        if observed != multiplicity:
+            reasons.append(
+                f"declared new record {key} was added {observed} time(s), "
+                f"expected {multiplicity}"
+            )
+
+    # (3) THE GUARD: a record the action added that no declared member names.
+    extra = [
+        r
+        for r in added
+        if not any(
+            record_matches(r, {k: str(v) for k, v in selector.items()})
+            for selector in effect.new_records
+        )
+    ]
+    if extra:
+        sample = ", ".join(
+            _selector_text({k: str(v) for k, v in r.items() if k != identity})
+            for r in extra[:_SAMPLE_LIMIT]
+        )
+        hidden = len(extra) - _SAMPLE_LIMIT
+        more = "" if hidden <= 0 else f", and {hidden} more"
+        reasons.append(
+            f"{len(extra)} record(s) the action added are NOT in the declared "
+            f"set -- unintended write(s) to the system of record: {sample}{more}"
+        )
+
+    # (4) Removals inside the scope this effect speaks for.
+    if effect.forbid_collateral_loss:
+        current_identities = {str(r[identity]) for r in current}
+        lost = [
+            r
+            for r in before.records
+            if record_matches(r, effect.match)
+            and str(r[identity]) not in current_identities
+        ]
+        if lost:
+            reasons.append(
+                f"{len(lost)} pre-existing record(s) inside the declared scope "
+                "vanished -- collateral loss"
+            )
+
+    if reasons:
+        return EffectVerdict(
+            verdict=Verdict.REFUTED,
+            kind=effect.kind,
+            substrate=substrate,
+            reason="; ".join(reasons),
+            observed_count=len(added),
+            expected_count=effect.expected_count,
+            matched_records=added,
+        )
+    return EffectVerdict(
+        verdict=Verdict.CONFIRMED,
+        kind=effect.kind,
+        substrate=substrate,
+        reason=(
+            f"the action added exactly the {effect.expected_count} declared "
+            "record(s) to the scoped read set, and nothing else"
+        ),
+        observed_count=len(added),
+        expected_count=effect.expected_count,
+        matched_records=added,
+    )
+
+
 def _judge_record_written(
     effect: Effect,
     before: EffectState,

openadapt_flow/runtime/effects/adapter.py

@@ -734,13 +734,13 @@ def requires_readable_pre_state_for(self, effect: Effect) -> bool:
         requirement = getattr(candidate, "requires_readable_pre_state_for", None)
         if callable(requirement):
             return bool(requirement(effect))
-        return bool(effect.count_new_only or effect.forbid_collateral_loss)
+        return bool(effect.requires_baseline or effect.forbid_collateral_loss)
 
     @staticmethod
     def _pre_state_requirement(candidate: Any, effect: Effect) -> bool:
         requirement = getattr(candidate, "requires_readable_pre_state_for", None)
         if not callable(requirement):
-            return bool(effect.count_new_only or effect.forbid_collateral_loss)
+            return bool(effect.requires_baseline or effect.forbid_collateral_loss)
         isolated = effect.model_copy(deep=True)
         original = isolated.model_dump(mode="json")
         required = bool(requirement(isolated))