feat(demo): publish exact-bound MockMed v3 presentation evidence

openadapt_flow/runtime/control_overlay.py

@@ -49,6 +49,9 @@
     "ROLLED_BACK",
 ]
 OverlayFrameSink: TypeAlias = Callable[[ControlOverlayFrameV2], None]
+OverlayObservationSink: TypeAlias = Callable[
+    [ControlOverlayFrameV2, bytes | None], None
+]
 MillisecondsClock: TypeAlias = Callable[[], int | float]
 ObservationKeyFactory: TypeAlias = Callable[[], bytes]
 
@@ -86,12 +89,14 @@ def __init__(
         observation_key_factory: ObservationKeyFactory = lambda: secrets.token_bytes(
             32
         ),
+        observation_sink: OverlayObservationSink | None = None,
     ) -> None:
         self._sink = sink
         self._mode = ControlOverlayMode(mode)
         self._unix_ms_clock = unix_ms_clock
         self._monotonic_ms_clock = monotonic_ms_clock
         self._observation_key_factory = observation_key_factory
+        self._observation_sink = observation_sink
         self._observation_hmac_key: bytes | None = None
         self._profile: ControlOverlayProfile | None = None
         self._next_sequence = 0
@@ -121,6 +126,7 @@ def emit_phase(
         current_step: int | None = None,
         total_steps: int | None = None,
         target_tracking: ControlOverlayTargetTrackingV2 | None = None,
+        observation_png: bytes | None = None,
     ) -> ControlOverlayFrameV2:
         """Emit an exact runtime phase; all presentation strings are canonical."""
 
@@ -143,10 +149,16 @@ def emit_phase(
             current_step=current_step,
             total_steps=total_steps,
             target_tracking=target_tracking,
+            observation_png=observation_png,
             terminal=False,
         )
 
-    def emit_terminal(self, outcome: ExecutionOutcome | str) -> ControlOverlayFrameV2:
+    def emit_terminal(
+        self,
+        outcome: ExecutionOutcome | str,
+        *,
+        observation_png: bytes | None = None,
+    ) -> ControlOverlayFrameV2:
         """Emit only an exact evidence-qualified ``RunReport.execution_outcome``.
 
         Generic success booleans and legacy strings such as ``SUCCESS`` are
@@ -164,6 +176,7 @@ def emit_terminal(self, outcome: ExecutionOutcome | str) -> ControlOverlayFrameV
             current_step=None,
             total_steps=None,
             target_tracking=None,
+            observation_png=observation_png,
             terminal=True,
         )
 
@@ -254,6 +267,7 @@ def _emit(
         current_step: int | None,
         total_steps: int | None,
         target_tracking: ControlOverlayTargetTrackingV2 | None,
+        observation_png: bytes | None,
         terminal: bool,
     ) -> ControlOverlayFrameV2:
         if self._profile is None:
@@ -292,6 +306,13 @@ def _emit(
         if terminal:
             self._terminal = True
         self._sink(frame)
+        if self._observation_sink is not None:
+            # Private presentation capture is deliberately out-of-band: the
+            # screenshot never enters the canonical event or an external
+            # transport.  Exporters/viewers can retain the exact already-held
+            # observation without taking a new browser screenshot between
+            # final revalidation and one-shot actuation.
+            self._observation_sink(frame, observation_png)
         return frame
 
 

openadapt_flow/runtime/replayer.py

@@ -522,6 +522,7 @@ def __init__(
         self.control_overlay = control_overlay
         self.control_overlay_error: Optional[str] = None
         self._control_overlay_disabled = False
+        self._control_overlay_last_observation_png: Optional[bytes] = None
 
     # -- public API ----------------------------------------------------------
 
@@ -911,6 +912,7 @@ def _begin_control_overlay(self, profile: Optional[str]) -> None:
 
         self.control_overlay_error = None
         self._control_overlay_disabled = False
+        self._control_overlay_last_observation_png = None
         if self.control_overlay is None:
             return
         try:
@@ -934,6 +936,7 @@ def _emit_control_overlay_phase(
         current_step: Optional[int] = None,
         total_steps: Optional[int] = None,
         target_tracking: Optional[Any] = None,
+        observation_png: Optional[bytes] = None,
     ) -> None:
         if self.control_overlay is None or self._control_overlay_disabled:
             return
@@ -943,7 +946,10 @@ def _emit_control_overlay_phase(
                 current_step=current_step,
                 total_steps=total_steps,
                 target_tracking=target_tracking,
+                observation_png=observation_png,
             )
+            if observation_png is not None:
+                self._control_overlay_last_observation_png = observation_png
         except Exception as exc:
             self.control_overlay_error = type(exc).__name__
             self._control_overlay_disabled = True
@@ -1012,7 +1018,10 @@ def _emit_control_overlay_terminal(self, outcome: Optional[str]) -> None:
         try:
             if outcome is None:
                 raise ValueError("run report has no exact execution outcome")
-            self.control_overlay.emit_terminal(outcome)
+            self.control_overlay.emit_terminal(
+                outcome,
+                observation_png=self._control_overlay_last_observation_png,
+            )
         except Exception as exc:
             self.control_overlay_error = type(exc).__name__
             self._control_overlay_disabled = True
@@ -2761,12 +2770,6 @@ def _run_step(
         overlay_current, overlay_total = self._control_overlay_progress(
             workflow, step_index, graph_ctx
         )
-        self._emit_control_overlay_phase(
-            "observing",
-            current_step=overlay_current,
-            total_steps=overlay_total,
-        )
-
         # API/tool tier -- the TOP of the capability ladder. When the step
         # carries an api_binding and an ApiActuator is configured, PERFORM the
         # write via the API (deterministic, $0, no GUI), CONFIRM it with the
@@ -2805,6 +2808,12 @@ def _run_step(
         )
         result.before_png = self._save_step_png(run_dir, step.id, "before", before_png)
         last_frame = before_png
+        self._emit_control_overlay_phase(
+            "observing",
+            current_step=overlay_current,
+            total_steps=overlay_total,
+            observation_png=before_png,
+        )
         # System-of-record pre-state, snapshotted just before the action when
         # the step declares effects (see the block guarding self._act below).
         effect_pre_state: Optional[EffectState] = None
@@ -3038,6 +3047,7 @@ def _run_step(
                     current_step=overlay_current,
                     total_steps=overlay_total,
                     target_tracking=overlay_target,
+                    observation_png=before_png,
                 )
                 try:
                     error = self._act(
@@ -3070,14 +3080,15 @@ def _run_step(
                     result.safety_halt = True
 
             if error is None:
-                self._emit_control_overlay_phase(
-                    "verifying",
-                    current_step=overlay_current,
-                    total_steps=overlay_total,
-                )
                 try:
                     after_png = self.vision.wait_settled(self.backend)
                     last_frame = after_png
+                    self._emit_control_overlay_phase(
+                        "verifying",
+                        current_step=overlay_current,
+                        total_steps=overlay_total,
+                        observation_png=after_png,
+                    )
                     postconditions_ok, last_frame, failed = self._check_postconditions(
                         step, after_png, bundle_dir, start_state, result
                     )