feat: bind production delivery to signed permits and receipts (#383)

Terminal-verification v2: separate Ed25519 pre-delivery permit and post-delivery receipt artifacts; exact canonical permit bytes retained before each managed backend edge and acknowledged after the backend call returns; missing or invalid receipts stay pending and refuse all later input delivery; terminal permit chain rebuilt from exact retained bytes; acceptor compares runner, delivery-session, and globally one-use claim identities with independently loaded Cloud state; one deterministic cross-language vector for the Cloud implementation.

Author: abrichr

openadapt_flow/runtime/durable/authority.py

@@ -10,7 +10,7 @@
 from contextvars import ContextVar
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
-from typing import Iterator, Literal, Optional
+from typing import Any, Iterator, Literal, Optional
 
 from pydantic import BaseModel, ConfigDict
 
@@ -274,14 +274,14 @@ def lease(
                 self.release(token)
             return
 
-    def before_delivery(self, token: ContinuationToken) -> None:
+    def before_delivery(self, token: ContinuationToken) -> Any:
         """Linearize Reject against the first resumed delivery boundary."""
 
         manifest = self.store.read_manifest()
         if manifest is None:
             raise ContinuationBusy("the durable manifest disappeared before delivery")
         try:
-            self.authority.before_delivery(
+            remote_permit = self.authority.before_delivery(
                 manifest,
                 attempt_id=token.attempt_id,
                 owner_nonce_sha256=self._nonce_digest(token.owner_nonce),
@@ -313,7 +313,7 @@ def before_delivery(self, token: ContinuationToken) -> None:
                         }
                     )
                 )
-                return
+                return remote_permit
             if record.phase != "validating":
                 raise ContinuationBusy(
                     "the continuation attempt is not eligible for delivery"
@@ -327,6 +327,27 @@ def before_delivery(self, token: ContinuationToken) -> None:
                     }
                 )
             )
+        return remote_permit
+
+    def acknowledge_delivery(
+        self, token: ContinuationToken, remote_permit: Any
+    ) -> None:
+        """Commit the receipt for the exact backend edge that just returned."""
+
+        if remote_permit is None:
+            return
+        manifest = self.store.read_manifest()
+        if manifest is None:
+            raise ContinuationBusy("the durable manifest disappeared after delivery")
+        try:
+            self.authority.acknowledge_remote_delivery(
+                manifest,
+                remote_permit,
+                attempt_id=token.attempt_id,
+                owner_nonce_sha256=self._nonce_digest(token.owner_nonce),
+            )
+        except StateDiverged as exc:
+            raise ContinuationBusy(str(exc)) from exc
 
     def bind_approval(self, token: ContinuationToken, approval: ApprovalRecord) -> None:
         """Bind the exact admitted approval in the external authority."""
@@ -720,9 +741,19 @@ def __init__(
     ) -> None:
         self.coordinator = coordinator
         self.token = token
+        self._pending_remote_permit: Any = None
 
     def before_delivery(self) -> None:
-        self.coordinator.before_delivery(self.token)
+        if self._pending_remote_permit is not None:
+            raise ContinuationBusy(
+                "a prior production delivery lacks an acknowledgment receipt"
+            )
+        self._pending_remote_permit = self.coordinator.before_delivery(self.token)
+
+    def acknowledge_delivery(self) -> None:
+        pending = self._pending_remote_permit
+        self.coordinator.acknowledge_delivery(self.token, pending)
+        self._pending_remote_permit = None
 
     def bind_approval(self, approval: ApprovalRecord) -> None:
         self.coordinator.bind_approval(self.token, approval)

openadapt_flow/runtime/durable/continuation.py

