From 2ab9e45548b20f3b9ed476220839b9f4afd226f9 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Thu, 30 Jul 2026 00:51:14 +0300 Subject: [PATCH] fix(replay): stream sealed perception overlays --- docs/01_IMPLEMENTATION_PLAN.md | 3 +- ...AVNOVES00_REFERENCE_SOURCE_PRODUCT_PLAN.md | 51 ++++++++-- src/k1link/compute/__init__.py | 2 + src/k1link/compute/fusion_epoch.py | 76 +++++++++++++-- src/k1link/compute/integrated_perception.py | 93 +++++++++++++++---- src/k1link/compute/results.py | 9 ++ src/k1link/web/session_api.py | 35 ++++++- tests/test_compute_results.py | 52 +++++++++++ tests/test_e10_integrated_perception.py | 36 +++++-- tests/test_session_api.py | 67 ++++++++++++- 10 files changed, 380 insertions(+), 44 deletions(-) diff --git a/docs/01_IMPLEMENTATION_PLAN.md b/docs/01_IMPLEMENTATION_PLAN.md index 99fd399..097dadd 100644 --- a/docs/01_IMPLEMENTATION_PLAN.md +++ b/docs/01_IMPLEMENTATION_PLAN.md @@ -17,7 +17,8 @@ Each gate produces evidence and an explicit GO, PAUSE or BLOCKED result. | E31 binding sensitivity | MEASURED — E45 closes accounting for 87/87 accepted correspondences and finds no material monotonic residual association with represented image radius, rig speed or pose age. It does not supply calibration-target truth or outer-fisheye coverage. | | Detector Truth Island | PREPARED FOR HUMAN REVIEW — E46 freezes 32 references with no prelabels, predictions, scores or candidate identity. Immutable handoff `e46-review-handoff-652f7409c8a529946a26264cea006f3afdd714e19529c731e0f66d0f03fd5668` provides two separately assigned, hash-verified CVAT/COCO reviewer packages and a fail-closed E48 converter. Two real independent reviews and adjudication are still required; the handoff is not ground truth. | | Detector candidate comparison | FROZEN BEFORE TRUTH — E47 freezes raw-KB4 and fixed-valid-FOV-fill predictions from the same exact Mask R-CNN checkpoint. No accuracy result or winner exists before the E46 truth seal. | -| Detector truth/evaluation executors | READY, NOT RUN — E48 fail-closed review/adjudication sealing and separate post-seal E49 scoring are implemented. Neither result exists because real independent reviews are absent. | +| Detector truth/evaluation executors | DEFERRED BY OWNER, READY, NOT RUN — E48 fail-closed review/adjudication sealing and separate post-seal E49 scoring are implemented. Neither result exists because real independent reviews are absent. The deferral does not convert E37–E40 into blind truth and does not block R2/R4 operational work. | +| RAVNOVES00 recorded AI publication | GO — after a real backend restart, the admitted E10 overlay `e10-integrated-perception-36964643f1c0727a434671e3cddc2e76c536b9c28b8c56845f18941ae791c39a` recovered from its sealed cache without model execution or source revalidation. Recording `f83e29dc-5d25-42a5-824e-228cdbb83d21` streamed `69,227,327` bytes with verified SHA-256 `c2bb73fca8bb616a3929df236c6dc63d071b6466a6ca57d06487ef18ed985b57`; a warm localhost request completed in `0.391 s`, and its steady-state server RSS delta was `48 KiB`. | | Product interface | DEFERRED — no new windows, page anatomy or design changes are part of this stabilization increment. | The governing decision is diff --git a/docs/20_RAVNOVES00_REFERENCE_SOURCE_PRODUCT_PLAN.md b/docs/20_RAVNOVES00_REFERENCE_SOURCE_PRODUCT_PLAN.md index ba3d44d..379091b 100644 --- a/docs/20_RAVNOVES00_REFERENCE_SOURCE_PRODUCT_PLAN.md +++ b/docs/20_RAVNOVES00_REFERENCE_SOURCE_PRODUCT_PLAN.md @@ -68,7 +68,9 @@ failures remain blocking even when an aggregate percentage passes. | R4 — reference release candidate | Replay the complete RAVNOVES00 source through the accepted pipeline at recorded pace with regression, resource and recovery evidence | Source-scoped release decision for the known-location product envelope | | R5 — later transfer | Under the preregistered E43 protocol, record the same K1 and mount over a new route after explicit operator authorization; freeze predictions before independent labels are revealed | Generalization decision; no retuning before comparison | -R0–R2 and non-UI R4 preparation are the active stabilization path. R3 is +R2 and non-UI R4 preparation are the active stabilization path. The owner has +deferred the two-reviewer E46–E49 blind-acceptance branch without cancelling or +weakening it; no independent accuracy claim exists while it is deferred. R3 is explicitly deferred for separate discussion. R5 has a frozen protocol but no capture or labels yet. @@ -195,6 +197,42 @@ seal and computes the frozen AP/AR, miss, large-box, valid-FOV-centre leakage and temporal class-count metrics. It cannot select a candidate automatically. No E48/E49 result exists until two real reviews and adjudication are supplied. +### Recorded AI publication checkpoint · 2026-07-30 + +The existing accepted E10 generation is now usable as a stable recorded product +layer rather than a replay-time compute job: + +- source session: `20260720T065719Z_viewer_live` (`RAVNOVES00`); +- admitted result: + `e10-integrated-perception-36964643f1c0727a434671e3cddc2e76c536b9c28b8c56845f18941ae791c39a`; +- Rerun recording identity: + `f83e29dc-5d25-42a5-824e-228cdbb83d21`; +- sealed overlay length: `69,227,327` bytes; +- sealed overlay SHA-256: + `c2bb73fca8bb616a3929df236c6dc63d071b6466a6ca57d06487ef18ed985b57`. + +After a real restart of the canonical `127.0.0.1:8000` backend, the server +recovered the cache admission without running the detector, segmentation or +fusion pipeline and without revalidating the heavy source artifacts. The +publication path verifies the bounded sidecar, RRD magic, byte length and +streamed file SHA-256, then returns a file-backed HTTP response. It no longer +copies the complete 69 MB overlay into the Python heap before sending it. + +Acceptance evidence: + +- focused overlay, mux and session-API tests pass; +- the restarted product endpoint returned HTTP `200` with the exact length, + media type and SHA-bound ETag; +- a throttled `8.241 s` transfer remained bounded, and a subsequent warm + `69,227,327`-byte transfer completed in `0.391 s`; +- steady-state backend RSS changed by only `48 KiB` across that warm transfer; +- the canonical service remained on port `8000`; no service listened on + `8765`. + +This closes the availability and recovery part of R4 for the existing admitted +AI overlay. It does not close detector accuracy, blind review, navigation or +safety acceptance. + E43 immutable protocol `e43-future-capture-protocol-28f091b9648daffce988d44c183e21f56d77988061630de934f8003fb13701d8` preregisters the later same-K1/new-route transfer. It requires the same mount, @@ -231,11 +269,10 @@ The current non-UI priority order is: producers to adopt verified resolution before any separately authorized physical reclamation; 5. preserve E43 unchanged until the owner supplies the new capture. -6. complete two independent E46 reviews and adjudication without exposing - model prelabels or E47 predictions; -7. reveal the sealed truth only after the E47 prediction freeze, then calculate - the preregistered detector metrics and select or reject a preprocessing - candidate. +6. keep the E46 handoff, E47 prediction freeze and E48/E49 executors immutable + while the owner-requested review deferral is active; +7. continue R2/R4 deterministic replay, recovery and resource acceptance + without treating operational readiness as independent detector accuracy. The following remain outside the present stabilization increment: @@ -244,6 +281,8 @@ The following remain outside the present stabilization increment: - initiation of a new physical route/camera capture by an agent; - second K1 or changed mount generalization; - open-world and adversarial evaluation. +- two independent E46 reviews, adjudication and E49 candidate scoring until the + owner resumes that branch. The owner may provide a future recording under E43. Until it exists, current work neither depends on it nor simulates its result. The existing E36 audit and diff --git a/src/k1link/compute/__init__.py b/src/k1link/compute/__init__.py index c0645d2..05d1688 100644 --- a/src/k1link/compute/__init__.py +++ b/src/k1link/compute/__init__.py @@ -220,6 +220,7 @@ from .realtime_tracking_qualification import ( from .results import ( DetectionFrame, ObjectDetection, + RecordedPerceptionOverlayArtifact, RecordedPerceptionOverlayError, RecordedPerceptionOverlayStore, RecordedPerceptionResult, @@ -414,6 +415,7 @@ __all__ = [ "lidar_field_review_catalog_item", "DetectionFrame", "ObjectDetection", + "RecordedPerceptionOverlayArtifact", "RecordedPerceptionOverlayError", "RecordedPerceptionOverlayStore", "RecordedPerceptionResult", diff --git a/src/k1link/compute/fusion_epoch.py b/src/k1link/compute/fusion_epoch.py index e35ba36..3f80aaf 100644 --- a/src/k1link/compute/fusion_epoch.py +++ b/src/k1link/compute/fusion_epoch.py @@ -22,7 +22,10 @@ from k1link.sessions import SessionIntegrityError from .jobs import CameraComputeJob, validate_camera_compute_job from .perception_epoch import validate_recorded_perception_epoch_result -from .results import RecordedPerceptionOverlayError +from .results import ( + RecordedPerceptionOverlayArtifact, + RecordedPerceptionOverlayError, +) FUSION_SCHEMA = "missioncore.recorded-calibrated-fusion/v1" FUSION_IDENTITY_SCHEMA = "missioncore.recorded-calibrated-fusion-identity/v1" @@ -188,18 +191,77 @@ class RecordedPerceptionOverlayMux: application_id: str, recording_id: str, ) -> bytes | None: - payload = self.primary.render( + materialized = self.materialize( session_id, application_id=application_id, recording_id=recording_id, ) + if not isinstance(materialized, RecordedPerceptionOverlayArtifact): + return materialized + try: + payload = materialized.path.read_bytes() + except OSError as exc: + raise RecordedPerceptionOverlayError( + "recorded perception cache became unavailable" + ) from exc + if ( + len(payload) != materialized.byte_length + or hashlib.sha256(payload).hexdigest() != materialized.sha256 + ): + raise RecordedPerceptionOverlayError( + "recorded perception cache changed after validation" + ) + return payload + + def materialize( + self, + session_id: str, + *, + application_id: str, + recording_id: str, + ) -> RecordedPerceptionOverlayArtifact | bytes | None: + primary_materializer = getattr(self.primary, "materialize", None) + payload = ( + primary_materializer( + session_id, + application_id=application_id, + recording_id=recording_id, + ) + if callable(primary_materializer) + else self.primary.render( + session_id, + application_id=application_id, + recording_id=recording_id, + ) + ) + if payload is not None and not isinstance( + payload, (bytes, RecordedPerceptionOverlayArtifact) + ): + raise RecordedPerceptionOverlayError( + "recorded perception provider returned an invalid artifact" + ) if payload is not None or self.fallback is None: return payload - return self.fallback.render( - session_id, - application_id=application_id, - recording_id=recording_id, - ) + fallback_materializer = getattr(self.fallback, "materialize", None) + if callable(fallback_materializer): + fallback_payload = fallback_materializer( + session_id, + application_id=application_id, + recording_id=recording_id, + ) + else: + fallback_payload = self.fallback.render( + session_id, + application_id=application_id, + recording_id=recording_id, + ) + if fallback_payload is not None and not isinstance( + fallback_payload, (bytes, RecordedPerceptionOverlayArtifact) + ): + raise RecordedPerceptionOverlayError( + "recorded perception fallback returned an invalid artifact" + ) + return fallback_payload def status(self, session_id: str, *, recording_id: str) -> dict[str, Any]: primary_status = getattr(self.primary, "status", None) diff --git a/src/k1link/compute/integrated_perception.py b/src/k1link/compute/integrated_perception.py index 264119e..fd5b80b 100644 --- a/src/k1link/compute/integrated_perception.py +++ b/src/k1link/compute/integrated_perception.py @@ -26,7 +26,10 @@ from k1link.artifacts import write_json_atomic from k1link.sessions import SessionIntegrityError from .jobs import CameraComputeJob, validate_camera_compute_job -from .results import RecordedPerceptionOverlayError +from .results import ( + RecordedPerceptionOverlayArtifact, + RecordedPerceptionOverlayError, +) RESULT_SCHEMA = "missioncore.e10-integrated-perception-result/v1" IDENTITY_SCHEMA = "missioncore.e10-integrated-perception-identity/v1" @@ -397,6 +400,39 @@ class IntegratedPerceptionOverlayStore: application_id: str, recording_id: str, ) -> bytes | None: + """Compatibility adapter for in-process consumers that require bytes.""" + + artifact = self.materialize( + session_id, + application_id=application_id, + recording_id=recording_id, + ) + if artifact is None: + return None + try: + payload = artifact.path.read_bytes() + except OSError as exc: + raise RecordedPerceptionOverlayError( + "integrated perception cache became unavailable" + ) from exc + if ( + len(payload) != artifact.byte_length + or hashlib.sha256(payload).hexdigest() != artifact.sha256 + ): + raise RecordedPerceptionOverlayError( + "integrated perception cache changed after validation" + ) + return payload + + def materialize( + self, + session_id: str, + *, + application_id: str, + recording_id: str, + ) -> RecordedPerceptionOverlayArtifact | None: + """Return a verified file-backed overlay without copying it into heap.""" + if application_id != "nodedc_mission_core_recorded": raise ValueError("integrated perception application id is invalid") if ( @@ -414,7 +450,7 @@ class IntegratedPerceptionOverlayStore: key, state="ready", phase="ready", - byte_length=len(cached), + byte_length=cached.byte_length, ) return cached self._set_status(key, state="preparing", phase="queued") @@ -431,7 +467,7 @@ class IntegratedPerceptionOverlayStore: ) output = cache / f"{recording_id}.rrd" sidecar = output.with_suffix(".rrd.cache.json") - cached = _read_cache( + cached = _read_cache_artifact( output, sidecar, result_id=result.result_id, @@ -442,7 +478,7 @@ class IntegratedPerceptionOverlayStore: key, state="ready", phase="ready", - byte_length=len(cached), + byte_length=cached.byte_length, ) return cached self._set_status(key, state="preparing", phase="rendering") @@ -483,7 +519,17 @@ class IntegratedPerceptionOverlayStore: phase="ready", byte_length=len(payload), ) - return payload + published = _read_cache_artifact( + output, + sidecar, + result_id=result.result_id, + recording_id=recording_id, + ) + if published is None: + raise RecordedPerceptionOverlayError( + "integrated perception cache publication failed" + ) + return published except BaseException: self._set_status(key, state="error", phase="error") raise @@ -556,7 +602,7 @@ class IntegratedPerceptionOverlayStore: self, session_id: str, recording_id: str, - ) -> bytes | None: + ) -> RecordedPerceptionOverlayArtifact | None: cache_root = _private_directory(self.cache_root) session_cache = _private_child(cache_root, session_id) admission = self._load_admission(session_cache, session_id) @@ -570,15 +616,15 @@ class IntegratedPerceptionOverlayStore: recording_id, ) if recovered is not None: - _, payload = recovered - return payload + _, artifact = recovered + return artifact 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( + return _read_cache_artifact( output, sidecar, result_id=admission.result_id, @@ -631,17 +677,17 @@ class IntegratedPerceptionOverlayStore: session_cache: Path, descriptor: _ResultDescriptor, recording_id: str, - ) -> tuple[_OverlayAdmission, bytes] | None: + ) -> tuple[_OverlayAdmission, RecordedPerceptionOverlayArtifact] | 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( + artifact = _read_cache_artifact( output, sidecar, result_id=descriptor.result_id, recording_id=recording_id, ) - if payload is None: + if artifact is None: return None admission = _OverlayAdmission( session_id=descriptor.session_id, @@ -650,7 +696,7 @@ class IntegratedPerceptionOverlayStore: result_created_at_utc=descriptor.created_at_utc, ) self._write_admission_document(session_cache, admission) - return admission, payload + return admission, artifact def _write_admission(self, result: IntegratedPerceptionResult) -> None: descriptor = _read_result_descriptor(result.result_root) @@ -1183,16 +1229,19 @@ def _read_result_descriptor(result_root: Path) -> _ResultDescriptor: ) -def _read_cache( +def _read_cache_artifact( output: Path, sidecar: Path, *, result_id: str, recording_id: str, -) -> bytes | None: +) -> RecordedPerceptionOverlayArtifact | None: try: value = _read_object(sidecar, sidecar.parent) - payload = output.read_bytes() + metadata = _confined_file(output, sidecar.parent) + with output.open("rb") as stream: + magic = stream.read(4) + digest = _sha256(output) except (OSError, SessionIntegrityError): return None if ( @@ -1200,12 +1249,16 @@ def _read_cache( or value.get("renderer_version") != OVERLAY_RENDERER_VERSION 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() - or not payload.startswith(b"RRF2") + or value.get("byte_length") != metadata.st_size + or value.get("sha256") != digest + or magic != b"RRF2" ): return None - return payload + return RecordedPerceptionOverlayArtifact( + path=output.resolve(strict=True), + byte_length=metadata.st_size, + sha256=digest, + ) def _read_object(path: Path, root: Path) -> dict[str, Any]: diff --git a/src/k1link/compute/results.py b/src/k1link/compute/results.py index b704662..a87db27 100644 --- a/src/k1link/compute/results.py +++ b/src/k1link/compute/results.py @@ -53,6 +53,15 @@ class DetectionFrame: detections: tuple[ObjectDetection, ...] +@dataclass(frozen=True, slots=True) +class RecordedPerceptionOverlayArtifact: + """One validated file-backed RRD ready for bounded HTTP streaming.""" + + path: Path + byte_length: int + sha256: str + + @dataclass(frozen=True, slots=True) class RecordedPerceptionResult: result_id: str diff --git a/src/k1link/web/session_api.py b/src/k1link/web/session_api.py index 574fde9..f43186a 100644 --- a/src/k1link/web/session_api.py +++ b/src/k1link/web/session_api.py @@ -15,7 +15,11 @@ from pydantic import BaseModel, ConfigDict, Field, StrictBool, field_validator, from starlette.concurrency import run_in_threadpool from starlette.types import Receive, Scope, Send -from k1link.compute import RecordedPerceptionOverlayError, RecordedPerceptionVideo +from k1link.compute import ( + RecordedPerceptionOverlayArtifact, + RecordedPerceptionOverlayError, + RecordedPerceptionVideo, +) from k1link.sessions import ( LayoutConflictError, MaterializedRecording, @@ -899,8 +903,9 @@ def build_session_router( False, False, ) + materializer = getattr(perception_overlay_provider, "materialize", None) payload = await run_in_threadpool( - perception_overlay_provider.render, + materializer if callable(materializer) else perception_overlay_provider.render, session_id, application_id=request.application_id, recording_id=request.recording_id, @@ -921,6 +926,32 @@ def build_session_router( ) from exc if payload is None: return Response(status_code=204, headers={"Cache-Control": "no-store"}) + if isinstance(payload, RecordedPerceptionOverlayArtifact): + try: + metadata = payload.path.stat() + except OSError as exc: + raise HTTPException( + status_code=500, + detail="Не удалось подготовить слой распознавания.", + ) from exc + if not metadata.st_size == payload.byte_length: + raise HTTPException( + status_code=500, + detail="Не удалось подготовить слой распознавания.", + ) + return FileResponse( + path=payload.path, + media_type="application/vnd.rerun.rrd", + filename="perception.rrd", + content_disposition_type="inline", + stat_result=metadata, + headers={ + "Cache-Control": "private, no-cache, no-transform", + "Content-Length": str(payload.byte_length), + "ETag": f'"sha256:{payload.sha256}"', + "X-Content-Type-Options": "nosniff", + }, + ) return Response( content=payload, media_type="application/vnd.rerun.rrd", diff --git a/tests/test_compute_results.py b/tests/test_compute_results.py index ca587be..117f4c7 100644 --- a/tests/test_compute_results.py +++ b/tests/test_compute_results.py @@ -14,6 +14,8 @@ import k1link.compute.results as result_module from k1link.compute import ( RecordedCalibratedFusionStore, RecordedPerceptionEpochStore, + RecordedPerceptionOverlayArtifact, + RecordedPerceptionOverlayMux, RecordedPerceptionOverlayStore, prepare_camera_compute_job, validate_recorded_calibrated_fusion, @@ -500,6 +502,56 @@ def test_calibrated_fusion_is_bound_to_full_epoch_and_reuses_render_cache( assert calls == 1 +def test_perception_mux_preserves_file_backed_artifact(tmp_path: Path) -> None: + payload = b"RRF2sealed" + overlay = tmp_path / "sealed.rrd" + overlay.write_bytes(payload) + artifact = RecordedPerceptionOverlayArtifact( + path=overlay, + byte_length=len(payload), + sha256=hashlib.sha256(payload).hexdigest(), + ) + + class Primary: + def materialize( + self, + _session_id: str, + *, + application_id: str, + recording_id: str, + ) -> RecordedPerceptionOverlayArtifact: + assert application_id == "nodedc_mission_core_recorded" + assert recording_id == "recording-1" + return artifact + + def render( + self, + _session_id: str, + *, + application_id: str, + recording_id: str, + ) -> bytes: + raise AssertionError(f"byte rendering must not run: {application_id}/{recording_id}") + + mux = RecordedPerceptionOverlayMux(Primary(), None) + assert ( + mux.materialize( + "session-1", + application_id="nodedc_mission_core_recorded", + recording_id="recording-1", + ) + == artifact + ) + assert ( + mux.render( + "session-1", + application_id="nodedc_mission_core_recorded", + recording_id="recording-1", + ) + == payload + ) + + def test_calibrated_fusion_rejects_changed_array_artifact(tmp_path: Path) -> None: job = _job(tmp_path) perception_root = _epoch_result(tmp_path, job) diff --git a/tests/test_e10_integrated_perception.py b/tests/test_e10_integrated_perception.py index 357a800..7572180 100644 --- a/tests/test_e10_integrated_perception.py +++ b/tests/test_e10_integrated_perception.py @@ -134,14 +134,15 @@ def test_integrated_overlay_recovers_admission_from_sealed_cache_without_revalid ffmpeg_path=tmp_path / "ffmpeg", ) - assert ( - store.render( - "session-1", - application_id="nodedc_mission_core_recorded", - recording_id=recording_id, - ) - == payload + artifact = store.materialize( + "session-1", + application_id="nodedc_mission_core_recorded", + recording_id=recording_id, ) + assert artifact is not None + assert artifact.path == output.resolve() + assert artifact.byte_length == len(payload) + assert artifact.sha256 == hashlib.sha256(payload).hexdigest() 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) == { @@ -150,6 +151,27 @@ def test_integrated_overlay_recovers_admission_from_sealed_cache_without_revalid "elapsed_seconds": pytest.approx(0.0, abs=0.1), "byte_length": len(payload), } + restarted = IntegratedPerceptionOverlayStore( + jobs_root=jobs_root, + results_root=results_root, + lidar_packs_root=lidar_packs_root, + cache_root=cache_root, + ffmpeg_path=tmp_path / "ffmpeg", + ) + restarted_artifact = restarted.materialize( + "session-1", + application_id="nodedc_mission_core_recorded", + recording_id=recording_id, + ) + assert restarted_artifact == artifact + assert ( + restarted.render( + "session-1", + application_id="nodedc_mission_core_recorded", + recording_id=recording_id, + ) + == payload + ) def test_integrated_overlay_serializes_heavy_materialization_across_sessions( diff --git a/tests/test_session_api.py b/tests/test_session_api.py index 2daa0fa..c26a68b 100644 --- a/tests/test_session_api.py +++ b/tests/test_session_api.py @@ -13,10 +13,11 @@ from typing import Any import pytest from fastapi import APIRouter, HTTPException, Request, Response +from fastapi.responses import FileResponse from fastapi.routing import APIRoute import k1link.sessions.media as recorded_media_module -from k1link.compute import RecordedPerceptionVideo +from k1link.compute import RecordedPerceptionOverlayArtifact, RecordedPerceptionVideo from k1link.device_plugins.xgrids_k1 import xgrids_k1_archive_source from k1link.device_plugins.xgrids_k1.mqtt.capture import FRAME_HEADER, RAW_MAGIC from k1link.sessions import ( @@ -1383,6 +1384,70 @@ def test_recorded_perception_endpoint_returns_one_complete_optional_overlay( assert empty.status_code == 204 +def test_recorded_perception_endpoint_streams_validated_file_artifact( + tmp_path: Path, +) -> None: + repository = tmp_path / "repo" + sessions = repository / "sessions" + session = make_legacy_session(sessions, "20260716T205632Z_viewer_live") + store = SessionStore(repository, data_dir=tmp_path / "data") + store.reconcile_archive(xgrids_k1_archive_source(sessions)) + payload = b"RRF2file-backed-perception" + overlay = tmp_path / "sealed-perception.rrd" + overlay.write_bytes(payload) + digest = hashlib.sha256(payload).hexdigest() + calls: list[tuple[str, str, str]] = [] + + class Provider: + def materialize( + self, + session_id: str, + *, + application_id: str, + recording_id: str, + ) -> RecordedPerceptionOverlayArtifact: + calls.append((session_id, application_id, recording_id)) + return RecordedPerceptionOverlayArtifact( + path=overlay, + byte_length=len(payload), + sha256=digest, + ) + + def render( + self, + _session_id: str, + *, + application_id: str, + recording_id: str, + ) -> bytes: + raise AssertionError(f"heap rendering must not run: {application_id}/{recording_id}") + + router = build_session_router(store, perception_overlay_provider=Provider()) + perception_route = endpoint( + router, + "/api/v1/observation-sessions/{session_id}/perception.rrd", + "POST", + ) + response = asyncio.run( + perception_route( + session_id=session.name, + request=RecordedPerceptionRequest( + application_id="nodedc_mission_core_recorded", + recording_id="recording-001", + ), + ) + ) + + assert isinstance(response, FileResponse) + assert Path(response.path) == overlay + assert response.media_type == "application/vnd.rerun.rrd" + assert response.headers["content-length"] == str(len(payload)) + assert response.headers["etag"] == f'"sha256:{digest}"' + assert response.headers["content-disposition"] == 'inline; filename="perception.rrd"' + assert str(overlay) not in str(response.headers) + assert calls == [(session.name, "nodedc_mission_core_recorded", "recording-001")] + + def test_recorded_point_color_endpoint_is_strict_and_forwards_confined_command( tmp_path: Path, ) -> None: