feat: bind remote volatility comparison contract

docs/REMOTE_FRAME_CONTRACT.md

@@ -0,0 +1,11 @@
+# Remote frame contract
+
+`remote_frame_contract` is a versioned deployment field for remote RDP and
+remote-display settling. It binds exact frame dimensions and reviewed volatile
+rectangles. The runtime rejects a geometry change or an overlap with a declared
+protected region. It retains raw frame bytes and raw lease hashes. It masks only
+transient derived inputs used for pointer-settle and final pre-input content
+comparisons.
+
+The current deployment schema can declare protected regions. Desktop editing
+of these reviewed regions is a separate follow-up; no runtime learning occurs.

openadapt_flow/backend.py

@@ -528,7 +528,9 @@ class RemoteActuationBackend(Protocol):
     The backend owns a one-shot content lease for the returned frame.  Its next
     input method captures once more under the backend input lock and refuses
     before the first input edge if the window/session, dimensions, readiness,
-    or exact frame content changed.  The lease is consumed once so a
+    or exact frame content changed.  A sealed remote frame contract can exclude
+    reviewed volatile regions from a derived comparison.  Raw frame evidence
+    and the exact lease stay unmodified.  The lease is consumed once so a
     multi-character type or double-click gesture cannot invalidate itself.
     """
 
@@ -537,6 +539,15 @@ def acquire_actuation_frame(self) -> bytes:
         ...
 
 
+@runtime_checkable
+class RemoteFrameContractBackend(Protocol):
+    """Optional pre-input protected-region binding for remote comparison masks."""
+
+    def arm_remote_frame_contract(
+        self, *, protected_regions: tuple[tuple[int, int, int, int], ...]
+    ) -> None: ...
+
+
 @runtime_checkable
 class FreshActuationReacquisitionBackend(Protocol):
     """Reset one proved zero-input invalidation for bounded reacquisition.

openadapt_flow/backends/factory.py

@@ -196,6 +196,7 @@ def _build_rdp_backend(
             application_version_marker=cfg.rdp_application_version_marker,
             environment_marker=cfg.rdp_environment_marker,
             session_marker=cfg.rdp_session_marker,
+            remote_frame_contract=cfg.remote_frame_contract,
         )
 
     if rdp_transport is not None or has_host:
@@ -220,6 +221,7 @@ def _build_rdp_backend(
             application_version_marker=cfg.rdp_application_version_marker,
             environment_marker=cfg.rdp_environment_marker,
             session_marker=cfg.rdp_session_marker,
+            remote_frame_contract=cfg.remote_frame_contract,
         )
 
     if window_client is not None or has_window:
