fix(replay): serve admitted AI overlays without revalidation

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

@@ -10,6 +10,10 @@ import {
 } from "../core/observation/liveReceiverWatchdog";
 import { postLiveViewerDiagnostic } from "../core/observation/liveViewerDiagnostics";
 import type { LiveViewerFailureStage } from "../core/observation/liveViewerDiagnostics";
+import {
+  fetchPerceptionPreparationStatus,
+  perceptionPreparationMessage,
+} from "../core/observation/perceptionPreparation";
 import type { RecordedAdmissionPhase } from "../core/observation/recordedSessionAdmission";
 
 export type RerunViewportStatus = "idle" | "loading" | "ready" | "error";
@@ -1575,6 +1579,9 @@ export function RerunViewport({
     }
     const abort = new AbortController();
     let lastReportedPercent = -1;
+    let requestSettled = false;
+    let transferStarted = false;
+    let statusTimer: number | null = null;
     onPerceptionLoadChange?.({
       phase: "loading",
       receivedBytes: 0,
@@ -1582,10 +1589,51 @@ export function RerunViewport({
       progress: null,
       message: "Сервер готовит AI-слои.",
     });
+    const pollPreparationStatus = () => {
+      void fetchPerceptionPreparationStatus(
+        recordedPerceptionUrl,
+        identity.recordingId,
+        {
+          origin: window.location.origin,
+          signal: abort.signal,
+        },
+      ).then((status) => {
+        if (
+          abort.signal.aborted ||
+          requestSettled ||
+          transferStarted ||
+          perceptionChannelRef.current !== active
+        ) {
+          return;
+        }
+        const message = perceptionPreparationMessage(status.phase);
+        if (message) {
+          onPerceptionLoadChange?.({
+            phase: "loading",
+            receivedBytes: 0,
+            totalBytes: status.byteLength,
+            progress: null,
+            message,
+          });
+        }
+      }).catch(() => {
+        // The primary RRD request remains authoritative when status polling is unavailable.
+      }).finally(() => {
+        if (!abort.signal.aborted && !requestSettled && !transferStarted) {
+          statusTimer = window.setTimeout(pollPreparationStatus, 750);
+        }
+      });
+    };
+    pollPreparationStatus();
     void fetchRecordedPerceptionRrd(recordedPerceptionUrl, identity, {
       origin: window.location.origin,
       signal: abort.signal,
       onProgress: (receivedBytes, totalBytes) => {
+        transferStarted = true;
+        if (statusTimer !== null) {
+          window.clearTimeout(statusTimer);
+          statusTimer = null;
+        }
         const progress = totalBytes > 0 ? receivedBytes / totalBytes : null;
         const percent = progress === null ? -1 : Math.floor(progress * 100);
         if (percent === lastReportedPercent && receivedBytes !== totalBytes) return;
@@ -1640,8 +1688,18 @@ export function RerunViewport({
           : "AI-слои недоступны.",
       });
       // The base recording remains available when no admitted perception layer exists.
+    }).finally(() => {
+      requestSettled = true;
+      if (statusTimer !== null) {
+        window.clearTimeout(statusTimer);
+        statusTimer = null;
+      }
     });
-    return () => abort.abort();
+    return () => {
+      requestSettled = true;
+      if (statusTimer !== null) window.clearTimeout(statusTimer);
+      abort.abort();
+    };
   }, [
     onPerceptionLoadChange,
     perceptionChannelRevision,

apps/control-station/src/core/observation/perceptionPreparation.ts

@@ -0,0 +1,125 @@
+export type PerceptionPreparationPhase =
+  | "idle"
+  | "cache-lookup"
+  | "queued"
+  | "artifact-validation"
+  | "rendering"
+  | "cache-write"
+  | "ready"
+  | "unavailable"
+  | "error";
+
+export interface PerceptionPreparationStatus {
+  state: "idle" | "preparing" | "ready" | "unavailable" | "error";
+  phase: PerceptionPreparationPhase;
+  elapsedSeconds: number;
+  byteLength: number | null;
+}
+
+const RECORDED_PERCEPTION_PATH =
+  /^\/api\/v1\/observation-sessions\/[A-Za-z0-9][A-Za-z0-9._:-]{0,127}\/perception\.rrd$/;
+const SAFE_RECORDING_ID = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/;
+
+export async function fetchPerceptionPreparationStatus(
+  endpointUrl: string,
+  recordingId: string,
+  {
+    origin,
+    signal,
+    fetcher = globalThis.fetch,
+  }: {
+    origin: string;
+    signal?: AbortSignal;
+    fetcher?: typeof globalThis.fetch;
+  },
+): Promise<PerceptionPreparationStatus> {
+  const base = new URL(origin);
+  const endpoint = new URL(endpointUrl, base.origin);
+  if (
+    endpoint.origin !== base.origin ||
+    endpoint.search ||
+    endpoint.hash ||
+    !RECORDED_PERCEPTION_PATH.test(endpoint.pathname) ||
+    !SAFE_RECORDING_ID.test(recordingId)
+  ) {
+    throw new Error("Unsafe perception preparation status request");
+  }
+  endpoint.pathname = endpoint.pathname.replace(/\/perception\.rrd$/, "/perception/status");
+  endpoint.searchParams.set("recording_id", recordingId);
+  const response = await fetcher(endpoint.href, {
+    method: "GET",
+    cache: "no-store",
+    credentials: "same-origin",
+    headers: { Accept: "application/json" },
+    signal,
+  });
+  if (!response.ok) {
+    throw new Error("Perception preparation status is unavailable");
+  }
+  const value: unknown = await response.json();
+  if (!isPerceptionPreparationStatus(value)) {
+    throw new Error("Invalid perception preparation status");
+  }
+  return {
+    state: value.state,
+    phase: value.phase,
+    elapsedSeconds: value.elapsed_seconds,
+    byteLength: value.byte_length,
+  };
+}
+
+export function perceptionPreparationMessage(
+  phase: PerceptionPreparationPhase,
+): string | null {
+  switch (phase) {
+    case "cache-lookup":
+      return "Проверяем готовый AI-слой.";
+    case "queued":
+      return "AI-слой ожидает последовательной подготовки.";
+    case "artifact-validation":
+      return "Проверяем опубликованный AI-результат.";
+    case "rendering":
+      return "Формируем AI-слой.";
+    case "cache-write":
+      return "Сохраняем AI-слой.";
+    case "ready":
+      return "AI-слой готов. Загружаем.";
+    default:
+      return null;
+  }
+}
+
+function isPerceptionPreparationStatus(
+  value: unknown,
+): value is {
+  state: PerceptionPreparationStatus["state"];
+  phase: PerceptionPreparationPhase;
+  elapsed_seconds: number;
+  byte_length: number | null;
+} {
+  if (typeof value !== "object" || value === null) return false;
+  const candidate = value as Record<string, unknown>;
+  return (
+    ["idle", "preparing", "ready", "unavailable", "error"].includes(
+      String(candidate.state),
+    ) &&
+    [
+      "idle",
+      "cache-lookup",
+      "queued",
+      "artifact-validation",
+      "rendering",
+      "cache-write",
+      "ready",
+      "unavailable",
+      "error",
+    ].includes(String(candidate.phase)) &&
+    typeof candidate.elapsed_seconds === "number" &&
+    Number.isFinite(candidate.elapsed_seconds) &&
+    candidate.elapsed_seconds >= 0 &&
+    (candidate.byte_length === null ||
+      (typeof candidate.byte_length === "number" &&
+        Number.isSafeInteger(candidate.byte_length) &&
+        candidate.byte_length >= 0))
+  );
+}

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

@@ -0,0 +1,77 @@
+import assert from "node:assert/strict";
+import { after, before, test } from "node:test";
+
+import { createServer } from "vite";
+
+let server;
+let fetchPerceptionPreparationStatus;
+let perceptionPreparationMessage;
+
+before(async () => {
+  server = await createServer({
+    appType: "custom",
+    logLevel: "silent",
+    server: { middlewareMode: true },
+  });
+  ({
+    fetchPerceptionPreparationStatus,
+    perceptionPreparationMessage,
+  } = await server.ssrLoadModule(
+    "/src/core/observation/perceptionPreparation.ts",
+  ));
+});
+
+after(async () => {
+  await server?.close();
+});
+
+test("perception preparation status is same-origin, typed and phase-specific", async () => {
+  let requested;
+  const status = await fetchPerceptionPreparationStatus(
+    "/api/v1/observation-sessions/session-1/perception.rrd",
+    "recording-001",
+    {
+      origin: "http://127.0.0.1:8000",
+      fetcher: async (input, init) => {
+        requested = { url: String(input), method: init.method };
+        return Response.json({
+          state: "preparing",
+          phase: "artifact-validation",
+          elapsed_seconds: 2.5,
+          byte_length: null,
+        });
+      },
+    },
+  );
+
+  assert.deepEqual(requested, {
+    url: "http://127.0.0.1:8000/api/v1/observation-sessions/session-1/perception/status?recording_id=recording-001",
+    method: "GET",
+  });
+  assert.deepEqual(status, {
+    state: "preparing",
+    phase: "artifact-validation",
+    elapsedSeconds: 2.5,
+    byteLength: null,
+  });
+  assert.equal(
+    perceptionPreparationMessage(status.phase),
+    "Проверяем опубликованный AI-результат.",
+  );
+});
+
+test("perception preparation status rejects cross-origin endpoints", async () => {
+  await assert.rejects(
+    fetchPerceptionPreparationStatus(
+      "https://example.com/api/v1/observation-sessions/session-1/perception.rrd",
+      "recording-001",
+      {
+        origin: "http://127.0.0.1:8000",
+        fetcher: async () => {
+          throw new Error("must not fetch");
+        },
+      },
+    ),
+    /Unsafe perception preparation status request/,
+  );
+});

src/k1link/compute/fusion_epoch.py

@@ -201,6 +201,19 @@ class RecordedPerceptionOverlayMux:
             recording_id=recording_id,
         )
 
+    def status(self, session_id: str, *, recording_id: str) -> dict[str, Any]:
+        primary_status = getattr(self.primary, "status", None)
+        if callable(primary_status):
+            value = primary_status(session_id, recording_id=recording_id)
+            if isinstance(value, dict):
+                return value
+        return {
+            "state": "idle",
+            "phase": "idle",
+            "elapsed_seconds": 0.0,
+            "byte_length": None,
+        }
+
 
 def validate_recorded_calibrated_fusion(
     job_root: Path,

src/k1link/compute/integrated_perception.py

@@ -11,7 +11,9 @@ import stat
 import subprocess
 import tempfile
 import threading
-from contextlib import suppress
+import time
+from collections.abc import Iterator
+from contextlib import contextmanager, suppress
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Protocol, TypeGuard
@@ -39,10 +41,13 @@ MAX_JSON_BYTES = 64 * 1024 * 1024
 MAX_LINE_BYTES = 4 * 1024 * 1024
 MAX_SOURCE_BYTES = 512 * 1024 * 1024
 OVERLAY_RENDERER_VERSION = "4"
+OVERLAY_ADMISSION_SCHEMA = "missioncore.e10-overlay-admission/v1"
+OVERLAY_CACHE_SCHEMA = "missioncore.e10-overlay-cache/v1"
 CUBOID_PRESENTATION_HOLD_NS = 500_000_000
 
 _SAFE_RESULT_ID = re.compile(r"^e10-integrated-perception-[a-f0-9]{64}$")
 _SAFE_PACK_ID = re.compile(r"^e10-lidar-pack-[a-f0-9]{64}$")
+_SAFE_JOB_ID = re.compile(r"^recorded-camera-[a-f0-9]{24}$")
 _SAFE_RECORDING_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
 _SHA256 = re.compile(r"^[a-f0-9]{64}$")
 
@@ -67,6 +72,30 @@ class IntegratedPerceptionResult:
     report_path: Path
 
 
+@dataclass(frozen=True, slots=True)
+class _ResultDescriptor:
+    result_id: str
+    result_root: Path
+    result_json_sha256: str
+    session_id: str
+    job_id: str
+    created_at_utc: str
+
+
+@dataclass(frozen=True, slots=True)
+class _OverlayAdmission:
+    session_id: str
+    result_id: str
+    result_json_sha256: str
+    result_created_at_utc: str
+
+
+@dataclass(slots=True)
+class _FlightLock:
+    lock: threading.Lock
+    users: int = 0
+
+
 @dataclass(frozen=True, slots=True)
 class _PresentedCuboid:
     observed_ns: int
@@ -329,7 +358,13 @@ def validate_integrated_perception_result(
 
 
 class IntegratedPerceptionOverlayStore:
-    """Prefer the newest accepted E10 result for a recorded session."""
+    """Serve the newest admitted E10 overlay without revalidating it on replay.
+
+    Result validation is an admission concern. Once a complete overlay has been
+    rendered and sealed, its cache and admission sidecars are sufficient for
+    the presentation path. A cache miss still fails closed and validates the
+    exact selected result before rendering.
+    """
 
     def __init__(
         self,
@@ -345,9 +380,15 @@ class IntegratedPerceptionOverlayStore:
         self.lidar_packs_root = lidar_packs_root.expanduser().absolute()
         self.cache_root = cache_root.expanduser().absolute()
         self.ffmpeg_path = ffmpeg_path.expanduser().absolute()
-        self._lock = threading.Lock()
-        self._catalog_signature: tuple[Any, ...] | None = None
+        self._catalog_lock = threading.Lock()
+        self._descriptor_catalog_generation: tuple[int, int] | None = None
+        self._descriptors_by_session: dict[str, tuple[_ResultDescriptor, ...]] = {}
         self._latest_by_session: dict[str, IntegratedPerceptionResult | None] = {}
+        self._flight_guard = threading.Lock()
+        self._flights: dict[tuple[str, str], _FlightLock] = {}
+        self._materialization_lock = threading.Lock()
+        self._status_lock = threading.Lock()
+        self._status_by_recording: dict[tuple[str, str], dict[str, Any]] = {}
 
     def render(
         self,
@@ -363,146 +404,357 @@ class IntegratedPerceptionOverlayStore:
             or _SAFE_RECORDING_ID.fullmatch(recording_id) is None
         ):
             raise ValueError("integrated perception recording id is invalid")
-        with self._lock:
-            result = self._latest(session_id)
-            if result is None:
-                return None
-            cache = _private_child(
-                _private_child(_private_directory(self.cache_root), session_id),
-                result.result_id,
-            )
-            output = cache / f"{recording_id}.rrd"
-            sidecar = output.with_suffix(".rrd.cache.json")
-            cached = _read_cache(output, sidecar, result, recording_id)
-            if cached is not None:
-                return cached
-            payload = _render(
-                result,
-                application_id=application_id,
-                recording_id=recording_id,
-                ffmpeg_path=self.ffmpeg_path,
-                temporary_root=cache,
-            )
-            temporary = output.with_name(f".{output.name}.{os.getpid()}.tmp")
-            try:
-                with temporary.open("xb") as stream:
-                    stream.write(payload)
-                    stream.flush()
-                    os.fsync(stream.fileno())
-                os.chmod(temporary, 0o600)
-                os.replace(temporary, output)
-                write_json_atomic(
-                    sidecar,
-                    {
-                        "schema_version": "missioncore.e10-overlay-cache/v1",
-                        "renderer_version": OVERLAY_RENDERER_VERSION,
-                        "result_id": result.result_id,
-                        "recording_id": recording_id,
-                        "byte_length": len(payload),
-                        "sha256": hashlib.sha256(payload).hexdigest(),
-                    },
-                )
-            finally:
-                temporary.unlink(missing_ok=True)
-            return payload
-
-    def _latest(self, session_id: str) -> IntegratedPerceptionResult | None:
-        signature = _directory_catalog_signature(
-            self.jobs_root,
-            self.results_root,
-            self.lidar_packs_root,
-        )
-        if signature != self._catalog_signature:
-            self._catalog_signature = signature
-            self._latest_by_session.clear()
-        if session_id in self._latest_by_session:
-            return self._latest_by_session[session_id]
+        key = (session_id, recording_id)
+        self._set_status(key, state="preparing", phase="cache-lookup")
         try:
-            jobs = sorted(self.jobs_root.iterdir())
-            candidates = sorted(self.results_root.iterdir())
-        except FileNotFoundError:
-            self._latest_by_session[session_id] = None
+            with self._single_flight(key):
+                cached = self._read_admitted_cache(session_id, recording_id)
+                if cached is not None:
+                    self._set_status(
+                        key,
+                        state="ready",
+                        phase="ready",
+                        byte_length=len(cached),
+                    )
+                    return cached
+                self._set_status(key, state="preparing", phase="queued")
+                with self._materialization_lock:
+                    self._set_status(key, state="preparing", phase="artifact-validation")
+                    result = self._latest(session_id)
+                    if result is None:
+                        self._set_status(key, state="unavailable", phase="unavailable")
+                        return None
+                    self._write_admission(result)
+                    cache = _private_child(
+                        _private_child(_private_directory(self.cache_root), session_id),
+                        result.result_id,
+                    )
+                    output = cache / f"{recording_id}.rrd"
+                    sidecar = output.with_suffix(".rrd.cache.json")
+                    cached = _read_cache(
+                        output,
+                        sidecar,
+                        result_id=result.result_id,
+                        recording_id=recording_id,
+                    )
+                    if cached is not None:
+                        self._set_status(
+                            key,
+                            state="ready",
+                            phase="ready",
+                            byte_length=len(cached),
+                        )
+                        return cached
+                    self._set_status(key, state="preparing", phase="rendering")
+                    payload = _render(
+                        result,
+                        application_id=application_id,
+                        recording_id=recording_id,
+                        ffmpeg_path=self.ffmpeg_path,
+                        temporary_root=cache,
+                    )
+                    self._set_status(key, state="preparing", phase="cache-write")
+                    temporary = output.with_name(f".{output.name}.{os.getpid()}.tmp")
+                    try:
+                        with temporary.open("xb") as stream:
+                            stream.write(payload)
+                            stream.flush()
+                            os.fsync(stream.fileno())
+                        os.chmod(temporary, 0o600)
+                        os.replace(temporary, output)
+                        write_json_atomic(
+                            sidecar,
+                            {
+                                "schema_version": OVERLAY_CACHE_SCHEMA,
+                                "renderer_version": OVERLAY_RENDERER_VERSION,
+                                "session_id": session_id,
+                                "result_id": result.result_id,
+                                "result_created_at_utc": result.created_at_utc,
+                                "recording_id": recording_id,
+                                "byte_length": len(payload),
+                                "sha256": hashlib.sha256(payload).hexdigest(),
+                            },
+                        )
+                    finally:
+                        temporary.unlink(missing_ok=True)
+                    self._set_status(
+                        key,
+                        state="ready",
+                        phase="ready",
+                        byte_length=len(payload),
+                    )
+                    return payload
+        except BaseException:
+            self._set_status(key, state="error", phase="error")
+            raise
+
+    def status(self, session_id: str, *, recording_id: str) -> dict[str, Any]:
+        if (
+            _SAFE_RECORDING_ID.fullmatch(session_id) is None
+            or _SAFE_RECORDING_ID.fullmatch(recording_id) is None
+        ):
+            raise ValueError("integrated perception recording id is invalid")
+        key = (session_id, recording_id)
+        with self._status_lock:
+            value = self._status_by_recording.get(key)
+            if value is None:
+                return {
+                    "state": "idle",
+                    "phase": "idle",
+                    "elapsed_seconds": 0.0,
+                    "byte_length": None,
+                }
+            elapsed = max(0.0, time.monotonic() - float(value["started_monotonic"]))
+            return {
+                "state": value["state"],
+                "phase": value["phase"],
+                "elapsed_seconds": round(elapsed, 3),
+                "byte_length": value.get("byte_length"),
+            }
+
+    def _set_status(
+        self,
+        key: tuple[str, str],
+        *,
+        state: str,
+        phase: str,
+        byte_length: int | None = None,
+    ) -> None:
+        with self._status_lock:
+            previous = self._status_by_recording.get(key)
+            started = (
+                float(previous["started_monotonic"])
+                if previous is not None and previous.get("state") == "preparing"
+                else time.monotonic()
+            )
+            self._status_by_recording[key] = {
+                "state": state,
+                "phase": phase,
+                "started_monotonic": started,
+                "byte_length": byte_length,
+            }
+
+    @contextmanager
+    def _single_flight(self, key: tuple[str, str]) -> Iterator[None]:
+        with self._flight_guard:
+            flight = self._flights.get(key)
+            if flight is None:
+                flight = _FlightLock(lock=threading.Lock())
+                self._flights[key] = flight
+            flight.users += 1
+        flight.lock.acquire()
+        try:
+            yield
+        finally:
+            flight.lock.release()
+            with self._flight_guard:
+                flight.users -= 1
+                if flight.users == 0:
+                    self._flights.pop(key, None)
+
+    def _read_admitted_cache(
+        self,
+        session_id: str,
+        recording_id: str,
+    ) -> bytes | None:
+        cache_root = _private_directory(self.cache_root)
+        session_cache = _private_child(cache_root, session_id)
+        admission = self._load_admission(session_cache, session_id)
+        latest = self._latest_descriptor(session_id)
+        if latest is None:
             return None
-        if len(jobs) > MAX_SCAN or len(candidates) > MAX_SCAN:
+        if admission is None or admission.result_id != latest.result_id:
+            recovered = self._recover_admission_from_cache(
+                session_cache,
+                latest,
+                recording_id,
+            )
+            if recovered is not None:
+                _, payload = recovered
+                return payload
+            admission = None
+        if admission is None or admission.result_id != latest.result_id:
+            return None
+        result_cache = _private_child(session_cache, admission.result_id)
+        output = result_cache / f"{recording_id}.rrd"
+        sidecar = output.with_suffix(".rrd.cache.json")
+        return _read_cache(
+            output,
+            sidecar,
+            result_id=admission.result_id,
+            recording_id=recording_id,
+        )
+
+    def _load_admission(
+        self,
+        session_cache: Path,
+        session_id: str,
+    ) -> _OverlayAdmission | None:
+        path = session_cache / "admission.json"
+        try:
+            value = _read_object(path, session_cache)
+        except (OSError, SessionIntegrityError):
+            return None
+        result_id = value.get("result_id")
+        result_json_sha256 = value.get("result_json_sha256")
+        created_at_utc = value.get("result_created_at_utc")
+        if (
+            value.get("schema_version") != OVERLAY_ADMISSION_SCHEMA
+            or value.get("renderer_version") != OVERLAY_RENDERER_VERSION
+            or value.get("session_id") != session_id
+            or not isinstance(result_id, str)
+            or _SAFE_RESULT_ID.fullmatch(result_id) is None
+            or not isinstance(result_json_sha256, str)
+            or _SHA256.fullmatch(result_json_sha256) is None
+            or not isinstance(created_at_utc, str)
+        ):
+            return None
+        try:
+            descriptor = _read_result_descriptor(self.results_root / result_id)
+        except (OSError, SessionIntegrityError):
+            return None
+        if (
+            descriptor.session_id != session_id
+            or descriptor.result_json_sha256 != result_json_sha256
+            or descriptor.created_at_utc != created_at_utc
+        ):
+            return None
+        return _OverlayAdmission(
+            session_id=session_id,
+            result_id=result_id,
+            result_json_sha256=result_json_sha256,
+            result_created_at_utc=created_at_utc,
+        )
+
+    def _recover_admission_from_cache(
+        self,
+        session_cache: Path,
+        descriptor: _ResultDescriptor,
+        recording_id: str,
+    ) -> tuple[_OverlayAdmission, bytes] | None:
+        cache = _private_child(session_cache, descriptor.result_id)
+        output = cache / f"{recording_id}.rrd"
+        sidecar = output.with_suffix(".rrd.cache.json")
+        payload = _read_cache(
+            output,
+            sidecar,
+            result_id=descriptor.result_id,
+            recording_id=recording_id,
+        )
+        if payload is None:
+            return None
+        admission = _OverlayAdmission(
+            session_id=descriptor.session_id,
+            result_id=descriptor.result_id,
+            result_json_sha256=descriptor.result_json_sha256,
+            result_created_at_utc=descriptor.created_at_utc,
+        )
+        self._write_admission_document(session_cache, admission)
+        return admission, payload
+
+    def _write_admission(self, result: IntegratedPerceptionResult) -> None:
+        descriptor = _read_result_descriptor(result.result_root)
+        if (
+            descriptor.session_id != result.job.session_id
+            or descriptor.result_id != result.result_id
+            or descriptor.created_at_utc != result.created_at_utc
+        ):
+            raise RecordedPerceptionOverlayError(
+                "integrated perception admission identity changed"
+            )
+        cache_root = _private_directory(self.cache_root)
+        session_cache = _private_child(cache_root, descriptor.session_id)
+        self._write_admission_document(
+            session_cache,
+            _OverlayAdmission(
+                session_id=descriptor.session_id,
+                result_id=descriptor.result_id,
+                result_json_sha256=descriptor.result_json_sha256,
+                result_created_at_utc=descriptor.created_at_utc,
+            ),
+        )
+
+    @staticmethod
+    def _write_admission_document(
+        session_cache: Path,
+        admission: _OverlayAdmission,
+    ) -> None:
+        write_json_atomic(
+            session_cache / "admission.json",
+            {
+                "schema_version": OVERLAY_ADMISSION_SCHEMA,
+                "renderer_version": OVERLAY_RENDERER_VERSION,
+                "session_id": admission.session_id,
+                "result_id": admission.result_id,
+                "result_json_sha256": admission.result_json_sha256,
+                "result_created_at_utc": admission.result_created_at_utc,
+            },
+        )
+
+    def _latest_descriptor(self, session_id: str) -> _ResultDescriptor | None:
+        try:
+            metadata = self.results_root.stat()
+            entries = tuple(sorted(self.results_root.iterdir(), key=lambda path: path.name))
+        except FileNotFoundError:
+            return None
+        if len(entries) > MAX_SCAN:
             raise RecordedPerceptionOverlayError("integrated perception catalog is outside bounds")
-        matches = []
-        for job_root in jobs:
-            if job_root.is_symlink():
-                continue
-            try:
-                job = validate_camera_compute_job(job_root)
-            except (OSError, SessionIntegrityError):
-                continue
-            if job.session_id != session_id:
-                continue
-            for candidate in candidates:
+        generation = (metadata.st_mtime_ns, len(entries))
+        with self._catalog_lock:
+            if generation != self._descriptor_catalog_generation:
+                self._descriptor_catalog_generation = generation
+                self._descriptors_by_session.clear()
+                self._latest_by_session.clear()
+            existing = self._descriptors_by_session.get(session_id)
+            if existing is not None:
+                return existing[0] if existing else None
+            matches: list[_ResultDescriptor] = []
+            for candidate in entries:
                 if candidate.is_symlink() or _SAFE_RESULT_ID.fullmatch(candidate.name) is None:
                     continue
                 try:
-                    value = validate_integrated_perception_result(
-                        job_root,
-                        candidate,
-                        self.lidar_packs_root,
-                    )
+                    descriptor = _read_result_descriptor(candidate)
                 except (OSError, SessionIntegrityError):
                     continue
-                if (
-                    value.accepted
-                    and value.publication_scope == "recorded-integrated-realtime-qualification-only"
-                ):
-                    matches.append(value)
-        if not matches:
-            self._latest_by_session[session_id] = None
-            return None
-        latest = max(matches, key=lambda value: (value.created_at_utc, value.result_id))
-        self._latest_by_session[session_id] = latest
-        return latest
-
-
-def _directory_catalog_signature(*roots: Path) -> tuple[Any, ...]:
-    """Track immutable lab catalogs without rehashing every large artifact per request.
-
-    Accepted result directories are append-only. Their directory mtimes still
-    change when an atomic manifest replacement occurs, while a new experiment
-    changes the parent listing. This cheap signature therefore invalidates the
-    process-local validation memo without weakening the first full integrity
-    validation of every catalog generation.
-    """
-
-    signature: list[Any] = []
-    for root in roots:
-        try:
-            root_stat = root.stat()
-            entries = sorted(root.iterdir(), key=lambda path: path.name)
-        except FileNotFoundError:
-            signature.append((str(root), None))
-            continue
-        if len(entries) > MAX_SCAN:
-            raise RecordedPerceptionOverlayError("integrated perception catalog is outside bounds")
-        entry_signature = []
-        for entry in entries:
-            try:
-                metadata = entry.lstat()
-            except FileNotFoundError:
-                # A concurrent atomic publication will alter the parent mtime
-                # and be observed on the next request.
-                continue
-            entry_signature.append(
-                (
-                    entry.name,
-                    metadata.st_mode,
-                    metadata.st_size,
-                    metadata.st_mtime_ns,
+                if descriptor.session_id == session_id:
+                    matches.append(descriptor)
+            ordered = tuple(
+                sorted(
+                    matches,
+                    key=lambda value: (value.created_at_utc, value.result_id),
+                    reverse=True,
                 )
             )
-        signature.append(
-            (
-                str(root),
-                root_stat.st_mtime_ns,
-                tuple(entry_signature),
-            )
-        )
-    return tuple(signature)
+            self._descriptors_by_session[session_id] = ordered
+            return ordered[0] if ordered else None
+
+    def _latest(self, session_id: str) -> IntegratedPerceptionResult | None:
+        latest_descriptor = self._latest_descriptor(session_id)
+        if latest_descriptor is None:
+            return None
+        with self._catalog_lock:
+            if session_id in self._latest_by_session:
+                return self._latest_by_session[session_id]
+            descriptors = self._descriptors_by_session.get(session_id, ())
+        for descriptor in descriptors:
+            try:
+                value = validate_integrated_perception_result(
+                    self.jobs_root / descriptor.job_id,
+                    descriptor.result_root,
+                    self.lidar_packs_root,
+                )
+            except (OSError, SessionIntegrityError):
+                continue
+            if (
+                value.accepted
+                and value.publication_scope == "recorded-integrated-realtime-qualification-only"
+            ):
+                with self._catalog_lock:
+                    self._latest_by_session[session_id] = value
+                return value
+        with self._catalog_lock:
+            self._latest_by_session[session_id] = None
+        return None
 
 
 def _render(
@@ -890,10 +1142,52 @@ def _read_rows(path: Path, root: Path, schema: str) -> list[dict[str, Any]]:
     return rows
 
 
+def _read_result_descriptor(result_root: Path) -> _ResultDescriptor:
+    root = result_root.expanduser().resolve(strict=True)
+    if not root.is_dir() or _SAFE_RESULT_ID.fullmatch(root.name) is None:
+        raise SessionIntegrityError("integrated perception result descriptor is invalid")
+    result_path = root / "result.json"
+    result = _read_object(result_path, root)
+    identity = result.get("identity")
+    identity_sha256 = result.get("identity_sha256")
+    created_at_utc = result.get("created_at_utc")
+    session_id = identity.get("session_id") if isinstance(identity, dict) else None
+    job_id = identity.get("job_id") if isinstance(identity, dict) else None
+    if (
+        result.get("schema_version") != RESULT_SCHEMA
+        or result.get("result_id") != root.name
+        or not isinstance(identity, dict)
+        or identity.get("schema_version") != IDENTITY_SCHEMA
+        or not isinstance(identity_sha256, str)
+        or _SHA256.fullmatch(identity_sha256) is None
+        or root.name != f"e10-integrated-perception-{identity_sha256}"
+        or hashlib.sha256(_canonical_json(identity)).hexdigest() != identity_sha256
+        or result.get("acceptance_state") != "accepted"
+        or result.get("publication_scope")
+        != "recorded-integrated-realtime-qualification-only"
+        or not isinstance(session_id, str)
+        or _SAFE_RECORDING_ID.fullmatch(session_id) is None
+        or not isinstance(job_id, str)
+        or _SAFE_JOB_ID.fullmatch(job_id) is None
+        or not isinstance(created_at_utc, str)
+        or not 1 <= len(created_at_utc) <= 64
+    ):
+        raise SessionIntegrityError("integrated perception result is not admitted")
+    return _ResultDescriptor(
+        result_id=root.name,
+        result_root=root,
+        result_json_sha256=_sha256(result_path),
+        session_id=session_id,
+        job_id=job_id,
+        created_at_utc=created_at_utc,
+    )
+
+
 def _read_cache(
     output: Path,
     sidecar: Path,
-    result: IntegratedPerceptionResult,
+    *,
+    result_id: str,
     recording_id: str,
 ) -> bytes | None:
     try:
@@ -902,9 +1196,9 @@ def _read_cache(
     except (OSError, SessionIntegrityError):
         return None
     if (
-        value.get("schema_version") != "missioncore.e10-overlay-cache/v1"
+        value.get("schema_version") != OVERLAY_CACHE_SCHEMA
         or value.get("renderer_version") != OVERLAY_RENDERER_VERSION
-        or value.get("result_id") != result.result_id
+        or value.get("result_id") != result_id
         or value.get("recording_id") != recording_id
         or value.get("byte_length") != len(payload)
         or value.get("sha256") != hashlib.sha256(payload).hexdigest()

src/k1link/device_plugins/xgrids_k1/rrd_export.py

@@ -55,6 +55,7 @@ JS_MAX_SAFE_INTEGER = (1 << 53) - 1
 RECORDED_VIEW_POINT_DECIMATION_THRESHOLD = 100_000
 RECORDED_VIEW_POINT_STRIDE = 4
 RECORDED_VIEW_POINT_FRAME_STRIDE = 5
+RECORDED_RRD_IDENTITY_VERSION = "missioncore.recorded-rrd/v1"
 
 # Rerun keys viewer state by these IDs.  Reusing them for every settings-only
 # blueprint update makes the update overwrite the existing scene instead of
@@ -216,13 +217,26 @@ def export_k1mqtt_to_rrd(
     _validate_paths(source, destination)
     destination.parent.mkdir(parents=True, exist_ok=True)
 
-    recording_id = str(uuid4())
-    temporary = destination.with_name(f".{destination.name}.{recording_id}.tmp")
     source_sha256 = _sha256_file(
         source,
         cancel_event=cancel_event,
         activity_callback=activity_callback,
     )
+    capture_clock = _optional_capture_clock(source, capture_clock_path)
+    capture_clock_origin = _optional_capture_clock_origin(
+        source,
+        capture_clock_origin_path,
+    )
+    # Rerun uses this identity to merge independently delivered base,
+    # blueprint and perception streams. Bind it to immutable source evidence
+    # so regenerating an unchanged recording does not invalidate every derived
+    # overlay cache.
+    recording_id = _stable_recording_id(
+        source_sha256,
+        capture_clock,
+        capture_clock_origin,
+    )
+    temporary = destination.with_name(f".{destination.name}.{uuid4()}.tmp")
     settings = RerunSceneSettings()
     blueprint = _recorded_blueprint(settings)
     recording: rr.RecordingStream | None = None
@@ -231,11 +245,6 @@ def export_k1mqtt_to_rrd(
 
     counters = _ExportCounters()
     trajectory = _TrajectoryBuffer.empty()
-    capture_clock = _optional_capture_clock(source, capture_clock_path)
-    capture_clock_origin = _optional_capture_clock_origin(
-        source,
-        capture_clock_origin_path,
-    )
     if (
         capture_clock is not None
         and capture_clock_origin is not None
@@ -449,6 +458,31 @@ def _validate_paths(source: Path, destination: Path) -> None:
         raise RrdExportError("RRD export accepts only native K1MQTT captures")
 
 
+def _stable_recording_id(
+    source_sha256: str,
+    capture_clock: CaptureClockEnvelope | None,
+    capture_clock_origin: CaptureClockOrigin | None,
+) -> str:
+    identity = hashlib.sha256()
+    for value in (
+        RECORDED_RRD_IDENTITY_VERSION,
+        source_sha256,
+        str(capture_clock.started_at_epoch_ns) if capture_clock is not None else "-",
+        str(capture_clock.started_monotonic_ns) if capture_clock is not None else "-",
+        str(capture_clock.completed_at_epoch_ns) if capture_clock is not None else "-",
+        str(capture_clock.completed_monotonic_ns) if capture_clock is not None else "-",
+        str(capture_clock_origin.started_at_epoch_ns)
+        if capture_clock_origin is not None
+        else "-",
+        str(capture_clock_origin.started_monotonic_ns)
+        if capture_clock_origin is not None
+        else "-",
+    ):
+        identity.update(value.encode("ascii"))
+        identity.update(b"\0")
+    return f"k1-{identity.hexdigest()}"
+
+
 def _optional_capture_clock(
     source: Path,
     explicit_path: Path | None,

src/k1link/web/session_api.py

@@ -130,6 +130,23 @@ class RecordedPerceptionRequest(StrictApiModel):
     )
 
 
+class RecordedPerceptionPreparationStatus(StrictApiModel):
+    state: Literal["idle", "preparing", "ready", "unavailable", "error"]
+    phase: Literal[
+        "idle",
+        "cache-lookup",
+        "queued",
+        "artifact-validation",
+        "rendering",
+        "cache-write",
+        "ready",
+        "unavailable",
+        "error",
+    ]
+    elapsed_seconds: float = Field(ge=0.0)
+    byte_length: int | None = Field(default=None, ge=0)
+
+
 class RecordedPointColorsRequest(StrictApiModel):
     application_id: Literal["nodedc_mission_core_recorded"]
     recording_id: str = Field(
@@ -915,6 +932,61 @@ def build_session_router(
             },
         )
 
+    @router.get("/api/v1/observation-sessions/{session_id}/perception/status")
+    async def get_observation_session_perception_status(
+        session_id: str,
+        response: Response,
+        recording_id: Annotated[
+            str,
+            Query(
+                min_length=1,
+                max_length=128,
+                pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$",
+            ),
+        ],
+    ) -> dict[str, Any]:
+        try:
+            await run_in_threadpool(
+                _prepare_replay,
+                store,
+                catalog_refresher,
+                session_id,
+                1.0,
+                False,
+                False,
+            )
+        except SessionNotFoundError as exc:
+            raise HTTPException(status_code=404, detail=str(exc)) from exc
+        except (SessionNotReplayableError, SessionIntegrityError) as exc:
+            raise HTTPException(status_code=409, detail=str(exc)) from exc
+        status_provider = (
+            getattr(perception_overlay_provider, "status", None)
+            if perception_overlay_provider is not None
+            else None
+        )
+        value: object = (
+            await run_in_threadpool(
+                status_provider,
+                session_id,
+                recording_id=recording_id,
+            )
+            if callable(status_provider)
+            else {
+                "state": "idle",
+                "phase": "idle",
+                "elapsed_seconds": 0.0,
+                "byte_length": None,
+            }
+        )
+        response.headers["Cache-Control"] = "no-store"
+        try:
+            return RecordedPerceptionPreparationStatus.model_validate(value).model_dump()
+        except ValueError as exc:
+            raise HTTPException(
+                status_code=500,
+                detail="Сервис слоя распознавания вернул некорректный статус.",
+            ) from exc
+
     @router.post("/api/v1/observation-sessions/{session_id}/point-colors.rrd")
     async def get_observation_session_point_colors(
         session_id: str,

tests/test_e10_integrated_perception.py

@@ -1,7 +1,12 @@
 from __future__ import annotations
 
+import hashlib
 import importlib.util
+import json
 import sys
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 from types import SimpleNamespace
 
@@ -15,6 +20,44 @@ from k1link.compute.integrated_perception import (
 )
 
 
+def _write_result_descriptor(
+    results_root: Path,
+    *,
+    session_id: str,
+    created_at_utc: str,
+) -> Path:
+    identity = {
+        "schema_version": integrated_module.IDENTITY_SCHEMA,
+        "job_id": f"recorded-camera-{'1' * 24}",
+        "session_id": session_id,
+    }
+    identity_sha256 = hashlib.sha256(
+        json.dumps(
+            identity,
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+        ).encode()
+    ).hexdigest()
+    result_id = f"e10-integrated-perception-{identity_sha256}"
+    root = results_root / result_id
+    root.mkdir()
+    (root / "result.json").write_text(
+        json.dumps(
+            {
+                "schema_version": integrated_module.RESULT_SCHEMA,
+                "result_id": result_id,
+                "identity": identity,
+                "identity_sha256": identity_sha256,
+                "acceptance_state": "accepted",
+                "publication_scope": "recorded-integrated-realtime-qualification-only",
+                "created_at_utc": created_at_utc,
+            }
+        )
+    )
+    return root
+
+
 def _worker_modules() -> tuple[object, object]:
     root = Path(__file__).resolve().parents[1] / "experiments" / "perception"
     worker = root / "worker"
@@ -41,7 +84,7 @@ def _worker_modules() -> tuple[object, object]:
         sys.path.pop(0)
 
 
-def test_integrated_overlay_catalog_validation_is_memoized_until_publication(
+def test_integrated_overlay_recovers_admission_from_sealed_cache_without_revalidation(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -51,32 +94,37 @@ def test_integrated_overlay_catalog_validation_is_memoized_until_publication(
     cache_root = tmp_path / "cache"
     for root in (jobs_root, results_root, lidar_packs_root):
         root.mkdir()
-    job_root = jobs_root / "job-1"
-    job_root.mkdir()
-    first_result = results_root / f"e10-integrated-perception-{'a' * 64}"
-    first_result.mkdir()
-
-    validation_calls = {"job": 0, "result": 0}
-    job = SimpleNamespace(session_id="session-1")
-
-    def validate_job(_root: Path) -> object:
-        validation_calls["job"] += 1
-        return job
-
-    def validate_result(_job_root: Path, candidate: Path, _packs: Path) -> object:
-        validation_calls["result"] += 1
-        return SimpleNamespace(
-            accepted=True,
-            publication_scope="recorded-integrated-realtime-qualification-only",
-            created_at_utc=candidate.name,
-            result_id=candidate.name,
+    result = _write_result_descriptor(
+        results_root,
+        session_id="session-1",
+        created_at_utc="2026-07-29T12:00:00Z",
+    )
+    payload = b"RRF2sealed-overlay"
+    recording_id = "recording-1"
+    cache = cache_root / "session-1" / result.name
+    cache.mkdir(parents=True)
+    output = cache / f"{recording_id}.rrd"
+    output.write_bytes(payload)
+    (cache / f"{recording_id}.rrd.cache.json").write_text(
+        json.dumps(
+            {
+                "schema_version": integrated_module.OVERLAY_CACHE_SCHEMA,
+                "renderer_version": integrated_module.OVERLAY_RENDERER_VERSION,
+                "result_id": result.name,
+                "recording_id": recording_id,
+                "byte_length": len(payload),
+                "sha256": hashlib.sha256(payload).hexdigest(),
+            }
         )
+    )
+
+    def reject_revalidation(*_args: object, **_kwargs: object) -> object:
+        raise AssertionError("sealed presentation cache must not revalidate source artifacts")
 
-    monkeypatch.setattr(integrated_module, "validate_camera_compute_job", validate_job)
     monkeypatch.setattr(
         integrated_module,
         "validate_integrated_perception_result",
-        validate_result,
+        reject_revalidation,
     )
     store = IntegratedPerceptionOverlayStore(
         jobs_root=jobs_root,
@@ -86,13 +134,75 @@ def test_integrated_overlay_catalog_validation_is_memoized_until_publication(
         ffmpeg_path=tmp_path / "ffmpeg",
     )
 
-    assert store._latest("session-1") is not None
-    assert store._latest("session-1") is not None
-    assert validation_calls == {"job": 1, "result": 1}
+    assert (
+        store.render(
+            "session-1",
+            application_id="nodedc_mission_core_recorded",
+            recording_id=recording_id,
+        )
+        == payload
+    )
+    admission = json.loads((cache_root / "session-1" / "admission.json").read_text())
+    assert admission["result_id"] == result.name
+    assert store.status("session-1", recording_id=recording_id) == {
+        "state": "ready",
+        "phase": "ready",
+        "elapsed_seconds": pytest.approx(0.0, abs=0.1),
+        "byte_length": len(payload),
+    }
 
-    (results_root / f"e10-integrated-perception-{'b' * 64}").mkdir()
-    assert store._latest("session-1") is not None
-    assert validation_calls == {"job": 2, "result": 3}
+
+def test_integrated_overlay_serializes_heavy_materialization_across_sessions(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    for name in ("jobs", "results", "lidar-packs"):
+        (tmp_path / name).mkdir()
+    store = IntegratedPerceptionOverlayStore(
+        jobs_root=tmp_path / "jobs",
+        results_root=tmp_path / "results",
+        lidar_packs_root=tmp_path / "lidar-packs",
+        cache_root=tmp_path / "cache",
+        ffmpeg_path=tmp_path / "ffmpeg",
+    )
+    result_id = f"e10-integrated-perception-{'a' * 64}"
+    result = SimpleNamespace(
+        result_id=result_id,
+        created_at_utc="2026-07-29T12:00:00Z",
+    )
+    monkeypatch.setattr(store, "_read_admitted_cache", lambda *_args: None)
+    monkeypatch.setattr(store, "_latest", lambda _session_id: result)
+    monkeypatch.setattr(store, "_write_admission", lambda _result: None)
+
+    active = 0
+    maximum_active = 0
+    active_lock = threading.Lock()
+
+    def render_overlay(*_args: object, **_kwargs: object) -> bytes:
+        nonlocal active, maximum_active
+        with active_lock:
+            active += 1
+            maximum_active = max(maximum_active, active)
+        time.sleep(0.05)
+        with active_lock:
+            active -= 1
+        return b"RRF2materialized"
+
+    monkeypatch.setattr(integrated_module, "_render", render_overlay)
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        results = tuple(
+            executor.map(
+                lambda item: store.render(
+                    item,
+                    application_id="nodedc_mission_core_recorded",
+                    recording_id=f"recording-{item}",
+                ),
+                ("session-1", "session-2"),
+            )
+        )
+
+    assert results == (b"RRF2materialized", b"RRF2materialized")
+    assert maximum_active == 1
 
 
 def test_e10_profile_pins_integrated_realtime_budget() -> None:

tests/test_rrd_export.py

@@ -683,6 +683,10 @@ def test_export_preserves_every_decodable_frame_and_source_timeline(tmp_path: Pa
     assert summary["first_decoded_time_ns"] == 600_000_000
     assert summary["last_decoded_time_ns"] == 1_700_000_000
     assert summary["source_sha256"] == _sha256(capture)
+    repeated = export_k1mqtt_to_rrd(capture, tmp_path / "session-repeated.rrd")
+    assert repeated["recording_id"] == summary["recording_id"]
+    assert summary["recording_id"].startswith("k1-")
+    assert len(summary["recording_id"]) == 67
     assert summary["rrd_bytes"] == output.stat().st_size
     assert summary["rrd_sha256"] == _sha256(output)
     assert output.stat().st_size > 0

tests/test_session_api.py

@@ -1313,6 +1313,16 @@ def test_recorded_perception_endpoint_returns_one_complete_optional_overlay(
             calls.append((session_id, application_id, recording_id))
             return b"RRF2perception"
 
+        def status(self, session_id: str, *, recording_id: str) -> dict[str, object]:
+            assert session_id == session.name
+            assert recording_id == "recording-001"
+            return {
+                "state": "preparing",
+                "phase": "artifact-validation",
+                "elapsed_seconds": 2.5,
+                "byte_length": None,
+            }
+
     router = build_session_router(store, perception_overlay_provider=Provider())
     perception_route = endpoint(
         router,
@@ -1334,6 +1344,26 @@ def test_recorded_perception_endpoint_returns_one_complete_optional_overlay(
     assert calls == [
         (session.name, "nodedc_mission_core_recorded", "recording-001")
     ]
+    status_route = endpoint(
+        router,
+        "/api/v1/observation-sessions/{session_id}/perception/status",
+        "GET",
+    )
+    status_response = Response()
+    status = asyncio.run(
+        status_route(
+            session_id=session.name,
+            response=status_response,
+            recording_id="recording-001",
+        )
+    )
+    assert status == {
+        "state": "preparing",
+        "phase": "artifact-validation",
+        "elapsed_seconds": 2.5,
+        "byte_length": None,
+    }
+    assert status_response.headers["cache-control"] == "no-store"
 
     empty_router = build_session_router(store)
     empty_route = endpoint(