"""Reuse exact v2 inputs before decoding the source, without changing v2 identity. The legacy producer hashes its own file into every pack. Keep that producer unchanged: this adapter only selects and verifies an existing pack, or calls the original builder. A cache hit is not a streaming/cold-start qualification. """ from __future__ import annotations import hashlib import json import math import os import re import stat from pathlib import Path from typing import Any from . import lidar_replay from .lidar_contract import K1_LIDAR_PACK_V2_PROFILE from .lidar_replay import ( DEFAULT_POSE_COVERAGE_THRESHOLD_MS, LIDAR_MANIFEST_NAME, LIDAR_REPLAY_PACK_SCHEMA, LidarReplayError, LidarReplayPackV2, ) _PACK_ID = re.compile(r"^lidar-replay-pack-[a-f0-9]{64}$") _SESSION_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") _MAX_MANIFEST_BYTES = 128 * 1024 _HASH_CHUNK_BYTES = 1024 * 1024 type _FileStamp = tuple[int, int, int, int, int] def prepare_lidar_replay_pack_v2( capture_path: Path, output_root: Path, *, session_id: str | None = None, pose_coverage_threshold_ms: float = DEFAULT_POSE_COVERAGE_THRESHOLD_MS, ) -> Path: """Return the exact validated input, skipping source decode on a cache hit. Only current-producer packs with exact raw/metadata/clock-origin digests may be reused. Their ordinary strict reader still checks artifacts, arrays, logical content and equivalence. Corruption fails without replacing evidence. No source or pack array survives this call. """ source = capture_path.expanduser().resolve(strict=True) if source.name != "mqtt.raw.k1mqtt" or not source.is_file(): raise LidarReplayError("LiDAR replay source must be mqtt.raw.k1mqtt") metadata = source.with_name("mqtt.metadata.jsonl") if not metadata.is_file(): raise LidarReplayError("exact host timing requires mqtt.metadata.jsonl") if ( not math.isfinite(pose_coverage_threshold_ms) or not 0 < pose_coverage_threshold_ms <= 10_000 ): raise LidarReplayError("pose coverage threshold is invalid") resolved_session = session_id or source.parents[2].name if _SESSION_ID.fullmatch(resolved_session) is None: raise LidarReplayError("LiDAR replay session id is unsafe") parent = output_root.expanduser().absolute() if parent.is_symlink(): raise LidarReplayError("LiDAR preparation cache cannot be a symlink") producer = Path(lidar_replay.__file__).resolve(strict=True) producer_sha256, _ = _hash_regular_file(producer) candidates = _candidates(parent, resolved_session, producer_sha256) # A cold directory does not add another whole-source hash pass. if not candidates: return lidar_replay.build_lidar_replay_pack_v2( source, parent, session_id=resolved_session, pose_coverage_threshold_ms=pose_coverage_threshold_ms, ) evidence, stamps = _source_evidence(source, metadata) matches = [ (root, identity) for root, identity in candidates if identity.get("source_evidence") == evidence ] if len(matches) > 1: raise LidarReplayError("LiDAR preparation cache has ambiguous source identity") if not matches: result = lidar_replay.build_lidar_replay_pack_v2( source, parent, session_id=resolved_session, pose_coverage_threshold_ms=pose_coverage_threshold_ms, ) _check_source_stamps(source, stamps) return result root, identity = matches[0] pack = LidarReplayPackV2(root) try: if pack.identity != identity: raise LidarReplayError("LiDAR preparation cache changed during validation") pose_binding = pack.quality.get("pose_binding") if ( not isinstance(pose_binding, dict) or pose_binding.get("threshold_ms") != pose_coverage_threshold_ms ): # v2 did not include this report parameter in its identity. Never # silently return another report or overwrite the existing pack. raise LidarReplayError("LiDAR cached pose coverage threshold differs") _check_source_stamps(source, stamps) return root finally: pack.close() def _candidates( parent: Path, session_id: str, producer_sha256: str, ) -> list[tuple[Path, dict[str, Any]]]: if not parent.exists(): return [] result: list[tuple[Path, dict[str, Any]]] = [] for root in parent.iterdir(): if _PACK_ID.fullmatch(root.name) is None: continue if root.is_symlink() or not root.is_dir(): raise LidarReplayError("LiDAR preparation cache entry is unsafe") manifest_path = root / LIDAR_MANIFEST_NAME if manifest_path.is_symlink(): raise LidarReplayError("LiDAR preparation manifest cannot be a symlink") try: with manifest_path.open("rb") as stream: payload = stream.read(_MAX_MANIFEST_BYTES + 1) if len(payload) > _MAX_MANIFEST_BYTES: raise ValueError("manifest too large") manifest = json.loads(payload) except (OSError, UnicodeDecodeError, ValueError) as exc: raise LidarReplayError("LiDAR preparation manifest is invalid") from exc identity = manifest.get("identity") if isinstance(manifest, dict) else None if not isinstance(identity, dict): raise LidarReplayError("LiDAR preparation identity is invalid") # Unrelated profiles/producers are preserved, never eagerly decoded. if ( identity.get("session_id") != session_id or identity.get("producer_sha256") != producer_sha256 ): continue encoded = json.dumps( identity, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False ).encode() digest = hashlib.sha256(encoded).hexdigest() if ( manifest.get("schema_version") != LIDAR_REPLAY_PACK_SCHEMA or identity.get("schema_version") != LIDAR_REPLAY_PACK_SCHEMA or manifest.get("pack_id") != root.name or root.name != f"lidar-replay-pack-{digest}" or manifest.get("identity_sha256") != digest or identity.get("lidar_evidence_profile") != K1_LIDAR_PACK_V2_PROFILE.to_dict() ): raise LidarReplayError("LiDAR preparation identity changed") result.append((root.resolve(strict=True), identity)) return result def _stamp(value: os.stat_result) -> _FileStamp: return value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns, value.st_ctime_ns def _hash_regular_file(path: Path) -> tuple[str, _FileStamp]: before = path.lstat() if not stat.S_ISREG(before.st_mode): raise LidarReplayError("LiDAR source evidence must be a regular file") digest = hashlib.sha256() with path.open("rb") as stream: if _stamp(os.fstat(stream.fileno())) != _stamp(before): raise LidarReplayError("LiDAR source evidence changed before hashing") while chunk := stream.read(_HASH_CHUNK_BYTES): digest.update(chunk) if _stamp(os.fstat(stream.fileno())) != _stamp(before): raise LidarReplayError("LiDAR source evidence changed during hashing") if _stamp(path.lstat()) != _stamp(before): raise LidarReplayError("LiDAR source evidence changed after hashing") return digest.hexdigest(), _stamp(before) def _source_evidence( source: Path, metadata: Path, ) -> tuple[dict[str, object], dict[Path, _FileStamp]]: paths = {"raw": source, "metadata": metadata} origin = source.with_name("mqtt.timeline.origin.json") if origin.exists(): paths["clock_origin"] = origin evidence: dict[str, object] = {} stamps: dict[Path, _FileStamp] = {} for role, path in paths.items(): digest, stamp = _hash_regular_file(path) evidence[role] = {"sha256": digest, "byte_length": stamp[2]} stamps[path] = stamp return evidence, stamps def _check_source_stamps(source: Path, stamps: dict[Path, _FileStamp]) -> None: origin = source.with_name("mqtt.timeline.origin.json") if origin.exists() != (origin in stamps): raise LidarReplayError("LiDAR source clock origin changed during preparation") for path, expected in stamps.items(): if not stat.S_ISREG(path.lstat().st_mode) or _stamp(path.lstat()) != expected: raise LidarReplayError("LiDAR source evidence changed during preparation")