From 76c71a163190bede746a31c32ccad355c16aa5a3 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Thu, 30 Jul 2026 18:16:43 -0400 Subject: [PATCH 1/9] 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 2984aee112a026cf018113806fec013e0c0e1934 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Thu, 30 Jul 2026 19:20:09 -0400 Subject: [PATCH 2/9] 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 356d3627ad067deded0c48d7ca961ec249cffb98 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Thu, 30 Jul 2026 19:24:28 -0400 Subject: [PATCH 3/9] 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 From 56fd39a06a7f06c1545b04e777545cc017855fab Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Thu, 30 Jul 2026 20:16:07 -0400 Subject: [PATCH 4/9] fix: bind candidate verifier backends --- openadapt_flow/runtime/effects/adapter.py | 27 ++++++++++++ tests/test_effect_verifier_candidates.py | 50 ++++++++++++++++++++++- 2 files changed, 76 insertions(+), 1 deletion(-) diff --git a/openadapt_flow/runtime/effects/adapter.py b/openadapt_flow/runtime/effects/adapter.py index 80a5beb7..fdbbd8b0 100644 --- a/openadapt_flow/runtime/effects/adapter.py +++ b/openadapt_flow/runtime/effects/adapter.py @@ -628,6 +628,24 @@ def __init__(self, candidates: list[Any]) -> None: 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): @@ -742,6 +760,15 @@ 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: return _probe_adapter_connection(self._inner, context) diff --git a/tests/test_effect_verifier_candidates.py b/tests/test_effect_verifier_candidates.py index 14b01699..0129e08d 100644 --- a/tests/test_effect_verifier_candidates.py +++ b/tests/test_effect_verifier_candidates.py @@ -4,7 +4,11 @@ import pytest -from openadapt_flow.deployment import EffectsConfig, build_effect_verifier +from openadapt_flow.deployment import ( + EffectsConfig, + build_effect_verifier, + build_replayer, +) from openadapt_flow.runtime import Replayer from openadapt_flow.runtime.effects.adapter import ( CandidateEffectVerifier, @@ -256,3 +260,47 @@ def test_redacting_wrapper_keeps_selected_candidate_and_redacts_evidence(): 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 From 927b0fd752749c447994644a4eda8a1110f08c77 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Thu, 30 Jul 2026 21:18:09 -0400 Subject: [PATCH 5/9] fix: preserve legacy effect verifier pre-states --- openadapt_flow/runtime/effects/adapter.py | 8 +++++++- openadapt_flow/runtime/replayer.py | 8 +++++++- tests/test_effect_verifier_candidates.py | 14 ++++++++++++++ tests/test_replayer_api_actuator.py | 12 ++++++++++++ 4 files changed, 40 insertions(+), 2 deletions(-) diff --git a/openadapt_flow/runtime/effects/adapter.py b/openadapt_flow/runtime/effects/adapter.py index fdbbd8b0..8e9ea711 100644 --- a/openadapt_flow/runtime/effects/adapter.py +++ b/openadapt_flow/runtime/effects/adapter.py @@ -707,11 +707,17 @@ def capture_pre_state_for_effects( candidate, tier = self._select(effect) key = id(candidate) if key not in captured: - captured[key] = ( + 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], diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index 51c359ae..dc748057 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -6924,7 +6924,13 @@ def _required_effect_pre_state_unreadable( 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: + 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. Candidate deployments are + # stricter: CandidateEffectVerifier accepts only EffectState, so a + # missing reachability value cannot bypass its pre-actuation guard. + if required and getattr(state, "reachable", None) is False: return True return False diff --git a/tests/test_effect_verifier_candidates.py b/tests/test_effect_verifier_candidates.py index 0129e08d..2f665b00 100644 --- a/tests/test_effect_verifier_candidates.py +++ b/tests/test_effect_verifier_candidates.py @@ -108,6 +108,20 @@ def test_selected_candidate_pre_state_is_captured_before_verification(): 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" diff --git a/tests/test_replayer_api_actuator.py b/tests/test_replayer_api_actuator.py index 2fe1ab0c..a5d5bc46 100644 --- a/tests/test_replayer_api_actuator.py +++ b/tests/test_replayer_api_actuator.py @@ -272,6 +272,18 @@ def test_unreachable_effect_pre_state_refuses_api_actuation(tmp_path): 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 + ) + + 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 fe3c0bacf244c82390a3e6d9b02278f159544770 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Thu, 30 Jul 2026 21:25:05 -0400 Subject: [PATCH 6/9] fix: refuse invalid legacy effect reachability --- openadapt_flow/runtime/replayer.py | 19 ++++++++++---- tests/test_replayer_api_actuator.py | 40 +++++++++++++++++++++++++++++ 2 files changed, 54 insertions(+), 5 deletions(-) diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index dc748057..2ffa6ca6 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -6918,6 +6918,7 @@ def _required_effect_pre_state_unreadable( collateral checks still require their selected verifier baseline. """ requirement = getattr(verifier, "requires_readable_pre_state_for", None) + missing_reachability = object() for effect in effects: required = ( bool(requirement(effect)) @@ -6926,11 +6927,19 @@ def _required_effect_pre_state_unreadable( ) 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. Candidate deployments are - # stricter: CandidateEffectVerifier accepts only EffectState, so a - # missing reachability value cannot bypass its pre-actuation guard. - if required and getattr(state, "reachable", None) is False: + # 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 diff --git a/tests/test_replayer_api_actuator.py b/tests/test_replayer_api_actuator.py index a5d5bc46..127f1506 100644 --- a/tests/test_replayer_api_actuator.py +++ b/tests/test_replayer_api_actuator.py @@ -171,6 +171,20 @@ def verify(self, *args, **kwargs): raise RuntimeError("verifier unavailable") +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 ----------- @@ -284,6 +298,32 @@ def test_opaque_legacy_effect_pre_state_uses_its_existing_verification_path(): ) +@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. From f1b1873fef657cfb71d8acf38e0cb3fc56405192 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Fri, 31 Jul 2026 16:01:11 +0200 Subject: [PATCH 7/9] style: format effect verifier test --- tests/test_replayer_api_actuator.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tests/test_replayer_api_actuator.py b/tests/test_replayer_api_actuator.py index 127f1506..8483bdf7 100644 --- a/tests/test_replayer_api_actuator.py +++ b/tests/test_replayer_api_actuator.py @@ -291,9 +291,7 @@ def test_opaque_legacy_effect_pre_state_uses_its_existing_verification_path(): effect = _record_written() assert ( - Replayer._required_effect_pre_state_unreadable( - object(), object(), [effect] - ) + Replayer._required_effect_pre_state_unreadable(object(), object(), [effect]) is False ) From e054caa4c888e90c0732fccfeb9151e36051661f Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Fri, 31 Jul 2026 10:32:40 -0400 Subject: [PATCH 8/9] feat: carry managed Execute authority through BYOC Bind one managed Execute dispatch to the exact run, bundle, inputs, runtime, and control-plane origin. Keep the bearer token out of argv and the envelope. Refuse redirects, stale authority, and capability mismatch. Preserve ordinary customer-controlled jobs. --- openadapt_flow/connector/client.py | 9 +- openadapt_flow/connector/executor.py | 93 ++++++++++- openadapt_flow/connector/protocol.py | 52 ++++++- openadapt_flow/runner/dispatch_envelope.py | 39 +++-- openadapt_flow/runtime/durable/authority.py | 26 +++- tests/test_connector.py | 163 +++++++++++++++++++- tests/test_durable_authority_v13.py | 55 +++++++ 7 files changed, 409 insertions(+), 28 deletions(-) diff --git a/openadapt_flow/connector/client.py b/openadapt_flow/connector/client.py index 0b2a0d4f..93e5fb1f 100644 --- a/openadapt_flow/connector/client.py +++ b/openadapt_flow/connector/client.py @@ -28,6 +28,8 @@ from openadapt_flow.hosted import HostedError +MANAGED_DELIVERY_AUTHORITY_CAPABILITY = "managed_delivery_authority_v1" + class ConnectorClientError(HostedError): """An outbound control-plane call failed.""" @@ -88,7 +90,12 @@ def poll(self, wait_s: int) -> Optional[dict[str, Any]]: """Long-poll for the next leased job. Returns the poll envelope ``{"job": {...}}`` or None on a 204 (no work in the wait window).""" resp = self._client.post( - "/api/connector/poll", json={"wait": wait_s}, headers=self._bearer() + "/api/connector/poll", + json={ + "wait": wait_s, + "capabilities": [MANAGED_DELIVERY_AUTHORITY_CAPABILITY], + }, + headers=self._bearer(), ) if resp.status_code == 204: return None diff --git a/openadapt_flow/connector/executor.py b/openadapt_flow/connector/executor.py index 37fc40b3..1791372d 100644 --- a/openadapt_flow/connector/executor.py +++ b/openadapt_flow/connector/executor.py @@ -41,13 +41,21 @@ import tempfile from dataclasses import dataclass, field from pathlib import Path -from typing import Any, Callable, Optional +from typing import Any, Callable, Mapping, Optional +from urllib.parse import urlsplit from openadapt_flow.connector.config import ConnectorSettings from openadapt_flow.connector.protocol import ByocGovernanceError, ByocJob from openadapt_flow.connector.storage import CustomerStorage, extract_bundle_archive from openadapt_flow.failure_signals import automation_failure_signal from openadapt_flow.ir import RunReport +from openadapt_flow.runner.dispatch_envelope import ( + write_managed_dispatch_envelope_value, +) +from openadapt_flow.runtime.durable.authority import ( + REMOTE_AUTHORITY_TOKEN_ENV, + REMOTE_AUTHORITY_URL_ENV, +) #: A run-gate refusal (fail-closed admission denied) exits 2 before the replay #: creates report.json. @@ -64,7 +72,9 @@ class RunOutcome: #: A runner maps argv -> RunOutcome. The default shells the governed ``run`` CLI #: in a child process; tests inject a fake to avoid launching a GUI. -Runner = Callable[[list[str], Path], RunOutcome] +Runner = Callable[[list[str], Path, Mapping[str, str]], RunOutcome] + +_MANAGED_DELIVERY_AUTHORITY_PATH = "/api/internal/managed-delivery-permit" @dataclass @@ -104,6 +114,7 @@ def build_run_argv( bundle_dir: Path, run_dir: Path, params_file: Optional[Path], + managed_dispatch_file: Optional[Path] = None, ) -> list[str]: """The exact governed CLI invocation for a verified BYOC dispatch. @@ -132,14 +143,23 @@ def build_run_argv( argv += ["--params-file", str(params_file)] if job.target_url: argv += ["--url", job.target_url] + if managed_dispatch_file is not None: + argv += ["--managed-dispatch-file", str(managed_dispatch_file)] if settings.allow_unencrypted: # Local escape hatch, mirroring the governed run CLI; default OFF. argv.append("--allow-unencrypted") return argv -def _subprocess_runner(argv: list[str], run_dir: Path) -> RunOutcome: - proc = subprocess.run(argv, capture_output=True, text=True) # nosec - fixed argv +def _subprocess_runner( + argv: list[str], run_dir: Path, child_env: Mapping[str, str] +) -> RunOutcome: + proc = subprocess.run( # nosec - fixed argv and exact process-local env + argv, + capture_output=True, + text=True, + env=dict(child_env), + ) report_path = run_dir / "report.json" report: dict[str, Any] = {} if report_path.is_file(): @@ -267,6 +287,44 @@ def _write_policy_audit(job: ByocJob, run_dir: Path) -> None: pass +def _assert_managed_authority_is_pinned( + job: ByocJob, settings: ConnectorSettings +) -> None: + """Keep the run capability on the enrolled control-plane origin.""" + + if job.managed_dispatch is None: + return + try: + configured = urlsplit(settings.control_plane_url) + delivered = urlsplit(job.managed_delivery_authority_url or "") + configured_port = configured.port or ( + 443 if configured.scheme == "https" else None + ) + delivered_port = delivered.port or ( + 443 if delivered.scheme == "https" else None + ) + except ValueError as exc: + raise ByocGovernanceError( + "byoc managed delivery authority URL is invalid" + ) from exc + if ( + configured.scheme != "https" + or not configured.hostname + or configured.username is not None + or configured.password is not None + or bool(configured.query) + or bool(configured.fragment) + or delivered.scheme != "https" + or delivered.hostname != configured.hostname + or delivered_port != configured_port + or delivered.path != _MANAGED_DELIVERY_AUTHORITY_PATH + ): + raise ByocGovernanceError( + "byoc managed delivery authority is not pinned to the enrolled " + "control plane (fail closed)" + ) + + def execute_job( job: ByocJob, settings: ConnectorSettings, @@ -284,6 +342,7 @@ def execute_job( # 1. Governance gates (fail closed BEFORE any bundle is fetched or run). try: job.ensure_governed(require_run_token=require_run_token) + _assert_managed_authority_is_pinned(job, settings) except ByocGovernanceError as exc: return ExecutionResult("failed", {}, None, job.report_ref(), str(exc)) if not _grounding_env_available(job): @@ -332,10 +391,32 @@ def execute_job( try: params_file = _write_params_file(job.params, run_dir) + managed_dispatch_file = None + child_env = os.environ.copy() + # Never let a long-running Connector inherit stale authority from + # its service environment. Add the exact run capability only for + # the child that also receives the matching dispatch envelope. + child_env.pop(REMOTE_AUTHORITY_URL_ENV, None) + child_env.pop(REMOTE_AUTHORITY_TOKEN_ENV, None) + if job.managed_dispatch is not None: + managed_dispatch_file = write_managed_dispatch_envelope_value( + run_dir / "managed-dispatch.json", job.managed_dispatch + ) + child_env[REMOTE_AUTHORITY_URL_ENV] = str( + job.managed_delivery_authority_url + ) + child_env[REMOTE_AUTHORITY_TOKEN_ENV] = str(job.run_token) # 3. The governed, fail-closed child invocation. - argv = build_run_argv(job, settings, Path(bundle_dir), run_dir, params_file) - outcome = runner(argv, run_dir) + argv = build_run_argv( + job, + settings, + Path(bundle_dir), + run_dir, + params_file, + managed_dispatch_file, + ) + outcome = runner(argv, run_dir, child_env) except Exception as exc: return ExecutionResult( "failed", diff --git a/openadapt_flow/connector/protocol.py b/openadapt_flow/connector/protocol.py index be6734fe..5126d9ec 100644 --- a/openadapt_flow/connector/protocol.py +++ b/openadapt_flow/connector/protocol.py @@ -21,10 +21,12 @@ from __future__ import annotations from typing import Any, Optional +from urllib.parse import urlsplit from pydantic import BaseModel, ConfigDict, Field, ValidationError from openadapt_flow.ir import ExecutionTargetKind +from openadapt_flow.runner.dispatch_envelope import ManagedDispatchEnvelope class ByocJobParseError(ValueError): @@ -100,9 +102,15 @@ class ByocJob(BaseModel): bundle_download_url: Optional[str] = None # --- Governed callback binding (fail-closed) ------------------------------- - #: Run-scoped HMAC capability presented as ``x-run-token`` on the PHI-free - #: callback. Not a secret value — proves this run, forbids forging another. + #: Run-scoped bearer capability presented as ``x-run-token`` on the PHI-free + #: callback. It proves this run and must remain private. run_token: Optional[str] = None + #: Exact Cloud-issued, one-run authority. It stays PHI-free and is written + #: to a private local file before the governed child starts. + managed_dispatch: Optional[ManagedDispatchEnvelope] = None + #: HTTPS endpoint that issues monotonic per-action delivery permits for the + #: exact managed dispatch. The run-scoped ``run_token`` authenticates it. + managed_delivery_authority_url: Optional[str] = None bundle_version_id: Optional[str] = None runtime_validation_id: Optional[str] = None #: SHA-256 of the exact approved sanitized derivative ZIP bytes staged at @@ -162,6 +170,46 @@ def ensure_governed(self, *, require_run_token: bool = True) -> None: "byoc dispatch is missing a run-scoped callback token; refusing " "to run a job whose outcome we could not report (fail closed)" ) + if (self.managed_dispatch is None) != ( + self.managed_delivery_authority_url is None + ): + raise ByocGovernanceError( + "byoc managed dispatch and delivery authority must be supplied " + "together (fail closed)" + ) + if self.managed_dispatch is not None: + try: + authorization = self.managed_dispatch.exact_authorization() + except ValueError as exc: + raise ByocGovernanceError( + "byoc managed dispatch has an invalid authorization binding" + ) from exc + if self.managed_dispatch.run_id != self.run_id: + raise ByocGovernanceError( + "byoc managed dispatch belongs to a different run (fail closed)" + ) + if not authorization.approval_source.startswith("hosted:"): + raise ByocGovernanceError( + "byoc managed dispatch lacks hosted authorization provenance" + ) + try: + parsed = urlsplit(self.managed_delivery_authority_url or "") + except ValueError as exc: + raise ByocGovernanceError( + "byoc managed delivery authority URL is invalid" + ) from exc + if ( + parsed.scheme != "https" + or not parsed.hostname + or parsed.username is not None + or parsed.password is not None + or bool(parsed.query) + or bool(parsed.fragment) + ): + raise ByocGovernanceError( + "byoc managed delivery authority must be a credential-free " + "HTTPS endpoint without query or fragment" + ) if not self.bundle_version_id or not self.runtime_validation_id: raise ByocGovernanceError( "byoc dispatch is missing its immutable bundle-version or " diff --git a/openadapt_flow/runner/dispatch_envelope.py b/openadapt_flow/runner/dispatch_envelope.py index 5fdff6b1..693e2652 100644 --- a/openadapt_flow/runner/dispatch_envelope.py +++ b/openadapt_flow/runner/dispatch_envelope.py @@ -93,18 +93,18 @@ def exact_authorization(self) -> GovernedRunAuthorization: return self.authorization -def write_managed_dispatch_envelope(path: Path, verified: "VerifiedDispatch") -> Path: - """Write one already verified dispatch without following a path.""" - - run_id = verified.payload.run_id - authorization = verified.payload.authorization - envelope = ManagedDispatchEnvelope( - run_id=run_id, - bundle_content_digest=authorization.bundle_content_digest, - runtime_inputs_digest=authorization.runtime_inputs_digest, - authorization=authorization, - dispatch_binding_sha256=verified.payload.dispatch_binding_sha256, - ) +def write_managed_dispatch_envelope_value( + path: Path, envelope: ManagedDispatchEnvelope +) -> Path: + """Write one strictly parsed dispatch envelope without following a path. + + This is the shared process-boundary writer for both push runners and the + customer-controlled outbound-pull Connector. The caller must already hold + the strict :class:`ManagedDispatchEnvelope` model; this function rechecks + its internal authorization binding before any bytes reach disk. + """ + + envelope.exact_authorization() path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL @@ -136,6 +136,21 @@ def write_managed_dispatch_envelope(path: Path, verified: "VerifiedDispatch") -> return path +def write_managed_dispatch_envelope(path: Path, verified: "VerifiedDispatch") -> Path: + """Write one already verified runner dispatch without following a path.""" + + run_id = verified.payload.run_id + authorization = verified.payload.authorization + envelope = ManagedDispatchEnvelope( + run_id=run_id, + bundle_content_digest=authorization.bundle_content_digest, + runtime_inputs_digest=authorization.runtime_inputs_digest, + authorization=authorization, + dispatch_binding_sha256=verified.payload.dispatch_binding_sha256, + ) + return write_managed_dispatch_envelope_value(path, envelope) + + def _read_managed_dispatch_envelope(path: Path) -> ManagedDispatchEnvelope: """Read only a regular, private file owned by this effective user.""" diff --git a/openadapt_flow/runtime/durable/authority.py b/openadapt_flow/runtime/durable/authority.py index 130f4d59..ca42a0ce 100644 --- a/openadapt_flow/runtime/durable/authority.py +++ b/openadapt_flow/runtime/durable/authority.py @@ -31,7 +31,7 @@ from typing import Any, Iterator, Literal, Optional from urllib.error import HTTPError, URLError from urllib.parse import urlparse -from urllib.request import Request, urlopen +from urllib.request import HTTPRedirectHandler, Request, build_opener from pydantic import BaseModel, ConfigDict @@ -70,6 +70,27 @@ } +class _RefuseRemoteAuthorityRedirects(HTTPRedirectHandler): + """Keep the runner credential on the configured authority origin.""" + + def redirect_request( + self, + req: Request, + fp: Any, + code: int, + msg: str, + headers: Any, + newurl: str, + ) -> None: + raise HTTPError( + req.full_url, + code, + "remote delivery authority redirects are refused", + headers, + fp, + ) + + def _is_windows() -> bool: return os.name == "nt" @@ -1645,7 +1666,8 @@ def _require_remote_delivery_permit( response_bytes = self._remote_transport(url, headers, body) else: request = Request(url, data=body, headers=headers, method="POST") - with urlopen(request, timeout=10) as response: # nosec B310 - HTTPS above + opener = build_opener(_RefuseRemoteAuthorityRedirects()) + with opener.open(request, timeout=10) as response: # nosec B310 - HTTPS above if not 200 <= response.status < 300: raise DurableAuthorityBusy("remote delivery authority refused") response_bytes = response.read( diff --git a/tests/test_connector.py b/tests/test_connector.py index 6d23edf0..79a39a3e 100644 --- a/tests/test_connector.py +++ b/tests/test_connector.py @@ -20,6 +20,8 @@ import hashlib import io import json +import os +import stat import zipfile from pathlib import Path @@ -44,6 +46,13 @@ from openadapt_flow.connector.protocol import ByocGovernanceError from openadapt_flow.connector.storage import LocalCustomerStorage from openadapt_flow.failure_signals import automation_failure_signal +from openadapt_flow.runner.dispatch_envelope import ManagedDispatchEnvelope +from openadapt_flow.runner.protocol import dispatch_binding_sha256 +from openadapt_flow.runtime.authorization import GovernedRunAuthorization +from openadapt_flow.runtime.durable.authority import ( + REMOTE_AUTHORITY_TOKEN_ENV, + REMOTE_AUTHORITY_URL_ENV, +) def _bundle_archive_bytes() -> bytes: @@ -90,8 +99,27 @@ def _payload(**overrides): return payload +def _managed_dispatch(*, run_id: str | None = None) -> dict: + exact_run_id = run_id or _payload()["run_id"] + authorization = GovernedRunAuthorization( + bundle_content_digest="b" * 64, + runtime_inputs_digest="c" * 64, + admitted_policy_name="standard", + execution_profile="standard", + approval_source="hosted:execute-api", + ) + envelope = ManagedDispatchEnvelope( + run_id=exact_run_id, + bundle_content_digest=authorization.bundle_content_digest, + runtime_inputs_digest=authorization.runtime_inputs_digest, + authorization=authorization, + dispatch_binding_sha256=dispatch_binding_sha256(exact_run_id, authorization), + ) + return envelope.model_dump(mode="json") + + def _fake_success_runner(report): - def runner(argv, run_dir: Path) -> RunOutcome: + def runner(argv, run_dir: Path, _env) -> RunOutcome: (run_dir / "report.json").write_text(json.dumps(report)) return RunOutcome(returncode=0, report=report) @@ -134,6 +162,68 @@ def test_execute_job_success_writes_report_to_customer_storage_only(): assert storage.written["org_demo/run_5/report.json"]["success"] is True +def test_execute_job_consumes_exact_managed_dispatch_through_private_child_boundary( + monkeypatch, +): + authority_url = "https://app.openadapt.ai/api/internal/managed-delivery-permit" + token = "a" * 64 + managed_dispatch = _managed_dispatch() + monkeypatch.setenv(REMOTE_AUTHORITY_URL_ENV, "https://stale.invalid/permit") + monkeypatch.setenv(REMOTE_AUTHORITY_TOKEN_ENV, "stale-parent-token") + job = parse_job( + _payload( + managed_dispatch=managed_dispatch, + managed_delivery_authority_url=authority_url, + ), + lease_job_id="bjob_1", + ) + storage = InMemoryCustomerStorage(bundle_bytes=_BUNDLE_BYTES) + observed: dict[str, object] = {} + + def runner(argv, run_dir: Path, child_env) -> RunOutcome: + envelope_path = Path(argv[argv.index("--managed-dispatch-file") + 1]) + observed["argv"] = argv + observed["mode"] = stat.S_IMODE(envelope_path.stat().st_mode) + observed["envelope"] = json.loads(envelope_path.read_text()) + observed["authority_url"] = child_env[REMOTE_AUTHORITY_URL_ENV] + observed["authority_token"] = child_env[REMOTE_AUTHORITY_TOKEN_ENV] + observed["parent_token"] = os.environ.get(REMOTE_AUTHORITY_TOKEN_ENV) + (run_dir / "report.json").write_text(json.dumps(SUCCESS_REPORT)) + return RunOutcome(returncode=0, report=SUCCESS_REPORT) + + result = execute_job(job, ConnectorSettings(), storage, runner=runner) + + assert result.status == "success" + assert observed["mode"] == 0o600 + assert observed["envelope"] == managed_dispatch + assert observed["authority_url"] == authority_url + assert observed["authority_token"] == token + assert observed["parent_token"] == "stale-parent-token" + assert token not in json.dumps(observed["argv"]) + assert token not in json.dumps(observed["envelope"]) + + +def test_connector_drops_inherited_authority_from_a_job_without_an_envelope( + monkeypatch, +): + monkeypatch.setenv(REMOTE_AUTHORITY_URL_ENV, "https://stale.invalid/permit") + monkeypatch.setenv(REMOTE_AUTHORITY_TOKEN_ENV, "stale-parent-token") + job = parse_job(_payload(), lease_job_id="bjob_1") + storage = InMemoryCustomerStorage(bundle_bytes=_BUNDLE_BYTES) + + def runner(argv, run_dir: Path, child_env) -> RunOutcome: + assert REMOTE_AUTHORITY_URL_ENV not in child_env + assert REMOTE_AUTHORITY_TOKEN_ENV not in child_env + assert "--managed-dispatch-file" not in argv + (run_dir / "report.json").write_text(json.dumps(SUCCESS_REPORT)) + return RunOutcome(returncode=0, report=SUCCESS_REPORT) + + assert ( + execute_job(job, ConnectorSettings(), storage, runner=runner).status + == "success" + ) + + def test_execute_job_refuses_child_report_for_different_substrate(): job = parse_job(_payload(), lease_job_id="bjob_1") settings = ConnectorSettings(profile=None) @@ -281,7 +371,7 @@ def test_halt_maps_to_halt_status_and_present_flag(): settings = ConnectorSettings() storage = InMemoryCustomerStorage(bundle_bytes=_BUNDLE_BYTES) - def runner(argv, run_dir: Path) -> RunOutcome: + def runner(argv, run_dir: Path, _env) -> RunOutcome: (run_dir / "report.json").write_text(json.dumps(halt_report)) return RunOutcome(returncode=2, report=halt_report) @@ -365,6 +455,64 @@ def test_dispatch_without_run_token_is_refused(): job.ensure_governed() +def test_managed_dispatch_for_different_run_is_refused(): + job = parse_job( + _payload( + managed_dispatch=_managed_dispatch( + run_id="00000000-0000-4000-8000-000000000006" + ), + managed_delivery_authority_url=( + "https://app.openadapt.ai/api/internal/managed-delivery-permit" + ), + ), + lease_job_id="bjob_1", + ) + with pytest.raises(ByocGovernanceError, match="different run"): + job.ensure_governed() + + +def test_execute_refuses_managed_authority_on_a_different_https_origin(): + job = parse_job( + _payload( + managed_dispatch=_managed_dispatch(), + managed_delivery_authority_url=( + "https://attacker.invalid/api/internal/managed-delivery-permit" + ), + ), + lease_job_id="bjob_1", + ) + result = execute_job( + job, + ConnectorSettings(control_plane_url="https://app.openadapt.ai"), + InMemoryCustomerStorage(bundle_bytes=_BUNDLE_BYTES), + runner=_fake_success_runner(SUCCESS_REPORT), + ) + assert result.status == "failed" + assert "pinned to the enrolled control plane" in (result.error or "") + + +@pytest.mark.parametrize( + "overrides, match", + [ + ( + {"managed_dispatch": _managed_dispatch()}, + "must be supplied together", + ), + ( + { + "managed_dispatch": _managed_dispatch(), + "managed_delivery_authority_url": "http://localhost/permit", + }, + "credential-free HTTPS endpoint", + ), + ], +) +def test_incomplete_or_unsafe_managed_authority_is_refused(overrides, match): + job = parse_job(_payload(**overrides), lease_job_id="bjob_1") + with pytest.raises(ByocGovernanceError, match=match): + job.ensure_governed() + + @pytest.mark.parametrize("field", ["bundle_version_id", "runtime_validation_id"]) def test_dispatch_without_immutable_validation_binding_is_refused(field): job = parse_job(_payload(**{field: None}), lease_job_id="bjob_1") @@ -398,7 +546,7 @@ def test_execute_refuses_when_required_grounding_key_is_absent(monkeypatch): # A runner that would "succeed" — the governance gate must refuse BEFORE it. called = {"ran": False} - def runner(argv, run_dir: Path) -> RunOutcome: + def runner(argv, run_dir: Path, _env) -> RunOutcome: called["ran"] = True return RunOutcome(returncode=0, report=SUCCESS_REPORT) @@ -413,7 +561,7 @@ def test_execute_refuses_archive_digest_mismatch_before_runner(): storage = InMemoryCustomerStorage(bundle_bytes=_BUNDLE_BYTES) called = {"ran": False} - def runner(argv, run_dir: Path) -> RunOutcome: + def runner(argv, run_dir: Path, _env) -> RunOutcome: called["ran"] = True return RunOutcome(returncode=0, report=SUCCESS_REPORT) @@ -429,7 +577,7 @@ def test_child_failure_after_archive_verification_carries_verified_digest(): job = parse_job(_payload(), lease_job_id="bjob_1") storage = InMemoryCustomerStorage(bundle_bytes=_BUNDLE_BYTES) - def runner(argv, run_dir: Path) -> RunOutcome: + def runner(argv, run_dir: Path, _env) -> RunOutcome: raise RuntimeError("child unavailable") result = execute_job(job, ConnectorSettings(), storage, runner=runner) @@ -507,6 +655,7 @@ def __init__(self): self.tokens = {} # token -> org_id self.jobs = [] # queued jobs (each {id, org_id, payload, status, leased_by}) self.callbacks = [] + self.poll_bodies = [] self.callback_attempts = 0 self.callback_status = 200 self.acks = [] @@ -542,6 +691,7 @@ def handler(self, request: httpx.Request) -> httpx.Response: if path == "/api/connector/poll": if org is None: return httpx.Response(401, json={"error": "invalid connector token"}) + self.poll_bodies.append(body) for j in self.jobs: # ISOLATION: a connector only ever leases ITS OWN org jobs. if j["status"] == "queued" and j["org_id"] == org: @@ -611,6 +761,9 @@ def test_full_loop_dispatch_execute_callback_ack(): storage_factory=lambda job: storage, ) assert result["status"] == "success" + assert cp.poll_bodies == [ + {"wait": 0, "capabilities": ["managed_delivery_authority_v1"]} + ] # A PHI-free callback carrying the run-scoped token was posted. assert len(cp.callbacks) == 1 diff --git a/tests/test_durable_authority_v13.py b/tests/test_durable_authority_v13.py index f9f26953..bbc65f6e 100644 --- a/tests/test_durable_authority_v13.py +++ b/tests/test_durable_authority_v13.py @@ -17,6 +17,7 @@ import pytest +import openadapt_flow.runtime.durable.authority as durable_authority_module from openadapt_flow.ir import ActionKind, RunReport, Step, StepResult, Workflow from openadapt_flow.runtime.authorization import GovernedRunAuthorization from openadapt_flow.runtime.durable.approval import ( @@ -296,6 +297,60 @@ def transport(_url: str, _headers: dict[str, str], _body: bytes) -> bytes: assert authority.validate(manifest).delivery_sequence == 0 +def test_remote_authority_refuses_redirect_before_forwarding_bearer_token( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + manifest, authority, owner = _remote_ready_authority( + tmp_path, monkeypatch, lambda *_args: b"{}" + ) + authority._remote_transport = None + monkeypatch.setenv( + REMOTE_AUTHORITY_URL_ENV, + "https://control.example/api/internal/managed-delivery-permit", + ) + attempted_urls: list[str] = [] + + class RedirectingOpener: + def __init__(self, handler: object) -> None: + self.handler = handler + + def open(self, request: object, *, timeout: int) -> object: + assert timeout == 10 + attempted_urls.append(request.full_url) # type: ignore[attr-defined] + assert request.get_header("Authorization") == "Bearer secret-token" # type: ignore[attr-defined] + self.handler.redirect_request( # type: ignore[attr-defined] + request, + None, + 307, + "Temporary Redirect", + {}, + "https://attacker.invalid/permit", + ) + raise AssertionError("redirect refusal must stop the request") + + def build_redirecting_opener(handler: object) -> RedirectingOpener: + assert isinstance( + handler, durable_authority_module._RefuseRemoteAuthorityRedirects + ) + return RedirectingOpener(handler) + + monkeypatch.setattr( + durable_authority_module, + "build_opener", + build_redirecting_opener, + ) + with pytest.raises(DurableAuthorityBusy, match="unavailable or refused"): + authority.before_delivery( + manifest, + attempt_id="aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + owner_nonce_sha256=owner, + ) + assert attempted_urls == [ + "https://control.example/api/internal/managed-delivery-permit" + ] + assert authority.validate(manifest).delivery_sequence == 0 + + def test_initial_customer_local_delivery_needs_no_cloud_credentials( tmp_path: Path, ) -> None: From c9618ccb22753395d2fbe2c1a2dfb2424610730a Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Sun, 2 Aug 2026 13:39:49 +0200 Subject: [PATCH 9/9] fix: deliver consequential remote clicks through the frame lease A consequential remote pointer edge required GuardedRemotePointerActionBackend, so every opaque remote surface that can offer only the documented one-shot lease -- the no-DOM HTML5-canvas class that Citrix Workspace-web presents -- halted on its write step. The lease is the safety property; the typed receipt is retained evidence. Deliver through the lease when no typed receipt exists, keep the refusal for governed runs and qualification-fault proofs, and leave the result unlabeled so the outcome classifier still returns COMPLETED_UNVERIFIED rather than a production-eligible success. The canvas-nodom-ladder qualification is accepted 3/3 again on real CI. --- openadapt_flow/runtime/replayer.py | 144 ++++++++++++++++++++------- tests/test_replayer.py | 151 +++++++++++++++++++++++++++++ 2 files changed, 261 insertions(+), 34 deletions(-) diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index 513e64dd..0bc8d141 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -7543,6 +7543,33 @@ def _requires_atomic_identity_keyboard( and self._step_has_identity_contract(step, workflow) ) + def _typed_remote_receipt_required(self) -> bool: + """Whether a remote pointer edge must carry a typed delivery receipt. + + :class:`RemoteActuationBackend` already owns the exact-frame contract: + after ``acquire_actuation_frame`` the backend captures once more under + its own input lock and refuses *before the first input edge* when the + window/session, dimensions, readiness, or exact frame content changed. + :class:`GuardedRemotePointerActionBackend` adds an explicit + expected-frame hash plus a typed, target-bound receipt on top of that + contract; its absence is therefore missing *evidence*, not a missing + safety property. + + A governed (Standard/Regulated) run and an exact qualification-fault + proof both consume that receipt as retained evidence, so they must + refuse before delivery rather than actuate without it. Ordinary replay + keeps the documented lease path — refusing there over-halts every + conforming opaque remote backend, including a pixel-only no-DOM canvas + that can only expose the lease. Such a result stays unlabeled, which + the outcome classifier already maps to ``COMPLETED_UNVERIFIED`` instead + of a production-eligible success. + """ + + return ( + self.governed_authorization is not None + or self.qualification_fault_driver is not None + ) + def _cancel_guarded_coordinate(self) -> None: """Clean an unconsumed local coordinate lease, if one was armed.""" @@ -8577,39 +8604,58 @@ def _act( self._step_is_consequential(step, workflow) or self.qualification_fault_driver is not None ) + typed_remote = isinstance( + self.backend, + GuardedRemotePointerActionBackend, + ) + if ( + remote_consequential + and not typed_remote + and self._typed_remote_receipt_required() + ): + result.safety_halt = True + result.failure_category = "safety_halt" + return ( + f"Step '{step.id}' ({step.intent}) is a consequential " + "remote click, but this backend cannot bind its exact " + "fresh frame and target to delivery; run aborted" + ) if remote_consequential: - if not isinstance( - self.backend, - GuardedRemotePointerActionBackend, - ): - result.safety_halt = True - result.failure_category = "safety_halt" - return ( - f"Step '{step.id}' ({step.intent}) is a consequential " - "remote click, but this backend cannot bind its exact " - "fresh frame and target to delivery; run aborted" - ) refusal = self._delivery_authorization_refusal( workflow, params, step, result ) if refusal is not None: return refusal self._require_qualification_environment_current() - remote_pointer = cast( - GuardedRemotePointerActionBackend, self.backend - ) - result.delivery_receipt = self._deliver_backend_call( - result, - lambda: remote_pointer.click_guarded( - x, - y, - expected_frame_sha256=hashlib.sha256( - before_png - ).hexdigest(), - double=step.action is ActionKind.DOUBLE_CLICK, - ), - ) - result.actuation = "remote_guarded" + if typed_remote: + remote_pointer = cast( + GuardedRemotePointerActionBackend, self.backend + ) + result.delivery_receipt = self._deliver_backend_call( + result, + lambda: remote_pointer.click_guarded( + x, + y, + expected_frame_sha256=hashlib.sha256( + before_png + ).hexdigest(), + double=step.action is ActionKind.DOUBLE_CLICK, + ), + ) + result.actuation = "remote_guarded" + else: + # See ``_typed_remote_receipt_required``: the exact + # fresh frame is already bound by the backend's own + # one-shot lease, armed immediately above by + # ``_revalidate_consequential_actuation``. + self._deliver_backend_call( + result, + lambda: self.backend.click( + x, + y, + double=step.action is ActionKind.DOUBLE_CLICK, + ), + ) elif requires_atomic_identity: if isinstance(self.backend, GuardedCoordinateActionBackend): refusal = self._delivery_authorization_refusal( @@ -8683,16 +8729,39 @@ def _act( self._step_is_consequential(step, workflow) or self.qualification_fault_driver is not None ) - if remote_consequential: - if not isinstance( - self.backend, - GuardedRemotePointerActionBackend, - ): + if remote_consequential and not isinstance( + self.backend, + GuardedRemotePointerActionBackend, + ): + if self._typed_remote_receipt_required(): return ( f"Step '{step.id}' ({step.intent}) is a consequential " "remote right click, but this backend cannot bind its " "exact fresh frame and target to delivery; run aborted" ) + # See ``_typed_remote_receipt_required``: the exact fresh frame + # is already bound by the backend's own one-shot lease, and its + # first input edge consumes it. + if not isinstance(self.backend, RichPointerActionBackend): + return ( + f"Step '{step.id}' ({step.intent}) requires a right " + "click, but this backend has no bounded right-click " + "operation" + ) + refusal = self._delivery_authorization_refusal( + workflow, params, step, result + ) + if refusal is not None: + return refusal + self._require_qualification_environment_current() + self._deliver_backend_call( + result, + lambda: cast(RichPointerActionBackend, self.backend).right_click( + x, y + ), + ) + return None + if remote_consequential: refusal = self._delivery_authorization_refusal( workflow, params, step, result ) @@ -8825,9 +8894,16 @@ def _act( ), ) result.actuation = "dom" - elif isinstance(self.backend, RemoteActuationBackend) and ( - self._step_is_consequential(step, workflow) - or self.qualification_fault_driver is not None + elif ( + isinstance(self.backend, RemoteActuationBackend) + and ( + self._step_is_consequential(step, workflow) + or self.qualification_fault_driver is not None + ) + and ( + isinstance(self.backend, GuardedRemotePointerActionBackend) + or self._typed_remote_receipt_required() + ) ): if not isinstance( self.backend, diff --git a/tests/test_replayer.py b/tests/test_replayer.py index 0b6771be..609e3c50 100644 --- a/tests/test_replayer.py +++ b/tests/test_replayer.py @@ -462,6 +462,69 @@ def click(self, x, y, *, double=False): ) +class PixelOnlyRemoteBackend: + """Opaque remote surface exposing ONLY the two-phase actuation lease. + + The exact protocol surface of the no-DOM HTML5-canvas backend in + ``benchmark/canvas_ladder``: pixels in, coordinates out, no structural + tree, no identity seam, and no typed delivery receipt. It implements + :class:`RemoteActuationBackend` and + :class:`PreparedPointerActuationBackend` and nothing else, so the exact + frame is bound by the backend's own one-shot lease, which its next input + method consumes and validates before the first input edge. + """ + + def __init__(self, *, frame=None, viewport=VIEWPORT): + self._frame = frame if frame is not None else make_png(viewport) + self._viewport = viewport + self.actions: list = [] + self.prepared_pointer_points: list = [] + self.acquire_count = 0 + self._leased_frame_sha256 = None + self.frame_after_lease = None + + @property + def viewport(self): + return self._viewport + + def screenshot(self): + return self._frame + + def prepare_pointer_actuation(self, x, y): + self._leased_frame_sha256 = None + self.prepared_pointer_points.append((int(x), int(y))) + + def acquire_actuation_frame(self) -> bytes: + self.acquire_count += 1 + self._leased_frame_sha256 = hashlib.sha256(self._frame).hexdigest() + if self.frame_after_lease is not None: + self._frame = self.frame_after_lease + return self._frame + + def _consume_lease(self): + leased = self._leased_frame_sha256 + self._leased_frame_sha256 = None + if leased is None: + return + if hashlib.sha256(self._frame).hexdigest() != leased: + raise RuntimeError("remote frame content changed before the input edge") + + def click(self, x, y, *, double=False): + self._consume_lease() + self.actions.append(("click", x, y, double)) + + def type_text(self, text): + self._consume_lease() + self.actions.append(("type", text)) + + def press(self, key): + self._consume_lease() + self.actions.append(("press", key)) + + def scroll(self, dx, dy): + self.actions.append(("scroll", dx, dy)) + + def click_step( step_id="s1", *, @@ -639,6 +702,94 @@ def test_consequential_remote_click_re_resolves_on_fresh_frame(bundle, run_dir): assert report.results[0].resolution.point == (110, 105) +def test_consequential_click_uses_lease_when_backend_has_no_typed_receipt( + bundle, run_dir +): + """A pixel-only opaque remote surface must actuate, not over-halt. + + ``GuardedRemotePointerActionBackend`` adds an explicit expected-frame hash + and a typed receipt; a plain ``RemoteActuationBackend`` already refuses + before the first input edge when the leased frame changed. Refusing the + latter halts every no-DOM canvas/VDI workflow on its write step. + """ + backend = PixelOnlyRemoteBackend() + vision = FakeVision() + vision.template_results = [ + Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95), + Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95), + ] + + report = Replayer(backend, vision=vision).run( + Workflow(name="wf", steps=[click_step(risk="irreversible")]), + bundle_dir=bundle, + run_dir=run_dir, + ) + + assert report.success is True + assert backend.prepared_pointer_points == [(110, 105)] + assert backend.acquire_count == 1 + assert backend.actions == [("click", 110, 105, False)] + assert report.rung_counts == {"template": 1} + assert report.model_calls == 0 + # No typed receipt exists, so the result must not claim a closed + # production actuation path. + assert report.results[0].actuation is None + assert report.results[0].delivery_receipt is None + + +def test_consequential_lease_click_still_refuses_a_changed_frame(bundle, run_dir): + """The lease is the safety property: a changed frame must stop delivery.""" + backend = PixelOnlyRemoteBackend() + backend.frame_after_lease = make_png(color=(10, 20, 30)) + vision = FakeVision() + vision.template_results = [ + Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95), + Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95), + ] + + report = Replayer(backend, vision=vision).run( + Workflow(name="wf", steps=[click_step(risk="irreversible")]), + bundle_dir=bundle, + run_dir=run_dir, + ) + + assert report.success is False + assert backend.actions == [] + + +def test_governed_consequential_click_requires_a_typed_remote_receipt(bundle, run_dir): + """A governed run still refuses a remote click it cannot evidence.""" + backend = PixelOnlyRemoteBackend() + vision = FakeVision() + vision.template_results = [ + Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95), + Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95), + ] + workflow = Workflow(name="wf", steps=[click_step(risk="irreversible")]) + workflow.save(bundle) + workflow = Workflow.load(bundle) + assert workflow.manifest is not None + authorization = GovernedRunAuthorization( + bundle_content_digest=workflow.manifest.content_digest, + runtime_inputs_digest=runtime_inputs_digest(workflow, None, None), + admitted_policy_name="test", + ) + + report = Replayer( + backend, + vision=vision, + governed_authorization=authorization, + ).run( + workflow, + bundle_dir=bundle, + run_dir=run_dir, + ) + + assert report.success is False + assert backend.actions == [] + assert "cannot bind its exact" in (report.results[0].error or "") + + def test_consequential_remote_hover_target_movement_halts_before_input(bundle, run_dir): backend = RemoteLeaseBackend( initial_frame=make_png(color=(240, 240, 240)),