from __future__ import annotations import hashlib import json from dataclasses import dataclass from pathlib import Path from types import SimpleNamespace from typing import cast import numpy as np import pytest from k1link.laboratory import m48_raw_evidence as raw_module from k1link.laboratory.m48_object_quality import M48ObjectQualityPack from k1link.laboratory.m48_raw_evidence import ( M48_EXPECTED_FRAME_COUNT, M48_EXPECTED_SESSION_ID, M48_EXPECTED_SOURCE_ID, M48_RAW_SPATIAL_FRAME_SCHEMA, M48RawEvidenceError, M48RawEvidenceReader, ) from k1link.perception.geometry import RecordedFrameTemporalBinding from k1link.perception.threat import ReplayBodyFrame, load_replay_threat_profile REPOSITORY_ROOT = Path(__file__).resolve().parents[1] PROFILE_PATH = REPOSITORY_ROOT / "config/perception/m4-replay-threat-v3.json" def _canonical_sha256(value: object) -> str: return hashlib.sha256( json.dumps( value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, ).encode() ).hexdigest() def _sealed_e10_pack(tmp_path: Path) -> tuple[Path, str, str]: payload = b"sealed-e10-lidar-pack" artifact_sha256 = hashlib.sha256(payload).hexdigest() identity = { "available_lidar_frames": 3928, "calibration_sha256": "1" * 64, "camera_slot": "camera_1", "e6_profile_sha256": "2" * 64, "e6_result_id": "e6-fixture", "frame_count": M48_EXPECTED_FRAME_COUNT, "input_sha256": "3" * 64, "job_id": "recorded-camera-fixture", "point_count": 5, "producer_sha256": "4" * 64, "projection": { "height": 600, "model": "kb4", "source_coordinates": "k1-map", "target_camera": "sensor.camera.right", "width": 800, }, "schema_version": "missioncore.e10-lidar-replay-pack/v1", "semantic_timeline_result_id": "result-fixture", "session_id": M48_EXPECTED_SESSION_ID, "source_end_frame_index": M48_EXPECTED_FRAME_COUNT - 1, "source_id": "sensor.camera.right", "source_start_frame_index": 0, "temporal_binding": "accepted-e6-nearest-host-arrival-best-effort", "temporal_policy": { "binding": "nearest-host-arrival-best-effort", "clock_source": "recorded-host-monotonic-arrival", "maximum_lidar_camera_delta_ms": 100.0, "maximum_pose_point_delta_ms": 100.0, }, "timeline_end_seconds": 484.0, "timeline_start_seconds": 35.0, } identity_sha256 = _canonical_sha256(identity) pack_id = f"e10-lidar-pack-{identity_sha256}" root = tmp_path / pack_id root.mkdir() (root / "lidar-pack.npz").write_bytes(payload) manifest = { "artifact": { "byte_length": len(payload), "media_type": "application/x-npz", "path": "lidar-pack.npz", "sha256": artifact_sha256, }, "classification": "private-recorded-sensor-replay-input", "created_at_utc": "2026-07-22T06:05:22.515Z", "ground_truth": False, "identity": identity, "identity_sha256": identity_sha256, "pack_id": pack_id, "schema_version": "missioncore.e10-lidar-replay-pack/v1", } (root / "manifest.json").write_text( json.dumps(manifest, sort_keys=True, separators=(",", ":")), encoding="utf-8", ) return root, pack_id, artifact_sha256 def test_e10_pack_validation_binds_identity_path_length_and_sha256(tmp_path: Path) -> None: root, pack_id, artifact_sha256 = _sealed_e10_pack(tmp_path) artifact = raw_module._validate_e10_pack( root, expected_pack_id=pack_id, expected_artifact_sha256=artifact_sha256, ) assert artifact == (root / "lidar-pack.npz").resolve() (root / "lidar-pack.npz").write_bytes(b"tampered") with pytest.raises(M48RawEvidenceError, match="artifact content changed"): raw_module._validate_e10_pack( root, expected_pack_id=pack_id, expected_artifact_sha256=artifact_sha256, ) @pytest.mark.parametrize( ("mutation", "message"), [ (lambda manifest: manifest["identity"].update(session_id="other"), "identity changed"), ( lambda manifest: manifest["artifact"].update(path="../lidar-pack.npz"), "identity changed", ), (lambda manifest: manifest.update(pack_id="e10-lidar-pack-wrong"), "identity changed"), ], ) def test_e10_pack_validation_rejects_manifest_escape( tmp_path: Path, mutation: object, message: str, ) -> None: root, pack_id, artifact_sha256 = _sealed_e10_pack(tmp_path) manifest_path = root / "manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) assert callable(mutation) mutation(manifest) manifest_path.write_text(json.dumps(manifest), encoding="utf-8") with pytest.raises(M48RawEvidenceError, match=message): raw_module._validate_e10_pack( root, expected_pack_id=pack_id, expected_artifact_sha256=artifact_sha256, ) @dataclass class _Store: source_time_ns: int = 2_000_000_000 source_available: bool = True def temporal_binding_for_index(self, frame_index: int) -> RecordedFrameTemporalBinding: return RecordedFrameTemporalBinding( frame_index=frame_index, source_time_ns=self.source_time_ns, source_available=self.source_available, lidar_camera_delta_ms=1.0 if self.source_available else None, pose_point_delta_ms=1.0 if self.source_available else None, ) def current_points_for_frame(self, frame_index: int) -> np.ndarray: del frame_index return np.asarray( [ [1.0, 0.0, 0.0], [2.0, 0.0, 0.0], [3.0, 0.0, 0.0], [4.0, 0.0, 0.0], [5.0, 0.0, 0.0], ], dtype=np.float64, ) @dataclass class _BodyFrames: available: bool = True def body_frame_for_frame(self, frame_id: str) -> ReplayBodyFrame | None: if not self.available: return None return ReplayBodyFrame( frame_id=frame_id, origin_map_xyz_m=(1.0, 0.0, 0.0), basis_map_from_body=((1.0, 0.0, 0.0), (0.0, 1.0, 0.0), (0.0, 0.0, 1.0)), sensor_height_m=1.25, surface_slope_deg=0.0, forward_source="fixture", camera_forward_alignment_deg=0.0, ) class _PredictionTrapPack: result_id = "m48-object-quality-pack-" + "a" * 64 result_root = REPOSITORY_ROOT manifest = { "identity": { "source": { "source_id": M48_EXPECTED_SOURCE_ID, "source_session_id": M48_EXPECTED_SESSION_ID, } } } report: dict[str, object] = {} clips: tuple[dict[str, object], ...] = () frame_references = ( { "clip_id": "clip-01", "sequence": 2, "source_time_ns": 2_000_000_000, }, ) @property def predictions(self) -> object: raise AssertionError("neutral raw reader opened frozen predictions") def _reader( tmp_path: Path, *, store: _Store | None = None, body_frames: _BodyFrames | None = None, ) -> M48RawEvidenceReader: profile = load_replay_threat_profile(PROFILE_PATH) timeline = SimpleNamespace( store=store or _Store(), body_frames=body_frames or _BodyFrames(), profile=profile, ) threat = SimpleNamespace( result_id="m4-threat-replay-fixture", result_root=tmp_path, manifest={"identity": {"frames_sha256": "f" * 64}}, ) return M48RawEvidenceReader( repository_root=tmp_path, threat_result=threat, timeline=timeline, point_limit=2, ) def test_raw_reader_is_one_based_bounded_body_frame_and_prediction_free( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr(raw_module, "_validate_m47_pack_binding", lambda **_: None) reader = _reader(tmp_path) pack = cast(M48ObjectQualityPack, _PredictionTrapPack()) frame = reader(pack, 2) assert set(frame) == { "schema_version", "pack_id", "clip_id", "sequence", "source_time_ns", "source_available", "body_frame_available", "point_cloud_body_xyz_m", "rig", "corridor", "occupied_voxel_size_m", "candidate_identity_included", "graph_boxes_ids_scores_included", "frozen_predictions_included", "strata_included", "authority", } assert frame["schema_version"] == M48_RAW_SPATIAL_FRAME_SCHEMA assert frame["clip_id"] == "clip-01" assert frame["sequence"] == 2 assert frame["source_available"] is True assert frame["body_frame_available"] is True assert frame["point_cloud_body_xyz_m"] == [[0.0, 0.0, 0.0], [3.0, 0.0, 0.0]] assert len(cast(list[object], frame["point_cloud_body_xyz_m"])) <= 2 assert frame["candidate_identity_included"] is False assert frame["graph_boxes_ids_scores_included"] is False assert frame["frozen_predictions_included"] is False assert frame["strata_included"] is False assert "metric_obstacles" not in frame assert "camera_proposals" not in frame assert "decision_counts" not in frame assert "body_frame" not in frame def test_raw_reader_fails_closed_on_clip_or_source_time_escape( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr(raw_module, "_validate_m47_pack_binding", lambda **_: None) pack = cast(M48ObjectQualityPack, _PredictionTrapPack()) reader = _reader(tmp_path) with pytest.raises(M48RawEvidenceError, match="outside the selected neutral clips"): reader(pack, 1) mismatched = _reader(tmp_path, store=_Store(source_time_ns=2_000_000_001)) with pytest.raises(M48RawEvidenceError, match="source time escaped"): mismatched(pack, 2) def test_raw_reader_emits_empty_cloud_when_body_frame_is_unavailable( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr(raw_module, "_validate_m47_pack_binding", lambda **_: None) reader = _reader( tmp_path, store=_Store(source_available=False), body_frames=_BodyFrames(available=False), ) pack = cast(M48ObjectQualityPack, _PredictionTrapPack()) frame = reader.frame(pack=pack, sequence=2) assert frame["source_available"] is False assert frame["body_frame_available"] is False assert frame["point_cloud_body_xyz_m"] == [] assert isinstance(frame["rig"], dict) assert isinstance(frame["corridor"], dict) @pytest.mark.parametrize("point_limit", [0, 4097, True]) def test_raw_reader_rejects_unbounded_point_limits( tmp_path: Path, point_limit: int, ) -> None: profile = load_replay_threat_profile(PROFILE_PATH) timeline = SimpleNamespace(store=_Store(), body_frames=_BodyFrames(), profile=profile) threat = SimpleNamespace(result_id="fixture", result_root=tmp_path, manifest={}) with pytest.raises(M48RawEvidenceError, match="point limit"): M48RawEvidenceReader( repository_root=tmp_path, threat_result=threat, timeline=timeline, point_limit=point_limit, )