from __future__ import annotations import hashlib import json from pathlib import Path from typing import Any import numpy as np import pytest import k1link.compute.fusion_epoch as fusion_module import k1link.compute.perception_epoch as epoch_module import k1link.compute.results as result_module from k1link.compute import ( RecordedCalibratedFusionStore, RecordedPerceptionEpochStore, RecordedPerceptionOverlayStore, prepare_camera_compute_job, validate_recorded_calibrated_fusion, validate_recorded_perception_epoch_result, validate_recorded_perception_result, ) from k1link.sessions import SessionIntegrityError from k1link.web.camera_archive import CameraArchiveWriter def _canonical_json(value: object) -> bytes: return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ).encode("utf-8") def _job(tmp_path: Path) -> Any: origin_epoch_ns = 1_000_000_000 origin_monotonic_ns = 2_000_000_000 session = tmp_path / "session-1" session.mkdir() def box(box_type: bytes, payload: bytes = b"") -> bytes: return (8 + len(payload)).to_bytes(4, "big") + box_type + payload def full_box(box_type: bytes, payload: bytes = b"", *, flags: int = 0) -> bytes: return box(box_type, bytes([0]) + flags.to_bytes(3, "big") + payload) track_id = 1 tkhd = full_box(b"tkhd", b"\x00" * 8 + track_id.to_bytes(4, "big") + b"\x00" * 4) mdhd = full_box( b"mdhd", b"\x00" * 8 + (1_000).to_bytes(4, "big") + b"\x00" * 4, ) hdlr = full_box(b"hdlr", b"\x00" * 4 + b"vide") trak = box(b"trak", tkhd + box(b"mdia", mdhd + hdlr)) trex = full_box( b"trex", track_id.to_bytes(4, "big") + (1).to_bytes(4, "big") + (500).to_bytes(4, "big") + b"\x00" * 8, ) init = box(b"ftyp", b"isom") + box( b"moov", trak + box(b"mvex", trex) + box(b"avcC", b"\x01\x64\x00\x28"), ) tfhd = full_box(b"tfhd", track_id.to_bytes(4, "big"), flags=0x020000) tfdt = full_box(b"tfdt", (0).to_bytes(4, "big")) trun = full_box(b"trun", (1).to_bytes(4, "big")) fragment = box(b"moof", box(b"traf", tfhd + tfdt + trun)) + box(b"mdat", b"frame") writer = CameraArchiveWriter(session, "sensor.camera.left", 1) writer.append( "init", init, host_epoch_ns=origin_epoch_ns + 100_000_000, host_monotonic_ns=origin_monotonic_ns + 100_000_000, ) writer.append( "media", fragment, host_epoch_ns=origin_epoch_ns + 200_000_000, host_monotonic_ns=origin_monotonic_ns + 200_000_000, ) writer.close() return prepare_camera_compute_job( session_root=session, source_id="sensor.camera.left", codec_epoch=1, origin_epoch_ns=origin_epoch_ns, origin_monotonic_ns=origin_monotonic_ns, output_root=tmp_path / "jobs", ) def _result(tmp_path: Path, job: Any) -> Path: parameters = { "score_threshold": 0.25, "nms_threshold": 0.45, "input_shape": [1, 3, 640, 640], } pipeline = {"id": "recorded-camera-coco-detection", "version": 1} model = { "id": "synthetic", "version": 1, "sha256": "a" * 64, "source_url": "https://example.invalid/model.onnx", "license": "test-only", "classes": "synthetic", } identity = { "schema_version": "missioncore.compute-result-identity/v1", "job_id": job.job_id, "input_sha256": job.input_sha256, "pipeline": pipeline, "model": {key: model[key] for key in ("id", "version", "sha256")}, "parameters": parameters, } identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() result_id = f"result-{identity_sha256}" root = tmp_path / "results" / job.job_id / result_id root.mkdir(parents=True) detections = { "schema_version": "missioncore.object-detections/v1", "result_id": result_id, "job_id": job.job_id, "input_sha256": job.input_sha256, "timestamp_basis": "session-time-seconds", "frames": [ { "frame_index": 0, "epoch_seconds": 0.0, "session_seconds": job.timeline_start_seconds, "detections": [ { "class_id": 0, "class_name": "person", "score": 0.75, "bbox_xyxy": [10.0, 20.0, 30.0, 60.0], } ], } ], } detection_payload = json.dumps(detections, indent=2).encode() + b"\n" (root / "detections.json").write_bytes(detection_payload) result = { "schema_version": "missioncore.compute-result/v1", "result_id": result_id, "identity_sha256": identity_sha256, "job_id": job.job_id, "input_sha256": job.input_sha256, "created_at_utc": "2026-07-19T00:00:00.000Z", "pipeline": pipeline, "runtime": {"kind": "synthetic"}, "model": model, "parameters": parameters, "metrics": {"frames_processed": 1, "detections": 1}, "artifacts": [ { "kind": "object-detections", "path": "detections.json", "byte_length": len(detection_payload), "sha256": hashlib.sha256(detection_payload).hexdigest(), "schema_version": "missioncore.object-detections/v1", } ], "warnings": [], } (root / "result.json").write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8") return root def _epoch_result(tmp_path: Path, job: Any) -> Path: identity = { "schema_version": "missioncore.recorded-perception-identity/v2", "job_id": job.job_id, "input_sha256": job.input_sha256, "calibration": { "content_identity_sha256": "b" * 64, "camera_slot": "camera_1", }, "configuration": {"pipeline": "synthetic-panoptic/v1"}, "models": {"instance": {"id": "synthetic"}}, "publication": { "video_encoder": "synthetic", "video_media_type": "video/mp4", "mask_archive_media_type": "application/gzip", }, } identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() result_id = f"result-{identity_sha256}" root = tmp_path / "results" / job.job_id / result_id root.mkdir(parents=True) frame = { "schema_version": "missioncore.panoptic-frame/v1", "frame_index": 0, "sequence": 1, "session_seconds": job.timeline_start_seconds, "instances": [], "semantic_classes": [], } (root / "frames.jsonl").write_text( json.dumps(frame, separators=(",", ":")) + "\n", encoding="utf-8", ) (root / "perception.mp4").write_bytes(b"synthetic-mp4") (root / "masks.tar.gz").write_bytes(b"synthetic-masks") (root / "gpu-telemetry.jsonl").write_text("{}\n", encoding="utf-8") report = { "schema_version": "missioncore.perception-run-report/v1", "state": "published", "result_id": result_id, "input": {"job_id": job.job_id, "input_sha256": job.input_sha256}, "metrics": { "frames_expected": 1, "frames_processed": 1, "frames_failed": 0, "frames_skipped": 0, }, } (root / "run-report.json").write_text(json.dumps(report) + "\n", encoding="utf-8") artifact_contracts = ( ("panoptic-overlay-video", "perception.mp4", "video/mp4", None), ("panoptic-mask-archive", "masks.tar.gz", "application/gzip", None), ( "panoptic-frame-metadata", "frames.jsonl", "application/x-ndjson", "missioncore.panoptic-frame/v1", ), ("worker-gpu-telemetry", "gpu-telemetry.jsonl", "application/x-ndjson", None), ( "perception-run-report", "run-report.json", "application/json", "missioncore.perception-run-report/v1", ), ) artifacts = [] for kind, name, media_type, schema_version in artifact_contracts: path = root / name descriptor = { "kind": kind, "path": name, "media_type": media_type, "byte_length": path.stat().st_size, "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), } if schema_version is not None: descriptor["schema_version"] = schema_version artifacts.append(descriptor) result = { "schema_version": "missioncore.recorded-perception-result/v2", "result_id": result_id, "identity_sha256": identity_sha256, "identity": identity, "created_at_utc": "2026-07-20T00:00:00.000Z", "job_id": job.job_id, "input_sha256": job.input_sha256, "session_id": job.session_id, "source_id": job.source_id, "codec_epoch": job.codec_epoch, "timestamp_basis": "session-time-seconds", "timeline_start_seconds": job.timeline_start_seconds, "timeline_end_seconds": job.timeline_end_seconds, "frames_processed": job.segment_count, "artifacts": artifacts, } (root / "result.json").write_text(json.dumps(result) + "\n", encoding="utf-8") return root def _fusion_result(tmp_path: Path, job: Any, perception_root: Path) -> Path: perception_result_id = perception_root.name identity = { "schema_version": "missioncore.recorded-calibrated-fusion-identity/v1", "job_id": job.job_id, "input_sha256": job.input_sha256, "perception_result_id": perception_result_id, "calibration_sha256": "b" * 64, "camera_slot": "camera_1", "configuration": {"pipeline": "synthetic-fusion/v1"}, } identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() fusion_id = f"fusion-{identity_sha256}" root = tmp_path / "fusions" / job.job_id / fusion_id root.mkdir(parents=True) np.savez_compressed( root / "fusion.npz", frame_times_ns=np.asarray([0], dtype=np.int64), point_offsets=np.asarray([0, 2], dtype=np.int64), points=np.asarray([[1, 2, 3], [4, 5, 6]], dtype=np.float32), point_colors=np.asarray([[1, 2, 3], [4, 5, 6]], dtype=np.uint8), box_offsets=np.asarray([0, 1], dtype=np.int64), box_centers=np.asarray([[1, 2, 3]], dtype=np.float32), box_half_sizes=np.asarray([[0.5, 0.5, 0.5]], dtype=np.float32), box_colors=np.asarray([[1, 2, 3, 96]], dtype=np.uint8), ) (root / "box-labels.json").write_text('["person · 2.0 m · 5 pts"]', encoding="utf-8") (root / "fusion-frames.jsonl").write_text( '{"frame_index":0,"state":"fused"}\n', encoding="utf-8", ) artifacts = [ { "name": path.name, "byte_length": path.stat().st_size, "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), } for path in ( root / "fusion.npz", root / "box-labels.json", root / "fusion-frames.jsonl", ) ] manifest = { "schema_version": "missioncore.recorded-calibrated-fusion/v1", "fusion_id": fusion_id, "identity_sha256": identity_sha256, "identity": identity, "created_at_utc": "2026-07-20T01:00:00.000Z", "session_id": job.session_id, "source_id": job.source_id, "frame_count": job.segment_count, "timeline_start_seconds": job.timeline_start_seconds, "timeline_end_seconds": job.timeline_end_seconds, "artifacts": artifacts, } (root / "manifest.json").write_text(json.dumps(manifest) + "\n", encoding="utf-8") return root def test_result_is_bound_to_job_and_cache_reuses_complete_overlay( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: job = _job(tmp_path) result_root = _result(tmp_path, job) result = validate_recorded_perception_result(job.job_root, result_root) assert result.job.session_id == "session-1" assert result.frames[0].detections[0].class_name == "person" calls = 0 def render(*_: object, **__: object) -> bytes: nonlocal calls calls += 1 return b"RRF2synthetic-overlay" monkeypatch.setattr(result_module, "_render_overlay", render) store = RecordedPerceptionOverlayStore( jobs_root=tmp_path / "jobs", results_root=tmp_path / "results", cache_root=tmp_path / "cache", ffmpeg_path=Path(__file__), ffprobe_path=Path(__file__), ) first = store.render( "session-1", application_id="nodedc_mission_core_recorded", recording_id="recording-1", ) second = store.render( "session-1", application_id="nodedc_mission_core_recorded", recording_id="recording-1", ) assert first == second == b"RRF2synthetic-overlay" assert calls == 1 def test_result_rejects_changed_detection_payload(tmp_path: Path) -> None: job = _job(tmp_path) result_root = _result(tmp_path, job) detection_path = result_root / "detections.json" detection_path.write_bytes(detection_path.read_bytes().replace(b"0.75", b"0.76")) with pytest.raises(SessionIntegrityError, match="artifact identity changed"): validate_recorded_perception_result(job.job_root, result_root) def test_full_epoch_result_publishes_native_panoptic_video( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: job = _job(tmp_path) result_root = _epoch_result(tmp_path, job) result = validate_recorded_perception_epoch_result(job.job_root, result_root) assert result.calibration_slot == "camera_1" assert result.artifact("panoptic-overlay-video").byte_length == len(b"synthetic-mp4") monkeypatch.setattr( epoch_module.RecordedPerceptionEpochStore, "_probe_video", lambda *_: (0.5, "h264", 800, 600), ) store = RecordedPerceptionEpochStore( jobs_root=tmp_path / "jobs", results_root=tmp_path / "results", ffprobe_path=Path(__file__), ) video = store.video("session-1") assert video is not None assert video.result_id == result.result_id assert video.public_source_id == "recorded.perception.left" assert video.timeline_start_seconds == 0 assert video.timeline_end_seconds == 0.5 def test_full_epoch_store_reuses_unchanged_validated_result( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: job = _job(tmp_path) _epoch_result(tmp_path, job) validation_calls = 0 validate = epoch_module.validate_recorded_perception_epoch_result def count_validation(job_root: Path, result_root: Path) -> Any: nonlocal validation_calls validation_calls += 1 return validate(job_root, result_root) monkeypatch.setattr( epoch_module, "validate_recorded_perception_epoch_result", count_validation, ) monkeypatch.setattr( epoch_module.RecordedPerceptionEpochStore, "_probe_video", lambda *_: (0.5, "h264", 800, 600), ) store = RecordedPerceptionEpochStore( jobs_root=tmp_path / "jobs", results_root=tmp_path / "results", ffprobe_path=Path(__file__), ) assert store.video("session-1") is not None assert store.video("session-1") is not None assert validation_calls == 1 def test_full_epoch_result_rejects_changed_frame_metadata(tmp_path: Path) -> None: job = _job(tmp_path) result_root = _epoch_result(tmp_path, job) frame_path = result_root / "frames.jsonl" frame_path.write_text("{}\n", encoding="utf-8") with pytest.raises(SessionIntegrityError, match="artifact identity changed"): validate_recorded_perception_epoch_result(job.job_root, result_root) def test_calibrated_fusion_is_bound_to_full_epoch_and_reuses_render_cache( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: job = _job(tmp_path) perception_root = _epoch_result(tmp_path, job) fusion_root = _fusion_result(tmp_path, job, perception_root) fusion = validate_recorded_calibrated_fusion( job.job_root, tmp_path / "results", fusion_root, ) assert fusion.perception_result_id == perception_root.name assert fusion.frame_count == 1 rendered = fusion_module._render_fusion( fusion, application_id="nodedc_mission_core_recorded", recording_id="recording-render-proof", ) assert rendered.startswith(b"RRF2") calls = 0 def render(*_: object, **__: object) -> bytes: nonlocal calls calls += 1 return b"RRF2synthetic-fusion" monkeypatch.setattr(fusion_module, "_render_fusion", render) store = RecordedCalibratedFusionStore( jobs_root=tmp_path / "jobs", perception_results_root=tmp_path / "results", fusion_results_root=tmp_path / "fusions", cache_root=tmp_path / "fusion-cache", ) first = store.render( "session-1", application_id="nodedc_mission_core_recorded", recording_id="recording-1", ) second = store.render( "session-1", application_id="nodedc_mission_core_recorded", recording_id="recording-1", ) assert first == second == b"RRF2synthetic-fusion" assert calls == 1 def test_calibrated_fusion_rejects_changed_array_artifact(tmp_path: Path) -> None: job = _job(tmp_path) perception_root = _epoch_result(tmp_path, job) fusion_root = _fusion_result(tmp_path, job, perception_root) with (fusion_root / "fusion.npz").open("ab") as stream: stream.write(b"changed") with pytest.raises(SessionIntegrityError, match="artifact identity changed"): validate_recorded_calibrated_fusion( job.job_root, tmp_path / "results", fusion_root, )