From 673d5a9e2665e7b7ae6493e647bff90fd7f76e87 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Thu, 30 Jul 2026 18:16:43 -0400 Subject: [PATCH 1/3] feat: select the strongest qualified effect verifier --- docs/EFFECT_KIT.md | 42 ++++- openadapt_flow/deployment.py | 79 ++++++++- openadapt_flow/runtime/effects/__init__.py | 6 + openadapt_flow/runtime/effects/adapter.py | 160 +++++++++++++++++- openadapt_flow/runtime/effects/onscreen.py | 5 + openadapt_flow/runtime/replayer.py | 131 ++++++++++++--- tests/test_effect_kit_config.py | 42 +++++ tests/test_effect_verifier_candidates.py | 178 +++++++++++++++++++++ tests/test_replayer_api_actuator.py | 25 +++ 9 files changed, 636 insertions(+), 32 deletions(-) create mode 100644 tests/test_effect_verifier_candidates.py diff --git a/docs/EFFECT_KIT.md b/docs/EFFECT_KIT.md index b3bde4b0..6f9164a9 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 91c4ee2e..de11c715 100644 --- a/openadapt_flow/deployment.py +++ b/openadapt_flow/deployment.py @@ -177,6 +177,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 +321,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). @@ -905,15 +939,54 @@ 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 + + candidate_selector = CandidateEffectVerifier(built) + 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 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..70ebecf8 100644 --- a/openadapt_flow/runtime/effects/adapter.py +++ b/openadapt_flow/runtime/effects/adapter.py @@ -88,7 +88,9 @@ from __future__ import annotations import hashlib +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 +104,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 +513,162 @@ def verify( raise NotImplementedError +@dataclass(frozen=True) +class CandidatePreState: + """The pre-action verifier and snapshot selected for one effect.""" + + verifier: Any + state: EffectState + tier: VerificationTier + + +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 _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) + + 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: list[ConnectionProbe] = [] + for candidate in self._candidates: + try: + probe = getattr(candidate, "test_connection", None) + if callable(probe): + result = probe(context) if context is not None else probe() + if isinstance(result, ConnectionProbe): + probes.append(result) + continue + state = ( + candidate.capture_pre_state() + if context is None + else candidate.capture_pre_state(context) + ) + probes.append( + ConnectionProbe( + ok=bool(state.reachable), + substrate=state.substrate or getattr(candidate, "substrate", ""), + reason="reachable" if state.reachable else "unreachable", + ) + ) + except Exception as exc: # noqa: BLE001 - preflight must not raise + probes.append( + ConnectionProbe( + ok=False, + substrate=str(getattr(candidate, "substrate", "")), + reason=f"connection probe raised: {type(exc).__name__}", + ) + ) + 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] = {} + captured: dict[int, EffectState] = {} + for effect in effects: + candidate, tier = self._select(effect) + key = id(candidate) + if key not in captured: + captured[key] = ( + candidate.capture_pre_state() + if context is None + else candidate.capture_pre_state(context) + ) + selections[_candidate_effect_key(effect)] = CandidatePreState( + verifier=candidate, + state=captured[key], + tier=tier, + ) + 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`. diff --git a/openadapt_flow/runtime/effects/onscreen.py b/openadapt_flow/runtime/effects/onscreen.py index fcff9ce7..0b712253 100644 --- a/openadapt_flow/runtime/effects/onscreen.py +++ b/openadapt_flow/runtime/effects/onscreen.py @@ -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, diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index 513e64dd..ef832085 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -3899,7 +3899,7 @@ 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" @@ -3926,8 +3926,9 @@ def revalidate_retained_effects( "forbid_collateral_loss": False, } ) + effect_current = self._effect_pre_state_for(current, effect) baseline = EffectState( - substrate=current.substrate, + substrate=effect_current.substrate, reachable=True, records=[], detail={"durable_resume_persistence_readback": True}, @@ -3935,8 +3936,8 @@ def revalidate_retained_effects( verdict = judge_records( persistence_effect, baseline, - current.records, - substrate=current.substrate, + effect_current.records, + substrate=effect_current.substrate, ) if not verdict.confirmed: raise StateDiverged( @@ -4262,7 +4263,7 @@ def revalidate_attended_completion( run_dir, evidence_step_id, "attended-after", frame ) return result - current = self.effect_verifier.capture_pre_state() + current = self._capture_effect_pre_state(self.effect_verifier, effects) tier_refusal = self._profile_effect_tier_refusal( profile_workflow, step, @@ -4302,6 +4303,7 @@ def revalidate_attended_completion( ) return result for effect in effects: + effect_current = self._effect_pre_state_for(current, effect) if effect.needs_operator_confirmation: result.effect_verified = False result.error = ( @@ -4318,7 +4320,7 @@ def revalidate_attended_completion( ) break baseline = EffectState( - substrate=current.substrate, + substrate=effect_current.substrate, reachable=True, records=[], detail={"attended_current_state_readback": True}, @@ -4326,8 +4328,8 @@ def revalidate_attended_completion( verdict = judge_records( effect, baseline, - current.records, - substrate=current.substrate, + effect_current.records, + substrate=effect_current.substrate, ) result.effect_contract_hashes.append(effect.contract_hash()) result.effect_results.append( @@ -5494,14 +5496,24 @@ 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, + effect_pre_state = self._capture_effect_pre_state( + active_verifier, resolved_effects ) + if 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 @@ -5795,16 +5807,28 @@ 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, - ) + effect_pre_state = self._capture_effect_pre_state( + active_verifier, resolved_effects ) + if 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: @@ -6427,7 +6451,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( @@ -6441,6 +6465,20 @@ def _try_api_tier( "send the request — run aborted" ) 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, @@ -6851,6 +6889,49 @@ 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) + 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 + + @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) + for effect in effects: + required = ( + bool(requirement(effect)) + if callable(requirement) + else bool(effect.count_new_only or effect.forbid_collateral_loss) + ) + if required and not cls._effect_pre_state_for(before, effect).reachable: + return True + return False + def _profile_effect_tier_refusal( self, workflow: Workflow, 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..c9a05d57 --- /dev/null +++ b/tests/test_effect_verifier_candidates.py @@ -0,0 +1,178 @@ +"""Per-effect candidate verifier selection remains fail-closed.""" + +from __future__ import annotations + +import pytest + +from openadapt_flow.deployment import EffectsConfig, build_effect_verifier +from openadapt_flow.runtime.effects.adapter import ( + CandidateEffectVerifier, + RedactingVerifier, + RedactionPolicy, + register_verifier_factory, +) +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_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_bool_plugin_tier_is_rejected(): + register_verifier_factory( + "candidate-bool-tier-test", lambda cfg, params: _Verifier(True), replace=True + ) + with pytest.raises(ValueError, match="no valid verification tier"): + 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 capture_pre_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_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]"}] diff --git a/tests/test_replayer_api_actuator.py b/tests/test_replayer_api_actuator.py index 758c3596..2fe1ab0c 100644 --- a/tests/test_replayer_api_actuator.py +++ b/tests/test_replayer_api_actuator.py @@ -247,6 +247,31 @@ 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_post_send_protocol_error_is_proven_by_the_complete_contract(tmp_path): """A lost response after a committed API write is uncertain delivery. From 1c57df6c8c5359d1a12276affb4baa5aa1510730 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Thu, 30 Jul 2026 19:20:09 -0400 Subject: [PATCH 2/3] fix: harden effect verifier candidate preflight --- openadapt_flow/deployment.py | 13 ++- openadapt_flow/runtime/effects/adapter.py | 127 +++++++++++++++------- tests/test_effect_verifier_candidates.py | 94 ++++++++++++++-- 3 files changed, 183 insertions(+), 51 deletions(-) diff --git a/openadapt_flow/deployment.py b/openadapt_flow/deployment.py index de11c715..3df11a2f 100644 --- a/openadapt_flow/deployment.py +++ b/openadapt_flow/deployment.py @@ -951,7 +951,9 @@ def build_effect_verifier( (fail loud rather than wire a broken verifier). """ if cfg.candidates: - built = [build_effect_verifier(candidate, params) for candidate in 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") @@ -1205,11 +1207,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/runtime/effects/adapter.py b/openadapt_flow/runtime/effects/adapter.py index 70ebecf8..80a5beb7 100644 --- a/openadapt_flow/runtime/effects/adapter.py +++ b/openadapt_flow/runtime/effects/adapter.py @@ -513,6 +513,82 @@ 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) + ) + 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" + ) + + +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.""" @@ -582,35 +658,10 @@ def test_connection(self, context: Any = None) -> ConnectionProbe: conservative aggregate is deterministic and cannot advertise a weak fallback as readiness for a stronger effect-specific selection. """ - probes: list[ConnectionProbe] = [] - for candidate in self._candidates: - try: - probe = getattr(candidate, "test_connection", None) - if callable(probe): - result = probe(context) if context is not None else probe() - if isinstance(result, ConnectionProbe): - probes.append(result) - continue - state = ( - candidate.capture_pre_state() - if context is None - else candidate.capture_pre_state(context) - ) - probes.append( - ConnectionProbe( - ok=bool(state.reachable), - substrate=state.substrate or getattr(candidate, "substrate", ""), - reason="reachable" if state.reachable else "unreachable", - ) - ) - except Exception as exc: # noqa: BLE001 - preflight must not raise - probes.append( - ConnectionProbe( - ok=False, - substrate=str(getattr(candidate, "substrate", "")), - reason=f"connection probe raised: {type(exc).__name__}", - ) - ) + 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", @@ -619,7 +670,11 @@ def test_connection(self, context: Any = None) -> ConnectionProbe: else "one or more candidate verifiers are unreachable", detail={ "candidates": [ - {"substrate": probe.substrate, "ok": probe.ok, "reason": probe.reason} + { + "substrate": probe.substrate, + "ok": probe.ok, + "reason": probe.reason, + } for probe in probes ] }, @@ -688,17 +743,7 @@ def __getattr__(self, name: str) -> Any: return getattr(self._inner, name) 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) diff --git a/tests/test_effect_verifier_candidates.py b/tests/test_effect_verifier_candidates.py index c9a05d57..14b01699 100644 --- a/tests/test_effect_verifier_candidates.py +++ b/tests/test_effect_verifier_candidates.py @@ -5,6 +5,7 @@ import pytest from openadapt_flow.deployment import EffectsConfig, build_effect_verifier +from openadapt_flow.runtime import Replayer from openadapt_flow.runtime.effects.adapter import ( CandidateEffectVerifier, RedactingVerifier, @@ -79,8 +80,14 @@ def test_selection_refines_onscreen_tier_per_resolved_effect(): 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 + 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(): @@ -98,7 +105,9 @@ def test_selected_candidate_pre_state_is_captured_before_verification(): def test_unavailable_selected_candidate_never_downgrades_after_actuation(): - strong = _Verifier(VerificationTier.INDEPENDENT_SYSTEM, reachable=False, name="export") + strong = _Verifier( + VerificationTier.INDEPENDENT_SYSTEM, reachable=False, name="export" + ) weak = _Verifier(VerificationTier.IMMEDIATE_SCREEN, name="screen") effect = _effect() selector = CandidateEffectVerifier([strong, weak]) @@ -112,11 +121,35 @@ def test_unavailable_selected_candidate_never_downgrades_after_actuation(): 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_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: _Verifier(True), replace=True + "candidate-bool-tier-test", + lambda cfg, params: _CompletePlugin(True), + replace=True, ) - with pytest.raises(ValueError, match="no valid verification tier"): + with pytest.raises( + ValueError, match="verification_tier must be a VerificationTier" + ): build_effect_verifier( EffectsConfig(candidates=[EffectsConfig(kind="candidate-bool-tier-test")]) ) @@ -129,7 +162,10 @@ class _TierOnly: register_verifier_factory( "candidate-tier-only-test", lambda cfg, params: _TierOnly(), replace=True ) - with pytest.raises(ValueError, match="missing capture_pre_state, verify"): + 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")]) ) @@ -137,7 +173,9 @@ class _TierOnly: 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") + unavailable = _Verifier( + VerificationTier.IMMEDIATE_SCREEN, reachable=False, name="screen" + ) verifier = CandidateEffectVerifier([readable, unavailable]) probe = verifier.test_connection() @@ -164,6 +202,48 @@ def test_redacting_wrapper_delegates_candidate_connection_probe(): 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_redacting_wrapper_keeps_selected_candidate_and_redacts_evidence(): strong = _Verifier(VerificationTier.INDEPENDENT_SYSTEM, name="export") effect = _effect() From b9664f0c426aa78f72df0374ebccb7a1bd0f034a Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Thu, 30 Jul 2026 19:24:28 -0400 Subject: [PATCH 3/3] style: format effect verifier prestate guard --- openadapt_flow/runtime/replayer.py | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index ef832085..51c359ae 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -6465,9 +6465,7 @@ def _try_api_tier( "send the request — run aborted" ) return True - if self._required_effect_pre_state_unreadable( - effect_verifier, before, effects - ): + 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" @@ -6890,9 +6888,7 @@ def _resolve_effects( ] @staticmethod - def _capture_effect_pre_state( - verifier: Any, effects: list["Effect"] - ) -> Any: + 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