From 6fc2e273da86d24ea7de4cd62de0ce4e4b861e85 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Fri, 14 Aug 2026 22:48:10 +0200 Subject: [PATCH 1/2] feat(effects): exact-new-set guard closes the over-write false-pass gap `_judge_record_written` verifies that the DECLARED record matches at the expected cardinality and, via `forbid_collateral_loss`, that no pre-existing record vanished. Nothing verified that the action added no records BEYOND the declared set. An actuation that wrote the intended record AND N unintended ones therefore satisfied every declared contract and was CONFIRMED: a FALSE PASS in the shipping runtime, the one error direction this design must never take. Measured, not hypothetical -- in a 150-trial benchmark study an agent asked to download 6 records downloaded 37, and every declared-record contract confirmed. Add EffectKind.EXACT_NEW_SET: one table-scoped effect declaring the FULL set of records an action may add (`new_records`, `expected_count == len(new_records)`, `identity_field`, `match` as SCOPE). It refutes an addition no member names, a missing or duplicated member, a wrong cardinality, and collateral loss inside the scope. It requires a REAL pre-action baseline and returns a structured INDETERMINATE refusal -- never a guess -- when the baseline is unreachable or a record carries no `identity_field` value. It rides flow's existing machinery unchanged: contract loading, the shared judge, aggregation, coverage, evidence tiers, receipts, the effect journal, and `EffectVerdict`. `Effect.requires_baseline` replaces the open-coded `count_new_only or forbid_collateral_loss` pre-state test in the candidate selector and the replayer, and the current-state read-back paths (durable resume, attended qualified read-back) now refuse a delta claim instead of judging it against a synthesized empty baseline. Backward compatibility: the kind is additive and OPT-IN. Flow contracts are operator-authored -- there is no derivation step that could default it on -- so every existing contract judges exactly as before and keeps a byte-identical `contract_hash` (the new fields enter the digest only on the new kind), pinned by a regression test. The honest boundary is stated in docs/EFFECT_KIT.md: an existing contract does not detect an over-write unless the operator declares this effect. The three shared oracle modules are ported so they stay a VERBATIM extraction of openadapt-attest's `openadapt_attest/oracle/` (attest main 3056285), modulo the import rewrites its UPSTREAM.lock declares, keeping the next re-extraction clean and its drift job meaningful. Co-Authored-By: Claude Fable 5 --- docs/EFFECT_KIT.md | 64 ++++- docs/design/EFFECT_VERIFIER.md | 6 +- openadapt_flow/runtime/effects/_common.py | 150 +++++++++++- openadapt_flow/runtime/effects/adapter.py | 4 +- openadapt_flow/runtime/effects/effect.py | 124 +++++++++- openadapt_flow/runtime/replayer.py | 19 +- tests/test_effect_verifier.py | 276 ++++++++++++++++++++++ tests/test_replayer_effects.py | 43 ++++ 8 files changed, 669 insertions(+), 17 deletions(-) diff --git a/docs/EFFECT_KIT.md b/docs/EFFECT_KIT.md index 9dd3a7ea..9a63128a 100644 --- a/docs/EFFECT_KIT.md +++ b/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, diff --git a/docs/design/EFFECT_VERIFIER.md b/docs/design/EFFECT_VERIFIER.md index 273bdc26..c9a86807 100644 --- a/docs/design/EFFECT_VERIFIER.md +++ b/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 diff --git a/openadapt_flow/runtime/effects/_common.py b/openadapt_flow/runtime/effects/_common.py index 26421e9b..457c9923 100644 --- a/openadapt_flow/runtime/effects/_common.py +++ b/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, diff --git a/openadapt_flow/runtime/effects/adapter.py b/openadapt_flow/runtime/effects/adapter.py index a79e6844..559e7de1 100644 --- a/openadapt_flow/runtime/effects/adapter.py +++ b/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)) diff --git a/openadapt_flow/runtime/effects/effect.py b/openadapt_flow/runtime/effects/effect.py index 7f055403..87bd6abe 100644 --- a/openadapt_flow/runtime/effects/effect.py +++ b/openadapt_flow/runtime/effects/effect.py @@ -166,6 +166,15 @@ class EffectKind(str, Enum): #: (and catches a partial save); for a read-only workflow it independently #: verifies the declared business outcome against the system of record. FIELD_EQUALS = "field_equals" + #: The set of records ADDED to the read set (scoped by :attr:`Effect.match`) + #: must be EXACTLY the declared set :attr:`Effect.new_records` -- every + #: declared record present once, and NO additional new record. This is the + #: over-write guard: a per-record ``record_written`` effect answers "is my + #: record there?" and is silent about the records it never named, so an + #: agent that wrote the 6 intended rows AND 31 unintended ones satisfies + #: every per-record contract. Requires a REAL pre-state baseline (a set + #: delta against an unknown baseline is never guessed -> INDETERMINATE). + EXACT_NEW_SET = "exact_new_set" #: A screen region ``(x, y, w, h)`` in the recorded/live frame's pixel space. @@ -274,7 +283,28 @@ class Effect(BaseModel): value: Optional[ValueExpr] = None #: ``record_written`` only: how many matching records must exist. 1 is the #: at-most-once contract for a consequential write; 0 asserts absence. + #: ``exact_new_set``: the declared CARDINALITY of the added set, which must + #: equal ``len(new_records)`` (validated) -- an explicit, hashed number so + #: a hand edit that deletes one member of the set fails loud instead of + #: silently weakening the contract. expected_count: int = 1 + #: ``exact_new_set`` only: the declared set of records the action may add, + #: one selector per intended record (same matching rules as :attr:`match`, + #: each value a literal or a run-``param`` reference). Repeating an + #: identical selector declares that many identical additions. Empty with + #: ``expected_count: 0`` is the meaningful assertion "this action adds + #: NOTHING to this read set". + new_records: list[dict[str, ValueExpr]] = Field(default_factory=list) + #: ``exact_new_set`` only: the record field whose value gives a record its + #: stable identity, used to tell records ADDED by the action from records + #: that were already there. It must be present on every record of both + #: snapshots; a record missing it makes the verdict INDETERMINATE (the + #: added set cannot be enumerated, so it is never guessed). An + #: environment-assigned surrogate key is the RIGHT choice here even though + #: it is the wrong thing to pin in a selector: it cannot identify the + #: INTENDED record across runs, but it does distinguish a new row from an + #: old one within one run. + identity_field: str = "id" #: Optional idempotency / at-most-once key. When set, ``record_written`` #: counts records bearing THIS key (via :attr:`key_field`) and requires #: exactly :attr:`expected_count` -- so a duplicate submission that reused @@ -310,7 +340,11 @@ class Effect(BaseModel): #: pre-state (``before``) and does NOT match :attr:`match` has since #: vanished -- collateral loss. This is what catches a stale / lost-update #: (last-write-wins) fault: our row lands (count 1, looks fine) while a - #: concurrent actor's row was silently destroyed. + #: concurrent actor's row was silently destroyed. On an ``exact_new_set`` + #: effect :attr:`match` is a SCOPE rather than a target, so the rule reads + #: the other way round: a pre-state record INSIDE the scope that has + #: vanished is the collateral loss (an exact-set claim about additions + #: must not quietly tolerate removals inside the same scope). forbid_collateral_loss: bool = True #: Consequential-write flag (mirrors ``Step.risk`` / RFC ``State.risk``). #: Compensation (``effects.compensation``) only fires for irreversible @@ -373,6 +407,20 @@ def _coerce_match(cls, v: Any) -> Any: def _coerce_value(cls, v: Any) -> Any: return cls._coerce_expr(v) + @field_validator("new_records", mode="before") + @classmethod + def _coerce_new_records(cls, v: Any) -> Any: + if isinstance(v, list): + return [ + ( + {k: cls._coerce_expr(val) for k, val in selector.items()} + if isinstance(selector, dict) + else selector + ) + for selector in v + ] + return v + @model_validator(mode="after") def _count_new_only_scope(self) -> "Effect": """``count_new_only`` is a ``record_written`` guard; refuse (fail @@ -380,10 +428,57 @@ def _count_new_only_scope(self) -> "Effect": if self.count_new_only and self.kind is not EffectKind.RECORD_WRITTEN: raise ValueError( "count_new_only applies only to record_written effects " - "(a field_equals read-back has no newness delta)" + "(a field_equals read-back has no newness delta; an " + "exact_new_set effect always counts new records only)" + ) + return self + + @model_validator(mode="after") + def _exact_new_set_shape(self) -> "Effect": + """Refuse an ``exact_new_set`` whose declared set is unusable, and + refuse ``new_records`` on any other kind rather than ignore it.""" + if self.kind is not EffectKind.EXACT_NEW_SET: + if self.new_records: + raise ValueError( + "new_records applies only to exact_new_set effects " + f"(this effect is {self.kind.value})" + ) + return self + for position, selector in enumerate(self.new_records, start=1): + if not selector: + raise ValueError( + f"exact_new_set member {position} has an EMPTY selector, " + "which matches every record -- every member must name at " + "least one field" + ) + if self.expected_count != len(self.new_records): + raise ValueError( + "an exact_new_set declares expected_count == len(new_records) " + f"(got expected_count={self.expected_count} for " + f"{len(self.new_records)} declared record(s)); the cardinality " + "is stated explicitly so an edit that drops a member of the " + "set fails loud instead of silently weakening the contract" + ) + if not self.identity_field: + raise ValueError( + "an exact_new_set requires a non-empty identity_field naming " + "the record field that distinguishes a new record from a " + "pre-existing one" ) return self + @property + def requires_baseline(self) -> bool: + """Whether judging this effect needs a REAL pre-action snapshot. + + True for a ``count_new_only`` delta and for every ``exact_new_set`` + guard: both answer "what did THIS action add?", which is unanswerable + against an unknown or synthetic-empty baseline. Callers that would + otherwise supply a post-hoc empty baseline (``check.run_check``, the + gate) must refuse instead -- a set delta is never guessed. + """ + return self.count_new_only or self.kind is EffectKind.EXACT_NEW_SET + # -- run-time parameter binding (P0-3) ----------------------------------- def resolve( self, @@ -403,6 +498,10 @@ def resolve( deep=True, update={ "match": {k: v.resolved(params) for k, v in self.match.items()}, + "new_records": [ + {k: v.resolved(params) for k, v in selector.items()} + for selector in self.new_records + ], "value": None if self.value is None else self.value.resolved(params), "idempotency_key": ( None @@ -473,6 +572,14 @@ def value(expr: ValueExpr | None) -> object: } if self.count_new_only: payload["count_new_only"] = True + # Bound ONLY on the new kind, so every pre-existing effect's digest + # (and every ledger entry that pins it) stays byte-identical. + if self.kind is EffectKind.EXACT_NEW_SET: + payload["new_records"] = [ + {k: value(v) for k, v in sorted(selector.items())} + for selector in self.new_records + ] + payload["identity_field"] = self.identity_field digest = hashlib.sha256( json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") ).hexdigest() @@ -481,7 +588,12 @@ def value(expr: ValueExpr | None) -> object: def referenced_params(self) -> set[str]: """Return the parameter names that determine this effect contract.""" - expressions = [*self.match.values(), self.value, self.idempotency_key] + expressions = [ + *self.match.values(), + *(expr for selector in self.new_records for expr in selector.values()), + self.value, + self.idempotency_key, + ] return { expression.param for expression in expressions @@ -530,6 +642,12 @@ def _contract_payload(self) -> dict[str, object]: # its hash — PR #129) keeps its exact digest. if self.count_new_only: payload["count_new_only"] = True + if self.kind is EffectKind.EXACT_NEW_SET: + payload["new_records"] = [ + {k: str(v) for k, v in sorted(selector.items())} + for selector in self.new_records + ] + payload["identity_field"] = self.identity_field return payload def _semantic_contract_sha256(self) -> str: diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index e585d595..e55aaca2 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -4653,7 +4653,7 @@ def revalidate_attended_completion( "statement cannot replace independent verification" ) break - if effect.count_new_only or effect.forbid_collateral_loss: + if effect.requires_baseline or effect.forbid_collateral_loss: result.effect_verified = False result.error = ( "the effect requires a pre-delivery delta or collateral-" @@ -7377,11 +7377,22 @@ def _verify_current_effect( else: from openadapt_flow.runtime.effects._common import judge_records + # An effect whose CLAIM is a delta (``exact_new_set``) cannot be + # judged against a synthesized empty baseline: every record in + # scope would read as an addition, so the verdict would be + # fabricated rather than proved. Mark the synthetic baseline + # UNREACHABLE for those effects and let the judge return its own + # structured INDETERMINATE refusal. Every pre-existing kind keeps + # the readable empty baseline and is judged exactly as before. + needs_baseline = bool(candidate.requires_baseline) + detail: dict[str, Any] = {"current_state_readback": True} + if needs_baseline: + detail["baseline_unavailable_for_delta"] = True baseline = EffectState( substrate=current.substrate, - reachable=True, + reachable=not needs_baseline, records=[], - detail={"current_state_readback": True}, + detail=detail, ) verdict = judge_records( candidate, @@ -7415,7 +7426,7 @@ def _required_effect_pre_state_unreadable( required = ( bool(requirement(effect)) if callable(requirement) - else bool(effect.count_new_only or effect.forbid_collateral_loss) + else bool(effect.requires_baseline or effect.forbid_collateral_loss) ) state = cls._effect_pre_state_for(before, effect) # Legacy in-process verifiers may retain an opaque pre-state for diff --git a/tests/test_effect_verifier.py b/tests/test_effect_verifier.py index 54966567..7229cda3 100644 --- a/tests/test_effect_verifier.py +++ b/tests/test_effect_verifier.py @@ -317,3 +317,279 @@ def test_compensation_escalates_when_indeterminate(): ) assert result.outcome is CompensationOutcome.ESCALATED assert not result.proceed + + +# -- exact_new_set: the over-write guard ------------------------------------ +# +# The gap this closes, measured in a 150-trial benchmark study: a contract set +# declares one ``record_written`` per intended new record and says NOTHING +# about records nobody declared. An agent asked to download 6 records +# downloaded 37; all 6 declared rows exist, so every per-record contract +# CONFIRMS while the system of record holds 31 writes nobody asked for. That +# is a FALSE PASS -- the one error direction the design must never take. + +SONG_ROWS = [ + {"id": 1, "user_id": "32", "song_id": "199"}, + {"id": 2, "user_id": "32", "song_id": "9"}, +] +DECLARED_SONGS = [ + {"user_id": "32", "song_id": "199"}, + {"user_id": "32", "song_id": "9"}, +] + + +def _exact(**kwargs): + kwargs.setdefault("new_records", DECLARED_SONGS) + kwargs.setdefault("expected_count", len(kwargs["new_records"])) + return Effect(kind=EffectKind.EXACT_NEW_SET, **kwargs) + + +def test_exact_new_set_confirms_when_only_declared_records_were_added(): + pre = [{"id": 0, "user_id": "7", "song_id": "1"}] + v = judge_records(_exact(), _state(pre), [*pre, *SONG_ROWS], substrate="test") + assert v.verdict is Verdict.CONFIRMED + assert v.observed_count == 2 + assert v.expected_count == 2 + + +def test_exact_new_set_refutes_extra_undeclared_records(): + """THE REGRESSION: every declared record is present AND unintended rows + were added to the same read set -- REFUTED, never CONFIRMED.""" + strays = [ + {"id": 10 + n, "user_id": "32", "song_id": str(500 + n)} for n in range(31) + ] + v = judge_records(_exact(), _state([]), [*SONG_ROWS, *strays], substrate="test") + assert v.verdict is Verdict.REFUTED + assert v.observed_count == 33 + assert v.expected_count == 2 + assert "added 33 record(s)" in v.reason + assert "declares exactly 2" in v.reason + assert "31 record(s) the action added are NOT in the declared set" in v.reason + # Every declared record IS present, so each per-record contract passes: + # that is exactly why the per-record contracts cannot see this fault. + for declared in DECLARED_SONGS: + assert any(record_matches(r, declared) for r in SONG_ROWS) + + +def test_exact_new_set_refutes_one_unintended_record_and_names_the_surplus(): + stray = {"id": 7, "user_id": "32", "song_id": "404"} + v = judge_records(_exact(), _state([]), [*SONG_ROWS, stray], substrate="test") + assert v.verdict is Verdict.REFUTED + assert "added 3 record(s)" in v.reason + assert "1 record(s) the action added are NOT in the declared set" in v.reason + assert "song_id=404" in v.reason + + +def test_exact_new_set_ignores_records_that_predate_the_action(): + """A pre-existing row inside the scope that no member names is NOT an + extra -- telling it apart from a new row is the identity_field's job.""" + pre = [{"id": 99, "user_id": "32", "song_id": "77"}] + v = judge_records(_exact(), _state(pre), [*pre, *SONG_ROWS], substrate="test") + assert v.verdict is Verdict.CONFIRMED + assert v.observed_count == 2 + + +def test_exact_new_set_refutes_a_missing_declared_record(): + v = judge_records(_exact(), _state([]), SONG_ROWS[:1], substrate="test") + assert v.verdict is Verdict.REFUTED + assert "was added 0 time(s), expected 1" in v.reason + + +def test_exact_new_set_scope_limits_what_the_guard_speaks_for(): + other = {"id": 50, "user_id": "41", "song_id": "3"} + v = judge_records( + _exact(match={"user_id": "32"}), + _state([]), + [*SONG_ROWS, other], + substrate="test", + ) + assert v.verdict is Verdict.CONFIRMED + + +def test_exact_new_set_refutes_collateral_loss_inside_the_scope(): + pre = [{"id": 99, "user_id": "32", "song_id": "77"}] + v = judge_records(_exact(), _state(pre), SONG_ROWS, substrate="test") + assert v.verdict is Verdict.REFUTED + assert "collateral loss" in v.reason + + +def test_exact_new_set_without_a_baseline_is_indeterminate(): + """No baseline, no delta: refuse loudly rather than call every record new + (which would REFUTE for the wrong reason) or guess.""" + unreachable = EffectState(substrate="test", reachable=False) + v = judge_records(_exact(), unreachable, SONG_ROWS, substrate="test") + assert v.verdict is Verdict.INDETERMINATE + assert "readable pre-state baseline" in v.reason + assert v.should_halt + + +def test_exact_new_set_without_an_identity_is_indeterminate(): + """A record with no identity_field: the added set cannot be enumerated, + so the judge issues a structured refusal instead of a verdict.""" + rows = [{"user_id": "7", "song_id": "1"}] + v = judge_records(_exact(), _state(rows), rows, substrate="test") + assert v.verdict is Verdict.INDETERMINATE + assert "cannot enumerate the added set" in v.reason + assert "identity_field" in v.reason + assert v.should_halt + + +def test_exact_new_set_repeated_member_declares_that_many_additions(): + twice = [{"user_id": "32", "song_id": "199"}] * 2 + rows = [ + {"id": 1, "user_id": "32", "song_id": "199"}, + {"id": 2, "user_id": "32", "song_id": "199"}, + ] + assert ( + judge_records( + _exact(new_records=twice), _state([]), rows, substrate="test" + ).verdict + is Verdict.CONFIRMED + ) + assert ( + judge_records( + _exact(new_records=twice), _state([]), rows[:1], substrate="test" + ).verdict + is Verdict.REFUTED + ) + + +def test_exact_new_set_zero_members_asserts_nothing_was_added(): + empty = Effect(kind=EffectKind.EXACT_NEW_SET, new_records=[], expected_count=0) + assert ( + judge_records(empty, _state([{"id": 1}]), [{"id": 1}], substrate="test").verdict + is Verdict.CONFIRMED + ) + assert ( + judge_records( + empty, _state([{"id": 1}]), [{"id": 1}, {"id": 2}], substrate="test" + ).verdict + is Verdict.REFUTED + ) + + +# -- exact_new_set: contract shape (operator-authored, Path A) --------------- + + +def test_exact_new_set_loads_from_a_bundle_style_mapping(): + """Flow contracts are operator-authored in the bundle; the new kind loads + through the same ``Effect`` model with no extra plumbing.""" + effect = Effect.model_validate( + { + "kind": "exact_new_set", + "match": {"user_id": "32"}, + "new_records": [ + {"user_id": "32", "song_id": "199"}, + {"user_id": "32", "song_id": "9"}, + ], + "expected_count": 2, + "identity_field": "id", + } + ) + assert effect.kind is EffectKind.EXACT_NEW_SET + assert effect.identity_field == "id" + assert effect.requires_baseline + + +def test_exact_new_set_cardinality_must_match_the_declared_set(): + with pytest.raises(ValueError, match=r"expected_count == len\(new_records\)"): + Effect( + kind=EffectKind.EXACT_NEW_SET, + new_records=DECLARED_SONGS, + expected_count=1, + ) + + +def test_exact_new_set_refuses_an_empty_member_selector(): + with pytest.raises(ValueError, match="EMPTY selector"): + Effect( + kind=EffectKind.EXACT_NEW_SET, + new_records=[{}], + expected_count=1, + ) + + +def test_new_records_on_another_kind_is_refused(): + with pytest.raises(ValueError, match="new_records applies only to exact_new_set"): + Effect( + kind=EffectKind.RECORD_WRITTEN, + match=TARGET, + new_records=DECLARED_SONGS, + ) + + +def test_exact_new_set_binds_its_declared_set_in_the_contract_hash(): + """A receipt must not be able to claim a set the judge did not judge.""" + original = _exact().contract_hash() + altered = _exact( + new_records=[ + {"user_id": "32", "song_id": "199"}, + {"user_id": "32", "song_id": "0"}, + ] + ).contract_hash() + assert original != altered + assert ( + _exact(identity_field="row_id").contract_hash() + != _exact(identity_field="id").contract_hash() + ) + + +def test_exact_new_set_resolves_param_references_in_its_declared_set(): + effect = Effect( + kind=EffectKind.EXACT_NEW_SET, + new_records=[{"song_id": {"param": "first"}}, {"song_id": {"param": "second"}}], + expected_count=2, + ) + assert effect.referenced_params() == {"first", "second"} + resolved = effect.resolve({"first": "199", "second": "9"}) + assert [dict(s) for s in resolved.new_records] == [ + {"song_id": "199"}, + {"song_id": "9"}, + ] + + +# -- backward compatibility: pre-existing kinds are judged as before --------- + + +def test_pre_existing_effect_contract_hashes_are_unchanged(): + """The new fields enter ``contract_hash`` ONLY on the new kind, so every + contract written before this option keeps its exact digest (pinned from + the parent commit).""" + assert Effect( + kind=EffectKind.RECORD_WRITTEN, match=TARGET, expected_count=1 + ).contract_hash() == ( + "sha256:ee92689f85689bbd45ea87b6c554857d0f2f80dea54711581bf79ca6953a698a" + ) + assert Effect( + kind=EffectKind.RECORD_WRITTEN, + match=TARGET, + expected_count=1, + count_new_only=True, + ).contract_hash() == ( + "sha256:f0341dde936ba727389d9b55e6adebaf66e2a9a4d1de5777b6b46b23bca951a0" + ) + assert Effect( + kind=EffectKind.FIELD_EQUALS, match=TARGET, field="note", value=NOTE + ).contract_hash() == ( + "sha256:d4453d6a40b045d781cd1d1f6a4c294017aad850e8b74e06a618ada3b9903f1d" + ) + assert Effect( + kind=EffectKind.RECORD_WRITTEN, + match=TARGET, + expected_count=1, + idempotency_key="abc", + ).contract_hash() == ( + "sha256:86fa2aaab213d36378032804e3243ce9f449e7f7a927a0dfb7c6b02f58a78a9f" + ) + + +def test_pre_existing_kinds_do_not_require_a_baseline(): + """``requires_baseline`` is additive: it is True exactly where + ``count_new_only`` already was, plus the new kind.""" + assert not Effect(kind=EffectKind.RECORD_WRITTEN, match=TARGET).requires_baseline + assert not Effect( + kind=EffectKind.FIELD_EQUALS, match=TARGET, field="note", value=NOTE + ).requires_baseline + assert Effect( + kind=EffectKind.RECORD_WRITTEN, match=TARGET, count_new_only=True + ).requires_baseline diff --git a/tests/test_replayer_effects.py b/tests/test_replayer_effects.py index 4fbce9d2..98f54541 100644 --- a/tests/test_replayer_effects.py +++ b/tests/test_replayer_effects.py @@ -421,3 +421,46 @@ def test_no_effects_bundle_replays_unchanged(tmp_path): assert r.effect_verified is None assert r.effect_results == [] assert ("press", "Enter") in backend.actions + + +# -- exact_new_set: a delta claim is never judged against a fabricated +# -- baseline (current-state read-back path) -------------------------------- + + +def test_current_state_readback_refuses_to_judge_an_exact_new_set(): + """``_verify_current_effect`` synthesizes an EMPTY baseline for adapters + with no ``verify_current_state``. For a kind whose CLAIM is a delta every + in-scope record would then read as an addition, so the verdict would be + fabricated. The judge must refuse (INDETERMINATE) instead.""" + from openadapt_flow.runtime.effects.effect import EffectState, Verdict + + effect = Effect( + kind=EffectKind.EXACT_NEW_SET, + new_records=[{"song_id": "199"}], + expected_count=1, + ) + current = EffectState( + substrate="test", + reachable=True, + records=[{"id": 1, "song_id": "199"}, {"id": 2, "song_id": "9"}], + ) + verdict = Replayer._verify_current_effect(object(), effect, current) + assert verdict.verdict is Verdict.INDETERMINATE + assert "readable pre-state baseline" in verdict.reason + + +def test_current_state_readback_still_judges_pre_existing_kinds(): + """The same path is unchanged for every kind that existed before.""" + from openadapt_flow.runtime.effects.effect import EffectState, Verdict + + effect = Effect( + kind=EffectKind.RECORD_WRITTEN, + match={"song_id": "199"}, + expected_count=1, + forbid_collateral_loss=False, + ) + current = EffectState( + substrate="test", reachable=True, records=[{"id": 1, "song_id": "199"}] + ) + verdict = Replayer._verify_current_effect(object(), effect, current) + assert verdict.verdict is Verdict.CONFIRMED From f5a1e6b139212e7dbb7ff7ed188936ea8421d535 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Fri, 14 Aug 2026 23:36:02 +0200 Subject: [PATCH 2/2] fix(sealing): keep pre-field sealed v2 digests valid for the new effect fields `_workflow_content()` dumps the whole workflow into the sealed content digest, so `Effect.new_records` and `Effect.identity_field` entered every bundle's digest the moment they existed. Any bundle sealed BEFORE this change recomputed to a different digest and failed integrity verification with `BundleIntegrityError`. That is a backward-compatibility break for every customer holding a sealed bundle whose steps declare an effect contract, not merely a fixture that needs regenerating -- so the committed fixture is left untouched. Add the two reviewed, version-scoped `_SealedCanonicalOmission` entries the registry's own comment prescribes, owned by `Effect` and scoped to schema version 2, each omitted only at its semantically-empty default (`[]` and `"id"`). They follow the existing entries' shape exactly: exact model ownership, explicit omitted values, and a reason. No recursive key-name removal and no global `exclude_defaults`. Regressions, in `tests/test_bundle_schema_v2.py`: - `test_default_exact_new_set_fields_preserve_pre_field_sealed_v2_digest` seals a synthetic bundle whose step carries a `record_written` contract and loads it with integrity verification. Its pinned digest (89f3648282f337a0...) was computed by the engine as it stood BEFORE these fields existed, so the test cannot merely agree with the new code. - `test_a_declared_exact_new_set_is_sealed_content` proves the omission does not weaken the digest for a contract that USES the guard: a different declared set, an extra declared member, and a non-default identity column each change the digest. - The cross-version test now also asserts both fields survive at schema version 3, so the rules never carry into a future schema implicitly. - The pre-field rendering helpers navigate to the exact owning `Effect` objects rather than stripping a key name anywhere it appears. Co-Authored-By: Claude Fable 5 --- openadapt_flow/bundle_validation.py | 21 +++++ tests/test_bundle_schema_v2.py | 127 +++++++++++++++++++++++++++- 2 files changed, 147 insertions(+), 1 deletion(-) diff --git a/openadapt_flow/bundle_validation.py b/openadapt_flow/bundle_validation.py index 07f0c9a7..2202f276 100644 --- a/openadapt_flow/bundle_validation.py +++ b/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" + ), + ), ), } diff --git a/tests/test_bundle_schema_v2.py b/tests/test_bundle_schema_v2.py index 2784db7f..53a310af 100644 --- a/tests/test_bundle_schema_v2.py +++ b/tests/test_bundle_schema_v2.py @@ -105,6 +105,12 @@ def _good_program_workflow() -> Workflow: "program-encrypted": ( "3173669a4a0c649a6cb8922306f3a30886b35cd73061484d2384a10900899986" ), + # A step carrying ONE pre-existing record_written contract. This digest was + # computed by the engine as it stood BEFORE the exact_new_set fields were + # added, so the regression cannot merely agree with the new code. + "linear-effect": ( + "89f3648282f337a085891bf2d9deaa23b70f3037d6427ff008a848fafbeccaf6" + ), } _SYNTHETIC_BUNDLE_KEY = "synthetic-sealed-v2-compatibility-key" @@ -151,6 +157,16 @@ def _step_payloads(content: dict, shape: str) -> list[dict]: ] +def _effect_payloads(content: dict, shape: str) -> list[dict]: + """The exact Effect JSON objects owned by the steps in either IR shape.""" + + return [ + effect + for step in _step_payloads(content, shape) + for effect in step.get("effects", []) + ] + + def _pre_frame_path_content(wf: Workflow, shape: str) -> dict: """Render the digest content emitted immediately before frame_path existed. @@ -173,6 +189,12 @@ def _pre_frame_path_content(wf: Workflow, shape: str) -> dict: assert step.pop("drag_end_anchor") is None assert step.pop("selection_commit_key") is None assert step.pop("selection_region") is None + for effect in _effect_payloads(content, shape): + # Reviewed v2 omission: the exact_new_set over-write guard postdates + # these seals. Navigated to the exact owning Effect, never stripped by + # key name -- see the note above. + assert effect.pop("new_records") == [] + assert effect.pop("identity_field") == "id" return content @@ -219,6 +241,9 @@ def _write_synthetic_pre_field_bundle( assert step.pop("field_label") is None assert step.pop("selection_commit_key") is None assert step.pop("selection_region") is None + for effect in _effect_payloads(raw, shape): + assert effect.pop("new_records") == [] + assert effect.pop("identity_field") == "id" serialized = json.dumps(raw, sort_keys=True).encode("utf-8") if encrypted: @@ -388,8 +413,105 @@ def test_pre_field_encrypted_certified_bundle_loads_with_original_digest(tmp_pat assert loaded.decrypted_template("templates/btn.png") is not None -def test_sealed_empty_defaults_do_not_implicitly_cross_schema_versions(): +def _effect_workflow(effects) -> Workflow: + """A linear structural workflow whose single step carries ``effects``.""" + wf = _structural_workflow("linear") + wf.steps[0].effects = list(effects) + return wf + + +def test_default_exact_new_set_fields_preserve_pre_field_sealed_v2_digest(tmp_path): + """A bundle sealed BEFORE the over-write guard existed still validates. + + Its steps carry effect contracts, so the additive ``new_records`` / + ``identity_field`` fields land inside the sealed content unless the + reviewed v2 omission rules apply. Without them every customer bundle with + a declared effect would fail integrity verification. + """ + b = _write_bundle_dir(tmp_path) + wf = _effect_workflow( + [ + Effect( + kind=EffectKind.RECORD_WRITTEN, + match={"patient_id": "p1"}, + expected_count=1, + ) + ] + ) + + file_hashes = bv.compute_file_hashes(wf, b) + legacy_digest = _synthetic_legacy_digest(wf, "linear", file_hashes) + assert legacy_digest == _SYNTHETIC_LEGACY_DIGESTS["linear-effect"] + assert bv.compute_content_digest(wf, file_hashes) == legacy_digest + assert ( + _write_synthetic_pre_field_bundle(b, wf, "linear", encrypted=False) + == legacy_digest + ) + + loaded = Workflow.load(b, verify_integrity=True) + assert loaded.manifest is not None + assert loaded.manifest.content_digest == legacy_digest + assert loaded.steps[0].effects[0].kind is EffectKind.RECORD_WRITTEN + + +def test_a_declared_exact_new_set_is_sealed_content(tmp_path): + """The omission must not weaken the digest for a contract USING the guard. + + The rules fire only at the semantically-empty default. A declared + over-write guard, and any identity column other than the default, are + ordinary sealed content: editing either changes the digest. + """ + b = _write_bundle_dir(tmp_path) + + def _guard(new_records, identity_field="id"): + return _effect_workflow( + [ + Effect( + kind=EffectKind.EXACT_NEW_SET, + new_records=new_records, + expected_count=len(new_records), + identity_field=identity_field, + ) + ] + ) + + declared = [{"user_id": "32", "song_id": "199"}] + wf = _guard(declared) + file_hashes = bv.compute_file_hashes(wf, b) + digest = bv.compute_content_digest(wf, file_hashes) + + # The declared set survives canonicalization -- the rule did NOT fire. + effect = _effect_payloads(bv._workflow_content(wf), "linear")[0] + assert len(effect["new_records"]) == 1 + # The rules are VALUE-scoped, exactly like every other entry in the + # registry: the 'id' default is omitted even here. That does not weaken + # this contract, because `kind`, `expected_count` and the declared set are + # all sealed, and no other identity column can reach the default. + assert "identity_field" not in effect + + # A different declared set is different sealed content. + altered = _guard([{"user_id": "32", "song_id": "9"}]) + assert bv.compute_content_digest(altered, file_hashes) != digest + + # So is a second, undeclared addition. + widened = _guard([*declared, {"user_id": "32", "song_id": "9"}]) + assert bv.compute_content_digest(widened, file_hashes) != digest + + # And so is a non-default identity column. + rekeyed = _guard(declared, identity_field="row_id") + assert bv.compute_content_digest(rekeyed, file_hashes) != digest + rekeyed_effect = _effect_payloads(bv._workflow_content(rekeyed), "linear")[0] + assert rekeyed_effect["identity_field"] == "row_id" + + # A guard declaring "this action adds NOTHING" is still distinguishable + # from the pre-field record_written seal: its kind is sealed content. + empty_guard = _guard([]) + assert bv.compute_content_digest(empty_guard, file_hashes) != digest + + +def test_sealed_empty_defaults_do_not_implicitly_cross_schema_versions(): + wf = _effect_workflow([Effect(kind=EffectKind.RECORD_WRITTEN, match={"a": "b"})]) rendered = wf.model_dump(mode="json", exclude={"manifest"}) bv._apply_sealed_canonical_omissions(wf, rendered, schema_version=3) @@ -399,6 +521,9 @@ def test_sealed_empty_defaults_do_not_implicitly_cross_schema_versions(): step = _step_payloads(rendered, "linear")[0] assert "selection_commit_key" in step assert "selection_region" in step + effect = _effect_payloads(rendered, "linear")[0] + assert "new_records" in effect + assert "identity_field" in effect def test_legacy_landmark_without_match_mode_keeps_digest_and_loads_fuzzy(tmp_path):