refactor(lab): перевести RAV004 на канонический Rerun pipeline

src/k1link/laboratory/canonical_rerun_overlay.py

@@ -0,0 +1,626 @@
+"""Native Rerun sidecar for immutable recorded laboratory evidence.
+
+The sidecar deliberately contains only evidence missing from the canonical K1
+recording: camera video, semantic images and diagnostic 2D boxes. The base RRD
+continues to own poses, point clouds and trajectory. Both files use the same
+Rerun recording id and ``session_time`` timeline, so the upstream viewer is the
+only playback clock and the only spatial renderer.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import io
+import json
+import os
+import subprocess
+import tempfile
+import threading
+import zipfile
+from contextlib import suppress
+from dataclasses import dataclass
+from pathlib import Path, PurePosixPath
+from typing import Any, Final
+from uuid import uuid4
+
+import numpy as np
+import rerun as rr
+import rerun_bindings as bindings
+from PIL import Image
+
+APPLICATION_ID: Final = "nodedc_mission_core_recorded"
+SESSION_TIMELINE: Final = "session_time"
+RENDERER_VERSION: Final = "upstream-rerun-0.36.3-v2"
+MAX_SOURCE_BYTES: Final = 768 * 1024 * 1024
+MAX_OVERLAY_BYTES: Final = 512 * 1024 * 1024
+
+
+class CanonicalLabOverlayError(RuntimeError):
+    """A sealed LAB result could not be projected into a native Rerun sidecar."""
+
+
+@dataclass(frozen=True, slots=True)
+class CanonicalLabOverlayArtifact:
+    path: Path
+    byte_length: int
+    sha256: str
+
+
+_render_lock = threading.Lock()
+_memory_cache: dict[tuple[str, str, str], CanonicalLabOverlayArtifact] = {}
+
+
+def canonical_recording_id(path: Path) -> str:
+    """Read the single data-store identity from a sealed base RRD."""
+
+    try:
+        entries = bindings.RrdReaderInternal(str(path.resolve(strict=True))).store_entries()
+        matches = [
+            entry.recording_id
+            for entry in entries
+            if entry.kind == "recording" and entry.application_id == APPLICATION_ID
+        ]
+    except Exception as exc:
+        raise CanonicalLabOverlayError("canonical recording identity is unavailable") from exc
+    if len(matches) != 1 or not matches[0] or len(matches[0]) > 128:
+        raise CanonicalLabOverlayError("canonical recording identity is ambiguous")
+    return str(matches[0])
+
+
+def canonical_lab_overlay(
+    result_root: Path,
+    manifest: dict[str, Any],
+    *,
+    recording_id: str,
+    base_generation_sha256: str,
+    jobs_root: Path,
+    cache_root: Path,
+    ffmpeg_path: Path,
+) -> CanonicalLabOverlayArtifact:
+    """Return one cached, digest-bound RRD sidecar for a full-route LAB result."""
+
+    root = result_root.expanduser().resolve(strict=True)
+    jobs = jobs_root.expanduser().resolve(strict=True)
+    ffmpeg = ffmpeg_path.expanduser().resolve(strict=True)
+    cache = cache_root.expanduser().absolute()
+    result_id = str(manifest.get("result_id", ""))
+    result_path = root / "result.json"
+    result_sha256 = _sha256(result_path)
+    if (
+        root.name != result_id
+        or not result_id.startswith("lab-v1-vegetation-shadow-")
+        or len(result_id) != len("lab-v1-vegetation-shadow-") + 64
+        or not recording_id
+        or len(recording_id) > 128
+        or not _is_sha256(base_generation_sha256)
+        or not ffmpeg.is_file()
+        or not os.access(ffmpeg, os.X_OK)
+    ):
+        raise CanonicalLabOverlayError("canonical LAB overlay identity is invalid")
+    key = (result_id, recording_id, base_generation_sha256)
+    cached = _memory_cache.get(key)
+    if cached is not None and _artifact_is_regular(cached):
+        return cached
+
+    with _render_lock:
+        cached = _memory_cache.get(key)
+        if cached is not None and _artifact_is_regular(cached):
+            return cached
+        cache.mkdir(parents=True, exist_ok=True)
+        if cache.is_symlink() or not cache.is_dir():
+            raise CanonicalLabOverlayError("canonical LAB overlay cache is invalid")
+        identity = hashlib.sha256(
+            "\0".join(
+                (RENDERER_VERSION, result_sha256, recording_id, base_generation_sha256)
+            ).encode()
+        ).hexdigest()
+        output = cache / f"{identity}.rrd"
+        sidecar = cache / f"{identity}.json"
+        restored = _restore_cached(
+            output,
+            sidecar,
+            result_id=result_id,
+            result_sha256=result_sha256,
+            recording_id=recording_id,
+            base_generation_sha256=base_generation_sha256,
+        )
+        if restored is not None:
+            _memory_cache[key] = restored
+            return restored
+
+        temporary = cache / f".{identity}.{uuid4().hex}.rrd"
+        proxy: Path | None = None
+        source: Path | None = None
+        try:
+            route = _full_route(manifest)
+            source = _verified_camera_source(root, route, jobs)
+            proxy = _camera_proxy(source, int(route["frame_count"]), ffmpeg, cache)
+            _render_overlay(temporary, root, route, recording_id, proxy)
+            stat = temporary.stat()
+            if stat.st_size < 4 or stat.st_size > MAX_OVERLAY_BYTES:
+                raise CanonicalLabOverlayError("canonical LAB overlay size is invalid")
+            digest = _sha256(temporary)
+            os.chmod(temporary, 0o600)
+            os.replace(temporary, output)
+            _write_json_atomic(
+                sidecar,
+                {
+                    "schema_version": "missioncore.canonical-lab-rerun-overlay/v1",
+                    "renderer_version": RENDERER_VERSION,
+                    "result_id": result_id,
+                    "result_sha256": result_sha256,
+                    "recording_id": recording_id,
+                    "base_generation_sha256": base_generation_sha256,
+                    "byte_length": stat.st_size,
+                    "sha256": digest,
+                },
+            )
+            artifact = CanonicalLabOverlayArtifact(output, stat.st_size, digest)
+            _memory_cache[key] = artifact
+            return artifact
+        except CanonicalLabOverlayError:
+            raise
+        except Exception as exc:
+            raise CanonicalLabOverlayError("failed to render canonical LAB overlay") from exc
+        finally:
+            temporary.unlink(missing_ok=True)
+            if proxy is not None:
+                proxy.unlink(missing_ok=True)
+            if source is not None:
+                source.unlink(missing_ok=True)
+
+
+def _full_route(manifest: dict[str, Any]) -> dict[str, Any]:
+    route = manifest.get("route_full_review")
+    layers = route.get("layers") if isinstance(route, dict) else None
+    if (
+        not isinstance(route, dict)
+        or route.get("source_id") != "RAVNOVES004TREE"
+        or route.get("frame_count") != 6830
+        or route.get("width") != 800
+        or route.get("height") != 600
+        or not isinstance(layers, dict)
+        or set(layers) != {"city", "vegetation"}
+    ):
+        raise CanonicalLabOverlayError("full-route LAB contract is unavailable")
+    return route
+
+
+def _verified_camera_source(root: Path, route: dict[str, Any], jobs_root: Path) -> Path:
+    source_job_id = route.get("source_job_id")
+    proof = route.get("proofs", {}).get("job") if isinstance(route.get("proofs"), dict) else None
+    if (
+        not isinstance(source_job_id, str)
+        or not source_job_id.startswith("recorded-camera-")
+        or not isinstance(proof, dict)
+        or proof.get("path") != "proofs/job.json"
+        or not _is_sha256(proof.get("sha256"))
+    ):
+        raise CanonicalLabOverlayError("camera job binding is invalid")
+    proof_path = root / "proofs" / "job.json"
+    job_root = (jobs_root / source_job_id).resolve(strict=True)
+    job_path = job_root / "job.json"
+    if (
+        not job_root.is_relative_to(jobs_root)
+        or _sha256(proof_path) != proof["sha256"]
+        or _sha256(job_path) != proof["sha256"]
+    ):
+        raise CanonicalLabOverlayError("camera job proof changed")
+    job = json.loads(job_path.read_text(encoding="utf-8"))
+    source = job.get("input") if isinstance(job, dict) else None
+    files = source.get("files") if isinstance(source, dict) else None
+    frame_count = int(route["frame_count"])
+    if (
+        not isinstance(source, dict)
+        or source.get("session_id") != route.get("session_id")
+        or source.get("source_id") != "sensor.camera.right"
+        or source.get("segment_count") != frame_count
+        or source.get("byte_length", 0) > MAX_SOURCE_BYTES
+        or not isinstance(files, list)
+    ):
+        raise CanonicalLabOverlayError("camera job source contract changed")
+    epoch_prefix = PurePosixPath(
+        "input/camera/sensor.camera.right"
+    ) / f"epoch-{source.get('codec_epoch')}"
+    required = [epoch_prefix / "init.mp4"] + [
+        epoch_prefix / "segments" / f"{index}.m4s"
+        for index in range(1, frame_count + 1)
+    ]
+    descriptors = {
+        item.get("path"): item
+        for item in files
+        if isinstance(item, dict) and isinstance(item.get("path"), str)
+    }
+    temporary_descriptor, temporary_name = tempfile.mkstemp(
+        prefix=".canonical-lab-source-", suffix=".mp4", dir=root.parent
+    )
+    output = Path(temporary_name)
+    total = 0
+    try:
+        with os.fdopen(temporary_descriptor, "wb") as stream:
+            for relative in required:
+                descriptor = descriptors.get(str(relative))
+                path = (job_root / relative).resolve(strict=True)
+                if (
+                    descriptor is None
+                    or not path.is_relative_to(job_root)
+                    or path.is_symlink()
+                    or descriptor.get("byte_length") != path.stat().st_size
+                    or not _is_sha256(descriptor.get("sha256"))
+                ):
+                    raise CanonicalLabOverlayError("camera fragment contract changed")
+                digest = hashlib.sha256()
+                with path.open("rb") as source_stream:
+                    while chunk := source_stream.read(1024 * 1024):
+                        digest.update(chunk)
+                        stream.write(chunk)
+                        total += len(chunk)
+                if digest.hexdigest() != descriptor["sha256"]:
+                    raise CanonicalLabOverlayError("camera fragment digest changed")
+            stream.flush()
+            os.fsync(stream.fileno())
+        if total <= 0 or total > MAX_SOURCE_BYTES:
+            raise CanonicalLabOverlayError("camera source size is invalid")
+        return output
+    except BaseException:
+        output.unlink(missing_ok=True)
+        raise
+
+
+def _camera_proxy(source: Path, frame_count: int, ffmpeg: Path, root: Path) -> Path:
+    descriptor, name = tempfile.mkstemp(prefix=".canonical-lab-video-", suffix=".mp4", dir=root)
+    os.close(descriptor)
+    output = Path(name)
+    output.unlink(missing_ok=True)
+    completed = subprocess.run(
+        [
+            str(ffmpeg),
+            "-hide_banner",
+            "-loglevel",
+            "error",
+            "-i",
+            str(source),
+            "-frames:v",
+            str(frame_count),
+            "-an",
+            "-c:v",
+            "libx264",
+            "-preset",
+            "veryfast",
+            "-crf",
+            "28",
+            "-g",
+            "20",
+            "-keyint_min",
+            "20",
+            "-pix_fmt",
+            "yuv420p",
+            "-fps_mode",
+            "passthrough",
+            "-movflags",
+            "+faststart",
+            str(output),
+        ],
+        check=False,
+        capture_output=True,
+        timeout=900,
+    )
+    if completed.returncode != 0 or not output.is_file() or output.stat().st_size <= 0:
+        output.unlink(missing_ok=True)
+        raise CanonicalLabOverlayError(
+            f"canonical LAB video proxy failed: {completed.stderr[-1000:]!r}"
+        )
+    os.chmod(output, 0o600)
+    return output
+
+
+def _render_overlay(
+    output: Path,
+    root: Path,
+    route: dict[str, Any],
+    recording_id: str,
+    proxy: Path,
+) -> None:
+    frame_times = _frame_times(root, route)
+    layers = route["layers"]
+    archives: dict[str, zipfile.ZipFile] = {}
+    recording = rr.RecordingStream(APPLICATION_ID, recording_id=recording_id)
+    try:
+        recording.set_sinks(rr.FileSink(output, write_footer=True))
+        recording.log(
+            "/perception/camera/image",
+            rr.AssetVideo(path=proxy),
+            static=True,
+        )
+        video = rr.AssetVideo(path=proxy)
+        video_timestamps = video.read_frame_timestamps_nanos()
+        # The sealed fMP4 source has one fragment per LAB tick, but not every
+        # fragment contains a decodable video sample. Preserve the source PTS
+        # in the proxy and bind every LAB tick to the latest available sample.
+        # Re-numbering decoded samples at 10 Hz shortens RAV004 by ~43 seconds
+        # and is exactly the camera/semantics drift this projection prevents.
+        video_references = _video_reference_timestamps(video_timestamps, frame_times)
+        for layer_id in ("city", "vegetation"):
+            layer = layers[layer_id]
+            taxonomy = layer.get("taxonomy")
+            classes = taxonomy.get("classes") if isinstance(taxonomy, dict) else None
+            if not isinstance(classes, list):
+                raise CanonicalLabOverlayError("semantic taxonomy is invalid")
+            context = rr.AnnotationContext(
+                [
+                    rr.ClassDescription(
+                        info=rr.AnnotationInfo(
+                            id=int(item["class_id"]),
+                            label=str(item["label"]),
+                            color=[*map(int, item["color_rgb"]), 255],
+                        )
+                    )
+                    for item in classes
+                    if isinstance(item, dict)
+                ]
+            )
+            recording.log(
+                f"/perception/camera/segmentation/{layer_id}",
+                context,
+                static=True,
+            )
+            archive = layer.get("mask_archive")
+            relative = archive.get("path") if isinstance(archive, dict) else None
+            if not isinstance(relative, str):
+                raise CanonicalLabOverlayError("semantic archive is unavailable")
+            path = (root / PurePosixPath(relative)).resolve(strict=True)
+            if not path.is_relative_to(root) or path.is_symlink():
+                raise CanonicalLabOverlayError("semantic archive path is unsafe")
+            archives[layer_id] = zipfile.ZipFile(path)
+
+        for index, timestamp in enumerate(frame_times):
+            recording.set_time(SESSION_TIMELINE, duration=np.timedelta64(timestamp, "ns"))
+            recording.log(
+                "/perception/camera/image",
+                rr.VideoFrameReference(nanoseconds=int(video_references[index])),
+            )
+            masks = {
+                layer_id: _read_mask(archive, index)
+                for layer_id, archive in archives.items()
+            }
+            for layer_id, mask in masks.items():
+                recording.log(
+                    f"/perception/camera/segmentation/{layer_id}",
+                    rr.SegmentationImage(mask, opacity=0.72, draw_order=1.0),
+                )
+            boxes, labels = semantic_component_boxes(masks["city"], index)
+            if boxes:
+                recording.log(
+                    "/perception/camera/detections",
+                    rr.Boxes2D(
+                        array=boxes,
+                        array_format=rr.Box2DFormat.XYXY,
+                        labels=labels,
+                        show_labels=True,
+                        colors=[[255, 210, 55, 255]] * len(boxes),
+                    ),
+                )
+            else:
+                recording.log(
+                    "/perception/camera/detections",
+                    rr.Clear(recursive=False),
+                )
+        recording.flush(timeout_sec=300.0)
+    finally:
+        for archive in archives.values():
+            with suppress(Exception):
+                archive.close()
+        with suppress(Exception):
+            recording.disconnect()
+
+
+def _frame_times(root: Path, route: dict[str, Any]) -> np.ndarray:
+    descriptor = route.get("timeline")
+    relative = descriptor.get("path") if isinstance(descriptor, dict) else None
+    if not isinstance(relative, str):
+        raise CanonicalLabOverlayError("LAB timeline is unavailable")
+    path = (root / PurePosixPath(relative)).resolve(strict=True)
+    payload = path.read_bytes()
+    if (
+        not path.is_relative_to(root)
+        or path.is_symlink()
+        or descriptor.get("byte_length") != len(payload)
+        or descriptor.get("sha256") != hashlib.sha256(payload).hexdigest()
+    ):
+        raise CanonicalLabOverlayError("LAB timeline changed")
+    values = np.frombuffer(payload, dtype="<u8").astype(np.int64, copy=False)
+    if values.shape != (route["frame_count"],) or np.any(np.diff(values) <= 0):
+        raise CanonicalLabOverlayError("LAB timeline order changed")
+    return values
+
+
+def _video_reference_timestamps(
+    video_timestamps: np.ndarray,
+    frame_times: np.ndarray,
+) -> np.ndarray:
+    """Bind every LAB tick to the latest decodable source video sample."""
+
+    relative_frame_times = frame_times - frame_times[0]
+    if (
+        len(video_timestamps) < int(len(frame_times) * 0.9)
+        or np.any(np.diff(video_timestamps) < 0)
+        or abs(int(video_timestamps[-1]) - int(relative_frame_times[-1]))
+        > 2_000_000_000
+    ):
+        raise CanonicalLabOverlayError("video proxy timeline changed")
+    indices = np.searchsorted(
+        video_timestamps,
+        relative_frame_times,
+        side="right",
+    ) - 1
+    return video_timestamps[np.clip(indices, 0, len(video_timestamps) - 1)]
+
+
+def _read_mask(archive: zipfile.ZipFile, sequence: int) -> np.ndarray:
+    member = f"masks/frame-{sequence + 1:06d}.png"
+    with Image.open(io.BytesIO(archive.read(member))) as image:
+        mask = np.asarray(image.convert("L"), dtype=np.uint8).copy()
+    if mask.shape != (600, 800):
+        raise CanonicalLabOverlayError("semantic mask dimensions changed")
+    return mask
+
+
+def semantic_component_boxes(
+    mask: np.ndarray,
+    _sequence: int,
+) -> tuple[list[list[int]], list[str]]:
+    labels = {
+        1: "person",
+        2: "bicycle",
+        3: "motorcycle",
+        4: "car",
+        5: "heavy vehicle",
+        13: "static obstacle",
+        14: "animal",
+    }
+    candidates: list[tuple[float, list[int], str]] = []
+    for class_id, label in labels.items():
+        minimum_pixels = 80 if class_id == 13 else 24
+        for left, top, right, bottom, pixels in _mask_component_boxes(
+            mask, class_id, minimum_pixels=minimum_pixels
+        )[:12]:
+            score = min(0.99, 0.5 + pixels / 20_000)
+            candidates.append(
+                (score, [left, top, right, bottom], f"{label} · {score:.0%} · semantic-only")
+            )
+    candidates.sort(key=lambda row: (-row[0], row[1][1], row[1][0]))
+    selected = candidates[:32]
+    return [row[1] for row in selected], [row[2] for row in selected]
+
+
+def _mask_component_boxes(
+    mask: np.ndarray,
+    class_id: int,
+    *,
+    minimum_pixels: int,
+) -> list[tuple[int, int, int, int, int]]:
+    """Return deterministic 8-connected run-length components."""
+
+    if mask.ndim != 2 or minimum_pixels < 1:
+        return []
+    parents: list[int] = []
+    runs: list[tuple[int, int, int, int]] = []
+
+    def root(index: int) -> int:
+        while parents[index] != index:
+            parents[index] = parents[parents[index]]
+            index = parents[index]
+        return index
+
+    def union(left: int, right: int) -> None:
+        left_root = root(left)
+        right_root = root(right)
+        if left_root != right_root:
+            parents[right_root] = left_root
+
+    previous: list[int] = []
+    for row_index, row in enumerate(mask):
+        matches = np.flatnonzero(row == class_id)
+        if matches.size == 0:
+            previous = []
+            continue
+        groups = np.split(matches, np.flatnonzero(np.diff(matches) > 1) + 1)
+        current: list[int] = []
+        previous_cursor = 0
+        for group in groups:
+            start = int(group[0])
+            stop = int(group[-1]) + 1
+            run_index = len(runs)
+            runs.append((row_index, start, stop, stop - start))
+            parents.append(run_index)
+            current.append(run_index)
+            while previous_cursor < len(previous) and runs[previous[previous_cursor]][2] < start:
+                previous_cursor += 1
+            candidate_cursor = previous_cursor
+            while candidate_cursor < len(previous):
+                previous_index = previous[candidate_cursor]
+                _, previous_start, previous_stop, _ = runs[previous_index]
+                if previous_start > stop:
+                    break
+                union(run_index, previous_index)
+                candidate_cursor += 1
+        previous = current
+
+    components: dict[int, list[int]] = {}
+    for run_index, (row, start, stop, count) in enumerate(runs):
+        component = components.setdefault(root(run_index), [start, row, stop, row + 1, 0])
+        component[0] = min(component[0], start)
+        component[1] = min(component[1], row)
+        component[2] = max(component[2], stop)
+        component[3] = max(component[3], row + 1)
+        component[4] += count
+    result = [
+        (left, top, right, bottom, count)
+        for left, top, right, bottom, count in components.values()
+        if count >= minimum_pixels and right - left >= 2 and bottom - top >= 3
+    ]
+    result.sort(key=lambda box: (-box[4], box[1], box[0]))
+    return result
+
+
+def _restore_cached(
+    output: Path,
+    sidecar: Path,
+    **identity: str,
+) -> CanonicalLabOverlayArtifact | None:
+    try:
+        value = json.loads(sidecar.read_text(encoding="utf-8"))
+        stat = output.stat()
+        if (
+            output.is_symlink()
+            or sidecar.is_symlink()
+            or value.get("schema_version") != "missioncore.canonical-lab-rerun-overlay/v1"
+            or value.get("renderer_version") != RENDERER_VERSION
+            or any(value.get(key) != expected for key, expected in identity.items())
+            or value.get("byte_length") != stat.st_size
+            or not _is_sha256(value.get("sha256"))
+            or _sha256(output) != value["sha256"]
+        ):
+            return None
+        return CanonicalLabOverlayArtifact(output, stat.st_size, value["sha256"])
+    except (OSError, ValueError, json.JSONDecodeError):
+        return None
+
+
+def _artifact_is_regular(artifact: CanonicalLabOverlayArtifact) -> bool:
+    try:
+        return (
+            not artifact.path.is_symlink()
+            and artifact.path.stat().st_size == artifact.byte_length
+            and _sha256(artifact.path) == artifact.sha256
+        )
+    except OSError:
+        return False
+
+
+def _write_json_atomic(path: Path, value: dict[str, Any]) -> None:
+    temporary = path.with_name(f".{path.name}.{uuid4().hex}.tmp")
+    try:
+        with temporary.open("x", encoding="utf-8") as stream:
+            json.dump(value, stream, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+            stream.write("\n")
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.chmod(temporary, 0o600)
+        os.replace(temporary, path)
+    finally:
+        temporary.unlink(missing_ok=True)
+
+
+def _sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as stream:
+        while chunk := stream.read(1024 * 1024):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _is_sha256(value: object) -> bool:
+    return isinstance(value, str) and len(value) == 64 and all(
+        character in "0123456789abcdef" for character in value
+    )

src/k1link/viewer/recorded.py

@@ -63,7 +63,7 @@ class _RecordedBlueprintStream:
         self.view_reset_generation = view_reset_generation
         self._lock = Lock()
         self._sequence = 0
-        self._follow_trajectory: bool | None = None
+        self._eye_contract: tuple[bool, bool] | None = None
         self._closed = False
         native = bindings.new_blueprint(
             application_id=application_id,
@@ -85,14 +85,13 @@ class _RecordedBlueprintStream:
         blueprint_factory: Callable[[bool], rrb.Blueprint],
         *,
         follow_trajectory: bool,
+        plan_view: bool,
     ) -> bytes:
         with self._lock:
             if self._closed:
                 raise RecordedBlueprintError("stable blueprint stream is closed")
-            update_eye_controls = (
-                self._follow_trajectory is None
-                or self._follow_trajectory != follow_trajectory
-            )
+            eye_contract = (follow_trajectory, plan_view)
+            update_eye_controls = self._eye_contract != eye_contract
             blueprint = blueprint_factory(update_eye_controls)
             self._blueprint_recording.set_time(
                 "blueprint",
@@ -110,7 +109,7 @@ class _RecordedBlueprintStream:
             payload = self._transport.read(flush=True, flush_timeout_sec=5.0)
             if not payload:
                 raise RecordedBlueprintError("stable blueprint stream produced no data")
-            self._follow_trajectory = follow_trajectory
+            self._eye_contract = eye_contract
             return payload
 
     def close(self) -> None:
@@ -165,6 +164,8 @@ def recorded_blueprint(
     active_view: RecordedView = "spatial",
     view_reset_generation: Literal[0, 1] = 0,
     unified_perception: bool = False,
+    semantic_layer: Literal["city", "vegetation"] | None = None,
+    plan_view: bool = False,
     show_detections_2d: bool = False,
     show_segmentation: bool = False,
     show_cuboids_3d: bool = False,
@@ -202,6 +203,27 @@ def recorded_blueprint(
     if accumulated_time_ranges is not None:
         point_overrides.append(accumulated_time_ranges)
         trajectory_overrides.append(accumulated_time_ranges)
+    selected_semantic_path = (
+        f"/perception/camera/segmentation/{semantic_layer}"
+        if semantic_layer is not None
+        else "/perception/camera/segmentation"
+    )
+    spatial_eye_controls = (
+        rrb.EyeControls3D.from_fields(
+            kind=rrb.Eye3DKind.Orbital,
+            position=[0.0, 0.0, 30.0],
+            look_target=[0.0, 0.0, 0.0],
+            eye_up=[0.0, 1.0, 0.0],
+            tracking_entity="/world/sensor_pose" if follow_trajectory else "",
+        )
+        if plan_view and update_eye_controls
+        else rrb.EyeControls3D.from_fields(
+            kind=rrb.Eye3DKind.Orbital if follow_trajectory else None,
+            tracking_entity="/world/sensor_pose" if follow_trajectory else "",
+        )
+        if update_eye_controls
+        else None
+    )
     spatial_view = rrb.Spatial3DView(
         origin="/world",
         name="Мир · LiDAR и объекты" if unified_perception else "Пространственная сцена",
@@ -232,14 +254,7 @@ def recorded_blueprint(
         # Keeping the view id stable preserves the current orbit offset when
         # tracking is toggled. An explicit empty path clears tracking without
         # overwriting the position/look-target saved by user interaction.
-        eye_controls=(
-            rrb.EyeControls3D.from_fields(
-                kind=rrb.Eye3DKind.Orbital if follow_trajectory else None,
-                tracking_entity="/world/sensor_pose" if follow_trajectory else "",
-            )
-            if update_eye_controls
-            else None
-        ),
+        eye_controls=spatial_eye_controls,
     )
     spatial_view.id = (
         RECORDED_SPATIAL_RESET_VIEW_ID
@@ -258,6 +273,12 @@ def recorded_blueprint(
             "/perception/camera/segmentation": rrb.EntityBehavior(
                 visible=show_segmentation,
             ),
+            "/perception/camera/segmentation/city": rrb.EntityBehavior(
+                visible=show_segmentation and selected_semantic_path.endswith("/city"),
+            ),
+            "/perception/camera/segmentation/vegetation": rrb.EntityBehavior(
+                visible=show_segmentation and selected_semantic_path.endswith("/vegetation"),
+            ),
         },
     )
     camera_view.id = (
@@ -292,14 +313,7 @@ def recorded_blueprint(
             # native cloud from /world/points.
             "/world/perception/lidar": rrb.EntityBehavior(visible=False),
         },
-        eye_controls=(
-            rrb.EyeControls3D.from_fields(
-                kind=rrb.Eye3DKind.Orbital if follow_trajectory else None,
-                tracking_entity="/world/sensor_pose" if follow_trajectory else "",
-            )
-            if update_eye_controls
-            else None
-        ),
+        eye_controls=spatial_eye_controls,
     )
     perception_3d_view.id = (
         RECORDED_PERCEPTION_3D_RESET_VIEW_ID
@@ -400,6 +414,8 @@ def recorded_blueprint_rrd(
     active_view: RecordedView = "spatial",
     view_reset_generation: Literal[0, 1] = 0,
     unified_perception: bool = False,
+    semantic_layer: Literal["city", "vegetation"] | None = None,
+    plan_view: bool = False,
     show_detections_2d: bool = False,
     show_segmentation: bool = False,
     show_cuboids_3d: bool = False,
@@ -414,6 +430,8 @@ def recorded_blueprint_rrd(
             active_view=active_view,
             view_reset_generation=view_reset_generation,
             unified_perception=unified_perception,
+            semantic_layer=semantic_layer,
+            plan_view=plan_view,
             show_detections_2d=show_detections_2d,
             show_segmentation=show_segmentation,
             show_cuboids_3d=show_cuboids_3d,
@@ -432,6 +450,7 @@ def recorded_blueprint_rrd(
             ).render(
                 build_blueprint,
                 follow_trajectory=follow_trajectory,
+                plan_view=plan_view,
             )
         except RecordedBlueprintError:
             raise

src/k1link/viewer/rerun_bridge.py

@@ -149,7 +149,7 @@ class RerunBridge:
                 # of record. Raw MQTT evidence is persisted independently. A large
                 # late-client backlog can block the native SDK and freeze preview.
                 server_memory_limit=LIVE_GRPC_BUFFER_LIMIT,
-                # Rerun 0.34.1 can replay ActivateStore before StoreInfo when an
+                # Rerun transport can replay ActivateStore before StoreInfo when an
                 # evicted buffer is served newest-first, leaving late viewers on the
                 # welcome screen. Preserve protocol order within the bounded cache.
                 newest_first=False,

src/k1link/web/app.py

@@ -1047,6 +1047,11 @@ app.include_router(
             if session_recorded_camera_frame_service is not None
             else None
         ),
+        jobs_root=REPOSITORY_ROOT / ".runtime" / "compute-jobs",
+        rerun_overlay_cache_root=(
+            session_store.data_dir / "laboratory-rerun-overlays"
+        ),
+        ffmpeg_path=_ffmpeg,
     )
 )
 app.include_router(

src/k1link/web/session_api.py

@@ -124,6 +124,8 @@ class RecordedBlueprintRequest(StrictApiModel):
     active_view: Literal["spatial", "perception", "perception3d", "metrics"] = "spatial"
     view_reset_generation: Literal[0, 1] = 0
     unified_perception: StrictBool = False
+    semantic_layer: Literal["city", "vegetation"] | None = None
+    plan_view: StrictBool = False
     show_detections_2d: StrictBool = False
     show_segmentation: StrictBool = False
     show_cuboids_3d: StrictBool = False
@@ -935,6 +937,8 @@ def build_session_router(
                 active_view=request.active_view,
                 view_reset_generation=request.view_reset_generation,
                 unified_perception=request.unified_perception,
+                semantic_layer=request.semantic_layer,
+                plan_view=request.plan_view,
                 show_detections_2d=request.show_detections_2d,
                 show_segmentation=request.show_segmentation,
                 show_cuboids_3d=request.show_cuboids_3d,

src/k1link/web/vegetation_shadow_lab_api.py

@@ -12,13 +12,21 @@ import zipfile
 from collections.abc import Callable
 from functools import lru_cache
 from pathlib import Path, PurePosixPath
-from typing import Any, Final
+from typing import Any, Final, Literal
 
 import numpy as np
 from fastapi import APIRouter, HTTPException
+from fastapi.concurrency import run_in_threadpool
 from fastapi.responses import FileResponse, JSONResponse, Response
 from PIL import Image
+from pydantic import BaseModel, ConfigDict, Field
 
+from k1link.laboratory.canonical_rerun_overlay import (
+    CanonicalLabOverlayError,
+    _mask_component_boxes,
+    canonical_lab_overlay,
+    canonical_recording_id,
+)
 from k1link.laboratory.evidence_registry import LaboratoryEvidenceDefinition
 from k1link.laboratory.evidence_report import (
     LaboratoryEvidenceReportError,
@@ -34,6 +42,17 @@ from k1link.sessions.canonical_lab_spatial import (
 RootProvider = Callable[[], Path | None]
 CanonicalRecordingProvider = Callable[[str], tuple[Path, str] | None]
 CameraFrameProvider = Callable[[str, int], RecordedCameraFrame]
+
+
+class CanonicalLabRerunRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    application_id: Literal["nodedc_mission_core_recorded"]
+    recording_id: str = Field(
+        min_length=1,
+        max_length=128,
+        pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$",
+    )
 _MAX_DOCUMENT_BYTES: Final = 1024 * 1024
 _CANONICAL_ROUTE_CHUNK_FRAMES: Final = 24
 _DEFINITION: Final = LaboratoryEvidenceDefinition(
@@ -57,6 +76,9 @@ def build_vegetation_shadow_lab_router(
     root_provider: RootProvider = lambda: None,
     canonical_recording_provider: CanonicalRecordingProvider | None = None,
     camera_frame_provider: CameraFrameProvider | None = None,
+    jobs_root: Path | None = None,
+    rerun_overlay_cache_root: Path | None = None,
+    ffmpeg_path: Path | None = None,
 ) -> APIRouter:
     return _build_vegetation_lab_router(
         prefix="/api/v1/laboratory/vegetation-shadow",
@@ -64,6 +86,9 @@ def build_vegetation_shadow_lab_router(
         root_provider=root_provider,
         canonical_recording_provider=canonical_recording_provider,
         camera_frame_provider=camera_frame_provider,
+        jobs_root=jobs_root,
+        rerun_overlay_cache_root=rerun_overlay_cache_root,
+        ffmpeg_path=ffmpeg_path,
     )
 
 
@@ -84,6 +109,9 @@ def _build_vegetation_lab_router(
     root_provider: RootProvider,
     canonical_recording_provider: CanonicalRecordingProvider | None = None,
     camera_frame_provider: CameraFrameProvider | None = None,
+    jobs_root: Path | None = None,
+    rerun_overlay_cache_root: Path | None = None,
+    ffmpeg_path: Path | None = None,
 ) -> APIRouter:
     router = APIRouter(
         prefix=prefix,
@@ -270,6 +298,61 @@ def _build_vegetation_lab_router(
             },
         )
 
+    @router.post("/{result_id}/canonical-overlay.rrd")
+    async def get_canonical_rerun_overlay(
+        result_id: str,
+        request: CanonicalLabRerunRequest,
+    ) -> FileResponse:
+        """Project LAB-only evidence into the base recording's native clock."""
+
+        if (
+            canonical_recording_provider is None
+            or jobs_root is None
+            or rerun_overlay_cache_root is None
+            or ffmpeg_path is None
+        ):
+            raise HTTPException(status_code=503, detail="Canonical LAB Rerun overlay unavailable")
+        candidate = _resolve_candidate(root_provider, definition, result_id)
+        manifest = _read_verified(candidate, definition)
+        route, _ = _full_route_context(candidate, manifest)
+        recording = canonical_recording_provider(str(route["session_id"]))
+        if recording is None:
+            raise HTTPException(status_code=409, detail="Canonical spatial recording is not ready")
+        recording_path, generation_sha256 = recording
+        try:
+            expected_recording_id = await run_in_threadpool(
+                canonical_recording_id,
+                recording_path,
+            )
+            if request.recording_id != expected_recording_id:
+                raise HTTPException(status_code=412, detail="Canonical recording identity changed")
+            artifact = await run_in_threadpool(
+                canonical_lab_overlay,
+                candidate,
+                manifest,
+                recording_id=request.recording_id,
+                base_generation_sha256=generation_sha256,
+                jobs_root=jobs_root,
+                cache_root=rerun_overlay_cache_root,
+                ffmpeg_path=ffmpeg_path,
+            )
+        except HTTPException:
+            raise
+        except CanonicalLabOverlayError as exc:
+            raise HTTPException(
+                status_code=503,
+                detail="Canonical LAB Rerun overlay failed verification",
+            ) from exc
+        return FileResponse(
+            artifact.path,
+            media_type="application/vnd.rerun.rrd",
+            headers={
+                "Cache-Control": "private, max-age=31536000, immutable",
+                "ETag": f'"{artifact.sha256}"',
+                "X-Content-Type-Options": "nosniff",
+            },
+        )
+
     @router.get("/{result_id}/timeline")
     def get_canonical_route_timeline(result_id: str) -> dict[str, object]:
         candidate = _resolve_candidate(root_provider, definition, result_id)