@@ -236,6 +238,7 @@ def _build_rdp_backend(
         kwargs["application_version_marker"] = cfg.rdp_application_version_marker
         kwargs["environment_marker"] = cfg.rdp_environment_marker
         kwargs["session_marker"] = cfg.rdp_session_marker
+        kwargs["remote_frame_contract"] = cfg.remote_frame_contract
         return RemoteDisplayBackend(window_client, **kwargs)
 
     raise ValueError(

openadapt_flow/backends/rdp_backend.py

@@ -69,6 +69,7 @@
     StructuralResolutionRefused,
 )
 from openadapt_flow.ir import ActionDeliveryReceipt, Point
+from openadapt_flow.remote_frame_contract import RemoteFrameContract
 from openadapt_flow.runtime.resolver import visual_resolution_point_fingerprint
 
 # What a transport may hand back as the current frame: a PIL image, or raw
@@ -307,8 +308,10 @@ def __init__(
         session_marker: Optional[str] = None,
         session_marker_probe: Optional[Callable[[bytes], bool]] = None,
         session_identity_observer: Optional[Callable[[], Optional[str]]] = None,
+        remote_frame_contract: Optional["RemoteFrameContract"] = None,
     ) -> None:
         self._transport = transport
+        self._remote_frame_contract = remote_frame_contract
         self._viewport = viewport
         self._max_frame_age_s = float(max_frame_age_s)
         if self._max_frame_age_s <= 0:
@@ -357,6 +360,7 @@ def __init__(
         self._session_identity_observer = session_identity_observer
         self._last_frame_monotonic: Optional[float] = None
         self._last_frame_digest: Optional[bytes] = None
+        self._last_comparison_digest: Optional[bytes] = None
         self._last_session_identity: Optional[str] = None
         self._qualification_environment: Optional[tuple[str, str, str, str]] = None
         self._qualification_input_guard: Optional[Callable[[], None]] = None
@@ -398,8 +402,15 @@ def screenshot(self) -> bytes:
             # and screenshot can never disagree.
             self._viewport = img.size
             png = self._png_bytes(img)
+            if self._remote_frame_contract is not None:
+                self._remote_frame_contract.require_geometry(img.size)
             self._last_frame_monotonic = time.monotonic()
             self._last_frame_digest = self._canonical_frame_digest(img)
+            self._last_comparison_digest = (
+                self._remote_frame_contract.comparison_digest(png)
+                if self._remote_frame_contract is not None
+                else self._last_frame_digest
+            )
             self._last_session_identity = self._session_identity_from_frame(png)
             if self._actuation_lease_state == _LEASE_ARMED:
                 self._invalidate_actuation_lease()
@@ -444,6 +455,12 @@ def acquire_actuation_frame(self) -> bytes:
             self._actuation_lease_state = _LEASE_ARMED
             return png
 
+    def arm_remote_frame_contract(
+        self, *, protected_regions: tuple[tuple[int, int, int, int], ...]
+    ) -> None:
+        if self._remote_frame_contract is not None:
+            self._remote_frame_contract.arm(protected_regions)
+
     def reset_fresh_actuation_state(self) -> None:
         """Reset only a typed zero-input content invalidation.
 
@@ -1102,8 +1119,16 @@ def _ensure_input_ready(
                     "target resolution; refusing input"
                 )
             if self._actuation_lease_state == _LEASE_ARMED:
-                digest = self._canonical_frame_digest(current_img)
-                if self._last_frame_digest is None or digest != self._last_frame_digest:
+                raw_digest = self._canonical_frame_digest(current_img)
+                digest = (
+                    self._remote_frame_contract.comparison_digest(current_png)
+                    if self._remote_frame_contract is not None
+                    else raw_digest
+                )
+                if (
+                    self._last_comparison_digest is None
+                    or digest != self._last_comparison_digest
+                ):
                     changed_pixel_count, changed_bbox = self._frame_difference(
                         self._actuation_frame_png,
                         current_img,

openadapt_flow/backends/remote_display.py

@@ -73,6 +73,7 @@
     StructuralResolutionRefused,
 )
 from openadapt_flow.ir import ActionDeliveryReceipt
+from openadapt_flow.remote_frame_contract import RemoteFrameContract
 from openadapt_flow.runtime.resolver import visual_resolution_point_fingerprint
 
 _PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"
@@ -526,8 +527,10 @@ def __init__(
         session_marker: Optional[str] = None,
         session_marker_probe: Optional[Callable[[bytes], bool]] = None,
         session_identity_observer: Optional[Callable[[], Optional[str]]] = None,
+        remote_frame_contract: Optional["RemoteFrameContract"] = None,
     ) -> None:
         self._client = client if client is not None else _default_window_client()
+        self._remote_frame_contract = remote_frame_contract
         self._owner_substr = owner_substr
         self._title_substr = title_substr
         self._require_input_trust = require_input_trust
@@ -589,6 +592,7 @@ def __init__(
         self._frame_window: Optional[WindowInfo] = None
         self._last_frame_monotonic: Optional[float] = None
         self._last_frame_digest: Optional[bytes] = None
+        self._last_comparison_digest: Optional[bytes] = None
         self._actuation_frame_png: Optional[bytes] = None
         self._last_session_identity: Optional[str] = None
         self._qualification_environment: Optional[tuple[str, str, str, str]] = None
@@ -714,11 +718,18 @@ def screenshot(self) -> bytes:
                     f"({scale_x:.4f}x vs {scale_y:.4f}y); refusing uncalibrated input"
                 )
             self._viewport = (w, h)
+            if self._remote_frame_contract is not None:
+                self._remote_frame_contract.require_geometry(self._viewport)
             self._scale_x, self._scale_y = scale_x, scale_y
             self._scale = scale_x  # compatibility for existing diagnostics
             self._frame_window = win
             self._last_frame_monotonic = time.monotonic()
             self._last_frame_digest = _canonical_rgb_digest(png)
+            self._last_comparison_digest = (
+                self._remote_frame_contract.comparison_digest(png)
+                if self._remote_frame_contract is not None
+                else self._last_frame_digest
+            )
             self._last_session_identity = self._session_identity_from_frame(png)
             # An ordinary observation is not permission to perform a
             # consequential remote action.  Only acquire_actuation_frame arms
@@ -798,6 +809,12 @@ def acquire_actuation_frame(self) -> bytes:
             self._actuation_frame_png = png
             return png
 
+    def arm_remote_frame_contract(
+        self, *, protected_regions: tuple[tuple[int, int, int, int], ...]
+    ) -> None:
+        if self._remote_frame_contract is not None:
+            self._remote_frame_contract.arm(protected_regions)
+
     def reset_fresh_actuation_state(self) -> None:
         """Reset only a typed zero-input content invalidation.
 
@@ -1281,8 +1298,9 @@ def _wait_for_pointer_settle(self) -> None:
         hover/cursor update reaches the remote framebuffer.  Sampling one frame
         after a fixed sleep can therefore arm a lease on the old pixels and
         invalidate it moments later.  This gate waits for consecutive,
-        byte-decoded RGB-identical frames; it never masks cursor regions or
-        weakens the exact post-resolution digest used at the delivery edge.
+        byte-decoded RGB-identical frames outside any reviewed volatile regions.
+        The contract applies only to derived comparisons. It never changes the
+        exact raw frame evidence or learns regions from ordinary runs.
         """
 
         deadline = time.monotonic() + self._pointer_settle_timeout_s
@@ -1291,7 +1309,7 @@ def _wait_for_pointer_settle(self) -> None:
         poll_s = max(0.01, self._settle_s)
         while time.monotonic() < deadline:
             self.screenshot()
-            digest = self._last_frame_digest
+            digest = self._last_comparison_digest
             if digest is not None and digest == previous_digest:
                 stable_frames += 1
             else:
@@ -1422,8 +1440,16 @@ def _ensure_input_ready(
                 # Consume once before the first input edge.  A double click or
                 # multi-character type is one gesture and must not invalidate
                 # itself after its first state-changing edge.
-                digest = _canonical_rgb_digest(png)
-                if self._last_frame_digest is None or digest != self._last_frame_digest:
+                raw_digest = _canonical_rgb_digest(png)
+                digest = (
+                    self._remote_frame_contract.comparison_digest(png)
+                    if self._remote_frame_contract is not None
+                    else raw_digest
+                )
+                if (
+                    self._last_comparison_digest is None
+                    or digest != self._last_comparison_digest
+                ):
                     changed_pixel_count, changed_bbox = self._frame_difference(
                         self._actuation_frame_png,
                         png,

openadapt_flow/deployment.py

@@ -26,6 +26,8 @@
 
 from pydantic import BaseModel, Field, field_validator, model_validator
 
+from openadapt_flow.remote_frame_contract import RemoteFrameContract
+
 # Import-light (pydantic only): the effect CONTRACT types double as the
 # declarative config vocabulary, so a deployment YAML binds run parameters
 # with the exact ``{param: ...}`` / ``{literal: ...}`` form bundles use.
@@ -141,6 +143,9 @@ class BackendConfig(BaseModel):
     #: Optional exact case-insensitive title. Zero or multiple exact matches are
     #: refused; the backend never selects a largest partial match.
     rdp_window_title: Optional[str] = None
+    # Exact reviewed geometry for derived remote settle comparisons. The
+    # enclosing DeploymentConfig serialization binds this value at admission.
+    remote_frame_contract: Optional[RemoteFrameContract] = None
 
 
 class EffectsConfig(BaseModel):

openadapt_flow/remote_frame_contract.py

@@ -0,0 +1,76 @@
+"""Immutable, deployment-bound comparison masks for remote frame settling."""
+
+from __future__ import annotations
+
+import hashlib
+import io
+from typing import Literal
+
+from PIL import Image, ImageDraw
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+Region = tuple[int, int, int, int]
+
+
+class RemoteFrameContract(BaseModel):
+    """Reviewed exact-geometry exclusions for derived settle inputs only."""
+
+    model_config = ConfigDict(extra="forbid", frozen=True)
+    schema_version: Literal["openadapt.remote-frame-contract/v1"] = (
+        "openadapt.remote-frame-contract/v1"
+    )
+    frame_width: int = Field(gt=0, le=32768)
+    frame_height: int = Field(gt=0, le=32768)
+    volatile_regions: tuple[Region, ...] = Field(min_length=1, max_length=32)
+    protected_regions: tuple[Region, ...] = Field(default_factory=tuple, max_length=128)
+
+    @model_validator(mode="after")
+    def validate_regions(self) -> "RemoteFrameContract":
+        for volatile in self.volatile_regions:
+            self._validate(volatile)
+            for protected in self.protected_regions:
+                self._validate(protected)
+                if _overlap(volatile, protected):
+                    raise ValueError("volatile region overlaps a protected region")
+        return self
+
+    def _validate(self, region: Region) -> None:
+        x, y, width, height = region
+        if (
+            width <= 0
+            or height <= 0
+            or x < 0
+            or y < 0
+            or x + width > self.frame_width
+            or y + height > self.frame_height
+        ):
+            raise ValueError("remote frame region is outside the exact qualified frame")
+
+    def require_geometry(self, size: tuple[int, int]) -> None:
+        if size != (self.frame_width, self.frame_height):
+            raise ValueError("remote frame contract geometry does not match live frame")
+
+    def arm(self, protected_regions: tuple[Region, ...]) -> None:
+        """Refuse a newly observed target/identity/effect overlap before input."""
+        for region in protected_regions:
+            self._validate(region)
+            if any(_overlap(region, volatile) for volatile in self.volatile_regions):
+                raise ValueError("volatile region overlaps a runtime protected region")
+
+    def comparison_digest(self, png: bytes) -> bytes:
+        """Hash a derived masked copy. Raw evidence and leases stay unmasked."""
+        image = Image.open(io.BytesIO(png)).convert("RGB")
+        self.require_geometry(image.size)
+        derived = image.copy()
+        draw = ImageDraw.Draw(derived)
+        for x, y, width, height in self.volatile_regions:
+            draw.rectangle((x, y, x + width - 1, y + height - 1), fill=(0, 0, 0))
+        out = io.BytesIO()
+        derived.save(out, format="PNG")
+        return hashlib.sha256(out.getvalue()).digest()
+
+
+def _overlap(left: Region, right: Region) -> bool:
+    x, y, width, height = left
+    a, b, c, d = right
+    return x < a + c and a < x + width and y < b + d and b < y + height

openadapt_flow/runtime/replayer.py

@@ -61,6 +61,7 @@
     GuardedSelectOptionBackend,
     PreparedPointerActuationBackend,
     RemoteActuationBackend,
+    RemoteFrameContractBackend,
     RichPointerActionBackend,
     SelectOptionBackend,
     StructuralResolutionRefused,
@@ -7921,6 +7922,36 @@ def _revalidate_consequential_actuation(
                 self._cancel_guarded_keyboard()
             if error is not None and focused_element_backend:
                 self._cancel_guarded_keyboard()
+            if error is None and isinstance(self.backend, RemoteFrameContractBackend):
+                protected: list[Region] = []
+                if fresh_region is not None:
+                    protected.append(fresh_region)
+                elif fresh_resolution is not None:
+                    x, y = fresh_resolution.point
+                    protected.append((x - 1, y - 1, 3, 3))
+                protected.extend(
+                    pc.region for pc in step.expect if pc.region is not None
+                )
+                if workflow.qualification is not None:
+                    policy = workflow.qualification.identity_policies.get(step.id)
+                    if policy is not None:
+                        protected.extend(
+                            s.region for s in policy.signals if s.region is not None
+                        )
+                try:
+                    self.backend.arm_remote_frame_contract(
+                        protected_regions=tuple(protected)
+                    )
+                except Exception as exc:
+                    return (
+                        fresh_resolution,
+                        fresh_region,
+                        fresh_png,
+                        (
+                            "Actuation preflight HALTED because the remote frame mask "
+                            f"overlaps protected evidence: {type(exc).__name__}"
+                        ),
+                    )
         # Retain the exact observation that authorizes the next input edge.
         # Composite TYPE/SELECT_OPTION and retry paths can re-resolve inside
         # ``_act`` after the outer scope captured its initial geometry. A typed