feat(lab): stabilize autonomous TGS playback

2026-08-27 20:34:20 +03:00
parent c8593207d3
commit 95a1ef5057
26 changed files with 3326 additions and 253 deletions
@@ -452,6 +452,32 @@ def build_threat_replay(


 def read_threat_replay_result(root: Path) -> ThreatReplayResult:
+    """Read and deeply verify a sealed threat replay result.
+
+    This is the acceptance/build seam.  It deliberately walks every ledger row
+    before returning and therefore must not sit on the synchronous LAB open
+    path for a hundreds-of-megabytes recorded result.
+    """
+
+    return _read_threat_replay_result(root, validate_ledgers=True)
+
+
+def read_threat_replay_result_metadata(root: Path) -> ThreatReplayResult:
+    """Read verified result metadata without eagerly parsing the full ledger.
+
+    Artifact bytes are still digest checked against the sealed manifest.  The
+    bounded timeline reader validates every requested row before projection;
+    only the redundant all-4489-row JSON accounting pass is deferred.
+    """
+
+    return _read_threat_replay_result(root, validate_ledgers=False)
+
+
+def _read_threat_replay_result(
+    root: Path,
+    *,
+    validate_ledgers: bool,
+) -> ThreatReplayResult:
    resolved = root.resolve(strict=True)
    if resolved.is_symlink() or not resolved.name.startswith(THREAT_REPLAY_RESULT_PREFIX):
        raise ThreatReplayError("threat replay result root is invalid")
@@ -498,7 +524,8 @@ def read_threat_replay_result(root: Path) -> ThreatReplayResult:
    for role, (name, identity_key) in expected.items():
        path = _validated_artifact(resolved, by_role[role], name)
        paths[role] = path
-        if identity_key is not None and _file_sha256(path) != identity.get(identity_key):
+        artifact = _object(by_role[role], "threat artifact")
+        if identity_key is not None and artifact.get("sha256") != identity.get(identity_key):
            raise ThreatReplayError("threat artifact identity changed")
    report = _read_json(paths["threat-replay-report"])
    metrics = _object(identity.get("metrics"), "threat metrics")
@@ -522,12 +549,13 @@ def read_threat_replay_result(root: Path) -> ThreatReplayResult:
        or requirements != expected_requirements
    ):
        raise ThreatReplayError("threat replay report or acceptance changed")
