fix(viewer): make recorded layers deterministic

apps/control-station/src/components/RerunViewport.tsx

@@ -49,6 +49,7 @@ export interface RerunViewportProps {
   presentationGate?: RecordedAdmissionPhase;
   expectedTimelineStartSeconds?: number;
   expectedTimelineEndSeconds?: number;
+  initialPlaybackStartSeconds?: number;
   onStatusChange?: (status: RerunViewportStatus, message?: string) => void;
   onSelectionChange?: (selection: RerunSelection | null) => void;
   onPlaybackChange?: (state: RerunPlaybackState | null) => void;
@@ -295,12 +296,21 @@ export function createRecordedAutoplayGate(): {
     rangeNs: { min: number; max: number } | null,
     seekToStart: (startNs: number) => void,
     startPlaying: () => void,
+    preferredStartNs?: number,
   ) => boolean;
   attempted: () => boolean;
 } {
   let consumed = false;
   return {
-    attempt(viewerStarted, fullyBuffered, presentationReady, rangeNs, seekToStart, startPlaying) {
+    attempt(
+      viewerStarted,
+      fullyBuffered,
+      presentationReady,
+      rangeNs,
+      seekToStart,
+      startPlaying,
+      preferredStartNs,
+    ) {
       if (
         consumed ||
         !viewerStarted ||
@@ -309,8 +319,11 @@ export function createRecordedAutoplayGate(): {
         !isUsableRecordedPlaybackRange(rangeNs)
       ) return false;
       consumed = true;
+      const startNs = Number.isFinite(preferredStartNs)
+        ? Math.min(Math.max(preferredStartNs as number, rangeNs.min), rangeNs.max)
+        : rangeNs.min;
       return attemptRecordedAutoplay(
-        () => seekToStart(rangeNs.min),
+        () => seekToStart(startNs),
         startPlaying,
       );
     },
@@ -515,7 +528,8 @@ export async function fetchRecordedBlueprintRrd(
       custom_color: settings.customColor,
       active_view: activeView,
       view_reset_generation: viewResetGeneration,
-      unified_perception: perceptionLayers.enabled,
+      unified_perception:
+        perceptionLayers.detections2d || perceptionLayers.segmentation,
       show_detections_2d: perceptionLayers.detections2d,
       show_segmentation: perceptionLayers.segmentation,
       show_cuboids_3d: perceptionLayers.cuboids3d,
@@ -616,6 +630,7 @@ export function RerunViewport({
   presentationGate = "ready",
   expectedTimelineStartSeconds,
   expectedTimelineEndSeconds,
+  initialPlaybackStartSeconds,
   onStatusChange,
   onSelectionChange,
   onPlaybackChange,
@@ -982,6 +997,9 @@ export function RerunViewport({
                   () => {
                     viewer.set_playing(event.recording_id, true);
                   },
+                  initialPlaybackStartSeconds === undefined
+                    ? undefined
+                    : initialPlaybackStartSeconds * 1_000_000_000,
                 );
                 if (autoplayStarted) {
                   playing = true;
@@ -1153,6 +1171,7 @@ export function RerunViewport({
     expectedTimelineEndSeconds,
     expectedTimelineStartSeconds,
     followLive,
+    initialPlaybackStartSeconds,
     onPlaybackChange,
     onPlaybackControllerChange,
     onSelectionChange,

apps/control-station/src/workspaces/Workspaces.tsx

@@ -336,10 +336,13 @@ function SpatialWorkspace({
     (source) => source.capabilities.overlay && source.modality !== "point-cloud",
   );
   const recordedPerceptionSupported = recordedSource && perceptionAvailability !== "unavailable";
-  // Keep one stable original-video + world composition for every admitted
+  const recordedPerceptionEnabled =
-  // perception recording. Layer buttons then only change entity visibility;
+    showDetections2d || showSegmentation || showCuboids3d;
-  // they never reparent the 3D view or invalidate the operator's camera.
+  // The native recorded camera remains the authoritative original. Only 2D
-  const unifiedPerception = recordedPerceptionSupported;
+  // image-space overlays need Rerun's paired camera/world composition; 3D
+  // cuboids are added directly to the stable spatial view.
+  const unifiedPerception = recordedPerceptionSupported &&
+    (showDetections2d || showSegmentation);
   const livePerceptionAvailable = !recordedSource && streamActive;
   const detections2dActive = recordedSource
     ? showDetections2d
@@ -354,6 +357,16 @@ function SpatialWorkspace({
     observationLayout.visibleSourceIds.has(source.id) &&
     !source.id.startsWith("recorded.perception."),
   );
+  const initialRecordedPlaybackStartSeconds = recordedSource
+    ? mediaSources.reduce<number | undefined>((earliest, source) => {
+        if (
+          source.id.startsWith("recorded.perception.") ||
+          source.delivery?.kind !== "recorded-fmp4-manifest"
+        ) return earliest;
+        const start = source.delivery.timelineStartSeconds;
+        return earliest === undefined ? start : Math.min(earliest, start);
+      }, undefined)
+    : undefined;
   const presentedMediaSourceCount = unifiedPerception ? 0 : visibleMediaSources.length;
   const pointCloudFocused = Boolean(
     pointCloudSource && observationLayout.focusedSourceId === pointCloudSource.id,
@@ -574,10 +587,11 @@ function SpatialWorkspace({
               expectedTimelineEndSeconds={recordedSource
                 ? state?.observationTimeline?.range?.endSeconds
                 : undefined}
+              initialPlaybackStartSeconds={initialRecordedPlaybackStartSeconds}
               sceneSettings={sceneSettings}
               recordedViewResetGeneration={recordedViewResetGeneration}
               recordedPerceptionLayers={{
-                enabled: unifiedPerception,
+                enabled: recordedPerceptionSupported && recordedPerceptionEnabled,
                 detections2d: showDetections2d,
                 segmentation: showSegmentation,
                 cuboids3d: showCuboids3d,

apps/control-station/test/observationSources.test.mjs

@@ -309,6 +309,39 @@ test("recorded blueprint fetch is bounded, strict and sends only display setting
     show_cuboids_3d: true,
   });
 
+  await fetchRecordedBlueprintRrd(
+    "http://127.0.0.1:5174/api/v1/observation-sessions/session-1/blueprint.rrd",
+    {
+      accumulationSeconds: 0,
+      showGrid: true,
+      showPoints: true,
+      showTrajectory: true,
+      pointSize: 2.5,
+      palette: "turbo",
+      customColor: "#ffffff",
+    },
+    { applicationId: "nodedc_mission_core_recorded", recordingId: "recording-001" },
+    {
+      origin: "http://127.0.0.1:5174",
+      blueprintSessionId: "b".repeat(32),
+      perceptionLayers: {
+        enabled: true,
+        detections2d: false,
+        segmentation: false,
+        cuboids3d: true,
+      },
+      fetcher: async (input, init) => {
+        calls.push({ input: String(input), init, body: JSON.parse(String(init.body)) });
+        return new Response(payload, {
+          status: 200,
+          headers: { "Content-Type": "application/vnd.rerun.rrd" },
+        });
+      },
+    },
+  );
+  assert.equal(calls[1].body.unified_perception, false);
+  assert.equal(calls[1].body.show_cuboids_3d, true);
+
   await assert.rejects(
     fetchRecordedBlueprintRrd(
       "https://outside.invalid/api/v1/observation-sessions/session-1/blueprint.rrd",

apps/control-station/test/rerunViewportProgressivePlayback.test.mjs

@@ -128,6 +128,51 @@ test("recorded autoplay waits for the full range and then runs exactly once", ()
   assert.equal(gate.attempted(), true);
 });
 
+test("recorded autoplay starts at the first presentable camera frame without shrinking the range", () => {
+  const gate = createRecordedAutoplayGate();
+  const seeks = [];
+  const range = { min: 0, max: 535_717_620_042 };
+
+  assert.equal(gate.attempt(
+    true,
+    true,
+    true,
+    range,
+    (value) => seeks.push(value),
+    () => {},
+    35_421_857_292,
+  ), true);
+  assert.deepEqual(seeks, [35_421_857_292]);
+  assert.deepEqual(range, { min: 0, max: 535_717_620_042 });
+});
+
+test("recorded autoplay clamps an invalid presentation start to the admitted archive", () => {
+  const before = createRecordedAutoplayGate();
+  const after = createRecordedAutoplayGate();
+  const seeks = [];
+  const range = { min: 5, max: 10 };
+
+  assert.equal(before.attempt(
+    true,
+    true,
+    true,
+    range,
+    (value) => seeks.push(value),
+    () => {},
+    -50,
+  ), true);
+  assert.equal(after.attempt(
+    true,
+    true,
+    true,
+    range,
+    (value) => seeks.push(value),
+    () => {},
+    50,
+  ), true);
+  assert.deepEqual(seeks, [5, 10]);
+});
+
 test("a failed vendor autoplay attempt is consumed instead of rewinding later", () => {
   const gate = createRecordedAutoplayGate();
   let seeks = 0;

src/k1link/viewer/recorded.py

@@ -55,48 +55,58 @@ class _RecordedBlueprintStream:
     def __init__(
         self,
         application_id: str,
+        recording_id: str,
         *,
         view_reset_generation: Literal[0, 1],
     ) -> None:
         self.view_reset_generation = view_reset_generation
         self._lock = Lock()
-        self._pending: list[bytes] = []
+        self._sequence = 0
-        self._initial_payload: bytes | None = None
+        self._closed = False
-        self._latest_payload = b""
         native = bindings.new_blueprint(
             application_id=application_id,
             make_default=False,
             make_thread_default=False,
             default_enabled=True,
         )
-        self._recording = rr.RecordingStream._from_native(native)
+        self._blueprint_recording = rr.RecordingStream._from_native(native)
-        bindings.set_callback_sink_blueprint(
+        self._blueprint_memory = self._blueprint_recording.memory_recording()
-            lambda chunk: self._pending.append(bytes(chunk)),
+        self._transport_recording = rr.RecordingStream(
-            True,
+            application_id,
-            False,
+            recording_id=recording_id,
-            native,
+            send_properties=False,
         )
-        self._recording.set_time("blueprint", sequence=0)
+        self._transport = rr.binary_stream(self._transport_recording)
 
     def render(self, blueprint: rrb.Blueprint) -> bytes:
         with self._lock:
-            if self._initial_payload is not None:
+            if self._closed:
-                self._pending.clear()
+                raise RecordedBlueprintError("stable blueprint stream is closed")
-            blueprint._log_to_stream(self._recording)
+            self._blueprint_recording.set_time(
-            self._recording.flush(timeout_sec=5.0)
+                "blueprint",
-            delta = b"".join(self._pending)
+                sequence=self._sequence,
-            if not delta:
+            )
+            self._sequence += 1
+            blueprint._log_to_stream(self._blueprint_recording)
+            self._blueprint_recording.flush(timeout_sec=5.0)
+            bindings.send_blueprint(
+                self._blueprint_memory.storage,
+                True,
+                False,
+                self._transport_recording.to_native(),
+            )
+            payload = self._transport.read(flush=True, flush_timeout_sec=5.0)
+            if not payload:
                 raise RecordedBlueprintError("stable blueprint stream produced no data")
-            if self._initial_payload is None:
+            return payload
-                self._initial_payload = delta
-                self._latest_payload = b""
-            else:
-                self._latest_payload = delta
-            return self._initial_payload + self._latest_payload
 
     def close(self) -> None:
-        with suppress(Exception):
+        with self._lock:
-            self._recording.disconnect()
+            self._closed = True
+            with suppress(Exception):
+                self._blueprint_recording.disconnect()
+            with suppress(Exception):
+                self._transport_recording.disconnect()
 
 
 _MAX_RECORDED_BLUEPRINT_STREAMS = 32
@@ -123,6 +133,7 @@ def _stable_recorded_blueprint_stream(
         if stream is None:
             stream = _RecordedBlueprintStream(
                 application_id,
+                recording_id,
                 view_reset_generation=view_reset_generation,
             )
             _recorded_blueprint_streams[key] = stream
@@ -197,7 +208,13 @@ def recorded_blueprint(
             # object-history trail when the operator widens the cloud window.
             "/world/points": point_overrides,
             "/world/trajectory": trajectory_overrides,
-            "/world/perception": rrb.EntityBehavior(visible=False),
+            "/world/perception": rrb.EntityBehavior(visible=show_cuboids_3d),
+            "/world/perception/lidar": rrb.EntityBehavior(visible=False),
+            "/world/perception/support": rrb.EntityBehavior(visible=False),
+            "/world/perception/semantic_points": rrb.EntityBehavior(visible=False),
+            "/world/perception/boxes3d": rrb.EntityBehavior(
+                visible=show_cuboids_3d,
+            ),
         },
         # Log an explicit empty range set so a previous accumulated blueprint
         # for this stable view id is cleared instead of surviving in Rerun.

tests/test_rrd_export.py

@@ -12,6 +12,7 @@ from pathlib import Path
 import pytest
 
 import k1link.device_plugins.xgrids_k1.rrd_export as export_module
+import k1link.viewer.recorded as viewer_recorded_module
 from k1link.device_plugins.xgrids_k1.mqtt.capture import (
     CAPTURE_CLOCK_FILENAME,
     FRAME_HEADER,
@@ -299,6 +300,39 @@ def test_zero_accumulation_uses_latest_frame_instead_of_empty_time_range() -> No
     assert point_components == {"Points3D:radii"}
 
 
+def test_cuboids_are_overlaid_on_the_stable_spatial_view_without_unifying_video() -> None:
+    baseline = viewer_recorded_blueprint(
+        RerunSceneSettings(accumulation_seconds=12.0),
+        include_initial_playback_state=False,
+        unified_perception=False,
+        show_cuboids_3d=False,
+    )
+    blueprint = viewer_recorded_blueprint(
+        RerunSceneSettings(accumulation_seconds=12.0),
+        include_initial_playback_state=False,
+        unified_perception=False,
+        show_cuboids_3d=True,
+    )
+    spatial_view = blueprint.root_container.contents[0]
+
+    assert spatial_view.id == baseline.root_container.contents[0].id
+    assert spatial_view.visualizer_overrides[
+        "/world/perception"
+    ].visible.as_arrow_array().to_pylist() == [True]
+    assert spatial_view.visualizer_overrides[
+        "/world/perception/lidar"
+    ].visible.as_arrow_array().to_pylist() == [False]
+    assert spatial_view.visualizer_overrides[
+        "/world/perception/support"
+    ].visible.as_arrow_array().to_pylist() == [False]
+    assert spatial_view.visualizer_overrides[
+        "/world/perception/semantic_points"
+    ].visible.as_arrow_array().to_pylist() == [False]
+    assert spatial_view.visualizer_overrides[
+        "/world/perception/boxes3d"
+    ].visible.as_arrow_array().to_pylist() == [True]
+
+
 def test_dynamic_blueprint_reuses_scene_ids_without_playback_mutation() -> None:
     first = _recorded_blueprint(
         RerunSceneSettings(show_points=False, show_trajectory=True),
@@ -416,7 +450,26 @@ def test_viewer_blueprint_reset_is_bounded_and_recreates_render_views() -> None:
     assert perception_behavior.visible.as_arrow_array().to_pylist() == [False]
 
 
-def test_recorded_blueprint_layer_updates_preserve_store_until_explicit_reset() -> None:
+def test_recorded_blueprint_layer_updates_preserve_store_until_explicit_reset(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    activations: list[tuple[object, bool, bool]] = []
+    send_blueprint = viewer_recorded_module.bindings.send_blueprint
+
+    def capture_activation(
+        storage: object,
+        make_active: bool,
+        make_default: bool,
+        recording: object,
+    ) -> None:
+        activations.append((storage, make_active, make_default))
+        send_blueprint(storage, make_active, make_default, recording)
+
+    monkeypatch.setattr(
+        viewer_recorded_module.bindings,
+        "send_blueprint",
+        capture_activation,
+    )
     session_id = "b" * 32
     initial = viewer_recorded_blueprint_rrd(
         RerunSceneSettings(accumulation_seconds=12.0),
@@ -452,6 +505,13 @@ def test_recorded_blueprint_layer_updates_preserve_store_until_explicit_reset()
     assert layer_store_ids == initial_store_ids
     assert len(reset_store_ids) == 1
     assert reset_store_ids.isdisjoint(initial_store_ids)
+    assert [activation[1:] for activation in activations] == [
+        (True, False),
+        (True, False),
+        (True, False),
+    ]
+    assert activations[0][0] is activations[1][0]
+    assert activations[2][0] is not activations[0][0]
     assert len(initial) < 350_000
     assert len(layers_enabled) < 350_000
     assert len(reset) < 350_000