from __future__ import annotations import copy import json from pathlib import Path from typing import Any import pytest from k1link.compute.l3_pointpillars_admission import ( L3PointPillarsAdmissionError, build_l3_pointpillars_admission, read_l3_pointpillars_admission, ) SHA = "a" * 64 TRITON_SHA = "58df7489c3f2276f9591d500a012dee03e23d35543ce3c390b4c001e6bf90794" def _profile() -> dict[str, Any]: return { "schema_version": "missioncore.l3-pointpillars-benchmark-profile/v1", "profile_id": "l3-pointpillars-public-transfer-probe-v1", "detector": { "family": "nvidia-tao-pointpillars", "upstream_model_id": "nvidia/tao/pointpillarnet", "candidate_frozen": False, "candidate_model_version": None, "candidate_source_sha256": None, "candidate_label_sha256": None, "triton_model_name": "pointpillars", "required_source_format": "onnx", "input_representation": "native-sensor-scan", "input_coordinate_frame": "sensor/lidar", "input_fields": ["x", "y", "z", "intensity"], "batch_size": 1, "maximum_points": 204_800, "point_cloud_range": [ -51.20000076293945, -51.20000076293945, -1.399999976158142, 51.20000076293945, 51.20000076293945, 4.400000095367432, ], "training_domain": "proprietary-solid-state-lidar", "training_ground_truth_publicly_reproducible": False, "model_classes": ["Vehicle", "Pedestrian", "Cyclist"], "onnx_contract_sha256": ( "2fd29cd054ab058c2cfec3dfba305c71e123ef3f04b457d0c64de0c8dac2e1be" ), "postprocessing": { "reference_repository": ( "https://github.com/NVIDIA-AI-IOT/tao_toolkit_recipes" ), "reference_commit": "a540badc47812a17a94e924b537d49ad3969b5a8", "output_row_fields": [ "x", "y", "z", "length", "width", "height", "yaw", "class_id", "score", ], "class_agnostic_nms": True, "nms_iou_threshold": 0.01, "pre_nms_top_n": 4096, "embedded_score_threshold": 0.1, "embedded_contract_source": "onnx-node-attributes", }, }, "runtime_policy": { "existing_triton_only": True, "second_serving_stack_allowed": False, "engine_built_on_target_required": True, "precision": "strongly-typed", "triton_image": "nvcr.io/nvidia/tritonserver:26.06-py3", "triton_image_digest": TRITON_SHA, }, "public_cross_domain_probe": { "required_split": "validation", "required_ground_truth": "oriented-3d-boxes", "independent_ground_truth_required": True, "benchmark_classes": ["Car", "Pedestrian", "Cyclist"], "model_to_benchmark_class_mapping": { "Vehicle": "Car", "Pedestrian": "Pedestrian", "Cyclist": "Cyclist", }, "metrics": [ "bev-map", "3d-map", "center-error-m", "range-error-m", "yaw-error-rad", "distance-bucket-recall", "false-occupied-rate", "end-to-end-latency-ms", ], "metric_contract": { "official_kitti_server_metric": False, "evaluation_kind": "public-cross-domain-transfer-probe", "ap_interpolation": "40-point", "difficulty_filtering": False, "predictions_outside_shared_range_ignored": True, "iou_thresholds": { "Car": 0.7, "Pedestrian": 0.5, "Cyclist": 0.5, }, "distance_buckets_m": [[0, 20], [20, 40], [40, 70]], }, "native_accuracy_claim_allowed": False, "retuning_allowed": False, }, "k1_transfer_stability": { "requires_completed_public_cross_domain_probe": True, "metrics": [ "input-admission-rate", "output-schema-valid-rate", "deterministic-replay-rate", "end-to-end-latency-ms", "queue-wait-ms", "drop-rate", ], "accuracy_claim_allowed": False, "retuning_allowed": False, }, "authority": { "shadow_only": True, "commands_enabled": False, "navigation_or_safety_accepted": False, }, } def _semantic_dataset(dataset_id: str = "goose-3d/v2025-08-22") -> dict[str, Any]: return { "dataset_id": dataset_id, "installed": True, "release_identity_sha256": SHA, "license": "CC-BY-SA-4.0", "point_fields": ["x", "y", "z", "intensity"], "annotations": ["point-semantic-labels", "point-instance-labels"], "splits": ["validation"], "independent_ground_truth": True, } def _box_dataset(*, installed: bool = True) -> dict[str, Any]: return { "dataset_id": "kitti-3d-object-detection/v1", "installed": installed, "release_identity_sha256": SHA if installed else None, "license": "CC-BY-NC-SA-3.0", "point_fields": ["x", "y", "z", "intensity"], "annotations": ["oriented-3d-boxes"], "splits": ["validation"], "independent_ground_truth": True, } def _worker(*, pointpillars: bool = False) -> dict[str, Any]: models: list[dict[str, Any]] = [ { "name": "yolox_s", "backend": "onnxruntime", "artifact_sha256": SHA, } ] if pointpillars: models.append( { "name": "pointpillars", "upstream_version": "tao-6.26.03-test", "source_model_sha256": "c" * 64, "source_label_sha256": "e" * 64, "source_format": "onnx", "backend": "tensorrt", "precision": "strongly-typed", "artifact_sha256": "b" * 64, "engine_built_on_target": True, "provenance_verified": True, "input_fields": ["x", "y", "z", "intensity"], "maximum_points": 204_800, "point_cloud_range": [ -51.20000076293945, -51.20000076293945, -1.399999976158142, 51.20000076293945, 51.20000076293945, 4.400000095367432, ], "model_classes": ["Vehicle", "Pedestrian", "Cyclist"], "outputs": [ { "name": "output_boxes", "dtype": "FP32", "shape": [1, 393_216, 9], }, {"name": "num_boxes", "dtype": "INT32", "shape": [1]}, ], "representation_smoke": { "status": "engine-executed", "input_artifact_sha256": "d" * 64, "input_point_count": 169_883, "single_query_gpu_compute_ms": 55.0, "accuracy_evaluated": False, "navigation_or_safety_accepted": False, }, } ) return { "schema_version": "missioncore.l3-worker-inventory/v1", "host_id": "worker-006", "observed_at_utc": "2026-07-30T20:57:38Z", "serving_stack_count": 1, "staged_models": [], "triton": { "container_name": "ndc-mission-core-triton", "image": "nvcr.io/nvidia/tritonserver:26.06-py3", "image_digest": TRITON_SHA, "healthy": True, "strict_readiness": True, "model_control_mode": "explicit", "model_repository_read_only": True, "models": models, }, "gpu": { "name": "NVIDIA GeForce RTX 4090", "driver_version": "610.47", "memory_total_mib": 24_564, }, } def _write(path: Path, value: object) -> None: path.write_text(json.dumps(value), encoding="utf-8") def _build( tmp_path: Path, *, profile: dict[str, Any] | None = None, datasets: list[dict[str, Any]] | None = None, worker: dict[str, Any] | None = None, ): profile_path = tmp_path / "profile.json" datasets_path = tmp_path / "datasets.json" worker_path = tmp_path / "worker.json" _write(profile_path, profile or _profile()) _write( datasets_path, { "schema_version": "missioncore.l3-lidar-dataset-inventory/v1", "observed_at_utc": "2026-07-30T20:57:38Z", "datasets": datasets or [_semantic_dataset()], }, ) _write(worker_path, worker or _worker()) return build_l3_pointpillars_admission( profile_path=profile_path, dataset_inventory_path=datasets_path, worker_inventory_path=worker_path, output_root=tmp_path / "results", ) def test_semantic_point_truth_and_missing_model_block_public_probe( tmp_path: Path, ) -> None: result = _build(tmp_path) assert result.report["status"] == "blocked-foundation-assets" assert result.report["blocker_codes"] == [ "public-oriented-3d-box-truth-not-admitted", "pointpillars-compatible-candidate-not-frozen", ] assert result.report["next_gate"] == ( "admit-public-3d-box-split-and-freeze-compatible-pointpillars-candidate" ) assert result.public_transfer_probe_authorized is False finding = result.report["dataset_findings"][0] assert finding["semantic_or_instance_labels_are_not_boxes"] is True assert finding["oriented_3d_box_accuracy_eligible"] is False assert ( result.report["decision"]["semantic_point_labels_substitute_for_3d_boxes"] is False ) assert result.report["decision"]["k1_transfer_stability_authorized"] is False def test_box_truth_and_target_built_model_authorize_public_probe_only( tmp_path: Path, ) -> None: profile = _profile() profile["detector"]["candidate_frozen"] = True profile["detector"]["candidate_model_version"] = "tao-6.26.03-test" profile["detector"]["candidate_source_sha256"] = "c" * 64 profile["detector"]["candidate_label_sha256"] = "e" * 64 result = _build( tmp_path, profile=profile, datasets=[_semantic_dataset(), _box_dataset()], worker=_worker(pointpillars=True), ) assert result.report["status"] == "ready-for-public-cross-domain-probe" assert result.report["blocker_codes"] == [] assert result.report["eligible_public_probe_dataset_ids"] == [ "kitti-3d-object-detection/v1" ] assert result.public_transfer_probe_authorized is True assert result.report["decision"] == { "public_cross_domain_probe_authorized": True, "native_model_accuracy_claim_authorized": False, "k1_transfer_stability_authorized": False, "k1_transfer_quality_claim_authorized": False, "semantic_point_labels_substitute_for_3d_boxes": False, "fine_tuning_allowed": False, "second_serving_stack_allowed": False, "lab_publication_allowed": False, "centerpoint_comparison_allowed": False, } assert result.report["next_gate"] == ( "run-public-cross-domain-pointpillars-probe" ) def test_pointpillars_without_target_engine_provenance_remains_blocked( tmp_path: Path, ) -> None: profile = _profile() profile["detector"]["candidate_frozen"] = True profile["detector"]["candidate_model_version"] = "tao-6.26.03-test" profile["detector"]["candidate_source_sha256"] = "c" * 64 profile["detector"]["candidate_label_sha256"] = "e" * 64 worker = _worker(pointpillars=True) worker["triton"]["models"][1]["engine_built_on_target"] = False result = _build( tmp_path, profile=profile, datasets=[_box_dataset()], worker=worker, ) assert result.report["blocker_codes"] == [ "pointpillars-target-engine-or-provenance-not-verified" ] assert result.report["next_gate"] == "verify-target-engine-and-model-provenance" def test_verified_staged_engine_is_distinct_from_live_install(tmp_path: Path) -> None: profile = _profile() profile["detector"]["candidate_frozen"] = True profile["detector"]["candidate_model_version"] = "tao-6.26.03-test" profile["detector"]["candidate_source_sha256"] = "c" * 64 profile["detector"]["candidate_label_sha256"] = "e" * 64 worker = _worker(pointpillars=True) staged = worker["triton"]["models"].pop() worker["staged_models"] = [staged] result = _build( tmp_path, profile=profile, datasets=[_box_dataset(installed=False)], worker=worker, ) assert result.report["blocker_codes"] == [ "public-oriented-3d-box-truth-not-admitted", "pointpillars-model-not-installed-live", ] assert result.report["detector"]["staged_target_engine_ready"] is True assert result.report["detector"]["model_ready"] is False assert result.report["next_gate"] == ( "admit-public-3d-box-split-then-install-staged-pointpillars-model" ) def test_second_serving_stack_is_rejected(tmp_path: Path) -> None: profile = _profile() profile["detector"]["candidate_frozen"] = True profile["detector"]["candidate_model_version"] = "tao-6.26.03-test" profile["detector"]["candidate_source_sha256"] = "c" * 64 profile["detector"]["candidate_label_sha256"] = "e" * 64 worker = _worker(pointpillars=True) worker["serving_stack_count"] = 2 result = _build( tmp_path, profile=profile, datasets=[_box_dataset()], worker=worker, ) assert result.report["blocker_codes"] == [ "canonical-triton-runtime-policy-not-satisfied" ] assert result.report["runtime_checks"]["second_serving_stack_absent"] is False assert result.report["decision"]["second_serving_stack_allowed"] is False def test_profile_cannot_skip_the_public_transfer_probe(tmp_path: Path) -> None: profile = copy.deepcopy(_profile()) profile["k1_transfer_stability"][ "requires_completed_public_cross_domain_probe" ] = False with pytest.raises(L3PointPillarsAdmissionError): _build(tmp_path, profile=profile) def test_result_is_content_addressed_and_detects_tampering(tmp_path: Path) -> None: first = _build(tmp_path) second = _build(tmp_path) assert first.result_id == second.result_id report_path = first.result_root / "admission-report.json" report = json.loads(report_path.read_text(encoding="utf-8")) report["status"] = "changed" _write(report_path, report) with pytest.raises(L3PointPillarsAdmissionError): read_l3_pointpillars_admission(first.result_root)