"""Full-split, crash-resumable GOOSE ground-provider qualification. The source archive stays on the Simulation Worker D drive. Frames are streamed directly from ZIP members and only bounded evidence is published through the Polygon qualification-run journal. """ from __future__ import annotations import hashlib import json import math import os import tempfile import time import zipfile from concurrent.futures import ProcessPoolExecutor, as_completed from dataclasses import asdict, dataclass from datetime import UTC, datetime from pathlib import Path, PurePosixPath from typing import Any, Final import numpy as np from k1link.datasets.gateway import DatasetPointFrame, decode_semantic_kitti_frame from k1link.datasets.goose_admission import ( GOOSE_ARCHIVE_FILENAME, GOOSE_SOURCE_ID, GooseAdmissionError, read_goose_label_mapping, ) from k1link.datasets.goose_benchmark import _ground_metrics from k1link.datasets.goose_profile import ( DEFAULT_GOOSE_PATCHWORK_PROFILE, GoosePatchworkProfile, ) from k1link.ground_segmentation import ( DEFAULT_GROUND_BENCHMARK_PROFILE, PATCHWORKPP_SOURCE_COMMIT, GroundBenchmarkProfile, GroundSegmentation, GroundSegmenter, LocalPercentileGroundSegmenter, PatchworkPPGroundSegmenter, ) from k1link.simulation.contracts import ( AuthorityProfile, ProviderPin, QualificationArtifact, QualificationRun, ReproducibilityTier, RunKind, RunState, ) from k1link.simulation.run_store import ( QualificationRunConflictError, QualificationRunStore, ) GOOSE_QUALIFICATION_PROFILE_SCHEMA: Final = "missioncore.goose-ground-qualification-profile/v1" GOOSE_QUALIFICATION_FRAME_SCHEMA: Final = "missioncore.goose-ground-qualification-frame/v1" GOOSE_QUALIFICATION_REPORT_SCHEMA: Final = "missioncore.goose-ground-qualification-report/v1" GOOSE_QUALIFICATION_PREVIEW_SCHEMA: Final = ( "missioncore.goose-ground-qualification-failure-preview/v1" ) REPORT_ARTIFACT_KIND: Final = "goose-ground-qualification-report" FAILURE_ARTIFACT_KIND: Final = "goose-ground-qualification-failure-preview" MAX_FAILURE_PREVIEW_POINTS: Final = 12_000 EXPECTED_VALIDATION_FRAMES: Final = 961 DEFAULT_SEED: Final = 42 _WORKER_PATCHWORK: GroundSegmenter | None = None @dataclass(frozen=True, slots=True) class GroundAcceptancePolicy: """Predeclared gates for a shadow-only provider decision.""" minimum_ground_iou_gain: float = 0.07 minimum_natural_ground_recall_gain: float = 0.10 minimum_obstacle_non_ground_recall: float = 0.90 minimum_assigned_fraction: float = 0.999 maximum_patchwork_latency_p95_ms: float = 50.0 maximum_per_frame_iou_regression: float = 0.20 maximum_degraded_iou_loss: float = 0.15 maximum_degraded_natural_recall_loss: float = 0.20 def to_dict(self) -> dict[str, float]: return asdict(self) @dataclass(frozen=True, slots=True) class DegradationProfile: profile_id: str kind: str value: float def to_dict(self) -> dict[str, str | float]: return asdict(self) DEFAULT_DEGRADATIONS: Final[tuple[DegradationProfile, ...]] = ( DegradationProfile("range-20m", "maximum-range-m", 20.0), DegradationProfile("range-40m", "maximum-range-m", 40.0), DegradationProfile("density-50", "density-fraction", 0.50), DegradationProfile("density-25", "density-fraction", 0.25), DegradationProfile("noise-05m", "gaussian-xyz-sigma-m", 0.05), DegradationProfile("dropout-30", "dropout-fraction", 0.30), DegradationProfile("front-180", "horizontal-field-of-view-deg", 180.0), ) DEFAULT_ACCEPTANCE_POLICY: Final = GroundAcceptancePolicy() def qualify_goose_ground( dataset_root: Path, runs_root: Path, *, mission_core_commit: str, parallel_workers: int = 8, expected_frame_count: int = EXPECTED_VALIDATION_FRAMES, current_profile: GroundBenchmarkProfile = DEFAULT_GROUND_BENCHMARK_PROFILE, patchwork_profile: GoosePatchworkProfile = DEFAULT_GOOSE_PATCHWORK_PROFILE, acceptance: GroundAcceptancePolicy = DEFAULT_ACCEPTANCE_POLICY, degradations: tuple[DegradationProfile, ...] = DEFAULT_DEGRADATIONS, seed: int = DEFAULT_SEED, patchwork_module_name: str = "pypatchworkpp", current_segmenter: GroundSegmenter | None = None, patchwork_segmenter: GroundSegmenter | None = None, ) -> dict[str, Any]: """Qualify Current vs Patchwork++ across the immutable validation split. Custom segmenters are intentionally limited to the single-process path so unit tests can exercise the complete orchestration contract without native Patchwork++ bindings. """ root = dataset_root.expanduser().absolute() if not _is_worker_dataset_root(root): raise GooseAdmissionError("GOOSE qualification requires the canonical worker D root") if not 1 <= parallel_workers <= 32: raise GooseAdmissionError("parallel worker count is outside the admitted range") if expected_frame_count < 1: raise GooseAdmissionError("expected frame count must be positive") if (current_segmenter is not None or patchwork_segmenter is not None) and parallel_workers != 1: raise GooseAdmissionError("custom ground providers require one qualification worker") manifest_path = root / "state/goose-3d-v2025-08-22.json" try: manifest = json.loads(manifest_path.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: raise GooseAdmissionError("GOOSE admission manifest is unavailable") from exc archive_sha256 = str(manifest.get("archive", {}).get("sha256", "")) if len(archive_sha256) != 64: raise GooseAdmissionError("GOOSE archive identity is invalid") install_root = root / "goose-3d/v2025-08-22/installs" / archive_sha256 archive_path = root / "goose-3d/v2025-08-22/archives" / GOOSE_ARCHIVE_FILENAME label_mapping = read_goose_label_mapping(install_root / "goose_label_mapping.csv") frame_members = _validation_frame_members(archive_path) if len(frame_members) != expected_frame_count: raise GooseAdmissionError( f"GOOSE validation split contains {len(frame_members)} frames, " f"not the pinned {expected_frame_count}" ) local = current_segmenter or LocalPercentileGroundSegmenter(current_profile) candidate = patchwork_segmenter or PatchworkPPGroundSegmenter.load( patchwork_profile, module_name=patchwork_module_name, ) provider_identities = { "current": dict(local.identity), "patchworkpp": dict(candidate.identity), } profile_document = { "schema_version": GOOSE_QUALIFICATION_PROFILE_SCHEMA, "source_id": GOOSE_SOURCE_ID, "split": "validation", "expected_frame_count": expected_frame_count, "archive_sha256": archive_sha256, "current_profile": current_profile.to_dict(), "patchwork_profile": patchwork_profile.to_dict(), "acceptance": acceptance.to_dict(), "degradations": [profile.to_dict() for profile in degradations], "seed": seed, "reproducibility_tier": "R1", "authority": { "qualification_only": True, "navigation_or_safety_accepted": False, }, } identity_document = { "profile": profile_document, "mission_core_commit": mission_core_commit, "providers": provider_identities, } identity_sha256 = _canonical_sha256(identity_document) run_id = f"goose-ground-{identity_sha256[:20]}" work_root = install_root / "qualifications" / identity_sha256 / "working" frame_cache_root = work_root / "frames" frame_cache_root.mkdir(parents=True, exist_ok=True) run_store = QualificationRunStore(runs_root) run = _admit_or_resume_run( run_store, run_id=run_id, identity_sha256=identity_sha256, profile_document=profile_document, mission_core_commit=mission_core_commit, provider_identities=provider_identities, seed=seed, ) if run.state is RunState.COMPLETED: return _read_completed_report(run_store, run) run = _ensure_running(run_store, run) _append_progress( run_store, run_id, "qualification.profile-admitted", { "identity_sha256": identity_sha256, "frames_total": len(frame_members), "parallel_workers": parallel_workers, "cache_resume": True, }, ) cached: dict[str, dict[str, Any]] = {} pending: list[tuple[str, str, str]] = [] for frame_id, point_member, label_member in frame_members: cache_path = frame_cache_root / f"{frame_id}.json" record = _read_cached_frame(cache_path, identity_sha256) if record is None: pending.append((frame_id, point_member, label_member)) else: cached[frame_id] = record common = ( str(archive_path), label_mapping, current_profile, patchwork_profile, degradations, seed, identity_sha256, patchwork_module_name, ) completed_since_event = 0 if pending and parallel_workers == 1: for frame_id, point_member, label_member in pending: record = _qualify_frame( *common, frame_id, point_member, label_member, local, candidate, ) _write_json_once(frame_cache_root / f"{frame_id}.json", record) cached[frame_id] = record completed_since_event += 1 if completed_since_event >= 25 or len(cached) == len(frame_members): _append_frame_progress(run_store, run_id, len(cached), len(frame_members)) completed_since_event = 0 elif pending: with ProcessPoolExecutor( max_workers=parallel_workers, initializer=_initialize_patchwork_worker, initargs=(patchwork_profile, patchwork_module_name), ) as executor: futures = { executor.submit( _qualify_frame_worker, *common, frame_id, point_member, label_member, ): frame_id for frame_id, point_member, label_member in pending } for future in as_completed(futures): frame_id = futures[future] record = future.result() _write_json_once(frame_cache_root / f"{frame_id}.json", record) cached[frame_id] = record completed_since_event += 1 if completed_since_event >= 25 or len(cached) == len(frame_members): _append_frame_progress(run_store, run_id, len(cached), len(frame_members)) completed_since_event = 0 ordered_frames = [cached[frame_id] for frame_id, _, _ in frame_members] report = _build_report( identity_sha256=identity_sha256, run_id=run_id, profile=profile_document, provider_identities=provider_identities, frames=ordered_frames, acceptance=acceptance, degradations=degradations, ) run_path = runs_root.expanduser().absolute() / run_id evidence_root = run_path / "evidence" evidence_root.mkdir(parents=True, exist_ok=True) report_path = evidence_root / "qualification.json" _write_json_once(report_path, report) _register_file( run_store, run_id, report_path, run_path, artifact_id="qualification-report", kind=REPORT_ARTIFACT_KIND, ) member_lookup = {frame_id: (point, label) for frame_id, point, label in frame_members} for index, failure in enumerate(report["worst_frames"][:5], start=1): frame_id = str(failure["frame_id"]) point_member, label_member = member_lookup[frame_id] preview = _failure_preview( archive_path, frame_id, point_member, label_member, label_mapping, local, candidate, ) preview_path = evidence_root / "failures" / f"{frame_id}.json" _write_json_once(preview_path, preview) _register_file( run_store, run_id, preview_path, run_path, artifact_id=f"failure-preview-{index:02d}", kind=FAILURE_ARTIFACT_KIND, ) _append_progress( run_store, run_id, "qualification.decision-recorded", { "status": report["decision"]["status"], "passed": report["decision"]["passed"], "frames_completed": len(ordered_frames), }, ) run = run_store.load(run_id) run = run_store.transition( run_id, RunState.STOPPING, expected_revision=run.revision, observed_at_utc=_utc_now(), host_monotonic_ns=time.monotonic_ns(), ) run_store.transition( run_id, RunState.COMPLETED, expected_revision=run.revision, observed_at_utc=_utc_now(), host_monotonic_ns=time.monotonic_ns(), reason="qualification-evidence-sealed", ) return report def _qualify_frame_worker( archive_path: str, label_mapping: dict[int, dict[str, Any]], current_profile: GroundBenchmarkProfile, patchwork_profile: GoosePatchworkProfile, degradations: tuple[DegradationProfile, ...], seed: int, identity_sha256: str, patchwork_module_name: str, frame_id: str, point_member: str, label_member: str, ) -> dict[str, Any]: del patchwork_module_name if _WORKER_PATCHWORK is None: raise GooseAdmissionError("Patchwork++ worker was not initialized") return _qualify_frame( archive_path, label_mapping, current_profile, patchwork_profile, degradations, seed, identity_sha256, "pypatchworkpp", frame_id, point_member, label_member, LocalPercentileGroundSegmenter(current_profile), _WORKER_PATCHWORK, ) def _qualify_frame( archive_path: str, label_mapping: dict[int, dict[str, Any]], current_profile: GroundBenchmarkProfile, patchwork_profile: GoosePatchworkProfile, degradations: tuple[DegradationProfile, ...], seed: int, identity_sha256: str, patchwork_module_name: str, frame_id: str, point_member: str, label_member: str, current: GroundSegmenter, patchwork: GroundSegmenter, ) -> dict[str, Any]: del current_profile, patchwork_profile, patchwork_module_name frame = _read_archive_frame(Path(archive_path), point_member, label_member) categories = _challenge_categories(frame, label_mapping) evaluated = categories != 0 ground_truth = (categories == 2) | (categories == 3) xyzi = np.column_stack((frame.points_xyz_m, frame.remission)).astype( np.float32, copy=False, ) current_result = current.segment(xyzi) patchwork_result = patchwork.segment(xyzi) _validate_result(current_result, frame.point_count, "current") _validate_result(patchwork_result, frame.point_count, "patchworkpp") nominal = { "current": _provider_frame_result( current_result, ground_truth, evaluated, categories, frame.point_count, ), "patchworkpp": _provider_frame_result( patchwork_result, ground_truth, evaluated, categories, frame.point_count, ), } degraded: dict[str, Any] = {} for profile in degradations: derived_xyzi, source_indices = _degrade(xyzi, frame_id, profile, seed) result = patchwork.segment(derived_xyzi) _validate_result(result, derived_xyzi.shape[0], profile.profile_id) degraded[profile.profile_id] = _provider_frame_result( result, ground_truth[source_indices], evaluated[source_indices], categories[source_indices], frame.point_count, ) return { "schema_version": GOOSE_QUALIFICATION_FRAME_SCHEMA, "identity_sha256": identity_sha256, "frame_id": frame_id, "source_point_count": frame.point_count, "nominal": nominal, "degradations": degraded, } def _build_report( *, identity_sha256: str, run_id: str, profile: dict[str, Any], provider_identities: dict[str, dict[str, object]], frames: list[dict[str, Any]], acceptance: GroundAcceptancePolicy, degradations: tuple[DegradationProfile, ...], ) -> dict[str, Any]: current = _aggregate([frame["nominal"]["current"] for frame in frames]) patchwork = _aggregate([frame["nominal"]["patchworkpp"] for frame in frames]) degradation_results = { degradation.profile_id: _aggregate( [frame["degradations"][degradation.profile_id] for frame in frames] ) for degradation in degradations } regressions = [ { "frame_id": frame["frame_id"], "current_ground_iou": frame["nominal"]["current"]["metrics"]["ground_iou"], "patchwork_ground_iou": frame["nominal"]["patchworkpp"]["metrics"]["ground_iou"], "ground_iou_delta": ( frame["nominal"]["patchworkpp"]["metrics"]["ground_iou"] - frame["nominal"]["current"]["metrics"]["ground_iou"] ), "patchwork_natural_ground_recall": frame["nominal"]["patchworkpp"]["metrics"][ "natural_ground_recall" ], } for frame in frames ] worst_frames = sorted( regressions, key=lambda item: (item["ground_iou_delta"], item["patchwork_ground_iou"]), )[:20] checks = [ _check( "ground-iou-gain", patchwork["micro"]["ground_iou"] - current["micro"]["ground_iou"], acceptance.minimum_ground_iou_gain, ">=", ), _check( "natural-ground-recall-gain", patchwork["micro"]["natural_ground_recall"] - current["micro"]["natural_ground_recall"], acceptance.minimum_natural_ground_recall_gain, ">=", ), _check( "obstacle-non-ground-recall", patchwork["micro"]["obstacle_non_ground_recall"], acceptance.minimum_obstacle_non_ground_recall, ">=", ), _check( "assigned-fraction", patchwork["assigned_fraction"], acceptance.minimum_assigned_fraction, ">=", ), _check( "latency-p95-ms", patchwork["latency_ms"]["p95"], acceptance.maximum_patchwork_latency_p95_ms, "<=", ), _check( "catastrophic-regression-count", sum( 1 for item in regressions if item["ground_iou_delta"] < -acceptance.maximum_per_frame_iou_regression ), 0, "<=", ), ] degradation_checks: list[dict[str, Any]] = [] for profile_id, aggregate in degradation_results.items(): degradation_checks.extend( ( _check( f"{profile_id}:ground-iou-loss", patchwork["micro"]["ground_iou"] - aggregate["micro"]["ground_iou"], acceptance.maximum_degraded_iou_loss, "<=", ), _check( f"{profile_id}:natural-ground-recall-loss", patchwork["micro"]["natural_ground_recall"] - aggregate["micro"]["natural_ground_recall"], acceptance.maximum_degraded_natural_recall_loss, "<=", ), _check( f"{profile_id}:latency-p95-ms", aggregate["latency_ms"]["p95"], acceptance.maximum_patchwork_latency_p95_ms, "<=", ), ) ) all_checks = checks + degradation_checks passed = all(bool(check["passed"]) for check in all_checks) return { "schema_version": GOOSE_QUALIFICATION_REPORT_SCHEMA, "identity_sha256": identity_sha256, "run_id": run_id, "source_id": GOOSE_SOURCE_ID, "split": "validation", "frame_count": len(frames), "profile": profile, "providers": provider_identities, "aggregates": { "current": current, "patchworkpp": patchwork, }, "degradations": degradation_results, "checks": all_checks, "worst_frames": worst_frames, "frames": frames, "decision": { "status": "shadow-candidate" if passed else "qualification-rejected", "passed": passed, "promoted_to_navigation_or_safety": False, "reason": ( "all predeclared validation and degradation gates passed" if passed else "one or more predeclared gates failed" ), }, "safety": { "qualification_only": True, "actuator_authority": False, "navigation_or_safety_accepted": False, }, } def _aggregate(records: list[dict[str, Any]]) -> dict[str, Any]: count_keys = ( "true_positive", "false_positive", "false_negative", "true_negative", "artificial_ground_true_positive", "artificial_ground_count", "natural_ground_true_positive", "natural_ground_count", "obstacle_non_ground_true_positive", "obstacle_count", ) counts = {key: sum(int(record["counts"][key]) for record in records) for key in count_keys} tp = counts["true_positive"] fp = counts["false_positive"] fn = counts["false_negative"] tn = counts["true_negative"] micro = { "precision": _ratio(tp, tp + fp), "recall": _ratio(tp, tp + fn), "f1": _ratio(2 * tp, 2 * tp + fp + fn), "ground_iou": _ratio(tp, tp + fp + fn), "accuracy": _ratio(tp + tn, tp + tn + fp + fn), "artificial_ground_recall": _ratio( counts["artificial_ground_true_positive"], counts["artificial_ground_count"], ), "natural_ground_recall": _ratio( counts["natural_ground_true_positive"], counts["natural_ground_count"], ), "obstacle_non_ground_recall": _ratio( counts["obstacle_non_ground_true_positive"], counts["obstacle_count"], ), } macro: dict[str, dict[str, float]] = {} for key in ( "ground_iou", "natural_ground_recall", "obstacle_non_ground_recall", ): values = np.asarray([record["metrics"][key] for record in records], dtype=np.float64) macro[key] = { "mean": float(np.mean(values)), "p50": float(np.percentile(values, 50)), "p05": float(np.percentile(values, 5)), "minimum": float(np.min(values)), } latency = np.asarray([record["latency_ms"] for record in records], dtype=np.float64) total_source_points = sum(int(record["source_point_count"]) for record in records) return { "micro": micro, "macro": macro, "latency_ms": { "p50": float(np.percentile(latency, 50)), "p95": float(np.percentile(latency, 95)), "maximum": float(np.max(latency)), }, "assigned_fraction": _ratio( sum(int(record["assigned_point_count"]) for record in records), sum(int(record["retained_point_count"]) for record in records), ), "source_coverage": _ratio( sum(int(record["retained_point_count"]) for record in records), total_source_points, ), "frame_count": len(records), "counts": counts, } def _provider_frame_result( result: GroundSegmentation, ground_truth: np.ndarray[Any, Any], evaluated: np.ndarray[Any, Any], categories: np.ndarray[Any, Any], source_point_count: int, ) -> dict[str, Any]: assigned_evaluated = evaluated & result.assigned_mask metrics = _ground_metrics( result.ground_mask, ground_truth, assigned_evaluated, categories, ) counts = { key: int(metrics[key]) for key in ("true_positive", "false_positive", "false_negative", "true_negative") } counts.update( { "artificial_ground_true_positive": int( np.count_nonzero(result.ground_mask & (categories == 2)) ), "artificial_ground_count": int(np.count_nonzero(categories == 2)), "natural_ground_true_positive": int( np.count_nonzero(result.ground_mask & (categories == 3)) ), "natural_ground_count": int(np.count_nonzero(categories == 3)), "obstacle_non_ground_true_positive": int( np.count_nonzero(~result.ground_mask & (categories == 4)) ), "obstacle_count": int(np.count_nonzero(categories == 4)), } ) return { "metrics": metrics, "counts": counts, "latency_ms": float(result.latency_ms), "source_point_count": source_point_count, "retained_point_count": int(result.ground_mask.shape[0]), "assigned_point_count": int(np.count_nonzero(result.assigned_mask)), "assigned_fraction": float(np.mean(result.assigned_mask)), } def _degrade( xyzi: np.ndarray[Any, Any], frame_id: str, profile: DegradationProfile, seed: int, ) -> tuple[np.ndarray[Any, Any], np.ndarray[Any, Any]]: point_count = xyzi.shape[0] indices = np.arange(point_count, dtype=np.int64) radial = np.linalg.norm(xyzi[:, :3], axis=1) if profile.kind == "maximum-range-m": indices = indices[radial <= profile.value] elif profile.kind in {"density-fraction", "dropout-fraction"}: fraction = profile.value if profile.kind == "density-fraction" else 1.0 - profile.value keep = max(1, int(math.floor(point_count * fraction))) rng = np.random.default_rng(_frame_seed(frame_id, profile.profile_id, seed)) indices = np.sort(rng.choice(indices, size=keep, replace=False)) elif profile.kind == "horizontal-field-of-view-deg": half_angle = math.radians(profile.value / 2) angles = np.arctan2(xyzi[:, 1], xyzi[:, 0]) indices = indices[np.abs(angles) <= half_angle] elif profile.kind != "gaussian-xyz-sigma-m": raise GooseAdmissionError("unknown GOOSE degradation profile") derived = np.ascontiguousarray(xyzi[indices], dtype=np.float32) if profile.kind == "gaussian-xyz-sigma-m": rng = np.random.default_rng(_frame_seed(frame_id, profile.profile_id, seed)) derived[:, :3] += rng.normal(0.0, profile.value, derived[:, :3].shape).astype(np.float32) return derived, indices def _failure_preview( archive_path: Path, frame_id: str, point_member: str, label_member: str, label_mapping: dict[int, dict[str, Any]], current: GroundSegmenter, patchwork: GroundSegmenter, ) -> dict[str, Any]: frame = _read_archive_frame(archive_path, point_member, label_member) categories = _challenge_categories(frame, label_mapping) ground_truth = (categories == 2) | (categories == 3) evaluated = categories != 0 xyzi = np.column_stack((frame.points_xyz_m, frame.remission)).astype( np.float32, copy=False, ) current_result = current.segment(xyzi) patchwork_result = patchwork.segment(xyzi) sample_count = min(frame.point_count, MAX_FAILURE_PREVIEW_POINTS) indices = np.linspace(0, frame.point_count - 1, sample_count, dtype=np.int64) return { "schema_version": GOOSE_QUALIFICATION_PREVIEW_SCHEMA, "source_id": GOOSE_SOURCE_ID, "frame_id": frame_id, "source_point_count": frame.point_count, "point_count": sample_count, "sampling": "deterministic-even-index", "points_xyz_m": frame.points_xyz_m[indices].tolist(), "ground_truth_ground": ground_truth[indices].astype(np.uint8).tolist(), "evaluated": evaluated[indices].astype(np.uint8).tolist(), "current_ground": current_result.ground_mask[indices].astype(np.uint8).tolist(), "patchwork_ground": patchwork_result.ground_mask[indices].astype(np.uint8).tolist(), "current_disagreement": ( evaluated[indices] & (current_result.ground_mask[indices] != ground_truth[indices]) ) .astype(np.uint8) .tolist(), "patchwork_disagreement": ( evaluated[indices] & (patchwork_result.ground_mask[indices] != ground_truth[indices]) ) .astype(np.uint8) .tolist(), "safety": { "visualization_only": True, "navigation_or_safety_accepted": False, }, } def _validation_frame_members(archive_path: Path) -> list[tuple[str, str, str]]: try: with zipfile.ZipFile(archive_path) as source: points = { _frame_id(member.filename, "_vls128.bin"): member.filename for member in source.infolist() if not member.is_dir() and "/lidar/val/" in "/" + member.filename.replace("\\", "/") and member.filename.endswith("_vls128.bin") } labels = { _frame_id(member.filename, "_goose.label"): member.filename for member in source.infolist() if not member.is_dir() and "/labels/val/" in "/" + member.filename.replace("\\", "/") and member.filename.endswith("_goose.label") } except (OSError, zipfile.BadZipFile) as exc: raise GooseAdmissionError("GOOSE archive cannot be indexed") from exc if len(points) != len(labels) or points.keys() != labels.keys(): raise GooseAdmissionError("GOOSE validation point and label members are not aligned") return [(frame_id, points[frame_id], labels[frame_id]) for frame_id in sorted(points)] def _read_archive_frame( archive_path: Path, point_member: str, label_member: str, ) -> DatasetPointFrame: try: with zipfile.ZipFile(archive_path) as source: return decode_semantic_kitti_frame( source.read(point_member), source.read(label_member), ) except (OSError, KeyError, zipfile.BadZipFile) as exc: raise GooseAdmissionError("GOOSE frame cannot be streamed from the archive") from exc def _challenge_categories( frame: DatasetPointFrame, labels: dict[int, dict[str, Any]], ) -> np.ndarray[Any, Any]: try: return np.asarray( [labels[int(value)]["challenge_category_id"] for value in frame.semantic_labels], dtype=np.uint8, ) except KeyError as exc: raise GooseAdmissionError("GOOSE frame contains an unmapped semantic label") from exc def _initialize_patchwork_worker( profile: GoosePatchworkProfile, module_name: str, ) -> None: global _WORKER_PATCHWORK _WORKER_PATCHWORK = PatchworkPPGroundSegmenter.load(profile, module_name=module_name) def _admit_or_resume_run( store: QualificationRunStore, *, run_id: str, identity_sha256: str, profile_document: dict[str, Any], mission_core_commit: str, provider_identities: dict[str, dict[str, object]], seed: int, ) -> QualificationRun: scenario = { "source_id": GOOSE_SOURCE_ID, "split": "validation", "clock": "dataset-frame-index", } patchwork_digest = provider_identities["patchworkpp"].get("binary_sha256") run = QualificationRun( run_id=run_id, episode_id=f"episode-{identity_sha256[:20]}", kind=RunKind.REPLAY_SHADOW, state=RunState.ADMITTED, scenario_generation="goose-3d-validation-v2025-08-22", scenario_sha256=_canonical_sha256(scenario), profile_generation="goose-ground-current-vs-patchworkpp-v1", profile_sha256=_canonical_sha256(profile_document), mission_core_commit=mission_core_commit, providers=( ProviderPin( identifier="missioncore-local-percentile-ground", version="v1", revision=mission_core_commit, ), ProviderPin( identifier="patchworkpp", version="v1.4.1", revision=PATCHWORKPP_SOURCE_COMMIT, digest=str(patchwork_digest) if patchwork_digest is not None else None, ), ), host_profile_id="simulation-worker-goose-ground-v1", host_profile_sha256=_canonical_sha256( { "role": "simulation-worker", "storage": "worker-d-only", "parallelism": "process-pool", } ), seed=seed, reproducibility_tier=ReproducibilityTier.R1, authority=AuthorityProfile( generation=1, command_ttl_max_ns=1, heartbeat_timeout_monotonic_ns=1, ), clock_domain="dataset:frame-index", created_at_utc=_utc_now(), ) try: return store.create(run) except QualificationRunConflictError as exc: existing = store.load(run_id) if ( existing.profile_sha256 != run.profile_sha256 or existing.scenario_sha256 != run.scenario_sha256 or existing.mission_core_commit != mission_core_commit ): raise GooseAdmissionError("existing qualification run has another identity") from exc return existing def _ensure_running(store: QualificationRunStore, run: QualificationRun) -> QualificationRun: if run.state is RunState.ADMITTED: run = store.transition( run.run_id, RunState.STARTING, expected_revision=run.revision, observed_at_utc=_utc_now(), host_monotonic_ns=time.monotonic_ns(), ) if run.state is RunState.STARTING: run = store.transition( run.run_id, RunState.RUNNING, expected_revision=run.revision, observed_at_utc=_utc_now(), host_monotonic_ns=time.monotonic_ns(), ) if run.state is not RunState.RUNNING: raise GooseAdmissionError( "qualification run cannot resume from its current lifecycle state" ) return run def _read_completed_report( store: QualificationRunStore, run: QualificationRun, ) -> dict[str, Any]: artifact = next( (artifact for artifact in run.artifacts if artifact.kind == REPORT_ARTIFACT_KIND), None, ) if artifact is None: raise GooseAdmissionError("completed qualification run has no report artifact") path = store.root / run.run_id / artifact.relative_path if _sha256_file(path) != artifact.sha256: raise GooseAdmissionError("completed qualification report digest differs") try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: raise GooseAdmissionError("completed qualification report is invalid") from exc if not isinstance(value, dict): raise GooseAdmissionError("completed qualification report must be an object") return value def _register_file( store: QualificationRunStore, run_id: str, path: Path, run_path: Path, *, artifact_id: str, kind: str, ) -> None: try: relative_path = path.relative_to(run_path).as_posix() except ValueError as exc: raise GooseAdmissionError("qualification artifact escaped its run root") from exc digest = _sha256_file(path) size = path.stat().st_size current = store.load(run_id) existing = next( (artifact for artifact in current.artifacts if artifact.artifact_id == artifact_id), None, ) if existing is not None: if ( existing.kind != kind or existing.relative_path != relative_path or existing.sha256 != digest or existing.byte_length != size ): raise GooseAdmissionError("registered qualification artifact changed on resume") return store.register_artifact( run_id, QualificationArtifact( artifact_id=artifact_id, kind=kind, relative_path=relative_path, sha256=digest, byte_length=size, source_of_record=True, ), ) def _append_frame_progress( store: QualificationRunStore, run_id: str, completed: int, total: int, ) -> None: _append_progress( store, run_id, "qualification.frames-progress", { "completed": completed, "total": total, "fraction": completed / total, }, ) def _append_progress( store: QualificationRunStore, run_id: str, event_type: str, payload: dict[str, Any], ) -> None: store.append_event( run_id, event_type=event_type, observed_at_utc=_utc_now(), host_monotonic_ns=time.monotonic_ns(), payload=payload, ) def _read_cached_frame(path: Path, identity_sha256: str) -> dict[str, Any] | None: if not path.is_file(): return None try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError): return None if ( not isinstance(value, dict) or value.get("schema_version") != GOOSE_QUALIFICATION_FRAME_SCHEMA or value.get("identity_sha256") != identity_sha256 ): return None return value def _check( name: str, observed: float | int, threshold: float | int, operator: str, ) -> dict[str, Any]: passed = observed >= threshold if operator == ">=" else observed <= threshold return { "check_id": name, "observed": float(observed), "operator": operator, "threshold": float(threshold), "passed": bool(passed), } def _validate_result(result: GroundSegmentation, point_count: int, label: str) -> None: if ( result.ground_mask.shape != (point_count,) or result.assigned_mask.shape != (point_count,) or not math.isfinite(result.latency_ms) or result.latency_ms < 0 ): raise GooseAdmissionError(f"{label} ground result violates point alignment") def _frame_seed(frame_id: str, profile_id: str, seed: int) -> int: digest = hashlib.sha256(f"{seed}:{frame_id}:{profile_id}".encode()).digest() return int.from_bytes(digest[:8], "little") def _frame_id(filename: str, suffix: str) -> str: name = PurePosixPath(filename.replace("\\", "/")).name if not name.endswith(suffix) or len(name) <= len(suffix): raise GooseAdmissionError("GOOSE validation member has an invalid frame identity") return name[: -len(suffix)] def _is_worker_dataset_root(root: Path) -> bool: return str(root).replace("\\", "/").rstrip("/").lower() == ("/mnt/d/ndc_missioncore/datasets") def _ratio(numerator: int | float, denominator: int | float) -> float: return float(numerator / denominator) if denominator else 0.0 def _canonical_sha256(value: dict[str, Any]) -> str: return hashlib.sha256( json.dumps(value, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() def _sha256_file(path: Path) -> str: digest = hashlib.sha256() try: with path.open("rb") as source: for chunk in iter(lambda: source.read(1024**2), b""): digest.update(chunk) except OSError as exc: raise GooseAdmissionError("qualification artifact cannot be hashed") from exc return digest.hexdigest() def _write_json_once(path: Path, value: dict[str, Any]) -> None: encoded = json.dumps(value, sort_keys=True, separators=(",", ":")).encode() + b"\n" if path.exists(): try: if path.is_file() and path.read_bytes() == encoded: return except OSError as exc: raise GooseAdmissionError( "immutable qualification artifact cannot be verified" ) from exc raise GooseAdmissionError("immutable qualification artifact already exists") path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile(dir=path.parent, delete=False) as temporary: temporary_path = Path(temporary.name) temporary.write(encoded) temporary.flush() os.fsync(temporary.fileno()) os.replace(temporary_path, path) def _utc_now() -> str: return datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z")