from __future__ import annotations import copy import hashlib import json from pathlib import Path from types import SimpleNamespace from typing import Any import pytest import k1link.laboratory.m48_object_quality as m48 from k1link.laboratory.evidence_registry import LaboratoryEvidenceRegistry from k1link.laboratory.evidence_report import verify_laboratory_evidence_result def _write_json(path: Path, value: object) -> None: path.write_text(json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n") def _frame_catalog() -> list[dict[str, object]]: return [ { "sequence": sequence, "source_time_ns": (sequence - 1) * 100_000_000, "camera_fragment_sha256": hashlib.sha256(f"frame-{sequence}".encode()).hexdigest(), } for sequence in range(1, 4490) ] def _clips() -> list[dict[str, object]]: rows = [] for index in range(20): start = 1 + index * 100 split = "development" if index < 10 else "validation" split_index = index if index < 10 else index - 10 rows.append( { "clip_id": f"clip-{index:02d}", "component_id": f"component-{split}-{split_index // 2:02d}", "route_block": f"route-{split}-{split_index // 3:02d}", "time_block": f"time-{split}-{split_index // 2:02d}", "split": split, "start_sequence": start, "end_sequence": start + 50, } ) return rows def _clip_fixture_state(clip_id: str) -> dict[str, object]: local_index = int(clip_id.rsplit("-", 1)[1]) % 10 if local_index == 0: return { "extent_xyxy": [0.2, 0.2, 0.22, 0.22], "geometry_association": "unknown", "motion": "unsupported", "threat": "unknown", "unknown_causes": [ "insufficient-geometry-support", "threat-evidence-insufficient", ], } if local_index == 1: return { "extent_xyxy": [0.01, 0.2, 0.2, 0.4], "geometry_association": "associated", "motion": "static", "threat": "not-threat", "unknown_causes": [], } if local_index == 2: return { "extent_xyxy": [0.1, 0.1, 0.3, 0.4], "geometry_association": "associated", "motion": "moving", "threat": "threat", "unknown_causes": [], } return { "extent_xyxy": [0.1, 0.1, 0.3, 0.4], "geometry_association": "associated", "motion": "static", "threat": "not-threat", "unknown_causes": [], } def _preparation_provenance() -> dict[str, object]: return { "schema_version": m48.M48_PREPARATION_PROVENANCE_SCHEMA, "adapter": { "module": "k1link.laboratory.m48_ravnoves00_pack", "sha256": "1" * 64, }, "selection": { "selection_id": "m48-ravnoves00-balanced-connected-clips/v1", "sha256": "2" * 64, }, "camera_index": { "source_session_id": "20260720T065719Z_viewer_live", "sha256": "3" * 64, "byte_length": 1234, "frame_count": 4489, }, "graph": { "result_id": "m47-reference-graph-" + "4" * 64, "manifest_sha256": "5" * 64, "frames_sha256": "6" * 64, }, "threat": { "result_id": "m4-threat-replay-" + "7" * 64, "manifest_sha256": "8" * 64, "frames_sha256": "9" * 64, }, "geometry": { "result_id": "m4-geometry-replay-" + "a" * 64, "manifest_sha256": "b" * 64, "frames_sha256": "c" * 64, }, } def _prediction_rows( clips: list[dict[str, object]], *, unsafe_free_space: bool, unsafe_free_space_split: str | None = None, ) -> list[dict[str, object]]: rows: list[dict[str, object]] = [] for clip in clips: fixture = _clip_fixture_state(str(clip["clip_id"])) unsafe = unsafe_free_space and ( unsafe_free_space_split is None or clip["split"] == unsafe_free_space_split ) for sequence in range(int(clip["start_sequence"]), int(clip["end_sequence"]) + 1): objects = [ { "prediction_id": f"prediction-{sequence}", "extent_xyxy": fixture["extent_xyxy"], "geometry_association": fixture["geometry_association"], "freshness": "current", "motion": fixture["motion"], "threat": fixture["threat"], "unknown_causes": fixture["unknown_causes"], } ] rows.append( { "sequence": sequence, "source_time_ns": (sequence - 1) * 100_000_000, "terminal_outcome": "delivered", "terminal_reason": None, "free_space_claimed": unsafe, "objects": objects, } ) return rows def _fake_m47(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: root = tmp_path / f"m47-reference-graph-lab-{'a' * 64}" root.mkdir() manifest = { "schema_version": "missioncore.reference-perception-graph-lab/v2", "accepted": True, "ground_truth": False, } _write_json(root / "manifest.json", manifest) report = { "source": { "source_id": "RAVNOVES00", "source_session_id": "20260720T065719Z_viewer_live", "graph_result_id": "m47-reference-graph-" + "b" * 64, }, "method": { "graph_id": "reference-perception-graph/v2", "run_mode": "lossless-replay", "canonical_payload_sha256": "c" * 64, }, "decision": { "state": "accepted-reference-graph-replay", "next_gate": "independent-object-centric-detection-quality", }, "acceptance": {"accepted": True}, "authority": { "mode": "replay-simulated", "physical_live": False, "commands_enabled": False, "actuation_allowed": False, "navigation_or_safety_accepted": False, "ground_truth": False, }, } monkeypatch.setattr( m48, "read_m47_reference_graph_lab", lambda _: SimpleNamespace( result_id=root.name, result_root=root, manifest=manifest, report=report, ), ) def _review_clips(clips: tuple[dict[str, Any], ...], *, state: str) -> list[dict[str, Any]]: rows = [] for clip in clips: fixture = _clip_fixture_state(str(clip["clip_id"])) start = int(clip["start_sequence"]) end = int(clip["end_sequence"]) rows.append( { "clip_id": clip["clip_id"], "start_sequence": start, "end_sequence": end, "review_state": state, "no_object": False, "tracklets": [ { "object_id": "object-1", "first_sequence": start, "last_sequence": end, "keyframes": [ { "sequence": start, "extent_xyxy": fixture["extent_xyxy"], "visibility": "visible", }, { "sequence": end, "extent_xyxy": fixture["extent_xyxy"], "visibility": "partial", }, ], "state_segments": [ { "start_sequence": start, "end_sequence": end, "geometry_association": fixture["geometry_association"], "freshness": "current", "motion": fixture["motion"], "threat": fixture["threat"], "critical_corridor_obstacle": True, } ], "notes": None, } ], "notes": None, } ) return rows def _review(pack: m48.M48ObjectQualityPack, reviewer_id: str) -> dict[str, Any]: return { "schema_version": m48.M48_REVIEW_SCHEMA, "pack_id": pack.result_id, "state": "completed-independent-no-predictions", "reviewer_id": reviewer_id, "review_round": 1, "blindness": { "candidate_identity_seen": False, "model_predictions_seen": False, "model_scores_seen": False, "semantic_class_task_seen": False, }, "clips": _review_clips(pack.clips, state="reviewed"), "acceptance": { "all_clips_reviewed": True, "independent": True, "submitted_at_utc": "2026-08-24T11:00:00Z", }, } def _build_generations( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, *, unsafe_free_space: bool = False, unsafe_free_space_split: str | None = None, ) -> tuple[m48.M48ObjectQualityPack, m48.M48ObjectTruthSeal]: _fake_m47(tmp_path, monkeypatch) clips = _clips() pack = m48.build_m48_object_quality_pack( m47_lab_root=tmp_path / "ignored-m47", frame_catalog=_frame_catalog(), clips=clips, predictions=_prediction_rows( clips, unsafe_free_space=unsafe_free_space, unsafe_free_space_split=unsafe_free_space_split, ), preparation_provenance=_preparation_provenance(), frozen_at_utc="2026-08-24T10:00:00Z", output_root=tmp_path / "packs", ) review_a = _review(pack, "reviewer-a") review_b = _review(pack, "reviewer-b") review_a_path = tmp_path / "review-a.json" review_b_path = tmp_path / "review-b.json" _write_json(review_a_path, review_a) _write_json(review_b_path, review_b) adjudication = { "schema_version": m48.M48_ADJUDICATION_SCHEMA, "pack_id": pack.result_id, "state": "completed-adjudicated", "adjudicator_id": "adjudicator-1", "review_submission_sha256": sorted( (m48._canonical_sha256(review_a), m48._canonical_sha256(review_b)) ), "clips": _review_clips(pack.clips, state="adjudicated"), "acceptance": { "all_clips_adjudicated": True, "all_disagreements_resolved": True, "sealed_at_utc": "2026-08-24T12:00:00Z", }, } adjudication_path = tmp_path / "adjudication.json" _write_json(adjudication_path, adjudication) truth = m48.build_m48_object_truth_seal( pack_root=pack.result_root, reviewer_a_path=review_a_path, reviewer_b_path=review_b_path, adjudication_path=adjudication_path, output_root=tmp_path / "truth", ) return pack, truth def test_m48_pack_is_neutral_tracklet_review_evidence( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: pack, truth = _build_generations(tmp_path, monkeypatch) reviewer_package = json.loads( (pack.result_root / "reviewer-package.json").read_text(encoding="utf-8") ) assert reviewer_package["strata_included"] is False assert reviewer_package["frozen_predictions_included"] is False assert all( "strata" not in clip and "selection_hypotheses" not in clip for clip in reviewer_package["clips"] ) assert { hypothesis for clip in pack.clips if clip["split"] == "validation" for hypothesis in clip["selection_hypotheses"] } == set(pack.manifest["identity"]["profile"]["required_validation_hypotheses"]) review_template = json.loads( (pack.result_root / "review-template.json").read_text(encoding="utf-8") ) assert "clips" in review_template and "frames" not in review_template assert "tracklets" in review_template["clips"][0] assert len(truth.truth_rows) == len(pack.frame_references) assert truth.truth_rows[0]["objects"][0]["visibility"] == "visible" assert truth.truth_rows[50]["objects"][0]["visibility"] == "partial" def test_m48_perfect_class_free_result_passes_all_gates_and_registry( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: pack, truth = _build_generations(tmp_path, monkeypatch) result = m48.score_m48_object_quality( pack_root=pack.result_root, truth_seal_root=truth.result_root, output_root=tmp_path / "results", ) assert result.report["acceptance"]["accepted"] is True assert all(result.report["acceptance"]["gates"].values()) assert result.report["acceptance"]["scope"] == "validation-only" assert result.report["metrics"] == result.report["metrics_by_split"]["validation"] assert result.report["method"]["semantic_class_scored"] is False assert result.report["decision"]["next_gate"] == ("m4.9-recorded-realtime-release-candidate") repository_root = Path(__file__).resolve().parents[1] registry = LaboratoryEvidenceRegistry.from_directory(repository_root / "config/laboratories") definitions = {definition.work_id: definition for definition in registry.definitions} pack_proof = verify_laboratory_evidence_result( definitions["m48-object-centric-quality"], pack.result_root ) result_proof = verify_laboratory_evidence_result( definitions["m48-object-centric-quality"], result.result_root ) assert pack_proof["artifact_count"] == 7 assert result_proof["artifact_count"] == 3 def test_m48_review_rejects_semantic_class_and_same_reviewer( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: _fake_m47(tmp_path, monkeypatch) clips = _clips() pack = m48.build_m48_object_quality_pack( m47_lab_root=tmp_path / "ignored-m47", frame_catalog=_frame_catalog(), clips=clips, predictions=_prediction_rows(clips, unsafe_free_space=False), preparation_provenance=_preparation_provenance(), frozen_at_utc="2026-08-24T10:00:00Z", output_root=tmp_path / "packs", ) review_a = _review(pack, "reviewer-a") review_a["clips"][0]["tracklets"][0]["category"] = "car" review_a_path = tmp_path / "review-a.json" _write_json(review_a_path, review_a) with pytest.raises(m48.M48ObjectQualityError, match="fields"): m48.validate_m48_review_submission(pack_root=pack.result_root, review_path=review_a_path) review_a = _review(pack, "reviewer-a") review_b = copy.deepcopy(review_a) review_a_path = tmp_path / "review-a-clean.json" review_b_path = tmp_path / "review-b-same.json" _write_json(review_a_path, review_a) _write_json(review_b_path, review_b) adjudication = { "schema_version": m48.M48_ADJUDICATION_SCHEMA, "pack_id": pack.result_id, "state": "completed-adjudicated", "adjudicator_id": "adjudicator-1", "review_submission_sha256": [m48._canonical_sha256(review_a)] * 2, "clips": _review_clips(pack.clips, state="adjudicated"), "acceptance": { "all_clips_adjudicated": True, "all_disagreements_resolved": True, "sealed_at_utc": "2026-08-24T12:00:00Z", }, } adjudication_path = tmp_path / "adjudication.json" _write_json(adjudication_path, adjudication) with pytest.raises(m48.M48ObjectQualityError, match="must differ"): m48.build_m48_object_truth_seal( pack_root=pack.result_root, reviewer_a_path=review_a_path, reviewer_b_path=review_b_path, adjudication_path=adjudication_path, output_root=tmp_path / "truth", ) def test_m48_unsafe_free_space_fails_with_bounded_atlas( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: pack, truth = _build_generations(tmp_path, monkeypatch, unsafe_free_space=True) result = m48.score_m48_object_quality( pack_root=pack.result_root, truth_seal_root=truth.result_root, output_root=tmp_path / "results", ) assert result.report["acceptance"]["accepted"] is False assert result.report["acceptance"]["gates"]["false_free_space_claims"] is False assert result.report["acceptance"]["gates"]["critical_corridor_obstacle_recall"] is True assert result.failure_atlas assert any("false-free-space-claim" in row["causes"] for row in result.failure_atlas) assert result.report["decision"]["next_gate"] == ( "bounded-cause-remediation-on-failed-m48-clusters" ) def test_m48_development_failures_are_reported_but_cannot_fail_release( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: pack, truth = _build_generations( tmp_path, monkeypatch, unsafe_free_space=True, unsafe_free_space_split="development", ) result = m48.score_m48_object_quality( pack_root=pack.result_root, truth_seal_root=truth.result_root, output_root=tmp_path / "results", ) assert result.report["acceptance"]["accepted"] is True assert result.report["metrics_by_split"]["development"]["false_free_space_claims"] > 0 assert result.report["metrics"]["false_free_space_claims"] == 0 assert all(result.report["acceptance"]["gates"].values()) assert any(row["split"] == "development" for row in result.failure_atlas) def test_m48_rejects_incomplete_clip_contract( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: _fake_m47(tmp_path, monkeypatch) clips = _clips()[:19] with pytest.raises(m48.M48ObjectQualityError, match="20–30"): m48.build_m48_object_quality_pack( m47_lab_root=tmp_path / "ignored-m47", frame_catalog=_frame_catalog(), clips=clips, predictions=_prediction_rows(clips, unsafe_free_space=False), preparation_provenance=_preparation_provenance(), frozen_at_utc="2026-08-24T10:00:00Z", output_root=tmp_path / "packs", ) def test_m48_rejects_cross_split_and_vacuous_grouping( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: _fake_m47(tmp_path, monkeypatch) clips = _clips() clips[10]["route_block"] = clips[0]["route_block"] with pytest.raises(m48.M48ObjectQualityError, match="route_block crosses"): m48.build_m48_object_quality_pack( m47_lab_root=tmp_path / "ignored-m47", frame_catalog=_frame_catalog(), clips=clips, predictions=_prediction_rows(clips, unsafe_free_space=False), preparation_provenance=_preparation_provenance(), frozen_at_utc="2026-08-24T10:00:00Z", output_root=tmp_path / "packs-cross-split", ) clips = _clips() for index, clip in enumerate(clips): clip["component_id"] = f"unique-component-{index:02d}" with pytest.raises(m48.M48ObjectQualityError, match="non-vacuous"): m48.build_m48_object_quality_pack( m47_lab_root=tmp_path / "ignored-m47", frame_catalog=_frame_catalog(), clips=clips, predictions=_prediction_rows(clips, unsafe_free_space=False), preparation_provenance=_preparation_provenance(), frozen_at_utc="2026-08-24T10:00:00Z", output_root=tmp_path / "packs-vacuous", ) def test_m48_profile_config_matches_executable_contract() -> None: repository_root = Path(__file__).resolve().parents[1] document = json.loads( (repository_root / "config/perception/m48-object-quality-v1.json").read_text( encoding="utf-8" ) ) profile = m48.DEFAULT_M48_OBJECT_QUALITY_PROFILE assert document["schema_version"] == m48.M48_PROFILE_SCHEMA assert document["profile_id"] == profile.profile_id assert document["clip_contract"]["minimum_clip_count"] == profile.minimum_clip_count assert document["clip_contract"]["maximum_clip_count"] == profile.maximum_clip_count assert document["review_contract"]["review_unit"] == "clip-local-object-tracklet" assert document["review_contract"]["semantic_class_labels_allowed"] is False assert document["review_contract"]["selection_hypotheses_visible_to_reviewers"] is False assert document["clip_contract"]["release_gate_split"] == "validation" assert document["clip_contract"]["required_validation_hypotheses"] == sorted( m48.DEFAULT_M48_OBJECT_QUALITY_PROFILE.to_dict()["required_validation_hypotheses"] ) assert document["release_thresholds"]["obstacle_presence_precision"] == ( profile.obstacle_presence_precision ) assert document["release_thresholds"]["critical_corridor_obstacle_recall"] == ( profile.critical_corridor_obstacle_recall ) def test_m48_pack_identity_is_stable_across_clip_and_prediction_input_order( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: _fake_m47(tmp_path, monkeypatch) clips = _clips() predictions = _prediction_rows(clips, unsafe_free_space=False) first = m48.build_m48_object_quality_pack( m47_lab_root=tmp_path / "ignored-m47", frame_catalog=_frame_catalog(), clips=clips, predictions=predictions, preparation_provenance=_preparation_provenance(), frozen_at_utc="2026-08-24T10:00:00Z", output_root=tmp_path / "packs-a", ) second = m48.build_m48_object_quality_pack( m47_lab_root=tmp_path / "ignored-m47", frame_catalog=_frame_catalog(), clips=reversed(clips), predictions=reversed(predictions), preparation_provenance=_preparation_provenance(), frozen_at_utc="2026-08-24T10:00:00Z", output_root=tmp_path / "packs-b", ) assert first.result_id == second.result_id assert first.manifest["identity_sha256"] == second.manifest["identity_sha256"] changed_provenance = _preparation_provenance() changed_provenance["camera_index"]["sha256"] = "d" * 64 third = m48.build_m48_object_quality_pack( m47_lab_root=tmp_path / "ignored-m47", frame_catalog=_frame_catalog(), clips=clips, predictions=predictions, preparation_provenance=changed_provenance, frozen_at_utc="2026-08-24T10:00:00Z", output_root=tmp_path / "packs-c", ) assert third.result_id != first.result_id assert third.manifest["identity"]["preparation"]["camera_index"]["sha256"] == ("d" * 64)