"""Immutable full-source M4.4 geometry replay over the accepted M4.3 ledger.""" from __future__ import annotations import hashlib import json import os import shutil import time import uuid from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path from typing import Final import numpy as np from .baseline import BASELINE_RECORDED_JOB_ID from .contracts import EvidenceBasis, ObstacleObservation from .detector_replay_result import ( read_detector_replay_result, require_m4_detector_replay_acceptance, ) from .geometry import ( DEFAULT_GEOMETRY_PROFILE_PATH, Ravnoves00GeometryAssociationProvider, RecordedGeometryStore, load_geometry_profile, ) from .graph_validation import validate_observations from .providers import SourcePacket from .recorded_source import RECORDED_SOURCE_PACK_ID, RecordedFrameReference GEOMETRY_REPLAY_SCHEMA: Final = "missioncore.perception-geometry-replay-result/v1" GEOMETRY_REPLAY_FRAME_SCHEMA: Final = "missioncore.perception-geometry-replay-frame/v1" GEOMETRY_REPLAY_REPORT_SCHEMA: Final = "missioncore.perception-geometry-replay-report/v1" GEOMETRY_REPLAY_RESULT_PREFIX: Final = "m4-geometry-replay-" GEOMETRY_REPLAY_FRAMES_NAME: Final = "frames.jsonl" GEOMETRY_REPLAY_REPORT_NAME: Final = "report.json" GEOMETRY_REPLAY_MANIFEST_NAME: Final = "manifest.json" E32_RESULT_ID: Final = ( "e32-track-geometry-" "a14ca0e7fb3850ca0dfa3c41634e1b490a2d58ab74d101afc6d6921fbdb0e6fd" ) E32_MANIFEST_SHA256: Final = "f4b57c9f7619c43414adf1d10488b05df466226a523a0c001671102ced0e9ab8" E53_RESULT_ID: Final = ( "e53-camera-first-shadow-" "e6f03cf8bfb15db86100239b060e13f914532618b7b99e811866e4e6a555186c" ) E53_MANIFEST_SHA256: Final = "fc5b4ae69aae0098b7075c539d25b563dff1bbb09209a49030edda6cba9ee544" EXPECTED_SOURCE_AVAILABLE_FRAMES: Final = 3928 EXPECTED_SOURCE_UNAVAILABLE_FRAMES: Final = 561 class GeometryReplayError(RuntimeError): """A full-source geometry replay is incomplete, mutable or inconsistent.""" @dataclass(frozen=True, slots=True) class GeometryReplayResult: result_id: str result_root: Path accepted: bool metrics: dict[str, object] report: dict[str, object] manifest: dict[str, object] def build_geometry_replay( *, repository_root: Path, detector_result_root: Path, output_root: Path, ) -> GeometryReplayResult: """Run all accepted detector proposals through the canonical M4.4 provider.""" repository = repository_root.resolve() detector = read_detector_replay_result(detector_result_root) require_m4_detector_replay_acceptance(detector) profile_path = repository / DEFAULT_GEOMETRY_PROFILE_PATH profile = load_geometry_profile(profile_path) store = RecordedGeometryStore.from_repository(repository, profile=profile) provider = Ravnoves00GeometryAssociationProvider(store=store) references = _verified_historical_references(repository) root = output_root.expanduser().absolute() root.mkdir(mode=0o700, parents=True, exist_ok=True) staging = root / f".geometry-replay.{uuid.uuid4().hex}.tmp" staging.mkdir(mode=0o700, exist_ok=False) started_ns = time.perf_counter_ns() frame_latencies_ms: list[float] = [] source_available = 0 source_unavailable = 0 try: frames_path = staging / GEOMETRY_REPLAY_FRAMES_NAME with frames_path.open("wb") as output: for detector_frame in detector.frames: if detector_frame.outcome != "completed": raise GeometryReplayError("accepted detector ledger contains a failed frame") packet = _packet(detector_frame.envelope) frame_started_ns = time.perf_counter_ns() before = provider.snapshot() observations = provider.associate(packet, detector_frame.proposals) validate_observations(packet, detector_frame.proposals, observations) after = provider.snapshot() frame_latency_ms = (time.perf_counter_ns() - frame_started_ns) / 1_000_000 frame_latencies_ms.append(frame_latency_ms) source_available += packet.envelope.registered_point_increment.available source_unavailable += not packet.envelope.registered_point_increment.available proposal_observations = tuple(item for item in observations if item.proposal_ids) geometry_only = tuple(item for item in observations if not item.proposal_ids) eligible = sum(_range_eligible(item) for item in proposal_observations) ranged = sum(item.metric_geometry is not None for item in proposal_observations) conflict = sum( item.basis is EvidenceBasis.CONFLICT for item in proposal_observations ) frame_document = { "schema_version": GEOMETRY_REPLAY_FRAME_SCHEMA, "sequence": detector_frame.sequence, "frame_id": packet.envelope.frame_id, "source_available": packet.envelope.registered_point_increment.available, "proposal_count": len(detector_frame.proposals), "eligible_proposal_count": eligible, "ranged_proposal_count": ranged, "conflict_proposal_count": conflict, "camera_only_proposal_count": len(proposal_observations) - ranged - conflict, "geometry_only_observation_count": len(geometry_only), "overlapping_claims_removed": ( after.overlapping_claims_removed - before.overlapping_claims_removed ), "observations": [item.to_dict() for item in observations], "policy": { "one_owner_per_source_point": True, "absence_of_points_means_free": False, "geometry_can_invent_semantic_class": False, }, "authority": { "ground_truth": False, "physical_live": False, "commands_enabled": False, "actuation_allowed": False, "navigation_or_safety_accepted": False, }, } output.write(_canonical_json(frame_document) + b"\n") snapshot = provider.snapshot() metrics = _metrics( snapshot=snapshot, source_available=source_available, source_unavailable=source_unavailable, frame_latencies_ms=frame_latencies_ms, elapsed_ns=time.perf_counter_ns() - started_ns, ) requirements = _requirements(metrics, detector_frame_count=len(detector.frames)) accepted = all(requirements.values()) frames_sha256 = _file_sha256(frames_path) detector_identity = _object(detector.manifest["identity"], "detector identity") detector_frames_sha256 = detector_identity.get("frames_sha256") if not isinstance(detector_frames_sha256, str): raise GeometryReplayError("detector frame digest is unavailable") identity = { "schema_version": GEOMETRY_REPLAY_SCHEMA, "detector_result_id": detector.result_id, "detector_frames_sha256": detector_frames_sha256, "geometry_profile_id": profile.profile_id, "geometry_profile_sha256": profile.profile_sha256, "geometry_provider_id": provider.provider_id, "source_pack_id": profile.source_pack_id, "source_pack_sha256": profile.source_pack_sha256, "local_surface_model_id": profile.local_surface_model_id, "local_surface_sha256": profile.local_surface_sha256, "historical_references": references, "producer_sha256": { name: _file_sha256(repository / "src/k1link/perception" / name) for name in ("geometry.py", "geometry_math.py", "geometry_replay.py") }, "frames_sha256": frames_sha256, "metrics": metrics, "acceptance_requirements": requirements, "accepted": accepted, "authority": { "ground_truth": False, "physical_live": False, "commands_enabled": False, "actuation_allowed": False, "navigation_or_safety_accepted": False, }, } identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() result_id = f"{GEOMETRY_REPLAY_RESULT_PREFIX}{identity_sha256}" created = datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z") report = { "schema_version": GEOMETRY_REPLAY_REPORT_SCHEMA, "result_id": result_id, "identity_sha256": identity_sha256, "created_at_utc": created, "status": "accepted-product-geometry-replay" if accepted else "rejected-fail-closed", "accepted": accepted, "metrics": metrics, "acceptance_requirements": requirements, "decision": { "m4_4_geometry_provider_integrated": accepted, "semantic_class_quality_accepted": False, "physical_live_accepted": False, "next_gate": "M4.5 temporal retention and motion state", }, "authority": identity["authority"], } report_path = staging / GEOMETRY_REPLAY_REPORT_NAME _write_json(report_path, report) manifest = { "schema_version": GEOMETRY_REPLAY_SCHEMA, "result_id": result_id, "identity_sha256": identity_sha256, "identity": identity, "created_at_utc": created, "accepted": accepted, "artifacts": [ _artifact(frames_path, "geometry-replay-frames"), _artifact(report_path, "geometry-replay-report"), ], } _write_json(staging / GEOMETRY_REPLAY_MANIFEST_NAME, manifest) destination = root / result_id if destination.exists(): shutil.rmtree(staging) return read_geometry_replay_result(destination) os.replace(staging, destination) return read_geometry_replay_result(destination) except BaseException: shutil.rmtree(staging, ignore_errors=True) raise def read_geometry_replay_result(root: Path) -> GeometryReplayResult: resolved = root.resolve(strict=True) if resolved.is_symlink() or not resolved.name.startswith(GEOMETRY_REPLAY_RESULT_PREFIX): raise GeometryReplayError("geometry replay result root is invalid") manifest = _read_json(resolved / GEOMETRY_REPLAY_MANIFEST_NAME) _exact_keys( manifest, { "schema_version", "result_id", "identity_sha256", "identity", "created_at_utc", "accepted", "artifacts", }, "geometry replay manifest", ) identity = _object(manifest["identity"], "geometry replay identity") _exact_keys( identity, { "schema_version", "detector_result_id", "detector_frames_sha256", "geometry_profile_id", "geometry_profile_sha256", "geometry_provider_id", "source_pack_id", "source_pack_sha256", "local_surface_model_id", "local_surface_sha256", "historical_references", "producer_sha256", "frames_sha256", "metrics", "acceptance_requirements", "accepted", "authority", }, "geometry replay identity", ) identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() if ( manifest["schema_version"] != GEOMETRY_REPLAY_SCHEMA or manifest["result_id"] != resolved.name or manifest["identity_sha256"] != identity_sha256 or resolved.name != f"{GEOMETRY_REPLAY_RESULT_PREFIX}{identity_sha256}" ): raise GeometryReplayError("geometry replay identity changed") artifacts = manifest.get("artifacts") if not isinstance(artifacts, list) or len(artifacts) != 2: raise GeometryReplayError("geometry replay artifact inventory changed") by_role = {_object(item, "geometry artifact")["role"]: item for item in artifacts} if set(by_role) != {"geometry-replay-frames", "geometry-replay-report"}: raise GeometryReplayError("geometry replay artifact roles changed") frames_path = _validated_artifact( resolved, by_role["geometry-replay-frames"], GEOMETRY_REPLAY_FRAMES_NAME, ) report_path = _validated_artifact( resolved, by_role["geometry-replay-report"], GEOMETRY_REPLAY_REPORT_NAME, ) if _file_sha256(frames_path) != identity.get("frames_sha256"): raise GeometryReplayError("geometry frame ledger digest changed") report = _read_json(report_path) if ( report.get("schema_version") != GEOMETRY_REPLAY_REPORT_SCHEMA or report.get("result_id") != resolved.name or report.get("identity_sha256") != identity_sha256 or report.get("metrics") != identity.get("metrics") or report.get("acceptance_requirements") != identity.get("acceptance_requirements") or report.get("accepted") != identity.get("accepted") or report.get("authority") != identity.get("authority") ): raise GeometryReplayError("geometry replay report changed") metrics = _object(identity.get("metrics"), "geometry metrics") requirements = _object(identity.get("acceptance_requirements"), "geometry requirements") if identity.get("authority") != _false_authority(): raise GeometryReplayError("geometry replay authority changed") accepted = all(value is True for value in requirements.values()) if manifest.get("accepted") is not accepted or identity.get("accepted") is not accepted: raise GeometryReplayError("geometry replay acceptance changed") _validate_frame_ledger(frames_path, metrics) return GeometryReplayResult( result_id=resolved.name, result_root=resolved, accepted=accepted, metrics=metrics, report=report, manifest=manifest, ) def _packet(envelope: object) -> SourcePacket: from .contracts import SourceEnvelope if not isinstance(envelope, SourceEnvelope): raise GeometryReplayError("detector frame envelope is incompatible") image = RecordedFrameReference(BASELINE_RECORDED_JOB_ID, envelope.sequence) geometry = ( RecordedFrameReference(RECORDED_SOURCE_PACK_ID, envelope.sequence) if envelope.registered_point_increment.available else None ) return SourcePacket( envelope=envelope, image_payload=image, registered_point_increment_payload=geometry, pose_payload=geometry, ) def _range_eligible(observation: ObstacleObservation) -> bool: excluded = { "outside-projected-lidar-overlap", "registered-point-increment-unavailable", "registered-point-increment-stale", "local-surface-unavailable", } return not excluded.intersection(observation.reason_codes) def _metrics( *, snapshot: object, source_available: int, source_unavailable: int, frame_latencies_ms: list[float], elapsed_ns: int, ) -> dict[str, object]: from .geometry import GeometryProviderSnapshot if not isinstance(snapshot, GeometryProviderSnapshot): raise GeometryReplayError("geometry provider snapshot is incompatible") values = np.asarray(frame_latencies_ms, dtype=np.float64) total_coverage = snapshot.total_range_coverage eligible_coverage = snapshot.eligible_range_coverage return { "frames": { "total": snapshot.completed_frames, "failed": snapshot.failed_frames, "source_available": source_available, "source_unavailable": source_unavailable, }, "proposals": { "total": snapshot.proposal_count, "eligible_for_range": snapshot.eligible_proposal_count, "with_range": snapshot.ranged_proposal_count, "camera_only": snapshot.camera_only_proposal_count, "conflict": snapshot.conflict_proposal_count, "unavailable": snapshot.unavailable_proposal_count, "outside_overlap": snapshot.outside_overlap_proposal_count, "sparse": snapshot.sparse_proposal_count, "ownership_collision": snapshot.ownership_collision_proposal_count, "total_range_coverage": total_coverage, "eligible_range_coverage": eligible_coverage, }, "geometry_only_observations": snapshot.geometry_only_observation_count, "published_source_point_rows": snapshot.published_source_point_count, "overlapping_claims_removed": snapshot.overlapping_claims_removed, "runtime": { "elapsed_ms": elapsed_ns / 1_000_000, "provider_core_ms": snapshot.core_duration_ns / 1_000_000, "frame_latency_ms": { "minimum": float(np.min(values)), "p50": float(np.percentile(values, 50)), "p95": float(np.percentile(values, 95)), "maximum": float(np.max(values)), "mean": float(np.mean(values)), }, }, } def _requirements(metrics: dict[str, object], *, detector_frame_count: int) -> dict[str, bool]: frames = _object(metrics["frames"], "frame metrics") proposals = _object(metrics["proposals"], "proposal metrics") total = _integer(proposals["total"], "total proposals") ranged = _integer(proposals["with_range"], "ranged proposals") camera = _integer(proposals["camera_only"], "camera proposals") conflict = _integer(proposals["conflict"], "conflict proposals") eligible = _integer(proposals["eligible_for_range"], "eligible proposals") return { "detector_frame_accounting_complete": detector_frame_count == 4489, "geometry_frame_accounting_complete": ( frames.get("total") == 4489 and frames.get("failed") == 0 ), "source_accounting_reconciles_e32_e53": ( frames.get("source_available") == EXPECTED_SOURCE_AVAILABLE_FRAMES and frames.get("source_unavailable") == EXPECTED_SOURCE_UNAVAILABLE_FRAMES ), "proposal_accounting_closed": total == ranged + camera + conflict, "range_denominators_separated": 0 <= ranged <= eligible <= total, "exclusive_point_ownership_validated_every_frame": True, "missing_points_never_interpreted_as_free": True, "range_requires_current_source_points": True, "geometry_only_semantic_class_absent": True, "authority_remains_false": True, } def _verified_historical_references(repository: Path) -> dict[str, object]: references = { "e32": ( repository / ".runtime/compute-experiments/e32/results" / E32_RESULT_ID / "manifest.json", E32_MANIFEST_SHA256, ), "e53": ( repository / ".runtime/compute-experiments/e53/results" / E53_RESULT_ID / "manifest.json", E53_MANIFEST_SHA256, ), } result: dict[str, object] = {} for role, (path, digest) in references.items(): if not path.is_file() or path.is_symlink() or _file_sha256(path) != digest: raise GeometryReplayError(f"accepted {role.upper()} reference changed") result[role] = {"result_id": path.parent.name, "manifest_sha256": digest} return result def _validate_frame_ledger(path: Path, metrics: dict[str, object]) -> None: frame_count = 0 source_available = 0 proposal_count = 0 eligible_count = 0 ranged_count = 0 conflict_count = 0 camera_count = 0 geometry_count = 0 point_rows = 0 overlaps_removed = 0 for line_number, line in enumerate(path.read_text("utf-8").splitlines(), start=1): try: value = json.loads(line) except json.JSONDecodeError as exc: raise GeometryReplayError(f"geometry frame {line_number} is invalid JSON") from exc frame = _object(value, "geometry frame") _exact_keys( frame, { "schema_version", "sequence", "frame_id", "source_available", "proposal_count", "eligible_proposal_count", "ranged_proposal_count", "conflict_proposal_count", "camera_only_proposal_count", "geometry_only_observation_count", "overlapping_claims_removed", "observations", "policy", "authority", }, "geometry frame", ) if frame.get("schema_version") != GEOMETRY_REPLAY_FRAME_SCHEMA: raise GeometryReplayError("geometry frame schema changed") if frame.get("sequence") != frame_count: raise GeometryReplayError("geometry frame sequence is incomplete") if frame.get("policy") != { "one_owner_per_source_point": True, "absence_of_points_means_free": False, "geometry_can_invent_semantic_class": False, }: raise GeometryReplayError("geometry frame policy changed") if frame.get("authority") != _false_authority(): raise GeometryReplayError("geometry frame authority changed") observations_value = frame.get("observations") if not isinstance(observations_value, list): raise GeometryReplayError("geometry observations are not an array") observations = tuple(ObstacleObservation.from_dict(item) for item in observations_value) frame_id = frame.get("frame_id") proposal_observations = tuple(item for item in observations if item.proposal_ids) geometry_observations = tuple(item for item in observations if not item.proposal_ids) if ( not isinstance(frame_id, str) or any( item.frame_id != frame_id or item.source_id != "RAVNOVES00" for item in observations ) or any(len(item.proposal_ids) != 1 for item in proposal_observations) or len(proposal_observations) != _integer(frame.get("proposal_count"), "frame proposals") or len(geometry_observations) != _integer(frame.get("geometry_only_observation_count"), "geometry-only") or sum(_range_eligible(item) for item in proposal_observations) != _integer(frame.get("eligible_proposal_count"), "eligible proposals") or sum(item.metric_geometry is not None for item in proposal_observations) != _integer(frame.get("ranged_proposal_count"), "ranged proposals") or sum(item.basis is EvidenceBasis.CONFLICT for item in proposal_observations) != _integer(frame.get("conflict_proposal_count"), "conflicts") ): raise GeometryReplayError("geometry frame observation accounting changed") owners: set[int] = set() for observation in observations: if owners.intersection(observation.source_point_ids): raise GeometryReplayError("geometry frame has duplicate point ownership") owners.update(observation.source_point_ids) if not observation.proposal_ids and observation.semantic_hint is not None: raise GeometryReplayError("geometry-only observation invented a semantic class") if observation.metric_geometry is not None and not observation.source_point_ids: raise GeometryReplayError("metric range lost its source points") frame_count += 1 source_available += frame.get("source_available") is True proposal_count += _integer(frame.get("proposal_count"), "frame proposals") eligible_count += _integer(frame.get("eligible_proposal_count"), "eligible proposals") ranged_count += _integer(frame.get("ranged_proposal_count"), "ranged proposals") conflict_count += _integer(frame.get("conflict_proposal_count"), "conflicts") camera_count += _integer(frame.get("camera_only_proposal_count"), "camera-only") geometry_count += _integer( frame.get("geometry_only_observation_count"), "geometry-only", ) overlaps_removed += _integer(frame.get("overlapping_claims_removed"), "overlaps") point_rows += sum(len(item.source_point_ids) for item in observations) frames = _object(metrics.get("frames"), "frame metrics") proposals = _object(metrics.get("proposals"), "proposal metrics") if ( frame_count != frames.get("total") or source_available != frames.get("source_available") or frame_count - source_available != frames.get("source_unavailable") or proposal_count != proposals.get("total") or eligible_count != proposals.get("eligible_for_range") or ranged_count != proposals.get("with_range") or conflict_count != proposals.get("conflict") or camera_count != proposals.get("camera_only") or geometry_count != metrics.get("geometry_only_observations") or point_rows != metrics.get("published_source_point_rows") or overlaps_removed != metrics.get("overlapping_claims_removed") ): raise GeometryReplayError("geometry frame ledger and metrics disagree") def _validated_artifact(root: Path, value: object, name: str) -> Path: document = _object(value, "geometry artifact") _exact_keys(document, {"role", "path", "bytes", "sha256"}, "geometry artifact") if document.get("path") != name: raise GeometryReplayError("geometry artifact path changed") path = root / name if not path.is_file() or path.is_symlink(): raise GeometryReplayError("geometry artifact is missing") if document.get("bytes") != path.stat().st_size or document.get("sha256") != _file_sha256(path): raise GeometryReplayError("geometry artifact digest changed") return path def _artifact(path: Path, role: str) -> dict[str, object]: return { "role": role, "path": path.name, "bytes": path.stat().st_size, "sha256": _file_sha256(path), } def _read_json(path: Path) -> dict[str, object]: if not path.is_file() or path.is_symlink(): raise GeometryReplayError("geometry JSON artifact is missing") try: return _object(json.loads(path.read_text("utf-8")), "geometry JSON artifact") except json.JSONDecodeError as exc: raise GeometryReplayError("geometry JSON artifact is invalid") from exc def _write_json(path: Path, value: object) -> None: path.write_bytes(_canonical_json(value) + b"\n") def _canonical_json(value: object) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() def _file_sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _object(value: object, label: str) -> dict[str, object]: if not isinstance(value, dict) or any(not isinstance(key, str) for key in value): raise GeometryReplayError(f"{label} must be an object") return value def _exact_keys(document: dict[str, object], keys: set[str], label: str) -> None: if set(document) != keys: raise GeometryReplayError(f"{label} fields changed") def _integer(value: object, label: str) -> int: if not isinstance(value, int) or isinstance(value, bool) or value < 0: raise GeometryReplayError(f"{label} must be a nonnegative integer") return value def _false_authority() -> dict[str, bool]: return { "ground_truth": False, "physical_live": False, "commands_enabled": False, "actuation_allowed": False, "navigation_or_safety_accepted": False, } __all__ = [ "GEOMETRY_REPLAY_FRAME_SCHEMA", "GEOMETRY_REPLAY_MANIFEST_NAME", "GEOMETRY_REPLAY_REPORT_NAME", "GEOMETRY_REPLAY_RESULT_PREFIX", "GEOMETRY_REPLAY_SCHEMA", "GeometryReplayError", "GeometryReplayResult", "build_geometry_replay", "read_geometry_replay_result", ]