fix(observatory): preserve camera across follow transitions

src/k1link/viewer/recorded.py

@@ -53,7 +53,7 @@ class _RecordedBlueprintStream:
     Upstream 0.36.3 activates a clone, not this source store. Explicit refresh
     makes layer changes visible. Ordinary layer updates deliberately omit eye
     components so the active clone retains the operator's camera. Only an
-    explicit 3D/plan/follow transition may write a new spatial eye preset.
+    explicit 3D/plan/reset transition may write a new spatial eye preset.
     """
 
     def __init__(
@@ -98,6 +98,9 @@ class _RecordedBlueprintStream:
                 raise RecordedBlueprintError("stable blueprint stream is closed")
             eye_contract = (follow_trajectory, plan_view)
             update_eye_controls = self._eye_contract != eye_contract
+            use_spatial_preset = (
+                self._eye_contract is not None and self._eye_contract[1] != plan_view
+            )
             # Initial admission/reset uses native framing. Explicit presets
             # apply to mode transitions, after the viewer has a source cursor.
             # Keep one view identity for the lifetime of this browser owner.
@@ -108,7 +111,7 @@ class _RecordedBlueprintStream:
             view_instance_token = self.blueprint_session_id
             blueprint = blueprint_factory(
                 update_eye_controls,
-                self._eye_contract is not None,
+                use_spatial_preset,
                 view_instance_token,
             )
             # Appending rows alone does not refresh upstream's active clone.

src/k1link/viewer/recorded_camera_bounds.py

@@ -50,6 +50,7 @@ def _recorded_spatial_bounds_index(
     rows: dict[str, list[tuple[int, np.ndarray, np.ndarray]]] = {
         "/world/points": [],
         "/world/trajectory": [],
+        "/world/sensor_pose": [],
     }
     for chunk in reader.stream(recording):
         entity_path = str(chunk.entity_path)
@@ -59,9 +60,11 @@ def _recorded_spatial_bounds_index(
         names = batch.column_names
         if SESSION_TIMELINE not in names:
             continue
-        component = (
-            "Points3D:positions" if entity_path == "/world/points" else "LineStrips3D:strips"
-        )
+        component = {
+            "/world/points": "Points3D:positions",
+            "/world/trajectory": "LineStrips3D:strips",
+            "/world/sensor_pose": "Transform3D:translation",
+        }[entity_path]
         if component not in names:
             continue
         times = np.asarray(
@@ -100,6 +103,34 @@ def _recorded_spatial_bounds_index(
     return result
 
 
+def recorded_tracking_position(
+    recording_path: Path,
+    *,
+    current_time_ns: int,
+) -> tuple[float, float, float] | None:
+    """Return the latest scanner pivot at or before the playback cursor."""
+
+    if current_time_ns < 0:
+        raise ValueError("invalid recorded camera query")
+    path = recording_path.expanduser().absolute()
+    if path.is_symlink() or not path.is_file():
+        raise ValueError("recorded camera source is unavailable")
+
+    stat = path.stat()
+    series = _recorded_spatial_bounds_index(
+        str(path),
+        stat.st_size,
+        stat.st_mtime_ns,
+    ).get("/world/sensor_pose")
+    if series is None:
+        return None
+    eligible_end = int(np.searchsorted(series.times_ns, current_time_ns, side="right"))
+    if eligible_end == 0:
+        return None
+    position = series.lower[eligible_end - 1]
+    return tuple(float(value) for value in position)
+
+
 def recorded_orbital_radius_limit(
     recording_path: Path,
     *,