@@ -581,6 +581,7 @@ def __init__(
         self._durable_linear_snapshot: tuple[Any, ...] = ()
         self._durable_program_snapshot: tuple[ProgramCheckpoint, ...] = ()
         self._durable_pending_snapshot: Optional[Any] = None
+        self._active_delivery_acknowledgers: tuple[Any, ...] = ()
         self._durable_continuation_guard: Optional[Any] = None
         # API/tool actuator -- the TOP of the capability ladder (RFC section 4
         # `api` tier). When set, a step carrying an `api_binding` has its write
@@ -1183,15 +1184,36 @@ def run(
         ):
             from openadapt_flow.runtime.durable.authority import DurableAuthority
 
+            managed_durable_run = durable_run
+
             class _ManagedInitialDeliveryGuard:
+                def __init__(self_nonlocal) -> None:
+                    self_nonlocal.authority = DurableAuthority(
+                        run_dir, managed_durable_run.store
+                    )
+                    self_nonlocal.pending_remote_permit: Any = None
+
                 def before_delivery(self_nonlocal) -> None:
-                    assert durable_run is not None
-                    DurableAuthority(
-                        run_dir, durable_run.store
-                    ).before_initial_delivery(
-                        durable_run._manifest  # noqa: SLF001 - exact retained manifest
+                    if self_nonlocal.pending_remote_permit is not None:
+                        raise RuntimeError(
+                            "a prior production delivery lacks an acknowledgment receipt"
+                        )
+                    self_nonlocal.pending_remote_permit = (
+                        self_nonlocal.authority.before_initial_delivery(
+                            managed_durable_run._manifest  # noqa: SLF001 - exact retained manifest
+                        )
                     )
 
+                def acknowledge_delivery(self_nonlocal) -> None:
+                    pending = self_nonlocal.pending_remote_permit
+                    if pending is None:
+                        return
+                    self_nonlocal.authority.acknowledge_remote_delivery(
+                        managed_durable_run._manifest,  # noqa: SLF001
+                        pending,
+                    )
+                    self_nonlocal.pending_remote_permit = None
+
             self._durable_initial_delivery_guard = _ManagedInitialDeliveryGuard()
         (run_dir / "steps").mkdir(parents=True, exist_ok=True)
 
@@ -8790,8 +8812,8 @@ def _resolve_drag_end(
         )
         return resolution, region, error
 
-    @staticmethod
     def _deliver_backend_call(
+        self,
         result: StepResult,
         call: Callable[[], _DeliveryResultT],
     ) -> _DeliveryResultT:
@@ -8804,6 +8826,8 @@ def _deliver_backend_call(
         """
 
         attempted_before = result.delivery_attempted
+        acknowledgers = self._active_delivery_acknowledgers
+        self._active_delivery_acknowledgers = ()
         try:
             delivered = call()
         except (FreshActuationRequired, StructuralResolutionRefused):
@@ -8813,6 +8837,8 @@ def _deliver_backend_call(
             result.delivery_attempted = True
             raise
         result.delivery_attempted = True
+        for guard in acknowledgers:
+            guard.acknowledge_delivery()
         return delivered
 
     @staticmethod
@@ -9222,6 +9248,8 @@ def _delivery_authorization_refusal(
     ) -> Optional[str]:
         """Recheck exact authority at the last point before input delivery."""
 
+        self._active_delivery_acknowledgers = ()
+        acknowledgers: list[Any] = []
         refusal = self._managed_dispatch_refusal()
         if refusal is None:
             refusal = self._fresh_actuation_authorization_refusal(
@@ -9231,6 +9259,7 @@ def _delivery_authorization_refusal(
         if refusal is None and initial_guard is not None:
             try:
                 initial_guard.before_delivery()
+                acknowledgers.append(initial_guard)
             except Exception as exc:  # noqa: BLE001 - durable fencing boundary
                 refusal = (
                     f"managed initial delivery was preempted before delivery: {exc}"
@@ -9239,6 +9268,7 @@ def _delivery_authorization_refusal(
         if refusal is None and self._durable_continuation_guard is not None:
             try:
                 self._durable_continuation_guard.before_delivery()
+                acknowledgers.append(self._durable_continuation_guard)
             except Exception as exc:  # noqa: BLE001 - durable fencing boundary
                 refusal = f"durable continuation was preempted before delivery: {exc}"
                 result.failure_category = "continuation_preempted"
@@ -9252,6 +9282,8 @@ def _delivery_authorization_refusal(
                     if self.governed_authorization is not None
                     else "safety_halt"
                 )
+        else:
+            self._active_delivery_acknowledgers = tuple(acknowledgers)
         return refusal
 
     def _act(
@@ -10731,6 +10763,9 @@ def _handle_interstitials(
                 # well: every backend input edge must cross the same lease.
                 try:
                     self._durable_continuation_guard.before_delivery()
+                    self._active_delivery_acknowledgers = (
+                        self._durable_continuation_guard,
+                    )
                 except Exception as exc:  # noqa: BLE001 - fencing boundary
                     result.failure_category = "continuation_preempted"
                     result.safety_halt = True