diff --git a/docs/EFFECT_KIT.md b/docs/EFFECT_KIT.md index b3bde4b0..9dd3a7ea 100644 --- a/docs/EFFECT_KIT.md +++ b/docs/EFFECT_KIT.md @@ -26,9 +26,12 @@ from the reference apps. 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 - `deployment.yaml` wires exactly one `EffectVerifier` (REST / GraphQL / - FHIR / SQL / file / email / document / document-hash, or a registered - plugin adapter) plus its secret-isolated auth. + `deployment.yaml` wires one `EffectVerifier` (REST / GraphQL / FHIR / SQL / + file / email / document / document-hash, or a registered plugin adapter) + plus its secret-isolated auth. When more than one reviewed read boundary is + available, `candidates:` selects the strongest evidence tier for each resolved + effect before input. It does not downgrade after input. An unavailable + selected proof halts or enters reconciliation. 3. **The runtime refuses to guess.** Every verdict is CONFIRMED / REFUTED / INDETERMINATE; both non-confirmed verdicts HALT. A step that declares effects with no verifier configured HALTs. An escalated failure emits a @@ -123,6 +126,39 @@ Any kind additionally accepts the evidence-minimization fields `evidence_redact_fields` / `evidence_keep_fields` (see "Evidence minimization" below). +### Candidate selection when there is no database connection + +A database connection is not required. Configure the strongest qualified +read boundary that the workflow has: REST/FHIR/GraphQL, read-only SQL, a file +or report export, a separately authenticated read-only session through a +plugin, or a persisted-state re-acquisition. Do not configure a same-surface +screen read-back as proof of a consequential write. + +For more than one reviewed boundary, use `effects.candidates` instead of +`effects.kind`. Each candidate has the normal `EffectsConfig` fields. Flow +constructs every candidate before actuation, then selects the lowest numeric +`VerificationTier` for each resolved effect; declaration order resolves a tie. +This makes the choice deterministic and reviewable. A missing secret, an +invalid config, or an invalid plugin tier refuses the run before input. The +on-screen candidate is tier 3 only for that exact effect when its read-back +reopens persisted state through a different path. It is tier 4 for a +same-surface read-back. After the action, Flow does not fall back to a weaker +candidate if the selected verifier is unavailable. It records the unavailable +proof and halts or creates the normal reconciliation task. + +```yaml +effects: + candidates: + - kind: document # independent export arrival (tier 1) + root: /secure/exports + file_pattern: "confirmation-*.json" + document_format: json + - kind: onscreen # lower-tier persisted-state read-back +``` + +The single `kind:` form remains the recommended configuration when one +qualified verifier exists and remains fully compatible with prior deployments. + The `sql` kind refuses to construct unless `sql_query` passes the read-only statement filter (single statement, `SELECT`/`WITH` leading keyword, no comments, no mutating/DDL/control keywords or known side-effecting functions, diff --git a/openadapt_flow/deployment.py b/openadapt_flow/deployment.py index 8051ed3b..36fd3087 100644 --- a/openadapt_flow/deployment.py +++ b/openadapt_flow/deployment.py @@ -21,6 +21,8 @@ from __future__ import annotations +import hashlib +import json from pathlib import Path from typing import Any, Callable, Literal, Mapping, Optional @@ -177,6 +179,20 @@ class EffectsConfig(BaseModel): #: LOWER-CONFIDENCE consistency tier, never independent proof. kind: str = "none" + #: Reviewed alternatives for the same effect contract. This is useful + #: where the application exposes more than one read boundary (for example, + #: a file export and a separately authenticated read session) but no direct + #: database connection. The runtime builds every candidate before input, + #: then uses the candidate with the strongest declared evidence tier + #: (lowest numeric tier; declaration order breaks a tie). It never falls + #: back after actuation: an unavailable selected verifier is uncertain and + #: therefore halts or enters reconciliation. + #: + #: The legacy single ``kind`` form and this list are mutually exclusive. + #: Nested lists are refused to keep the selected verifier and its evidence + #: unambiguous. + candidates: list["EffectsConfig"] = Field(default_factory=list) + # -- onscreen (no-API screen read-back; auto-derived per-effect region) --- #: Explicit read-back region ``(x, y, w, h)`` for a hand-configured #: deployment. Normally left None — the compiler auto-derives a per-effect @@ -307,6 +323,26 @@ def _coerce_value_exprs(cls, v: Any) -> Any: } return v + @model_validator(mode="after") + def _validate_candidates(self) -> "EffectsConfig": + if not self.candidates: + return self + if (self.kind or "none").strip().lower() not in ("", "none"): + raise ValueError( + "effects.kind and effects.candidates are mutually exclusive" + ) + for candidate in self.candidates: + if candidate.candidates: + raise ValueError("effects.candidates cannot contain nested candidates") + if (candidate.kind or "none").strip().lower() in ("", "none"): + raise ValueError( + "every effects.candidates entry must configure a verifier kind" + ) + return self + + +EffectsConfig.model_rebuild() + class ActuationConfig(BaseModel): """The API/tool actuation tier (top of the capability ladder). @@ -915,18 +951,69 @@ def build_effect_verifier( Resolution order for ``kind``: built-in adapters, then plugin factories registered under the ``openadapt_flow.effect_verifiers`` entry-point group or via - ``register_verifier_factory`` (the customer adapter SDK seam). When the - config sets ``evidence_redact_fields`` / ``evidence_keep_fields``, the - built verifier is wrapped so every verdict's evidence is minimized. + ``register_verifier_factory`` (the customer adapter SDK seam). ``candidates`` + builds every reviewed alternative and selects the strongest declared tier; + it does not create a post-actuation fallback. When the config sets + ``evidence_redact_fields`` / ``evidence_keep_fields``, the built verifier + is wrapped so every verdict's evidence is minimized. Raises: ValueError: on an unknown ``kind``, a missing required field, an unresolved ``{param: ...}`` reference, or a missing secret env var (fail loud rather than wire a broken verifier). """ + if cfg.candidates: + built = [ + build_effect_verifier(candidate, params) for candidate in cfg.candidates + ] + for candidate in built: + if candidate is None: # guarded by EffectsConfig, retained fail-closed + raise ValueError("effects.candidates entry did not build a verifier") + missing = [ + name + for name in ("capture_pre_state", "verify") + if not callable(getattr(candidate, name, None)) + ] + if missing: + raise ValueError( + "effects.candidates entry is not an EffectVerifier; missing " + + ", ".join(missing) + ) + tier = getattr(candidate, "verification_tier", None) + tier_for = getattr(candidate, "verification_tier_for", None) + if isinstance(tier, bool) or (tier is None and not callable(tier_for)): + raise ValueError( + "effects.candidates entry has no valid verification tier" + ) + from openadapt_flow.runtime.effects.adapter import ( + CandidateEffectVerifier, + set_verifier_identity, + ) + + candidate_selector = CandidateEffectVerifier(built) + set_verifier_identity( + candidate_selector, + _effect_verifier_config_identity(cfg, candidate_selector), + ) + if cfg.evidence_redact_fields or cfg.evidence_keep_fields is not None: + from openadapt_flow.runtime.effects.adapter import ( + RedactingVerifier, + RedactionPolicy, + ) + + policy = RedactionPolicy( + redact_fields=list(cfg.evidence_redact_fields), + keep_fields=cfg.evidence_keep_fields, + ) + return RedactingVerifier(candidate_selector, policy) + return candidate_selector + verifier = _build_effect_verifier_unredacted(cfg, params) if verifier is None: return None + from openadapt_flow.runtime.effects.adapter import set_verifier_identity + + set_verifier_identity(verifier, _effect_verifier_config_identity(cfg, verifier)) if cfg.evidence_redact_fields or cfg.evidence_keep_fields is not None: from openadapt_flow.runtime.effects.adapter import ( RedactingVerifier, @@ -941,6 +1028,41 @@ def build_effect_verifier( return verifier +def _effect_verifier_config_identity(cfg: EffectsConfig, verifier: Any) -> str: + """Hash the non-secret deployment fields that identify one verifier.""" + + sensitive = ("credential", "password", "secret", "token") + + def sanitized(value: Any, *, key: str = "") -> Any: + lowered = key.lower() + if not lowered.endswith("_env") and any( + marker in lowered for marker in sensitive + ): + return "[secret-reference]" + if isinstance(value, dict): + return { + str(item_key): sanitized(item_value, key=str(item_key)) + for item_key, item_value in value.items() + } + if isinstance(value, list): + return [sanitized(item) for item in value] + return value + + payload = { + "schema": "openadapt.effect-verifier-identity/v1", + "implementation": ( + f"{type(verifier).__module__}.{type(verifier).__qualname__}" + ), + "implementation_version": getattr(verifier, "adapter_version", None), + "substrate": getattr(verifier, "substrate", None), + "config": sanitized(cfg.model_dump(mode="json")), + } + digest = hashlib.sha256( + json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest() + return f"sha256:{digest}" + + def _build_effect_verifier_unredacted( cfg: EffectsConfig, params: Optional[Mapping[str, str]] = None ) -> Optional[Any]: @@ -1142,11 +1264,16 @@ def _build_effect_verifier_unredacted( # Plugin seam: a customer adapter registered programmatically or under the # 'openadapt_flow.effect_verifiers' entry-point group serves its own kind. - from openadapt_flow.runtime.effects.adapter import resolve_verifier_factory + from openadapt_flow.runtime.effects.adapter import ( + resolve_verifier_factory, + validate_verifier_adapter, + ) factory = resolve_verifier_factory(kind) if factory is not None: - return factory(cfg, params) + verifier = factory(cfg, params) + validate_verifier_adapter(verifier) + return verifier raise ValueError( f"unknown effects.kind {cfg.kind!r} " diff --git a/openadapt_flow/ir.py b/openadapt_flow/ir.py index da5f10d5..28d50879 100644 --- a/openadapt_flow/ir.py +++ b/openadapt_flow/ir.py @@ -2805,6 +2805,13 @@ class EffectVerificationEvidence(BaseModel): effect_contract_hash: str = Field(pattern=r"^sha256:[0-9a-f]{64}$") substrate: str = Field(min_length=1) + #: Opaque digest of the exact deployment adapter configuration that + #: produced this evidence. Durable resume uses it with the retained tier to + #: refuse a changed verifier after restart. ``None`` remains valid for + #: checkpoints written before adapter binding was retained. + verifier_identity: Optional[str] = Field( + default=None, pattern=r"^sha256:[0-9a-f]{64}$" + ) verification_tier: Optional[int] = Field(default=None, ge=1, le=4) initial_verdict: Literal["confirmed", "refuted", "indeterminate"] final_verdict: Literal["confirmed", "refuted", "indeterminate"] diff --git a/openadapt_flow/runtime/durable/resume.py b/openadapt_flow/runtime/durable/resume.py index f904bf29..b503f46b 100644 --- a/openadapt_flow/runtime/durable/resume.py +++ b/openadapt_flow/runtime/durable/resume.py @@ -1014,6 +1014,7 @@ def _resume_under_lease( workflow=workflow, step=step, actuation_path=("api" if checkpoint.actuation == "api" else "gui"), + retained_evidence=list(checkpoint.effect_evidence), ) if last_linear is not None and last_linear.next_step_index < len(workflow.steps): replayer.revalidate_linear_checkpoint_state( diff --git a/openadapt_flow/runtime/effects/__init__.py b/openadapt_flow/runtime/effects/__init__.py index a6c314b8..9a577454 100644 --- a/openadapt_flow/runtime/effects/__init__.py +++ b/openadapt_flow/runtime/effects/__init__.py @@ -47,6 +47,9 @@ from openadapt_flow.runtime.effects.adapter import ( # noqa: F401 ENTRY_POINT_GROUP, AdapterResult, + CandidateEffectState, + CandidateEffectVerifier, + CandidatePreState, CollateralHook, ConnectionProbe, RedactingVerifier, @@ -127,6 +130,9 @@ # adapter platform "ENTRY_POINT_GROUP", "AdapterResult", + "CandidateEffectState", + "CandidateEffectVerifier", + "CandidatePreState", "CollateralHook", "ConnectionProbe", "RedactingVerifier", diff --git a/openadapt_flow/runtime/effects/adapter.py b/openadapt_flow/runtime/effects/adapter.py index 91f98a10..a79e6844 100644 --- a/openadapt_flow/runtime/effects/adapter.py +++ b/openadapt_flow/runtime/effects/adapter.py @@ -88,7 +88,10 @@ from __future__ import annotations import hashlib +import inspect +import json import time +from dataclasses import dataclass from datetime import datetime, timezone from enum import Enum from typing import Any, Callable, Mapping, Optional, Protocol, runtime_checkable @@ -102,7 +105,7 @@ EffectVerdict, Verdict, ) -from openadapt_flow.verification import VerificationTier +from openadapt_flow.verification import VerificationTier, verifier_effect_tier #: Entry-point group a customer verifier package registers its factory under. #: Each entry point's NAME is the ``effects.kind`` it serves; its value loads @@ -511,6 +514,352 @@ def verify( raise NotImplementedError +def validate_verifier_adapter(adapter: Any) -> None: + """Refuse an incomplete third-party adapter before a run can use it. + + Plugin factories run at deployment construction time. Validate their full + lifecycle there, rather than discovering a missing method after a delivery + boundary. Built-in compatibility adapters do not use this function; they + inherit the complete lifecycle from :class:`VerifierAdapterBase`. + """ + missing = [ + name + for name in ( + "test_connection", + "capture_pre_state", + "capture_post_state", + "verify", + ) + if not callable(getattr(adapter, name, None)) + ] + if missing: + raise ValueError( + "effect-verifier plugin is not a complete VerifierAdapter; missing " + + ", ".join(missing) + ) + call_shapes = { + "test_connection": ((),), + "capture_pre_state": ((),), + "capture_post_state": ((),), + "verify": ((object(), object()),), + } + invalid: list[str] = [] + for name, shapes in call_shapes.items(): + method = getattr(adapter, name) + try: + signature = inspect.signature(method) + except (TypeError, ValueError): + invalid.append(name) + continue + try: + for args in shapes: + signature.bind(*args) + except TypeError: + invalid.append(name) + if invalid: + raise ValueError( + "effect-verifier plugin has an incompatible lifecycle signature: " + + ", ".join(invalid) + ) + substrate = getattr(adapter, "substrate", None) + if not isinstance(substrate, str) or not substrate.strip(): + raise ValueError( + "effect-verifier plugin is not a complete VerifierAdapter; " + "substrate must be a non-empty string" + ) + tier = getattr(adapter, "verification_tier", None) + if not isinstance(tier, VerificationTier): + raise ValueError( + "effect-verifier plugin is not a complete VerifierAdapter; " + "verification_tier must be a VerificationTier" + ) + # The runtime uses this private marker to run the plugin's read-only + # connection probe before its first state capture. Built-in adapters are + # marked by deployment construction through the same boundary. + try: + setattr(adapter, "_openadapt_requires_preflight", True) + except (AttributeError, TypeError) as exc: + raise ValueError( + "effect-verifier plugin cannot retain its validated lifecycle state" + ) from exc + + +def set_verifier_identity(adapter: Any, identity: str) -> None: + """Bind one opaque deployment identity to an adapter instance.""" + + if not isinstance(identity, str) or not identity.startswith("sha256:"): + raise ValueError("effect verifier identity must be an opaque sha256 digest") + try: + setattr(adapter, "_openadapt_verifier_identity", identity) + setattr(adapter, "_openadapt_requires_preflight", True) + except (AttributeError, TypeError) as exc: + raise ValueError( + "effect verifier cannot retain its deployment identity" + ) from exc + + +def verifier_identity(adapter: Any) -> str: + """Return the stable opaque identity used in retained effect evidence.""" + + configured = getattr(adapter, "_openadapt_verifier_identity", None) + if isinstance(configured, str) and configured.startswith("sha256:"): + return configured + cls = type(adapter) + payload = { + "class": f"{cls.__module__}.{cls.__qualname__}", + "substrate": str(getattr(adapter, "substrate", "")), + } + digest = hashlib.sha256( + json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest() + return f"sha256:{digest}" + + +def _probe_adapter_connection(adapter: Any, context: Any = None) -> ConnectionProbe: + """Run an adapter readiness probe without letting a preflight exception out. + + This also supports older adapters that only expose ``capture_pre_state``. + The helper deliberately does not select a candidate or change a selection. + """ + try: + probe = getattr(adapter, "test_connection", None) + if callable(probe): + result = probe() if context is None else probe(context) + if isinstance(result, ConnectionProbe): + return result + return ConnectionProbe( + ok=False, + substrate=str(getattr(adapter, "substrate", "")), + reason="connection probe returned an invalid result", + ) + capture = getattr(adapter, "capture_pre_state", None) + if not callable(capture): + return ConnectionProbe( + ok=False, + substrate=str(getattr(adapter, "substrate", "")), + reason="adapter has no pre-state capture method", + ) + state = capture() if context is None else capture(context) + return ConnectionProbe( + ok=bool(state.reachable), + substrate=state.substrate or str(getattr(adapter, "substrate", "")), + reason="reachable" if state.reachable else "unreachable", + detail=dict(state.detail), + ) + except Exception as exc: # noqa: BLE001 - preflight must not raise + return ConnectionProbe( + ok=False, + substrate=str(getattr(adapter, "substrate", "")), + reason=f"connection probe raised: {type(exc).__name__}", + ) + + +@dataclass(frozen=True) +class CandidatePreState: + """The pre-action verifier and snapshot selected for one effect.""" + + verifier: Any + state: EffectState + tier: VerificationTier + verifier_identity: str + requires_readable_pre_state: bool + + +class CandidateEffectState: + """Effect-semantics-indexed pre-states from :class:`CandidateEffectVerifier`.""" + + def __init__(self, selections: Mapping[str, CandidatePreState]) -> None: + self._selections = dict(selections) + + @property + def reachable(self) -> bool: + return all(selection.state.reachable for selection in self._selections.values()) + + def for_effect(self, effect: Effect) -> CandidatePreState: + try: + return self._selections[_candidate_effect_key(effect)] + except KeyError as exc: + raise ValueError("effect has no captured candidate pre-state") from exc + + +class CandidateEffectVerifier: + """Select the strongest configured verifier separately for each effect. + + Selection happens before actuation and the returned ``CandidateEffectState`` + pins the verifier plus its baseline. ``verify`` uses that exact selection; + it never tries a weaker candidate after delivery. + """ + + def __init__(self, candidates: list[Any]) -> None: + if not candidates: + raise ValueError("candidate verifier list must not be empty") + self._candidates = tuple(candidates) + + def bind_backend(self, backend: Any) -> None: + """Bind the live replay backend to every backend-aware candidate. + + Selection is still per effect and pinned before actuation. Binding is + setup only: it does not select, probe, or replace a candidate. A + malformed backend-binding hook fails before replay starts rather than + leaving an on-screen candidate detached from the live backend. + """ + for candidate in self._candidates: + binder = getattr(candidate, "bind_backend", None) + if binder is None: + continue + if not callable(binder): + raise ValueError( + "effects.candidates entry exposes a non-callable bind_backend" + ) + binder(backend) + + def _select(self, effect: Effect) -> tuple[Any, VerificationTier]: + ranked: list[tuple[int, int, Any, VerificationTier]] = [] + for index, candidate in enumerate(self._candidates): + tier = verifier_effect_tier(candidate, effect) + if tier is None: + raise ValueError( + "effects.candidates entry has no valid verification tier for " + "this effect" + ) + ranked.append((int(tier), index, candidate, tier)) + _rank, _index, candidate, tier = min(ranked, key=lambda item: item[:2]) + return candidate, tier + + def verification_tier_for(self, effect: Effect) -> VerificationTier: + return self._select(effect)[1] + + def requires_readable_pre_state_for(self, effect: Effect) -> bool: + candidate, _tier = self._select(effect) + 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) + + @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) + isolated = effect.model_copy(deep=True) + original = isolated.model_dump(mode="json") + required = bool(requirement(isolated)) + if isolated.model_dump(mode="json") != original: + raise ValueError( + "selected effects.candidates verifier changed the effect while " + "declaring its pre-state requirement" + ) + return required + + def test_connection(self, context: Any = None) -> ConnectionProbe: + """Probe every candidate without selecting or downgrading one. + + ``ok`` is true only when every configured candidate is readable. This + conservative aggregate is deterministic and cannot advertise a weak + fallback as readiness for a stronger effect-specific selection. + """ + probes = [ + _probe_adapter_connection(candidate, context) + for candidate in self._candidates + ] + return ConnectionProbe( + ok=bool(probes) and all(probe.ok for probe in probes), + substrate="candidates", + reason="all candidate verifiers reachable" + if probes and all(probe.ok for probe in probes) + else "one or more candidate verifiers are unreachable", + detail={ + "candidates": [ + { + "substrate": probe.substrate, + "ok": probe.ok, + "reason": probe.reason, + } + for probe in probes + ] + }, + ) + + def capture_pre_state_for_effects( + self, effects: list[Effect], context: Any = None + ) -> CandidateEffectState: + selections: dict[str, CandidatePreState] = {} + selected: dict[str, tuple[Any, VerificationTier, bool, str]] = {} + selected_candidates: dict[int, Any] = {} + for effect in effects: + candidate, tier = self._select(effect) + selected[_candidate_effect_key(effect)] = ( + candidate, + tier, + self._pre_state_requirement(candidate, effect), + verifier_identity(candidate), + ) + selected_candidates[id(candidate)] = candidate + + # Deployment-built candidates carry this marker. Probe only the exact + # candidates selected above. An unavailable weaker alternative must + # not block a stronger selected verifier, and it must never become a + # fallback after input. + for candidate in selected_candidates.values(): + if not getattr(candidate, "_openadapt_requires_preflight", False): + continue + probe = _probe_adapter_connection(candidate, context) + if not probe.ok: + raise ValueError( + "selected effects.candidates verifier failed its read-only " + f"connection preflight ({probe.substrate}: {probe.reason})" + ) + + captured: dict[int, EffectState] = {} + for effect in effects: + candidate, tier, required, identity = selected[ + _candidate_effect_key(effect) + ] + key = id(candidate) + if key not in captured: + state = ( + candidate.capture_pre_state() + if context is None + else candidate.capture_pre_state(context) + ) + if not isinstance(state, EffectState): + raise ValueError( + "selected effects.candidates verifier returned an invalid " + "pre-state" + ) + captured[key] = state + selections[_candidate_effect_key(effect)] = CandidatePreState( + verifier=candidate, + state=captured[key], + tier=tier, + verifier_identity=identity, + requires_readable_pre_state=required, + ) + return CandidateEffectState(selections) + + def capture_pre_state(self, context: Any = None) -> EffectState: + raise ValueError( + "candidate verifier selection requires resolved effects before " + "pre-state capture" + ) + + def verify( + self, expected: Effect, before: CandidateEffectState, context: Any = None + ) -> EffectVerdict: + selection = before.for_effect(expected) + if context is None: + return selection.verifier.verify(expected, selection.state) + return selection.verifier.verify(expected, selection.state, context) + + +def _candidate_effect_key(effect: Effect) -> str: + """Key every resolved effect, including read-back semantics outside its contract hash.""" + payload = json.dumps( + effect.model_dump(mode="json"), sort_keys=True, separators=(",", ":") + ).encode("utf-8") + return hashlib.sha256(payload).hexdigest() + + class RedactingVerifier: """Protocol-transparent wrapper applying a :class:`RedactionPolicy`. @@ -529,18 +878,17 @@ def __init__(self, inner: Any, policy: RedactionPolicy) -> None: def __getattr__(self, name: str) -> Any: return getattr(self._inner, name) + def bind_backend(self, backend: Any) -> None: + """Forward live-backend binding without changing the verifier choice.""" + binder = getattr(self._inner, "bind_backend", None) + if binder is None: + return + if not callable(binder): + raise ValueError("wrapped verifier exposes a non-callable bind_backend") + binder(backend) + def test_connection(self, context: Any = None) -> ConnectionProbe: - probe = getattr(self._inner, "test_connection", None) - if callable(probe): - result = probe(context) - if isinstance(result, ConnectionProbe): - return result - state = self._inner.capture_pre_state(context) - return ConnectionProbe( - ok=state.reachable, - substrate=state.substrate, - reason="" if state.reachable else "unreachable", - ) + return _probe_adapter_connection(self._inner, context) def capture_pre_state(self, context: Any = None) -> EffectState: return self._inner.capture_pre_state(context) @@ -557,6 +905,27 @@ def verify( verdict = self._inner.verify(expected, before, context) return redact_verdict(verdict, self._policy, field=expected.field) + def verify_current_state( + self, expected: Effect, current: EffectState, context: Any = None + ) -> EffectVerdict: + callback = getattr(self._inner, "verify_current_state", None) + if callable(callback): + verdict = callback(expected, current, context) + else: + baseline = EffectState( + substrate=current.substrate, + reachable=True, + records=[], + detail={"current_state_readback": True}, + ) + verdict = judge_records( + expected, + baseline, + current.records if current.reachable is True else None, + substrate=current.substrate, + ) + return redact_verdict(verdict, self._policy, field=expected.field) + # -- plugin registry ---------------------------------------------------------- diff --git a/openadapt_flow/runtime/effects/onscreen.py b/openadapt_flow/runtime/effects/onscreen.py index fcff9ce7..1c96c97c 100644 --- a/openadapt_flow/runtime/effects/onscreen.py +++ b/openadapt_flow/runtime/effects/onscreen.py @@ -49,7 +49,7 @@ import difflib from typing import Any, Optional -from openadapt_flow.runtime.effects.adapter import VerifierAdapterBase +from openadapt_flow.runtime.effects.adapter import ConnectionProbe, VerifierAdapterBase from openadapt_flow.runtime.effects.effect import ( MIN_RENAV_ACTIONS, Effect, @@ -162,6 +162,11 @@ def verification_tier_for(effect: Effect) -> VerificationTier: return VerificationTier.PERSISTED_STATE_REACQUISITION return VerificationTier.IMMEDIATE_SCREEN + @staticmethod + def requires_readable_pre_state_for(effect: Effect) -> bool: + """Read-back proof occurs after input and has no pre-action delta baseline.""" + return False + def __init__( self, backend: Any = None, @@ -184,6 +189,34 @@ def bind_backend(self, backend: Any) -> None: where the verifier is built before the backend exists).""" self._backend = backend + def test_connection(self, context: Any = None) -> ConnectionProbe: + """Prove that the local observation boundary can provide a fresh frame.""" + + screenshot = getattr(self._backend, "screenshot", None) + if not callable(screenshot): + return ConnectionProbe( + ok=False, + substrate=self.substrate, + reason="the live backend has no screenshot boundary", + ) + try: + frame = screenshot() + except Exception as exc: # noqa: BLE001 - a probe never escapes + return ConnectionProbe( + ok=False, + substrate=self.substrate, + reason=f"screenshot probe raised: {type(exc).__name__}", + ) + return ConnectionProbe( + ok=isinstance(frame, bytes) and bool(frame), + substrate=self.substrate, + reason=( + "live observation boundary reachable" + if isinstance(frame, bytes) and bool(frame) + else "live observation boundary returned no frame" + ), + ) + # -- region read --------------------------------------------------------- def _read_region(self, region: Region) -> tuple[str, float]: @@ -390,6 +423,13 @@ def verify( } ) + def verify_current_state( + self, expected: Effect, current: EffectState, context: Any = None + ) -> EffectVerdict: + """Reacquire and verify persisted state after attended or durable input.""" + + return self.verify(expected, current, context) + @staticmethod def _expected_text(expected: Effect, params: dict) -> Optional[str]: if expected.value is not None: diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index 12d5cf73..410cb818 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -157,6 +157,12 @@ EffectVerifier, reconcile_or_escalate, ) +from openadapt_flow.runtime.effects.adapter import ( + _probe_adapter_connection, +) +from openadapt_flow.runtime.effects.adapter import ( + verifier_identity as effect_verifier_identity, +) from openadapt_flow.runtime.program_predicates import ( evaluate_program_predicate, exact_png_size, @@ -4152,6 +4158,7 @@ def revalidate_program_checkpoint_effects( workflow=workflow, step=source_step, actuation_path="api" if checkpoint.actuation == "api" else "gui", + retained_evidence=list(checkpoint.new_effect_evidence), ) def revalidate_retained_effects( @@ -4161,6 +4168,7 @@ def revalidate_retained_effects( workflow: Workflow, step: Step, actuation_path: Literal["gui", "api"], + retained_evidence: Optional[list[EffectVerificationEvidence]] = None, ) -> None: """Prove that previously confirmed effects still hold before resume.""" @@ -4181,45 +4189,82 @@ def revalidate_retained_effects( if refusal is not None: raise StateDiverged(refusal) try: - current = self.effect_verifier.capture_pre_state() + current = self._capture_effect_pre_state(self.effect_verifier, effects) except Exception as exc: raise StateDiverged( "the retained effects could not be read before resume" ) from exc - if not current.reachable: - raise StateDiverged("the retained effects could not be read before resume") - from openadapt_flow.runtime.effects._common import judge_records - + binding_refusal = self._candidate_binding_refusal(current, effects) + if binding_refusal is not None: + raise StateDiverged(binding_refusal) + remaining_evidence = list(retained_evidence or []) for effect in effects: if effect.needs_operator_confirmation: raise StateDiverged( "a retained effect still requires an operator-authored binding" ) + retained = next( + ( + item + for item in remaining_evidence + if item.effect_contract_hash == effect.contract_hash() + ), + None, + ) + if retained is not None: + remaining_evidence.remove(retained) + ( + current_verifier, + effect_current, + current_tier, + current_identity, + ) = self._effect_binding_for(self.effect_verifier, current, effect) + selected = getattr(current, "for_effect", None) + if callable(selected) and ( + retained is None or retained.verifier_identity is None + ): + raise StateDiverged( + "the retained candidate verifier identity is missing; refusing " + "to select a different verifier after restart" + ) + if retained is not None and retained.verifier_identity is not None: + if ( + retained.verifier_identity != current_identity + or retained.verification_tier + != (int(current_tier) if current_tier is not None else None) + ): + raise StateDiverged( + "the retained verifier identity or evidence tier changed " + "after restart; refusing to resume" + ) + # Historical evidence already proves the original delta and - # collateral-loss contract. Resume must prove that the intended - # effect still persists now. Re-read the current records through - # the qualified read-only verifier and judge the exact selector as - # an absolute persistence contract. Calling ``verify`` here would - # incorrectly use the current snapshot as a new pre-action - # baseline and can also invoke delivery-oriented verifier logic. + # collateral-loss contract. Resume now asks the exact retained + # adapter to prove that the intended effect still persists. The + # adapter owns its correct current-state read path, including + # different-path on-screen re-navigation. persistence_effect = effect.model_copy( update={ "count_new_only": False, "forbid_collateral_loss": False, } ) - baseline = EffectState( - substrate=current.substrate, - reachable=True, - records=[], - detail={"durable_resume_persistence_readback": True}, - ) - verdict = judge_records( - persistence_effect, - baseline, - current.records, - substrate=current.substrate, - ) + try: + verdict = self._verify_current_effect( + current_verifier, persistence_effect, effect_current + ) + except Exception as exc: # noqa: BLE001 - resume verifier boundary + raise StateDiverged( + "the retained effect current-state readback failed" + ) from exc + if callable(selected) and ( + verifier_effect_tier(current_verifier, effect) != current_tier + or effect_verifier_identity(current_verifier) != current_identity + ): + raise StateDiverged( + "the retained verifier identity or evidence tier changed " + "during current-state readback" + ) if not verdict.confirmed: raise StateDiverged( "an already-confirmed effect no longer holds " @@ -4407,8 +4452,6 @@ def revalidate_attended_completion( one is unavailable. """ from openadapt_flow.policy import effects_for_actuation - from openadapt_flow.runtime.effects import EffectState - from openadapt_flow.runtime.effects._common import judge_records if not _preserve_execution_snapshot: self._install_execution_snapshots(workflow) @@ -4544,7 +4587,29 @@ def revalidate_attended_completion( run_dir, evidence_step_id, "attended-after", frame ) return result - current = self.effect_verifier.capture_pre_state() + try: + current = self._capture_effect_pre_state(self.effect_verifier, effects) + except Exception as exc: # noqa: BLE001 - attended verifier boundary + result.effect_verified = False + result.error = ( + "the qualified current-state verifier failed its preflight or " + f"readback ({type(exc).__name__}); outcome is uncertain and " + "Continue is refused" + ) + result.after_png = self._save_step_png( + run_dir, evidence_step_id, "attended-after", frame + ) + return result + binding_refusal = self._candidate_binding_refusal(current, effects) + if binding_refusal is not None: + result.effect_verified = False + result.safety_halt = True + result.failure_category = "governed_refusal" + result.error = binding_refusal + result.after_png = self._save_step_png( + run_dir, evidence_step_id, "attended-after", frame + ) + return result tier_refusal = self._profile_effect_tier_refusal( profile_workflow, step, @@ -4573,17 +4638,13 @@ def revalidate_attended_completion( run_dir, evidence_step_id, "attended-after", frame ) return result - if not current.reachable: - result.effect_verified = False - result.error = ( - "the system of record is unreachable; outcome is uncertain " - "and Continue is refused" - ) - result.after_png = self._save_step_png( - run_dir, evidence_step_id, "attended-after", frame - ) - return result for effect in effects: + ( + current_verifier, + effect_current, + tier, + verifier_binding_identity, + ) = self._effect_binding_for(self.effect_verifier, current, effect) if effect.needs_operator_confirmation: result.effect_verified = False result.error = ( @@ -4599,35 +4660,46 @@ def revalidate_attended_completion( "outcome is uncertain and Continue is refused" ) break - baseline = EffectState( - substrate=current.substrate, - reachable=True, - records=[], - detail={"attended_current_state_readback": True}, - ) - verdict = judge_records( - effect, - baseline, - current.records, - substrate=current.substrate, - ) + try: + verdict = self._verify_current_effect( + current_verifier, effect, effect_current + ) + except Exception as exc: # noqa: BLE001 - post-input verifier boundary + result.effect_verified = False + result.error = ( + "the qualified current-state readback failed " + f"({type(exc).__name__}); outcome is uncertain and " + "Continue is refused" + ) + break + if callable(getattr(current, "for_effect", None)) and ( + verifier_effect_tier(current_verifier, effect) != tier + or effect_verifier_identity(current_verifier) + != verifier_binding_identity + ): + result.effect_verified = False + result.error = ( + "the selected verifier identity or evidence tier changed " + "during attended current-state readback; Continue is refused" + ) + break result.effect_contract_hashes.append(effect.contract_hash()) result.effect_results.append( - f"[attended-current-readback] {effect.kind.value}: " + f"[attended-qualified-readback] {effect.kind.value}: " f"{verdict.verdict.value} — {verdict.reason}" ) if not verdict.confirmed: result.effect_verified = False result.error = ( - "the independent current-state effect readback did not " + "the qualified current-state effect readback did not " f"confirm the outcome ({verdict.verdict.value})" ) break - tier = verifier_effect_tier(self.effect_verifier, effect) result.effect_evidence.append( EffectVerificationEvidence( effect_contract_hash=effect.contract_hash(), substrate=verdict.substrate, + verifier_identity=verifier_binding_identity, verification_tier=(int(tier) if tier is not None else None), initial_verdict=verdict.verdict.value, final_verdict=verdict.verdict.value, @@ -5784,14 +5856,40 @@ def _run_step( active_verifier, ) if error is None: - effect_pre_state = active_verifier.capture_pre_state() - error = self._profile_effect_tier_refusal( - workflow, - step, - "gui", - resolved_effects, - active_verifier, - ) + try: + effect_pre_state = self._capture_effect_pre_state( + active_verifier, resolved_effects + ) + except Exception as exc: # noqa: BLE001 - verifier boundary + error = ( + "the selected system-of-record effect verifier " + "failed its preflight or pre-state capture before " + f"actuation ({type(exc).__name__}); refusing input" + ) + else: + error = self._candidate_binding_refusal( + effect_pre_state, resolved_effects + ) + if ( + error is None + and self._required_effect_pre_state_unreadable( + active_verifier, + effect_pre_state, + resolved_effects, + ) + ): + error = ( + "the selected system-of-record effect verifier is " + "unreachable before actuation; refusing input" + ) + if error is None: + error = self._profile_effect_tier_refusal( + workflow, + step, + "gui", + resolved_effects, + active_verifier, + ) if error is not None: result.effect_verified = False result.safety_halt = True @@ -6085,16 +6183,46 @@ def _run_step( active_verifier, ) if effect_refresh_error is None: - effect_pre_state = active_verifier.capture_pre_state() - effect_refresh_error = ( - self._profile_effect_tier_refusal( - workflow, - step, - "gui", - resolved_effects, - active_verifier, + try: + effect_pre_state = self._capture_effect_pre_state( + active_verifier, resolved_effects + ) + except Exception as exc: # noqa: BLE001 - verifier boundary + effect_refresh_error = ( + "the selected system-of-record effect verifier " + "failed its preflight or pre-state capture " + f"before actuation ({type(exc).__name__}); " + "refusing input" + ) + else: + effect_refresh_error = ( + self._candidate_binding_refusal( + effect_pre_state, resolved_effects + ) + ) + if ( + effect_refresh_error is None + and self._required_effect_pre_state_unreadable( + active_verifier, + effect_pre_state, + resolved_effects, + ) + ): + effect_refresh_error = ( + "the selected system-of-record effect " + "verifier is unreachable before actuation; " + "refusing input" + ) + if effect_refresh_error is None: + effect_refresh_error = ( + self._profile_effect_tier_refusal( + workflow, + step, + "gui", + resolved_effects, + active_verifier, + ) ) - ) error = effect_refresh_error if error is not None: @@ -6717,7 +6845,7 @@ def _try_api_tier( # only what THIS actuation wrote (delta / at-most-once / collateral # loss), then actuate exactly once. try: - before = effect_verifier.capture_pre_state() + before = self._capture_effect_pre_state(effect_verifier, effects) except Exception as exc: # noqa: BLE001 - deployment verifier boundary result.effect_verified = False result.effect_results.append( @@ -6731,6 +6859,27 @@ def _try_api_tier( "send the request — run aborted" ) return True + binding_refusal = self._candidate_binding_refusal(before, effects) + if binding_refusal is not None: + result.effect_verified = False + result.effect_results.append(f"[api] {binding_refusal}") + result.ok = False + result.safety_halt = True + result.failure_category = "governed_refusal" + result.error = binding_refusal + return True + if self._required_effect_pre_state_unreadable(effect_verifier, before, effects): + result.effect_verified = False + result.effect_results.append( + "[api] effect pre-state is unreachable before request delivery" + ) + result.ok = False + result.error = ( + f"Step '{step.id}' ({step.intent}) could not read the " + "system-of-record pre-state before API actuation; refusing to " + "send the request — run aborted" + ) + return True refusal = self._profile_effect_tier_refusal( workflow, step, @@ -7141,6 +7290,150 @@ def _resolve_effects( for effect in effects ] + @staticmethod + def _capture_effect_pre_state(verifier: Any, effects: list["Effect"]) -> Any: + """Capture the pre-action baseline for resolved effect contracts. + + Multi-candidate deployments select an adapter per resolved effect. A + normal verifier keeps the established single snapshot behavior. + """ + capture_for_effects = getattr(verifier, "capture_pre_state_for_effects", None) + if callable(capture_for_effects): + return capture_for_effects(effects) + if getattr(verifier, "_openadapt_requires_preflight", False): + probe = _probe_adapter_connection(verifier) + if not probe.ok: + raise ValueError( + "the selected effect verifier failed its read-only connection " + f"preflight ({probe.substrate}: {probe.reason})" + ) + return verifier.capture_pre_state() + + @staticmethod + def _effect_pre_state_for(before: Any, effect: "Effect") -> Any: + """Return one selected effect baseline, or a normal shared baseline.""" + selected = getattr(before, "for_effect", None) + if callable(selected): + return selected(effect).state + return before + + @staticmethod + def _effect_binding_for( + verifier: Any, before: Any, effect: "Effect" + ) -> tuple[Any, Any, Optional[Any], str]: + """Return the exact adapter, state, tier, and identity bound to an effect.""" + + selected = getattr(before, "for_effect", None) + if callable(selected): + binding = selected(effect) + return ( + binding.verifier, + binding.state, + binding.tier, + binding.verifier_identity, + ) + return ( + verifier, + before, + verifier_effect_tier(verifier, effect), + effect_verifier_identity(verifier), + ) + + @staticmethod + def _candidate_binding_refusal( + before: Any, effects: list["Effect"] + ) -> Optional[str]: + """Refuse a candidate whose identity or tier changed after selection.""" + + selected = getattr(before, "for_effect", None) + if not callable(selected): + return None + for effect in effects: + binding = selected(effect) + current_tier = verifier_effect_tier(binding.verifier, effect) + current_identity = effect_verifier_identity(binding.verifier) + if ( + current_tier != binding.tier + or current_identity != binding.verifier_identity + ): + return ( + "the selected effect verifier identity or evidence tier changed " + "after pre-action binding; refusing to use a different proof" + ) + return None + + @staticmethod + def _verify_current_effect( + verifier: Any, effect: "Effect", current: EffectState + ) -> Any: + """Use an adapter's current-state path without inventing a new baseline.""" + + candidate = effect.model_copy(deep=True) + original = candidate.model_dump(mode="json") + callback = getattr(verifier, "verify_current_state", None) + if callable(callback): + verdict = callback(candidate, current) + else: + from openadapt_flow.runtime.effects._common import judge_records + + baseline = EffectState( + substrate=current.substrate, + reachable=True, + records=[], + detail={"current_state_readback": True}, + ) + verdict = judge_records( + candidate, + baseline, + current.records if current.reachable is True else None, + substrate=current.substrate, + ) + if candidate.model_dump(mode="json") != original: + raise EffectContractMutationError( + "effect verifier changed the resolved effect contract" + ) + return verdict + + @classmethod + def _required_effect_pre_state_unreadable( + cls, verifier: Any, before: Any, effects: list["Effect"] + ) -> bool: + """Return true only when a selected effect needs, but lacks, a baseline. + + Different-path on-screen read-back proves persistence after input and + deliberately has no readable pre-action state. Delta, duplicate, and + collateral checks still require their selected verifier baseline. + """ + requirement = getattr(verifier, "requires_readable_pre_state_for", None) + selected = getattr(before, "for_effect", None) + missing_reachability = object() + for effect in effects: + if callable(selected): + required = bool(selected(effect).requires_readable_pre_state) + else: + required = ( + bool(requirement(effect)) + if callable(requirement) + else bool(effect.count_new_only or effect.forbid_collateral_loss) + ) + state = cls._effect_pre_state_for(before, effect) + # Legacy in-process verifiers may retain an opaque pre-state for + # their own verify implementation. Their established verification + # path decides whether it is usable. If a legacy state exposes + # reachability, it must state exactly True; None, falsey values, + # and truthy non-bools are not evidence that its baseline is + # readable. Candidate deployments are stricter: their verifier + # accepts only EffectState, so a missing reachability value cannot + # bypass its pre-actuation guard. + reachable = getattr(state, "reachable", missing_reachability) + if ( + required + and reachable is not missing_reachability + and reachable is not True + ): + return True + return False + def _profile_effect_tier_refusal( self, workflow: Workflow, @@ -7290,7 +7583,33 @@ def _verify_effects( ) try: verdict = verify_effect_without_mutation(verifier, effect, before) - tier = verifier_effect_tier(verifier, effect) + selected_pre_state = getattr(before, "for_effect", None) + if callable(selected_pre_state): + # Candidate selection and its evidence strength are pinned + # before input. Do not recompute the tier after delivery: + # a stateful plugin must not change the receipt contract. + binding = selected_pre_state(effect) + tier = binding.tier + identity = binding.verifier_identity + if ( + verifier_effect_tier(binding.verifier, effect) != tier + or effect_verifier_identity(binding.verifier) != identity + ): + result.effect_verified = False + result.safety_halt = True + result.failure_category = "governed_refusal" + result.effect_results.append( + "selected effect verifier identity or evidence tier " + "changed after input; HALT" + ) + return ( + f"Effect verifier binding changed after input for step " + f"'{step.id}' — refusing evidence from a different " + "adapter or tier; reconciliation is required" + ) + else: + tier = verifier_effect_tier(verifier, effect) + identity = effect_verifier_identity(verifier) except EffectContractMutationError: result.effect_verified = False result.safety_halt = True @@ -7308,6 +7627,7 @@ def _verify_effects( EffectVerificationEvidence( effect_contract_hash=effect.contract_hash(), substrate=verdict.substrate, + verifier_identity=identity, verification_tier=(int(tier) if tier is not None else None), initial_verdict=verdict.verdict.value, final_verdict=verdict.verdict.value, @@ -7340,6 +7660,7 @@ def _verify_effects( EffectVerificationEvidence( effect_contract_hash=effect.contract_hash(), substrate=final_verdict.substrate, + verifier_identity=identity, verification_tier=(int(tier) if tier is not None else None), initial_verdict=verdict.verdict.value, final_verdict=final_verdict.verdict.value, @@ -7360,6 +7681,7 @@ def _verify_effects( EffectVerificationEvidence( effect_contract_hash=effect.contract_hash(), substrate=final_verdict.substrate, + verifier_identity=identity, verification_tier=(int(tier) if tier is not None else None), initial_verdict=verdict.verdict.value, final_verdict=final_verdict.verdict.value, @@ -7386,6 +7708,7 @@ def _verify_effects( EffectVerificationEvidence( effect_contract_hash=effect.contract_hash(), substrate=verdict.substrate, + verifier_identity=identity, verification_tier=(int(tier) if tier is not None else None), initial_verdict=verdict.verdict.value, final_verdict=verdict.verdict.value, diff --git a/tests/test_attended_actions.py b/tests/test_attended_actions.py index 20121241..00744e6a 100644 --- a/tests/test_attended_actions.py +++ b/tests/test_attended_actions.py @@ -102,6 +102,7 @@ EffectVerdict, Verdict, ) +from openadapt_flow.runtime.effects.adapter import CandidateEffectVerifier from openadapt_flow.runtime.replayer import Replayer from openadapt_flow.verification import VerificationTier from tests.test_replayer import ( @@ -930,6 +931,61 @@ def capture_pre_state(self, context=None): assert target is None +def test_program_attended_completion_uses_pinned_candidate_current_readback(tmp_path): + class PersistedReadback: + substrate = "persisted-readback" + verification_tier = VerificationTier.PERSISTED_STATE_REACQUISITION + _openadapt_verifier_identity = "sha256:" + "d" * 64 + + def __init__(self): + self.current_reads = 0 + + def capture_pre_state(self, context=None): + return EffectState(substrate=self.substrate, reachable=False) + + def requires_readable_pre_state_for(self, effect): + return False + + def verify(self, effect, before, context=None): + raise AssertionError("attended completion must use current-state readback") + + def verify_current_state(self, effect, current, context=None): + self.current_reads += 1 + return EffectVerdict( + verdict=Verdict.CONFIRMED, + kind=effect.kind, + substrate=self.substrate, + ) + + candidate = PersistedReadback() + result, target = Replayer( + FakeBackend(), + vision=FakeVision(), + effect_verifier=CandidateEffectVerifier([candidate]), + ).revalidate_attended_program_completion( + _attended_effect_program(), + graph_id="__program__", + state_id="human", + params={}, + bundle_dir=tmp_path, + run_dir=tmp_path / "run", + run_id="run-program-current-readback", + transition_baseline=TransitionObservation(), + transition_digest=lambda field, value: f"{field}:{value}", + ) + + assert result.ok is True + assert target == "done" + assert candidate.current_reads == 1 + assert ( + result.effect_evidence[0].verifier_identity + == candidate._openadapt_verifier_identity + ) + assert result.effect_evidence[0].verification_tier == int( + VerificationTier.PERSISTED_STATE_REACQUISITION + ) + + def test_attended_final_persistence_callback_cannot_verify_changed_semantics( tmp_path, monkeypatch ): diff --git a/tests/test_effect_kit_config.py b/tests/test_effect_kit_config.py index 1ab1f263..221b1623 100644 --- a/tests/test_effect_kit_config.py +++ b/tests/test_effect_kit_config.py @@ -286,6 +286,48 @@ def test_defaults_are_none_kind(self): assert build_effect_verifier(DeploymentConfig().effects) is None +class TestVerifierCandidates: + def test_selects_the_strongest_configured_candidate(self, tmp_path: Path): + """A file/export proof wins over a screen-consistency alternative.""" + from openadapt_flow.runtime.effects.adapter import CandidateEffectVerifier + + verifier = build_effect_verifier( + EffectsConfig( + candidates=[ + EffectsConfig(kind="onscreen"), + EffectsConfig(kind="file", root=str(tmp_path)), + ] + ) + ) + assert isinstance(verifier, CandidateEffectVerifier) + + def test_candidate_construction_never_skips_a_broken_stronger_proof( + self, monkeypatch, tmp_path: Path + ): + """A missing proof credential stops setup; it cannot silently downgrade.""" + monkeypatch.delenv("MISSING_ORACLE_TOKEN", raising=False) + cfg = EffectsConfig( + candidates=[ + EffectsConfig( + kind="rest", + base_url="https://oracle.invalid", + auth={"bearer_env": "MISSING_ORACLE_TOKEN"}, + ), + EffectsConfig(kind="file", root=str(tmp_path)), + ] + ) + with pytest.raises(ValueError, match="MISSING_ORACLE_TOKEN"): + build_effect_verifier(cfg) + + def test_candidates_reject_ambiguous_legacy_configuration(self, tmp_path: Path): + with pytest.raises(ValueError, match="mutually exclusive"): + EffectsConfig( + kind="file", + root=str(tmp_path), + candidates=[EffectsConfig(kind="onscreen")], + ) + + class TestOnScreenKitConfig: def test_onscreen_builds_unbound_readback_verifier(self): from openadapt_flow.runtime.effects.onscreen import OnScreenReadbackVerifier diff --git a/tests/test_effect_verifier_candidates.py b/tests/test_effect_verifier_candidates.py new file mode 100644 index 00000000..666e97aa --- /dev/null +++ b/tests/test_effect_verifier_candidates.py @@ -0,0 +1,538 @@ +"""Per-effect candidate verifier selection remains fail-closed.""" + +from __future__ import annotations + +import pytest + +from openadapt_flow.deployment import ( + EffectsConfig, + build_effect_verifier, + build_replayer, +) +from openadapt_flow.ir import ActionKind, EffectVerificationEvidence, Step, Workflow +from openadapt_flow.runtime import Replayer +from openadapt_flow.runtime.durable.approval import StateDiverged +from openadapt_flow.runtime.effects.adapter import ( + CandidateEffectVerifier, + RedactingVerifier, + RedactionPolicy, + register_verifier_factory, + verifier_identity, +) +from openadapt_flow.runtime.effects.effect import ( + Effect, + EffectKind, + EffectState, + EffectVerdict, + ReadbackNav, + ReadbackSpec, + Verdict, +) +from openadapt_flow.runtime.effects.onscreen import OnScreenReadbackVerifier +from openadapt_flow.verification import VerificationTier + + +class _Verifier: + def __init__(self, tier, *, reachable=True, name="test"): + self.verification_tier = tier + self.substrate = name + self.reachable = reachable + self.captures = 0 + self.verifies = 0 + + def capture_pre_state(self): + self.captures += 1 + return EffectState(substrate=self.substrate, reachable=self.reachable) + + def verify(self, effect, before, context=None): + self.verifies += 1 + return EffectVerdict( + verdict=Verdict.CONFIRMED if before.reachable else Verdict.INDETERMINATE, + kind=effect.kind, + substrate=self.substrate, + matched_records=[{"patient": "private"}], + unavailable=not before.reachable, + ) + + +def _effect(*, different_path=False): + return Effect( + kind=EffectKind.FIELD_EQUALS, + field="note", + value="saved", + readback=ReadbackSpec( + region=(0, 0, 10, 10), + different_path=different_path, + renavigation=( + [ + ReadbackNav(action="click", point=(1, 1)), + ReadbackNav(action="type", text="record"), + ReadbackNav(action="key", key="Enter"), + ] + if different_path + else [] + ), + ), + ) + + +def test_selection_refines_onscreen_tier_per_resolved_effect(): + onscreen = OnScreenReadbackVerifier(backend=None) + session = _Verifier(VerificationTier.PERSISTED_STATE_REACQUISITION, name="session") + selector = CandidateEffectVerifier([onscreen, session]) + persisted = _effect(different_path=True) + same_surface = _effect(different_path=False) + + state = selector.capture_pre_state_for_effects([persisted, same_surface]) + + assert state.for_effect(persisted).verifier is onscreen + assert state.for_effect(same_surface).verifier is session + assert ( + selector.verification_tier_for(persisted) + == VerificationTier.PERSISTED_STATE_REACQUISITION + ) + assert ( + selector.verification_tier_for(same_surface) + == VerificationTier.PERSISTED_STATE_REACQUISITION + ) + + +def test_selected_candidate_pre_state_is_captured_before_verification(): + strong = _Verifier(VerificationTier.INDEPENDENT_SYSTEM, name="export") + weak = _Verifier(VerificationTier.IMMEDIATE_SCREEN, name="screen") + effect = _effect() + selector = CandidateEffectVerifier([weak, strong]) + + state = selector.capture_pre_state_for_effects([effect]) + + assert state.reachable + assert strong.captures == 1 + assert weak.captures == 0 + assert state.for_effect(effect).state.substrate == "export" + + +def test_candidate_rejects_an_opaque_pre_state_before_actuation(): + class _OpaquePreStateVerifier(_Verifier): + def capture_pre_state(self): + return object() + + effect = _effect() + verifier = CandidateEffectVerifier( + [_OpaquePreStateVerifier(VerificationTier.INDEPENDENT_SYSTEM)] + ) + + with pytest.raises(ValueError, match="invalid pre-state"): + verifier.capture_pre_state_for_effects([effect]) + + +def test_unavailable_selected_candidate_never_downgrades_after_actuation(): + strong = _Verifier( + VerificationTier.INDEPENDENT_SYSTEM, reachable=False, name="export" + ) + weak = _Verifier(VerificationTier.IMMEDIATE_SCREEN, name="screen") + effect = _effect() + selector = CandidateEffectVerifier([strong, weak]) + state = selector.capture_pre_state_for_effects([effect]) + + verdict = selector.verify(effect, state) + + assert not state.reachable + assert verdict.verdict is Verdict.INDETERMINATE + assert strong.verifies == 1 + assert weak.verifies == 0 + + +def test_different_path_onscreen_candidate_does_not_require_prestate_readability(): + """Post-action reacquisition can prove a GUI-only write without a delta.""" + effect = _effect(different_path=True) + verifier = CandidateEffectVerifier([OnScreenReadbackVerifier(backend=None)]) + before = verifier.capture_pre_state_for_effects([effect]) + + assert before.for_effect(effect).state.reachable is False + assert ( + Replayer._required_effect_pre_state_unreadable(verifier, before, [effect]) + is False + ) + + +def test_prestate_requirement_stays_bound_when_candidate_tier_changes(): + class _Stateful(_Verifier): + def requires_readable_pre_state_for(self, effect): + return True + + def capture_pre_state(self): + self.verification_tier = VerificationTier.IMMEDIATE_SCREEN + return EffectState(substrate=self.substrate, reachable=False) + + effect = _effect() + strong = _Stateful(VerificationTier.INDEPENDENT_SYSTEM, name="strong") + weak = _Verifier(VerificationTier.INDEPENDENT_SESSION, name="weak") + verifier = CandidateEffectVerifier([strong, weak]) + + before = verifier.capture_pre_state_for_effects([effect]) + + assert before.for_effect(effect).verifier is strong + assert before.for_effect(effect).requires_readable_pre_state is True + assert Replayer._required_effect_pre_state_unreadable(verifier, before, [effect]) + assert Replayer._candidate_binding_refusal(before, [effect]) is not None + + +def test_only_selected_candidate_runs_connection_preflight(): + class _Probed(_Verifier): + def __init__(self, tier, *, ok, name): + super().__init__(tier, name=name) + self.ok = ok + self.probes = 0 + self._openadapt_requires_preflight = True + + def test_connection(self, context=None): + from openadapt_flow.runtime.effects.adapter import ConnectionProbe + + self.probes += 1 + return ConnectionProbe( + ok=self.ok, + substrate=self.substrate, + reason="ready" if self.ok else "unavailable", + ) + + effect = _effect() + selected = _Probed(VerificationTier.INDEPENDENT_SYSTEM, ok=True, name="selected") + unselected = _Probed(VerificationTier.IMMEDIATE_SCREEN, ok=False, name="unselected") + + CandidateEffectVerifier([selected, unselected]).capture_pre_state_for_effects( + [effect] + ) + + assert selected.probes == 1 + assert unselected.probes == 0 + + +def test_selected_candidate_connection_failure_refuses_before_capture(): + class _Unavailable(_Verifier): + _openadapt_requires_preflight = True + + def test_connection(self, context=None): + from openadapt_flow.runtime.effects.adapter import ConnectionProbe + + return ConnectionProbe( + ok=False, substrate=self.substrate, reason="unavailable" + ) + + selected = _Unavailable(VerificationTier.INDEPENDENT_SYSTEM, name="selected") + + with pytest.raises(ValueError, match="connection preflight"): + CandidateEffectVerifier([selected]).capture_pre_state_for_effects([_effect()]) + + assert selected.captures == 0 + + +def test_bool_plugin_tier_is_rejected(): + class _CompletePlugin(_Verifier): + def test_connection(self, context=None): + raise AssertionError("construction must reject the invalid tier first") + + def capture_post_state(self, context=None): + return self.capture_pre_state(context) + + register_verifier_factory( + "candidate-bool-tier-test", + lambda cfg, params: _CompletePlugin(True), + replace=True, + ) + with pytest.raises( + ValueError, match="verification_tier must be a VerificationTier" + ): + build_effect_verifier( + EffectsConfig(candidates=[EffectsConfig(kind="candidate-bool-tier-test")]) + ) + + +def test_tier_only_plugin_is_rejected_during_candidate_construction(): + class _TierOnly: + verification_tier = VerificationTier.INDEPENDENT_SYSTEM + + register_verifier_factory( + "candidate-tier-only-test", lambda cfg, params: _TierOnly(), replace=True + ) + with pytest.raises( + ValueError, + match="missing test_connection, capture_pre_state, capture_post_state, verify", + ): + build_effect_verifier( + EffectsConfig(candidates=[EffectsConfig(kind="candidate-tier-only-test")]) + ) + + +def test_connection_aggregates_all_candidates_without_selection_or_raising(): + readable = _Verifier(VerificationTier.INDEPENDENT_SYSTEM, name="export") + unavailable = _Verifier( + VerificationTier.IMMEDIATE_SCREEN, reachable=False, name="screen" + ) + verifier = CandidateEffectVerifier([readable, unavailable]) + + probe = verifier.test_connection() + + assert probe.ok is False + assert probe.substrate == "candidates" + assert probe.detail["candidates"] == [ + {"substrate": "export", "ok": True, "reason": "reachable"}, + {"substrate": "screen", "ok": False, "reason": "unreachable"}, + ] + assert readable.captures == 1 + assert unavailable.captures == 1 + + +def test_redacting_wrapper_delegates_candidate_connection_probe(): + verifier = RedactingVerifier( + CandidateEffectVerifier([_Verifier(VerificationTier.INDEPENDENT_SYSTEM)]), + RedactionPolicy(), + ) + + probe = verifier.test_connection() + + assert probe.ok is True + assert probe.substrate == "candidates" + + +def test_redacting_wrapper_connection_probe_never_raises(): + class _ExplodingConnection(_Verifier): + def test_connection(self): + raise RuntimeError("no connection") + + verifier = RedactingVerifier( + CandidateEffectVerifier( + [_ExplodingConnection(VerificationTier.INDEPENDENT_SYSTEM)] + ), + RedactionPolicy(), + ) + + probe = verifier.test_connection() + + assert probe.ok is False + assert ( + probe.detail["candidates"][0]["reason"] + == "connection probe raised: RuntimeError" + ) + + +def test_plugin_with_incomplete_lifecycle_fails_at_construction(): + class _IncompletePlugin: + substrate = "incomplete" + verification_tier = VerificationTier.INDEPENDENT_SYSTEM + + def capture_pre_state(self): + return EffectState(substrate=self.substrate, reachable=True) + + def verify(self, effect, before, context=None): + return EffectVerdict(verdict=Verdict.INDETERMINATE, kind=effect.kind) + + register_verifier_factory( + "candidate-incomplete-plugin-test", + lambda cfg, params: _IncompletePlugin(), + replace=True, + ) + + with pytest.raises(ValueError, match="test_connection, capture_post_state"): + build_effect_verifier(EffectsConfig(kind="candidate-incomplete-plugin-test")) + + +def test_plugin_with_incompatible_verify_signature_fails_at_construction(): + class _WrongSignature: + substrate = "wrong-signature" + verification_tier = VerificationTier.INDEPENDENT_SYSTEM + + def test_connection(self, context=None): + return None + + def capture_pre_state(self, context=None): + return EffectState(substrate=self.substrate, reachable=True) + + def capture_post_state(self, context=None): + return self.capture_pre_state(context) + + def verify(self): + return None + + register_verifier_factory( + "candidate-wrong-signature-test", + lambda cfg, params: _WrongSignature(), + replace=True, + ) + + with pytest.raises(ValueError, match="incompatible lifecycle signature: verify"): + build_effect_verifier(EffectsConfig(kind="candidate-wrong-signature-test")) + + +def test_verifier_identity_binds_config_but_not_literal_secret(): + first = build_effect_verifier( + EffectsConfig( + kind="fhir", base_url="https://records.example/a", access_token="one" + ) + ) + rotated_secret = build_effect_verifier( + EffectsConfig( + kind="fhir", base_url="https://records.example/a", access_token="two" + ) + ) + changed_boundary = build_effect_verifier( + EffectsConfig( + kind="fhir", base_url="https://records.example/b", access_token="two" + ) + ) + + assert verifier_identity(first) == verifier_identity(rotated_secret) + assert verifier_identity(first) != verifier_identity(changed_boundary) + + +@pytest.mark.parametrize( + ("retained_identity", "retained_tier"), + [ + ("sha256:" + "a" * 64, int(VerificationTier.INDEPENDENT_SYSTEM)), + ("sha256:" + "b" * 64, int(VerificationTier.INDEPENDENT_SESSION)), + ], +) +def test_durable_revalidation_refuses_changed_candidate_binding( + retained_identity, retained_tier +): + effect = _effect() + verifier = _Verifier(VerificationTier.INDEPENDENT_SYSTEM, name="current") + verifier._openadapt_verifier_identity = "sha256:" + "b" * 64 + selector = CandidateEffectVerifier([verifier]) + step = Step( + id="write", + intent="write", + action=ActionKind.KEY, + key="Enter", + effects=[effect], + ) + evidence = EffectVerificationEvidence( + effect_contract_hash=effect.contract_hash(), + substrate="retained", + verifier_identity=retained_identity, + verification_tier=retained_tier, + initial_verdict="confirmed", + final_verdict="confirmed", + ) + + with pytest.raises(StateDiverged, match="identity or evidence tier changed"): + Replayer(object(), effect_verifier=selector).revalidate_retained_effects( + [effect], + workflow=Workflow(name="resume", steps=[step]), + step=step, + actuation_path="gui", + retained_evidence=[evidence], + ) + + +def test_durable_revalidation_uses_selected_current_state_readback(): + class _PersistedReadback(_Verifier): + def __init__(self): + super().__init__( + VerificationTier.PERSISTED_STATE_REACQUISITION, + reachable=False, + name="onscreen-like", + ) + self.current_reads = 0 + + def requires_readable_pre_state_for(self, effect): + return False + + def verify(self, effect, before, context=None): + raise AssertionError("durable readback must use the current-state path") + + def verify_current_state(self, effect, current, context=None): + self.current_reads += 1 + return EffectVerdict( + verdict=Verdict.CONFIRMED, + kind=effect.kind, + substrate=self.substrate, + ) + + effect = _effect(different_path=True) + verifier = _PersistedReadback() + verifier._openadapt_verifier_identity = "sha256:" + "c" * 64 + step = Step( + id="write", + intent="write", + action=ActionKind.KEY, + key="Enter", + effects=[effect], + ) + evidence = EffectVerificationEvidence( + effect_contract_hash=effect.contract_hash(), + substrate=verifier.substrate, + verifier_identity=verifier._openadapt_verifier_identity, + verification_tier=int(verifier.verification_tier), + initial_verdict="confirmed", + final_verdict="confirmed", + ) + + Replayer( + object(), effect_verifier=CandidateEffectVerifier([verifier]) + ).revalidate_retained_effects( + [effect], + workflow=Workflow(name="resume", steps=[step]), + step=step, + actuation_path="gui", + retained_evidence=[evidence], + ) + + assert verifier.current_reads == 1 + + +def test_redacting_wrapper_keeps_selected_candidate_and_redacts_evidence(): + strong = _Verifier(VerificationTier.INDEPENDENT_SYSTEM, name="export") + effect = _effect() + verifier = RedactingVerifier( + CandidateEffectVerifier([strong]), RedactionPolicy(redact_fields=["patient"]) + ) + + state = verifier.capture_pre_state_for_effects([effect]) + verdict = verifier.verify(effect, state) + + assert state.for_effect(effect).verifier is strong + assert verdict.matched_records == [{"patient": "[redacted]"}] + + +def test_build_replayer_binds_backend_to_candidate_onscreen_verifier(): + backend = object() + verifier = build_effect_verifier( + EffectsConfig(candidates=[EffectsConfig(kind="onscreen")]) + ) + + replayer = build_replayer( + backend, + allow_egress=False, + effect_verifier=verifier, + api_actuator=None, + durable=False, + use_structural=True, + ) + + assert replayer.effect_verifier is verifier + assert isinstance(verifier, CandidateEffectVerifier) + assert verifier._candidates[0]._backend is backend + + +def test_build_replayer_binds_backend_through_redacting_candidate_wrapper(): + backend = object() + verifier = build_effect_verifier( + EffectsConfig( + candidates=[EffectsConfig(kind="onscreen")], + evidence_redact_fields=["value"], + ) + ) + + replayer = build_replayer( + backend, + allow_egress=False, + effect_verifier=verifier, + api_actuator=None, + durable=False, + use_structural=True, + ) + + assert replayer.effect_verifier is verifier + assert isinstance(verifier, RedactingVerifier) + assert isinstance(verifier._inner, CandidateEffectVerifier) + assert verifier._inner._candidates[0]._backend is backend diff --git a/tests/test_replayer_api_actuator.py b/tests/test_replayer_api_actuator.py index 758c3596..fad040ea 100644 --- a/tests/test_replayer_api_actuator.py +++ b/tests/test_replayer_api_actuator.py @@ -46,7 +46,9 @@ EffectKind, RestRecordVerifier, ) +from openadapt_flow.runtime.effects.adapter import CandidateEffectVerifier from openadapt_flow.runtime.replayer import Replayer +from openadapt_flow.verification import VerificationTier # Reuse the scripted fakes from the main replayer unit tests (pytest's prepend # import mode puts tests/ on sys.path). @@ -171,6 +173,54 @@ def verify(self, *args, **kwargs): raise RuntimeError("verifier unavailable") +class _TrackingRestVerifier(RestRecordVerifier): + """Count candidate use and optionally fail only after delivery.""" + + def __init__( + self, + url, + *, + substrate, + tier, + fail_verify=False, + tier_after_verify=None, + ): + super().__init__(url) + self.substrate = substrate + self.verification_tier = tier + self.fail_verify = fail_verify + self.tier_after_verify = tier_after_verify + self.captures = 0 + self.verifies = 0 + + def capture_pre_state(self, context=None): + self.captures += 1 + return super().capture_pre_state(context) + + def verify(self, expected, before, context=None): + self.verifies += 1 + if self.fail_verify: + raise RuntimeError("selected verifier unavailable after delivery") + verdict = super().verify(expected, before, context) + if self.tier_after_verify is not None: + self.verification_tier = self.tier_after_verify + return verdict + + +class _LegacyReachabilityState: + def __init__(self, reachable): + self.reachable = reachable + + +class _LegacyReachabilityVerifier(RestRecordVerifier): + def __init__(self, url, reachable): + super().__init__(url) + self._reachable = reachable + + def capture_pre_state(self): + return _LegacyReachabilityState(self._reachable) + + # -- ACTUATED + CONFIRMED: API performs the write, GUI is skipped ----------- @@ -247,6 +297,67 @@ def test_unreachable_api_halts_without_gui_fallback(tmp_path): stop() +def test_unreachable_effect_pre_state_refuses_api_actuation(tmp_path): + """A readable pre-action proof is required before any API write.""" + url, db, stop = _fault_server() + try: + backend = GuiWritingBackend(url) + workflow = _api_save_workflow(effects=[_record_written()]) + bundle, run_dir = _dirs(tmp_path) + replayer = Replayer( + backend, + vision=_vision_that_confirms_saved(), + effect_verifier=RestRecordVerifier("http://127.0.0.1:1"), + api_actuator=ApiActuator(url), + poll_interval_s=0.01, + ) + + report = replayer.run(workflow, bundle_dir=bundle, run_dir=run_dir) + + assert report.success is False + assert report.results[0].effect_verified is False + assert backend.actions == [] + assert db.snapshot()["records"] == [] + finally: + stop() + + +def test_opaque_legacy_effect_pre_state_uses_its_existing_verification_path(): + """The candidate guard does not reinterpret an older verifier snapshot.""" + effect = _record_written() + + assert ( + Replayer._required_effect_pre_state_unreadable(object(), object(), [effect]) + is False + ) + + +@pytest.mark.parametrize("reachable", [None, 0, "", False]) +def test_invalid_legacy_reachability_refuses_api_actuation_before_input( + tmp_path, reachable +): + """A present legacy reachability member must be exactly True.""" + url, db, stop = _fault_server() + try: + backend = GuiWritingBackend(url) + workflow = _api_save_workflow(effects=[_record_written()]) + bundle, run_dir = _dirs(tmp_path) + report = Replayer( + backend, + vision=_vision_that_confirms_saved(), + effect_verifier=_LegacyReachabilityVerifier(url, reachable), + api_actuator=ApiActuator(url), + poll_interval_s=0.01, + ).run(workflow, bundle_dir=bundle, run_dir=run_dir) + + assert report.success is False + assert report.results[0].effect_verified is False + assert backend.actions == [] + assert db.snapshot()["records"] == [] + finally: + stop() + + def test_post_send_protocol_error_is_proven_by_the_complete_contract(tmp_path): """A lost response after a committed API write is uncertain delivery. @@ -286,6 +397,96 @@ def test_post_send_protocol_error_is_proven_by_the_complete_contract(tmp_path): stop() +def test_response_loss_refuses_selected_candidate_tier_change_after_input(tmp_path): + """A selected verifier cannot relabel its evidence after one API send.""" + url, db, stop = _fault_server() + try: + strong = _TrackingRestVerifier( + url, + substrate="independent-records", + tier=VerificationTier.INDEPENDENT_SYSTEM, + tier_after_verify=VerificationTier.IMMEDIATE_SCREEN, + ) + weak = _TrackingRestVerifier( + url, + substrate="screen-readback", + tier=VerificationTier.IMMEDIATE_SCREEN, + ) + session = _ResponseLossSession() + bundle, run_dir = _dirs(tmp_path) + report = Replayer( + GuiWritingBackend(url), + vision=_vision_that_confirms_saved(), + effect_verifier=CandidateEffectVerifier([weak, strong]), + api_actuator=ApiActuator(url, session=session), + poll_interval_s=0.01, + ).run( + _api_save_workflow(effects=[_record_written()]), + bundle_dir=bundle, + run_dir=run_dir, + ) + + result = report.results[0] + assert report.success is False + assert report.transaction_outcome == "RECONCILIATION_REQUIRED" + assert result.delivery_uncertainty is not None + assert result.delivery_uncertainty.resolved_by_contract is False + assert result.effect_evidence == [] + assert "binding changed after input" in (result.error or "") + assert session.requests == 1 + assert strong.captures == 1 + assert strong.verifies == 1 + assert weak.captures == 0 + assert weak.verifies == 0 + assert len(db.snapshot()["records"]) == 1 + finally: + stop() + + +def test_response_loss_never_downgrades_an_unavailable_selected_candidate(tmp_path): + """A weaker verifier cannot convert uncertain delivery into success.""" + url, db, stop = _fault_server() + try: + strong = _TrackingRestVerifier( + url, + substrate="independent-records", + tier=VerificationTier.INDEPENDENT_SYSTEM, + fail_verify=True, + ) + weak = _TrackingRestVerifier( + url, + substrate="screen-readback", + tier=VerificationTier.IMMEDIATE_SCREEN, + ) + session = _ResponseLossSession() + bundle, run_dir = _dirs(tmp_path) + report = Replayer( + GuiWritingBackend(url), + vision=_vision_that_confirms_saved(), + effect_verifier=CandidateEffectVerifier([strong, weak]), + api_actuator=ApiActuator(url, session=session), + poll_interval_s=0.01, + ).run( + _api_save_workflow(effects=[_record_written()]), + bundle_dir=bundle, + run_dir=run_dir, + ) + + result = report.results[0] + assert report.success is False + assert report.transaction_outcome == "RECONCILIATION_REQUIRED" + assert result.delivery_uncertainty is not None + assert result.delivery_uncertainty.resolved_by_contract is False + assert session.requests == 1 + assert strong.captures == 1 + assert strong.verifies == 1 + assert weak.captures == 0 + assert weak.verifies == 0 + assert len(db.snapshot()["records"]) == 1 + finally: + stop() + + def test_post_send_protocol_error_with_refuted_effect_requires_reconciliation( tmp_path, ):