From cc6838ed2409d57986ab0bd8905054b748dd0dcb Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 29 Jul 2026 23:13:42 +0300 Subject: [PATCH] fix(replay): serve admitted AI overlays without revalidation --- .../src/components/RerunViewport.tsx | 60 +- .../core/observation/perceptionPreparation.ts | 125 ++++ .../test/perceptionPreparation.test.mjs | 77 +++ src/k1link/compute/fusion_epoch.py | 13 + src/k1link/compute/integrated_perception.py | 568 +++++++++++++----- .../device_plugins/xgrids_k1/rrd_export.py | 48 +- src/k1link/web/session_api.py | 72 +++ tests/test_e10_integrated_perception.py | 166 ++++- tests/test_rrd_export.py | 4 + tests/test_session_api.py | 30 + 10 files changed, 990 insertions(+), 173 deletions(-) create mode 100644 apps/control-station/src/core/observation/perceptionPreparation.ts create mode 100644 apps/control-station/test/perceptionPreparation.test.mjs diff --git a/apps/control-station/src/components/RerunViewport.tsx b/apps/control-station/src/components/RerunViewport.tsx index 5a198ed..0acb3bb 100644 --- a/apps/control-station/src/components/RerunViewport.tsx +++ b/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, diff --git a/apps/control-station/src/core/observation/perceptionPreparation.ts b/apps/control-station/src/core/observation/perceptionPreparation.ts new file mode 100644 index 0000000..f60262e --- /dev/null +++ b/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 { + 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; + 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)) + ); +} diff --git a/apps/control-station/test/perceptionPreparation.test.mjs b/apps/control-station/test/perceptionPreparation.test.mjs new file mode 100644 index 0000000..c7c7736 --- /dev/null +++ b/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/, + ); +}); diff --git a/src/k1link/compute/fusion_epoch.py b/src/k1link/compute/fusion_epoch.py index d4bba5b..e35ba36 100644 --- a/src/k1link/compute/fusion_epoch.py +++ b/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, diff --git a/src/k1link/compute/integrated_perception.py b/src/k1link/compute/integrated_perception.py index 69e8bc1..264119e 100644 --- a/src/k1link/compute/integrated_perception.py +++ b/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() diff --git a/src/k1link/device_plugins/xgrids_k1/rrd_export.py b/src/k1link/device_plugins/xgrids_k1/rrd_export.py index 17dbe6f..97afdb9 100644 --- a/src/k1link/device_plugins/xgrids_k1/rrd_export.py +++ b/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, diff --git a/src/k1link/web/session_api.py b/src/k1link/web/session_api.py index e5b765b..574fde9 100644 --- a/src/k1link/web/session_api.py +++ b/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, diff --git a/tests/test_e10_integrated_perception.py b/tests/test_e10_integrated_perception.py index 07420a6..357a800 100644 --- a/tests/test_e10_integrated_perception.py +++ b/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: diff --git a/tests/test_rrd_export.py b/tests/test_rrd_export.py index a5f1ffe..aa4cc0d 100644 --- a/tests/test_rrd_export.py +++ b/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 diff --git a/tests/test_session_api.py b/tests/test_session_api.py index ae3ab14..2daa0fa 100644 --- a/tests/test_session_api.py +++ b/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(