@@ -811,80 +894,6 @@ def _semantic_component_proposals_cached(
     return tuple(proposals[:32])
 
 
-def _mask_component_boxes(
-    mask: np.ndarray,
-    class_id: int,
-    *,
-    minimum_pixels: int,
-) -> list[tuple[int, int, int, int, int]]:
-    """Return 8-connected run-length components without an OpenCV dependency."""
-
-    if mask.ndim != 2 or minimum_pixels < 1:
-        return []
-    parents: list[int] = []
-    runs: list[tuple[int, int, int, int]] = []
-
-    def root(index: int) -> int:
-        while parents[index] != index:
-            parents[index] = parents[parents[index]]
-            index = parents[index]
-        return index
-
-    def union(left: int, right: int) -> None:
-        left_root = root(left)
-        right_root = root(right)
-        if left_root != right_root:
-            parents[right_root] = left_root
-
-    previous: list[int] = []
-    for row_index, row in enumerate(mask):
-        matches = np.flatnonzero(row == class_id)
-        if matches.size == 0:
-            previous = []
-            continue
-        split_at = np.flatnonzero(np.diff(matches) > 1) + 1
-        groups = np.split(matches, split_at)
-        current: list[int] = []
-        previous_cursor = 0
-        for group in groups:
-            start = int(group[0])
-            stop = int(group[-1]) + 1
-            run_index = len(runs)
-            runs.append((row_index, start, stop, stop - start))
-            parents.append(run_index)
-            current.append(run_index)
-            while (
-                previous_cursor < len(previous)
-                and runs[previous[previous_cursor]][2] < start
-            ):
-                previous_cursor += 1
-            candidate_cursor = previous_cursor
-            while candidate_cursor < len(previous):
-                previous_index = previous[candidate_cursor]
-                _, previous_start, previous_stop, _ = runs[previous_index]
-                if previous_start > stop:
-                    break
-                union(run_index, previous_index)
-                candidate_cursor += 1
-        previous = current
-
-    components: dict[int, list[int]] = {}
-    for run_index, (row, start, stop, count) in enumerate(runs):
-        component = components.setdefault(root(run_index), [start, row, stop, row + 1, 0])
-        component[0] = min(component[0], start)
-        component[1] = min(component[1], row)
-        component[2] = max(component[2], stop)
-        component[3] = max(component[3], row + 1)
-        component[4] += count
-    result = [
-        (left, top, right, bottom, count)
-        for left, top, right, bottom, count in components.values()
-        if count >= minimum_pixels and right - left >= 2 and bottom - top >= 3
-    ]
-    result.sort(key=lambda box: (-box[4], box[1], box[0]))
-    return result
-
-
 def _route_tgs_anchor_payload(path: Path, source_sequence: int) -> dict[str, object]:
     before = path.stat()
     with np.load(path, allow_pickle=False) as archive: