from __future__ import annotations import hashlib import json import shutil import zipfile from pathlib import Path from types import SimpleNamespace from fastapi import FastAPI from fastapi.testclient import TestClient import k1link.laboratory.vegetation_policy_review as policy_review_module import k1link.laboratory.vegetation_shadow_lab as vegetation_lab_module from k1link.laboratory import LaboratoryEvidenceRegistry from k1link.laboratory.evidence_report import verify_laboratory_evidence_result from k1link.laboratory.vegetation_policy_review import seal_vegetation_policy_review from k1link.laboratory.vegetation_shadow_lab import seal_vegetation_shadow_lab from k1link.web.vegetation_shadow_lab_api import build_vegetation_shadow_lab_router REPOSITORY_ROOT = Path(__file__).resolve().parents[1] def _sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _worker_result(root: Path, *, candidate: str, mode: str, vegetation_iou: float) -> None: root.mkdir(parents=True) visual_cases = [] for index in range(12): case_id = f"case-{index:02d}" case_root = root / "cases" / case_id case_root.mkdir(parents=True) keys = ["source", "prediction_semantic", "policy_urban", "policy_rural", "policy_offroad"] if mode == "goose": keys.extend(("truth_semantic", "vegetation_material_error")) files = {} for key in keys: path = case_root / f"{key}.png" path.write_bytes(b"\x89PNG\r\n\x1a\n" + f"{candidate}:{mode}:{case_id}:{key}".encode()) files[key] = { "relative_path": path.relative_to(root).as_posix(), "sha256": _sha256(path), } visual_cases.append( { "case_id": case_id, "source_width": 800 if mode == "ravnoves" else 512, "source_height": 600 if mode == "ravnoves" else 512, "center_crop_xyxy": [100, 0, 700, 600] if mode == "ravnoves" else [0, 0, 512, 512], "outside_crop_state": "undefined" if mode == "ravnoves" else "not-applicable", "focus": { "class_name": "high_grass", "label_id": 51, "truth_pixels": 16384, "truth_fraction": 0.0625, "stratum_rank": index + 1, } if mode == "goose" else None, "files": files, } ) payload = { "schema_version": "missioncore.lab-v1-goose-vegetation-run/v1", "result_id": f"lab-v1-{mode}-{candidate}-fixture", "mode": mode, "candidate": { "candidate_key": candidate, "loaded_model_name": "ddrnet_39" if candidate == "ddrnet" else "pp_lite_t_seg", "checkpoint_sha256": ("a" if candidate == "ddrnet" else "b") * 64, }, "metrics": { "mean_iou_percent": 44.0 + vegetation_iou, "published_mean_iou_percent": 46.53 if candidate == "ddrnet" else 45.09, "vegetation_mean_iou": vegetation_iou, }, "timing": { "latency_ms_p95": 20.0 if candidate == "ddrnet" else 15.0, "throughput_fps_from_mean_inference": 55.0, }, "resource": { "peak_reserved_vram_bytes": 2_000_000_000, "gpu_name": "fixture RTX 4090", }, "visual_cases": visual_cases, "authority": { "navigation_accepted": False, "safety_accepted": False, "actuation_accepted": False, "camera_semantics_can_clear_rigid_geometry": False, }, } (root / "result.json").write_text(json.dumps(payload), encoding="utf-8") def _video_worker_result(root: Path) -> None: root.mkdir(parents=True) archive = root / "semantic-masks.zip" mask = b"\x89PNG\r\n\x1a\n" with zipfile.ZipFile(archive, "x", compression=zipfile.ZIP_STORED) as frozen: for sequence in range(4489): frozen.writestr(f"masks/frame-{sequence + 1:06d}.png", mask) taxonomy = { "schema_version": "missioncore.lab-v1-vegetation-taxonomy/v1", "classes": [ { "class_id": class_id, "label": "undefined" if class_id == 0 else f"class-{class_id}", "color_rgb": [class_id, class_id, class_id], "disposition": "undefined" if class_id == 0 else "prediction", } for class_id in range(64) ], } payload = { "schema_version": "missioncore.lab-v1-goose-vegetation-run/v1", "result_id": f"lab-v1-ravnoves-video-ddrnet-{'e' * 64}", "mode": "ravnoves-video", "candidate": {"candidate_key": "ddrnet"}, "source": { "input_count": 4489, "ground_truth_available": False, }, "video_semantics": { "base_m4_result_id": f"m4-threat-replay-{'f' * 64}", "mask_archive": { "path": "semantic-masks.zip", "sha256": _sha256(archive), "byte_length": archive.stat().st_size, "frame_count": 4489, "width": 800, "height": 600, "encoding": "uint8-class-id-png", "media_type": "application/zip", "sequence_binding": "sequence-0-to-masks/frame-000001.png", }, "taxonomy": taxonomy, "aggregate_prediction_pixels": [4489 * 800 * 600, *([0] * 63)], "center_crop_xyxy": [100, 0, 700, 600], "outside_crop_state": "undefined", }, "authority": { "navigation_accepted": False, "safety_accepted": False, "actuation_accepted": False, "camera_semantics_can_clear_rigid_geometry": False, }, } (root / "result.json").write_text(json.dumps(payload), encoding="utf-8") def test_vegetation_shadow_lab_seals_autonomous_visual_evidence( tmp_path: Path, monkeypatch, ) -> None: roots = {} for candidate, vegetation_iou in (("ddrnet", 0.64), ("ppliteseg", 0.61)): for mode in ("goose", "ravnoves"): root = tmp_path / "worker" / f"{candidate}-{mode}" _worker_result(root, candidate=candidate, mode=mode, vegetation_iou=vegetation_iou) roots[(candidate, mode)] = root video_root = tmp_path / "worker" / "ddrnet-ravnoves-video" _video_worker_result(video_root) m47_root = tmp_path / f"m47-reference-graph-lab-{'a' * 64}" m47_root.mkdir() monkeypatch.setattr( vegetation_lab_module, "read_m47_reference_graph_lab", lambda _root: SimpleNamespace( result_id=m47_root.name, report={ "source": {"source_id": "RAVNOVES00"}, "visual_evidence": { "linked_result_id": f"m4-threat-replay-{'f' * 64}", "timeline_frames": 4489, }, }, ), ) result_root = seal_vegetation_shadow_lab( ddrnet_goose_root=roots[("ddrnet", "goose")], ppliteseg_goose_root=roots[("ppliteseg", "goose")], ddrnet_ravnoves_root=roots[("ddrnet", "ravnoves")], ppliteseg_ravnoves_root=roots[("ppliteseg", "ravnoves")], output_root=tmp_path / "results", ddrnet_ravnoves_video_root=video_root, m47_reference_graph_lab_root=m47_root, ) manifest = json.loads((result_root / "result.json").read_text("utf-8")) assert manifest["decision"]["selected_candidate"] == "ddrnet" assert manifest["ground_truth"] is False assert manifest["authority"]["commands_enabled"] is False assert manifest["authority"]["navigation_or_safety_accepted"] is False assert len(manifest["catalogs"]["ravnoves"]) == 0 assert len(manifest["catalogs"]["goose"]) == 12 assert len(manifest["artifacts"]) == 78 assert manifest["route_video"]["frame_count"] == 4489 assert manifest["route_video"]["outside_crop_state"] == "undefined" assert manifest["catalogs"]["goose"][0]["focus"]["class_name"] == "high_grass" assert "ddrnet_error" in manifest["catalogs"]["goose"][0]["assets"] assert "ppliteseg_error" in manifest["catalogs"]["goose"][0]["assets"] assert "all_classes" not in manifest["metrics"]["candidates"]["ddrnet"]["validation_metrics"] assert (result_root / "result.json").stat().st_size <= 64 * 1024 registry = LaboratoryEvidenceRegistry.from_directory(REPOSITORY_ROOT / "config/laboratories") definition = next( row for row in registry.definitions if row.work_id == "lab-v1-vegetation-shadow" ) proof = verify_laboratory_evidence_result(definition, result_root) assert proof["result_id"] == result_root.name assert proof["artifact_count"] == 78 app = FastAPI() app.include_router(build_vegetation_shadow_lab_router(root_provider=lambda: result_root.parent)) client = TestClient(app) response = client.get(f"/api/v1/laboratory/vegetation-shadow/{result_root.name}") assert response.status_code == 200 assert response.json()["access"] == "read-only" asset_path = manifest["catalogs"]["goose"][0]["assets"]["ddrnet_error"]["path"] asset = client.get( f"/api/v1/laboratory/vegetation-shadow/{result_root.name}/assets/{asset_path}" ) assert asset.status_code == 200 assert asset.headers["cache-control"].endswith("immutable") mask = client.get(f"/api/v1/laboratory/vegetation-shadow/{result_root.name}/masks/0") assert mask.status_code == 200 assert mask.content == b"\x89PNG\r\n\x1a\n" assert mask.headers["cache-control"].endswith("immutable") (result_root / asset_path).write_bytes(b"tampered") assert ( client.get(f"/api/v1/laboratory/vegetation-shadow/{result_root.name}").status_code == 503 ) def test_policy_review_reuses_sealed_video_and_links_yolox_tgs( tmp_path: Path, monkeypatch, ) -> None: roots = {} for candidate, vegetation_iou in (("ddrnet", 0.64), ("ppliteseg", 0.61)): for mode in ("goose", "ravnoves"): root = tmp_path / "worker" / f"{candidate}-{mode}" _worker_result(root, candidate=candidate, mode=mode, vegetation_iou=vegetation_iou) roots[(candidate, mode)] = root video_root = tmp_path / "worker" / "ddrnet-ravnoves-video" _video_worker_result(video_root) m47_root = tmp_path / f"m47-reference-graph-lab-{'a' * 64}" m47_root.mkdir() base_m4_result_id = f"m4-threat-replay-{'f' * 64}" monkeypatch.setattr( vegetation_lab_module, "read_m47_reference_graph_lab", lambda _root: SimpleNamespace( result_id=m47_root.name, report={ "source": {"source_id": "RAVNOVES00"}, "visual_evidence": { "linked_result_id": base_m4_result_id, "timeline_frames": 4489, }, }, ), ) base_root = seal_vegetation_shadow_lab( ddrnet_goose_root=roots[("ddrnet", "goose")], ppliteseg_goose_root=roots[("ppliteseg", "goose")], ddrnet_ravnoves_root=roots[("ddrnet", "ravnoves")], ppliteseg_ravnoves_root=roots[("ppliteseg", "ravnoves")], output_root=tmp_path / "results", ddrnet_ravnoves_video_root=video_root, m47_reference_graph_lab_root=m47_root, ) tgs_result_id = f"m49-tgs-full-shadow-{'9' * 64}" monkeypatch.setattr( policy_review_module, "read_m49_tgs_full_shadow", lambda _root: SimpleNamespace( result_id=tgs_result_id, report={ "source": { "source_id": "RAVNOVES00", "linked_visual_result_id": base_m4_result_id, }, "timeline": {"frame_count": 4489}, }, ), ) def fake_policy_archive(**kwargs) -> list[int]: shutil.copyfile(kwargs["source_archive"], kwargs["destination_archive"]) return [4489 * 800 * 600, *([0] * 8)] monkeypatch.setattr(policy_review_module, "build_policy_mask_archive", fake_policy_archive) result_root = seal_vegetation_policy_review( base_lab_root=base_root, mission_policy_path=REPOSITORY_ROOT / "config/perception/lab-v1-vegetation-mission-policy-v1.json", provider_label_map_path=REPOSITORY_ROOT / "config/perception/lab-v1-vegetation-provider-label-map-v1.json", m49_tgs_full_shadow_root=tmp_path / "sealed-tgs", output_root=tmp_path / "results", created_at_utc="2026-08-28T08:00:00+00:00", ) manifest = json.loads((result_root / "result.json").read_text("utf-8")) route = manifest["route_video"] assert route["view_kind"] == "coarse-material-policy-review" assert route["linked_tgs_result_id"] == tgs_result_id assert route["fusion"]["pixel_raster_fusion"] is False assert route["fusion"]["camera_semantic_temporal_filter"] == "none" assert route["taxonomy"]["schema_version"] == ( "missioncore.lab-v1-terrain-policy-taxonomy/v1" ) assert len(route["taxonomy"]["classes"]) == 9 assert len(manifest["artifacts"]) == 79 assert manifest["authority"]["commands_enabled"] is False assert manifest["decision"]["multilayer_policy_review_ready"] is True app = FastAPI() app.include_router(build_vegetation_shadow_lab_router(root_provider=lambda: result_root.parent)) response = TestClient(app).get( f"/api/v1/laboratory/vegetation-shadow/{result_root.name}/masks/0" ) assert response.status_code == 200 assert response.content == b"\x89PNG\r\n\x1a\n"