fix(rdp): bind public presentation timeline to reviewed artifacts

benchmark/rdp_ladder/render_presentation.py

@@ -17,9 +17,11 @@
 from __future__ import annotations
 
 import argparse
+import base64
 import hashlib
 import json
 import math
+import os
 import shutil
 import subprocess
 from pathlib import Path
@@ -54,6 +56,13 @@
         "outcome",
     }
 )
+PUBLICATION_APPROVAL_SCHEMA = "openadapt.rdp-publication-approval.v2"
+PUBLICATION_APPROVAL_SCOPE = "openadapt.rdp-publication.finalize.v1"
+CANDIDATE_VIDEO_NAME = "openadapt-rdp-demo.mp4"
+
+
+def _sha256(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
 
 
 class HybridTimeline:
@@ -233,11 +242,11 @@ def validate_hybrid_timeline(
         if source_frame is not None:
             if not isinstance(source_frame, dict):
                 raise RuntimeError("hybrid timeline source frame is invalid")
-            presentation_phase = source_frame.get("presentation_phase")
+            source_phase = source_frame.get("presentation_phase")
             file = source_frame.get("file")
-            if not isinstance(presentation_phase, str) or not isinstance(file, str):
+            if not isinstance(source_phase, str) or not isinstance(file, str):
                 raise RuntimeError("hybrid timeline source frame is incomplete")
-            source = source_frames.get((presentation_phase, file))
+            source = source_frames.get((source_phase, file))
             if source is None:
                 raise RuntimeError("hybrid timeline source frame is not retained")
             for key, source_key in (
@@ -940,6 +949,140 @@ def render(presentation_dir: Path, output: Path) -> dict:
     return render_manifest
 
 
+def _candidate_paths(candidate_dir: Path) -> tuple[Path, Path, Path]:
+    """Return the only files that a public RDP presentation may contain."""
+    video = candidate_dir / CANDIDATE_VIDEO_NAME
+    return (
+        video,
+        video.with_suffix(".timeline.json"),
+        video.with_suffix(".manifest.json"),
+    )
+
+
+def _candidate_inventory(candidate_dir: Path) -> tuple[Path, Path, Path]:
+    """Refuse a candidate directory with any unsigned extra file or directory."""
+    if candidate_dir.is_symlink() or not candidate_dir.is_dir():
+        raise RuntimeError(
+            "public artifact candidate directory is not a real directory"
+        )
+    video, timeline, manifest = _candidate_paths(candidate_dir)
+    expected = {video.name, timeline.name, manifest.name}
+    actual = {path.name for path in candidate_dir.iterdir()}
+    if actual != expected or not all(
+        path.is_file() and not path.is_symlink() for path in (video, timeline, manifest)
+    ):
+        raise RuntimeError("public artifact candidate inventory is not exact")
+    return video, timeline, manifest
+
+
+def render_candidate(presentation_dir: Path, candidate_dir: Path) -> dict:
+    """Render an isolated review candidate. This function never publishes it."""
+    if candidate_dir.exists():
+        raise RuntimeError("public artifact candidate directory already exists")
+    candidate_dir.mkdir(parents=True)
+    video, _timeline, _manifest = _candidate_paths(candidate_dir)
+    return render(presentation_dir, video)
+
+
+def approve_public_artifact_set(
+    candidate_dir: Path,
+    *,
+    approval_path: Path,
+    key_id: str,
+    private_key: bytes,
+) -> Path:
+    """Create a detached approval for a reviewed exact candidate manifest."""
+    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+    _video, _timeline, manifest = _candidate_inventory(candidate_dir)
+    unsigned = {
+        "schema_version": PUBLICATION_APPROVAL_SCHEMA,
+        "candidate_sha256": _sha256(manifest),
+        "signer_key_id": key_id,
+        "approval_scope": PUBLICATION_APPROVAL_SCOPE,
+    }
+    signature = Ed25519PrivateKey.from_private_bytes(private_key).sign(
+        json.dumps(unsigned, sort_keys=True, separators=(",", ":")).encode()
+    )
+    approval_path.write_text(
+        json.dumps(
+            {**unsigned, "signature": base64.b64encode(signature).decode()},
+            indent=2,
+            sort_keys=True,
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+    return approval_path
+
+
+def finalize_public_artifact_set(
+    presentation_dir: Path,
+    candidate_dir: Path,
+    output_dir: Path,
+    *,
+    approval_path: Path,
+    trusted_public_keys: dict[str, bytes],
+) -> None:
+    """Verify one closed candidate directory, then publish it by one rename."""
+    if candidate_dir.resolve() == output_dir.resolve():
+        raise RuntimeError("candidate and public artifact directories must differ")
+    if approval_path.resolve().is_relative_to(candidate_dir.resolve()):
+        raise RuntimeError("publication approval must be detached from the candidate")
+    if output_dir.exists():
+        raise RuntimeError("public artifact output directory already exists")
+    video, timeline_path, manifest_path = _candidate_inventory(candidate_dir)
+    approval = _load_json(approval_path)
+    required = {
+        "schema_version",
+        "candidate_sha256",
+        "signer_key_id",
+        "approval_scope",
+        "signature",
+    }
+    if (
+        not isinstance(approval, dict)
+        or set(approval) != required
+        or approval.get("schema_version") != PUBLICATION_APPROVAL_SCHEMA
+        or approval.get("approval_scope") != PUBLICATION_APPROVAL_SCOPE
+        or approval.get("candidate_sha256") != _sha256(manifest_path)
+        or not isinstance(approval.get("signer_key_id"), str)
+    ):
+        raise RuntimeError("publication approval does not bind the exact candidate")
+    key = trusted_public_keys.get(approval["signer_key_id"])
+    if key is None:
+        raise RuntimeError("publication approval signer is not trusted")
+    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
+
+    unsigned = {key: value for key, value in approval.items() if key != "signature"}
+    try:
+        Ed25519PublicKey.from_public_bytes(key).verify(
+            base64.b64decode(approval["signature"], validate=True),
+            json.dumps(unsigned, sort_keys=True, separators=(",", ":")).encode(),
+        )
+    except Exception as exc:
+        raise RuntimeError("publication approval signature is invalid") from exc
+
+    manifest = _load_json(manifest_path)
+    timeline = _load_json(timeline_path)
+    if (
+        not isinstance(manifest, dict)
+        or manifest.get("video") != video.name
+        or manifest.get("video_sha256") != _sha256(video)
+        or manifest.get("hybrid_timeline") != timeline_path.name
+        or manifest.get("hybrid_timeline_sha256") != _sha256(timeline_path)
+    ):
+        raise RuntimeError("public artifact candidate hashes do not match")
+    source_manifests = {
+        name: _load_json(presentation_dir / name / "manifest.json")
+        for name in PHASE_DIRS
+    }
+    graph = _load_json(presentation_dir / "02-compiled-workflow" / "program-graph.json")
+    validate_hybrid_timeline(timeline, manifests=source_manifests, graph=graph)
+    output_dir.parent.mkdir(parents=True, exist_ok=True)
+    os.replace(candidate_dir, output_dir)
+
+
 def main() -> int:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--presentation-dir", type=Path, required=True)

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,