from __future__ import annotations import hashlib import json from pathlib import Path from typing import Any import pytest import k1link.compute.results as result_module from k1link.compute import ( RecordedPerceptionOverlayStore, prepare_camera_compute_job, 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 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)