-    _validate_ledgers(
-        paths["threat-replay-frames"],
-        paths["threat-visual-frames"],
-        metrics,
-        is_v2=is_v2,
-    )
+    if validate_ledgers:
+        _validate_ledgers(
+            paths["threat-replay-frames"],
+            paths["threat-visual-frames"],
+            metrics,
+            is_v2=is_v2,
+        )
    return ThreatReplayResult(
        result_id=resolved.name,
        result_root=resolved,
@@ -13,6 +13,8 @@ from pathlib import Path
 from threading import RLock
 from typing import Final

+import numpy as np
+
 from .geometry import RecordedGeometryStore
 from .recorded_source import RECORDED_REPRESENTATION_ID
 from .spatial_evidence import (
@@ -78,24 +80,56 @@ class RecordedThreatTimeline:
        }
        if any(identity.get(key) != value for key, value in expected_identity.items()):
            raise RecordedThreatTimelineError("recorded timeline escaped the threat profile")
-        self.store = RecordedGeometryStore.from_repository(self.repository_root)
-        if (
-            self.store.profile.source_pack_id != self.profile.source_pack_id
-            or self.store.profile.source_pack_sha256 != self.profile.source_pack_sha256
-            or self.store.profile.frame_count != _EXPECTED_FRAME_COUNT
-        ):
-            raise RecordedThreatTimelineError("recorded timeline geometry identity changed")
-        if self.store.maximum_current_point_count > RECORDED_SPATIAL_POINT_LIMIT:
-            raise RecordedThreatTimelineError(
-                "recorded timeline exact point delivery exceeds its declared bound"
-            )
-        self.body_frames = RecordedReplayBodyFrameResolver(
-            self.store,
-            profile=self.profile.body_frame,
+        self._maximum_source_points_per_frame = _read_source_point_bound(
+            self.repository_root,
+            source_pack_id=self.profile.source_pack_id,
+            source_pack_sha256=self.profile.source_pack_sha256,
        )
+        self._store: RecordedGeometryStore | None = None
+        self._body_frames: RecordedReplayBodyFrameResolver | None = None
        self.index = _index_frame_ledger(self.frames_path)
        self._lock = RLock()

+    @property
+    def store(self) -> RecordedGeometryStore:
+        """Load and deeply verify the large geometry archives on first use.
+
+        Timeline metadata, camera playback and the independent TGS artifact can
+        become visible without waiting for the source cloud archive.  Any
+        endpoint that actually delivers source points still crosses the full
+        digest and shape validation in ``RecordedGeometryStore``.
+        """
+
+        with self._lock:
+            if self._store is None:
+                store = RecordedGeometryStore.from_repository(self.repository_root)
+                if (
+                    store.profile.source_pack_id != self.profile.source_pack_id
+                    or store.profile.source_pack_sha256 != self.profile.source_pack_sha256
+                    or store.profile.frame_count != _EXPECTED_FRAME_COUNT
+                    or store.maximum_current_point_count
+                    != self._maximum_source_points_per_frame
+                ):
+                    raise RecordedThreatTimelineError(
+                        "recorded timeline geometry identity changed"
+                    )
+                if store.maximum_current_point_count > RECORDED_SPATIAL_POINT_LIMIT:
+                    raise RecordedThreatTimelineError(
+                        "recorded timeline exact point delivery exceeds its declared bound"
+                    )
+                self._store = store
+            return self._store
+
+    @property
+    def body_frames(self) -> RecordedReplayBodyFrameResolver:
+        with self._lock:
+            if self._body_frames is None:
+                self._body_frames = RecordedReplayBodyFrameResolver(
+                    self.store,
+                    profile=self.profile.body_frame,
+                )
+            return self._body_frames
+
    def metadata(self) -> dict[str, object]:
        times = self.index.source_times_ns
        intervals = [(current - previous) / 1_000_000_000 for previous, current in pairwise(times)]
@@ -120,7 +154,7 @@ class RecordedThreatTimeline:
            "max_chunk_frames": RECORDED_SPATIAL_MAX_CHUNK_FRAMES,
            "point_sample_limit": RECORDED_SPATIAL_POINT_LIMIT,
            "point_delivery": "exact-current-increment",
-            "maximum_source_points_per_frame": self.store.maximum_current_point_count,
+            "maximum_source_points_per_frame": self._maximum_source_points_per_frame,
            "local_surface_visualization": {
                "derivation": "bounded-registered-increment-accumulation",
                "window_seconds": RECORDED_LOCAL_SURFACE_WINDOW_SECONDS,
@@ -258,6 +292,82 @@ class RecordedThreatTimeline:
        }


+def _read_source_point_bound(
+    repository_root: Path,
+    *,
+    source_pack_id: str,
+    source_pack_sha256: str,
+) -> int:
+    """Read the small offsets member without inflating the 70 MiB source pack.
+
+    This is metadata only.  The exact archive digest, every array shape and the
+    local-surface binding are still verified lazily by ``RecordedGeometryStore``
+    before any source point is delivered.
+    """
+
+    base = (
+        repository_root / ".runtime/compute-experiments/e10/lidar-packs"
+    ).resolve(strict=True)
+    pack_root = (base / source_pack_id).resolve(strict=True)
+    manifest_path = (pack_root / "manifest.json").resolve(strict=True)
+    pack_path = (pack_root / "lidar-pack.npz").resolve(strict=True)
+    if (
+        pack_root.parent != base
+        or pack_root.is_symlink()
+        or manifest_path.parent != pack_root
+        or manifest_path.is_symlink()
+        or pack_path.parent != pack_root
+        or pack_path.is_symlink()
+        or not pack_path.is_file()
+    ):
+        raise RecordedThreatTimelineError("recorded timeline source pack path changed")
+    try:
+        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+        if not isinstance(manifest, dict):
+            raise RecordedThreatTimelineError(
+                "recorded timeline source pack manifest is invalid"
+            )
+        identity = manifest.get("identity")
+        artifact = manifest.get("artifact")
+        if not isinstance(identity, dict) or not isinstance(artifact, dict):
+            raise RecordedThreatTimelineError(
+                "recorded timeline source pack manifest is invalid"
+            )
+        point_count = identity.get("point_count")
+        if (
+            manifest.get("pack_id") != source_pack_id
+            or identity.get("frame_count") != _EXPECTED_FRAME_COUNT
+            or not isinstance(point_count, int)
+            or isinstance(point_count, bool)
+            or point_count < 0
+            or artifact.get("path") != "lidar-pack.npz"
+            or artifact.get("sha256") != source_pack_sha256
+            or artifact.get("byte_length") != pack_path.stat().st_size
+        ):
+            raise RecordedThreatTimelineError(
+                "recorded timeline source pack identity changed"
+            )
+        with np.load(pack_path, allow_pickle=False) as archive:
+            offsets = np.asarray(archive["cloud_offsets"], dtype=np.int64)
+    except (KeyError, OSError, ValueError, json.JSONDecodeError) as error:
+        raise RecordedThreatTimelineError(
+            "recorded timeline source point offsets are unavailable"
+        ) from error
+    if (
+        offsets.shape != (_EXPECTED_FRAME_COUNT + 1,)
+        or int(offsets[0]) != 0
+        or int(offsets[-1]) != point_count
+        or np.any(np.diff(offsets) < 0)
+    ):
+        raise RecordedThreatTimelineError("recorded timeline source point offsets changed")
+    maximum = int(np.diff(offsets).max(initial=0))
+    if maximum > RECORDED_SPATIAL_POINT_LIMIT:
+        raise RecordedThreatTimelineError(
+            "recorded timeline exact point delivery exceeds its declared bound"
+        )
+    return maximum
+
+
 def _index_frame_ledger(path: Path) -> RecordedThreatTimelineIndex:
    offsets: list[int] = []
    source_times: list[int] = []