from __future__ import annotations import hashlib import json from pathlib import Path from typing import Any import pytest from k1link.compute.e30_review_pack import ( E30_REASON_TAXONOMY, E30ReviewPackError, E30ReviewSelectionProfile, build_e30_review_pack, ) from k1link.compute.semantic_geometry_fusion import ( CAMERA_GEOMETRY_FRAME_SCHEMA, CAMERA_GEOMETRY_FUSION_SCHEMA, CAMERA_GEOMETRY_REPORT_SCHEMA, ) def _canonical(value: object) -> bytes: return json.dumps( value, sort_keys=True, separators=(",", ":"), allow_nan=False, ).encode() def _sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _observation(status: str, index: int) -> dict[str, object]: return { "track_id": index, "label": "car" if index % 2 else "person", "association_group": "vehicle" if index % 2 else "person", "geometry_status": status, "geometry_reason": f"reason-{status}", "range_m": None if status in {"single-source-camera", "unknown"} else 2.5 + index, "support": {"connected_occupied_points": 0}, } def _source_result(root: Path) -> Path: result_id = "e29-camera-geometry-" + "a" * 64 result = root / result_id result.mkdir() frames = [ { "schema_version": CAMERA_GEOMETRY_FRAME_SCHEMA, "frame_index": 0, "source_frame_index": 10, "session_seconds": 1.0, "semantic_observations": [ _observation("agree", 1), _observation("single-source-camera", 2), _observation("conflict", 3), _observation("unknown", 4), ], "geometry_only_occupied": [ { "geometry_status": "single-source-geometry", "nearest_range_m": 8.0, "point_count": 10, } ], }, { "schema_version": CAMERA_GEOMETRY_FRAME_SCHEMA, "frame_index": 1, "source_frame_index": 11, "session_seconds": 2.0, "semantic_observations": [ _observation("agree", 5), _observation("single-source-camera", 6), _observation("conflict", 7), _observation("unknown", 8), ], "geometry_only_occupied": [ { "geometry_status": "single-source-geometry", "nearest_range_m": 2.0, "point_count": 12, } ], }, ] frames_path = result / "camera-geometry-frames.jsonl" frames_path.write_bytes(b"".join(_canonical(frame) + b"\n" for frame in frames)) identity: dict[str, Any] = { "frame_count": 2, "timeline_start_seconds": 1.0, "timeline_end_seconds": 2.0, "source_result_id": "e10-integrated-perception-" + "b" * 64, "source_pack_id": "e10-lidar-pack-" + "c" * 64, "local_surface_model_id": "k1-local-surface-" + "d" * 64, "profile": {"profile_id": "camera-first-local-surface-validation/v1"}, "authority": { "commands_enabled": False, "navigation_or_safety_accepted": False, }, } report = { "schema_version": CAMERA_GEOMETRY_REPORT_SCHEMA, "result_id": result_id, "status": "diagnostic-replay-complete", "ground_truth": False, "identity": identity, "metrics": { "semantic_observations": { "geometry_status": { "agree": 2, "single-source-camera": 2, "conflict": 2, "unknown": 2, } }, "geometry_only_occupied": {"cluster_count": 2}, }, "authority": { "commands_enabled": False, "navigation_or_safety_accepted": False, }, } report_path = result / "camera-geometry-report.json" report_path.write_bytes(_canonical(report)) identity_sha256 = hashlib.sha256(_canonical(identity)).hexdigest() result_id = f"e29-camera-geometry-{identity_sha256}" result_with_identity = root / result_id result.rename(result_with_identity) result = result_with_identity frames_path = result / frames_path.name report_path = result / report_path.name report["result_id"] = result_id report_path.write_bytes(_canonical(report)) manifest = { "schema_version": CAMERA_GEOMETRY_FUSION_SCHEMA, "result_id": result_id, "identity_sha256": identity_sha256, "identity": identity, "ground_truth": False, "artifacts": [ { "role": "camera-geometry-frames", "path": frames_path.name, "byte_length": frames_path.stat().st_size, "sha256": _sha256(frames_path), }, { "role": "camera-geometry-report", "path": report_path.name, "byte_length": report_path.stat().st_size, "sha256": _sha256(report_path), }, ], } (result / "manifest.json").write_bytes(_canonical(manifest)) return result def test_review_pack_binds_all_strata_and_keeps_human_decision_open( tmp_path: Path, ) -> None: source = _source_result(tmp_path) profile = E30ReviewSelectionProfile( agree_maximum=1, camera_only_maximum=1, unknown_maximum=1, geometry_only_maximum=1, temporal_bins=2, ) first = build_e30_review_pack( e29_result_root=source, output_root=tmp_path / "review-packs", profile=profile, ) second = build_e30_review_pack( e29_result_root=source, output_root=tmp_path / "review-packs", profile=profile, ) assert first.result_id == second.result_id assert first.manifest["human_review_complete"] is False assert first.manifest["lab_published"] is False assert first.manifest["source_counts"] == { "agree": 2, "camera-only": 2, "conflict": 2, "geometry-only": 2, "unknown": 2, } assert first.manifest["selected_counts"] == { "agree": 1, "camera-only": 1, "conflict": 2, "geometry-only": 1, "unknown": 1, } lines = (first.result_root / "review-items.jsonl").read_text().splitlines() items = [json.loads(line) for line in lines] assert len(items) == 6 assert all(item["review"]["state"] == "unreviewed" for item in items) assert all(item["materialization"]["source_reprojection_required"] for item in items) assert len({item["item_id"] for item in items}) == 6 def test_review_pack_taxonomy_is_fixed_and_source_tampering_rejects( tmp_path: Path, ) -> None: assert E30_REASON_TAXONOMY[0] == "no_lidar_observation" assert E30_REASON_TAXONOMY[-1] == "unknown" assert len(E30_REASON_TAXONOMY) == 19 source = _source_result(tmp_path) frames = source / "camera-geometry-frames.jsonl" frames.write_bytes(frames.read_bytes() + b"\n") with pytest.raises(E30ReviewPackError, match="byte length changed"): build_e30_review_pack( e29_result_root=source, output_root=tmp_path / "review-packs", )