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/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/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_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): 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