"""Compare the complete public-data LiDAR-IMU bag pipeline in fresh workers.

Freeze both source trees before timing. Each worker verifies its actual imports,
uses one BLAS thread, and saves the complete schema-valid calibration evidence.
Startup, bag decoding, initial odometry, map rounds, covariance and controls all
enter the parent wall time; per-round fit times are reported separately.
"""

from __future__ import annotations

import argparse
import ast
import hashlib
import io
import json
import os
import shutil
import statistics
import struct
import subprocess
import sys
import tarfile
import time
from dataclasses import replace
from pathlib import Path
from typing import Any


def _digest(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(4 * 1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _write(path: Path, value: object) -> None:
    if isinstance(value, dict) and value.get("schema_version"):
        import jsonschema

        schema = Path(__file__).with_name("schemas") / "bspline_lidar_imu_comparison.schema.json"
        jsonschema.validate(value, json.loads(schema.read_bytes()))
    path.write_text(json.dumps(value, indent=2, allow_nan=False) + "\n", encoding="utf-8")


def _manifest(root: Path) -> dict[str, str]:
    return {p.relative_to(root).as_posix(): _digest(p) for p in sorted(root.rglob("*.py"))}


def _verify_sources(root: Path) -> dict[str, str]:
    expected = json.loads((root / "manifest.json").read_bytes())
    if _manifest(root / "src") != expected:
        raise ValueError("frozen calibration source tree changed")
    return expected


def _freeze(root: Path, commit: str | None) -> None:
    root.mkdir()
    if commit is None:
        for path in sorted(Path("src/calibrex").rglob("*.py")):
            target = root / path
            target.parent.mkdir(parents=True, exist_ok=True)
            shutil.copyfile(path, target)
    else:
        archive = subprocess.check_output(["git", "archive", commit, "src/calibrex"])
        with tarfile.open(fileobj=io.BytesIO(archive)) as bundle:
            for member in bundle.getmembers():
                name = Path(member.name)
                if not member.isfile() or name.suffix != ".py":
                    continue
                if (
                    name.is_absolute()
                    or ".." in name.parts
                    or name.parts[:2] != ("src", "calibrex")
                ):
                    raise ValueError("unexpected Git source archive path")
                contents = bundle.extractfile(member)
                assert contents is not None
                target = root / name
                target.parent.mkdir(parents=True, exist_ok=True)
                target.write_bytes(contents.read())
    _write(root / "manifest.json", _manifest(root / "src"))


def _worker(args: argparse.Namespace) -> None:
    root = Path(args.source_root).resolve()
    expected = _verify_sources(root)
    launcher_digest = _digest(Path(__file__))
    backend = _backend(args.comparison, args.variant)
    cuda_cache_empty = None
    if args.comparison in ("cpu-cuda", "cuda-cuda"):
        cache = Path(os.environ["CUPY_CACHE_DIR"])
        cuda_cache_empty = not cache.exists() or not any(cache.iterdir())
        if not cuda_cache_empty:
            raise RuntimeError("worker CUDA compilation cache was not empty")
    from calibrex.data.livox_ros2 import LivoxStreamProfile
    from calibrex.evaluation import lidar_imu_bag_calibration as runner

    control_timing = _time_controls(runner)
    normal_timing = None
    if backend == "cuda":
        from calibrex.solvers.cuda_scan_odometry_adapter import CudaOdometryTarget

        normal_timing = _time_normal_preparation(CudaOdometryTarget)

    rounds: list[dict[str, object]] = []
    original = runner.fit_bspline_lidar_imu

    def measured_fit(problem, options):
        begin = time.perf_counter()
        result = original(problem, options)
        rounds.append(
            {
                "seconds": time.perf_counter() - begin,
                "status": result.status,
                "iterations": result.iterations,
                "initial_cost": result.initial_cost,
                "final_cost": result.final_cost,
                "residual_count": result.residual_count,
                "parameter_count": result.parameter_count,
                "calibration_covariance": result.calibration_covariance.tolist(),
                "residual_variance": result.residual_variance,
                "trajectory_rotations": result.trajectory.rotations.tolist(),
                "trajectory_positions": result.trajectory.positions.tolist(),
                "calibration": {
                    block: result.calibration.vector(block).tolist()
                    for block in runner.CALIBRATION_BLOCKS
                },
            }
        )
        if _plane_comparison(args):
            rounds[-1].update(
                execution_backend=getattr(result, "execution_backend", "numpy"),
                execution_statistics=getattr(result, "execution_statistics", {}),
            )
        return result

    runner.fit_bspline_lidar_imu = measured_fit
    begin = time.perf_counter()
    settings = runner.LidarImuBagOptions(max_rounds=args.max_rounds)
    fit_plane_backend = _plane_backends(args)[args.variant]
    if fit_plane_backend != "numpy":
        settings = replace(settings, fit_plane_backend=fit_plane_backend)
    if args.comparison != "source":
        windowing = settings.rotation.windowing
        settings = replace(settings, rotation=replace(
            settings.rotation, windowing=replace(
                windowing, odometry_backend=backend,
                odometry=replace(windowing.odometry, **_odometry_options(args)),
            ),
        ))
    if args.comparison in ("cpu-cpu", "cpu-source"):
        settings = replace(settings, rotation=replace(
            settings.rotation, windowing=replace(
                settings.rotation.windowing, odometry=replace(
                    settings.rotation.windowing.odometry,
                    query_workers=_query_workers(args)[args.variant],
                ),
            ),
        ))
    if _resident_backends(args)[args.variant]:
        settings = replace(settings, resident_odometry=True)
    result = runner.calibrate_lidar_imu_bag(
        args.bag,
        LivoxStreamProfile("public-glim-avia", "/livox/points", "/livox/imu", "t", "offset_s", "g"),
        settings,
        dataset_family="glim_versatile",
        dataset_license="upstream AIST GLIM dataset terms",
        max_seconds=args.max_seconds,
        command=[
            "python",
            "-m",
            "tools.benchmark_bspline_lidar_imu",
            "--worker",
            "--variant",
            args.variant,
            "--max-seconds",
            str(args.max_seconds),
            "--max-rounds",
            str(args.max_rounds),
            *(["--comparison", args.comparison,
               *_odometry_arguments(args)] if args.comparison != "source" else []),
            *(_worker_arguments(args)),
        ],
    )
    result.provenance.git_commit = args.source_commit
    out = Path(args.output)
    result.save(out / "result.json")
    pipeline_seconds = time.perf_counter() - begin
    imports = {}
    for name, module in list(sys.modules.items()):
        if name == "calibrex" or name.startswith("calibrex."):
            path = Path(module.__file__).resolve()
            if not path.is_relative_to(root / "src"):
                raise RuntimeError(f"calibration imported outside frozen tree: {name}")
            relative = path.relative_to(root / "src").as_posix()
            if _digest(path) != expected[relative]:
                raise RuntimeError(f"executed source digest differs: {name}")
            imports[relative] = expected[relative]
    if backend == "cpu" and "cupy" in sys.modules:
        raise RuntimeError("CPU comparison unexpectedly imported CuPy")
    if args.comparison != "source":
        _validate_backend_execution(result.model_dump(mode="json"), backend)
    if any(_resident_backends(args).values()):
        _validate_resident_execution(
            result.model_dump(mode="json"), _resident_backends(args)[args.variant],
            _resident_hashes(root) if _resident_backends(args)[args.variant] else None,
        )
    if args.comparison in ("cpu-cpu", "cpu-source"):
        _validate_query_workers(result.model_dump(mode="json"), _query_workers(args)[args.variant])
    if _plane_comparison(args):
        plane_path = root / "src/calibrex/solvers/cuda_bspline_plane_adapter.py"
        kernel_hash = (_plane_kernel_hash(plane_path.read_bytes())
                       if fit_plane_backend == "cuda" else None)
        _validate_plane_execution(result.model_dump(mode="json"), rounds,
                                  fit_plane_backend, kernel_hash)
    _verify_sources(root)
    if _digest(Path(__file__)) != launcher_digest:
        raise RuntimeError("benchmark launcher changed during execution")
    import numpy
    import scipy

    _write(
        out / "measurement.json",
        {
            "schema_version": "slac.bspline_lidar_imu_comparison/v0.1",
            "record_type": "worker",
            "variant": args.variant,
            "pipeline_seconds": pipeline_seconds,
            "fit_seconds": sum(float(r["seconds"]) for r in rounds),
            "control_evaluation_seconds": control_timing["seconds"],
            "control_evaluation_calls": control_timing["calls"],
            "control_score_calls": control_timing["scores"],
            **({"normal_preparation_seconds": normal_timing["seconds"],
                "normal_preparation_calls": normal_timing["calls"]}
               if normal_timing is not None else {}),
            "rounds": rounds,
            "provenance": {
                "bag_sha256": _digest(Path(args.bag)),
                "source_manifest_sha256": _digest(root / "manifest.json"),
                "executed_sources": imports,
                "launcher_sha256": launcher_digest,
                "result_sha256": _digest(out / "result.json"),
                "python": sys.version,
                "numpy": numpy.__version__,
                "scipy": scipy.__version__,
                "blas_threads": {
                    name: os.environ[name]
                    for name in ("OPENBLAS_NUM_THREADS", "OMP_NUM_THREADS", "MKL_NUM_THREADS")
                },
                **({"odometry_backend": backend} if args.comparison != "source" else {}),
                **({"cpu_query_workers": _query_workers(args)[args.variant]}
                   if args.comparison in ("cpu-cpu", "cpu-source") else {}),
                **({"fit_plane_backend": fit_plane_backend} if _plane_comparison(args) else {}),
                **({"resident_odometry": _resident_backends(args)[args.variant]}
                   if any(_resident_backends(args).values()) else {}),
                **({"cuda_cache_initially_empty": cuda_cache_empty,
                    "cupy": (sys.modules["cupy"].__version__ if backend == "cuda" else None)}
                   if args.comparison in ("cpu-cuda", "cuda-cuda") else {}),
            },
        },
    )


def _compare(prior: dict, current: dict) -> dict:
    """Fixed physical tolerances plus exact decisions; never compare timestamps."""
    import numpy as np
    from scipy.spatial.transform import Rotation

    rotation_error = float(
        np.degrees(
            (
                Rotation.from_quat(prior["t_lidar_imu_rotation_quat_xyzw"]).inv()
                * Rotation.from_quat(current["t_lidar_imu_rotation_quat_xyzw"])
            ).magnitude()
        )
    )
    errors = {
        "rotation_deg": rotation_error,
        "lever_arm_m": float(
            np.max(
                np.abs(
                    np.asarray(prior["t_lidar_imu_translation_m"])
                    - current["t_lidar_imu_translation_m"]
                )
            )
        ),
        "clock_offset_s": abs(prior["imu_clock_offset_s"] - current["imu_clock_offset_s"]),
    }
    decisions = (
        prior["policy_status"] == current["policy_status"]
        and [(c["parameter"], c.get("axis"), c["sign"], c["detected"]) for c in prior["controls"]]
        == [(c["parameter"], c.get("axis"), c["sign"], c["detected"]) for c in current["controls"]]
        and [(b["name"], b["verdict"]) for b in prior["blocks"]]
        == [(b["name"], b["verdict"]) for b in current["blocks"]]
        and [(s["block_a"], s["block_b"], s["entangled"]) for s in prior["separability"]]
        == [(s["block_a"], s["block_b"], s["entangled"]) for s in current["separability"]]
        and prior["converged"] == current["converged"]
    )
    numeric_evidence_equal = (
        all(
            np.allclose(left[field], right[field], rtol=1e-4, atol=1e-8)
            for left, right in zip(prior["blocks"], current["blocks"], strict=True)
            for field in ("estimate", "formal_std", "last_round_change")
        )
        and np.allclose(
            [c["delta_chi2"] for c in prior["controls"]],
            [c["delta_chi2"] for c in current["controls"]],
            rtol=1e-5,
            atol=1e-5,
        )
        and np.allclose(
            [s["max_abs_correlation"] for s in prior["separability"]],
            [s["max_abs_correlation"] for s in current["separability"]],
            atol=1e-5,
        )
        and all(
            np.isclose(prior["noise_sigmas"][k], current["noise_sigmas"][k], rtol=1e-5, atol=1e-8)
            for k in prior["noise_sigmas"]
        )
    )
    holdout = max(
        abs(prior["holdout_whitened_rms"][k] - current["holdout_whitened_rms"][k])
        for k in prior["holdout_whitened_rms"]
    )
    passed = (
        decisions
        and numeric_evidence_equal
        and errors["rotation_deg"] <= 0.001
        and errors["lever_arm_m"] <= 1e-5
        and errors["clock_offset_s"] <= 1e-7
        and holdout <= 1e-5
    )
    return {
        "passed": passed,
        "decisions_equal": decisions,
        "numeric_evidence_equal": numeric_evidence_equal,
        "errors": errors,
        "max_holdout_whitened_rms_difference": holdout,
    }


def _without_run_metadata(value: object) -> object:
    if isinstance(value, dict):
        result = {}
        for name, item in value.items():
            if name == "provenance":
                item = {k: v for k, v in item.items()
                        if k not in ("created_at", "command", "git_commit")}
            result[name] = _without_run_metadata(item)
        return result
    if isinstance(value, list):
        return [_without_run_metadata(item) for item in value]
    return value


def _validate_exact_pair(prior: dict, current: dict, left: list[dict], right: list[dict]) -> None:
    """Require complete numeric/evidence equality in addition to physical tolerances."""
    if _without_run_metadata(prior) != _without_run_metadata(current):
        raise ValueError("complete calibration/evidence fields differ")
    if len(left) != len(right) or any(
        {k: v for k, v in a.items() if k != "seconds"}
        != {k: v for k, v in b.items() if k != "seconds"}
        for a, b in zip(left, right, strict=True)
    ):
        raise ValueError("complete fit fields differ")


def _preparation_result(result: dict, *, asynchronous: bool) -> dict:
    """Remove only the declared scheduling identity/counter from copied evidence."""
    _validate_backend_execution(result, "cuda")
    value = _without_run_metadata(result)
    assert isinstance(value, dict)
    suffix = ";normal_schedule=async-v1"
    for name in ("initial_odometry_execution", "segment_odometry_execution"):
        for report in value["options"][name]:
            statistics = report["statistics"]
            batches = statistics.pop("gpu_async_normal_batches", 0)
            if (type(batches) is not int or
                    (not 0 < batches <= statistics.get("gpu_normal_kernel_calls", 0)
                     if asynchronous else batches != 0)):
                raise ValueError("normal preparation did not execute the declared schedule")
            identity = report["executed_backends"][0]
            if identity.endswith(suffix) != asynchronous:
                raise ValueError("normal preparation schedule identity differs")
            if asynchronous:
                report["executed_backends"][0] = identity.removesuffix(suffix)
    return value


def _validate_preparation_pair(
    prior: dict, current: dict, left: list[dict], right: list[dict],
) -> None:
    """Require every result, old work counter and fit field to remain exact."""
    _validate_exact_pair(
        _preparation_result(prior, asynchronous=False),
        _preparation_result(current, asynchronous=True), left, right,
    )


def _scan_spatial_hashes(root: Path) -> tuple[str, str]:
    """Bind both available query paths to their literal frozen CUDA sources."""
    def constants(name: str) -> dict[str, str]:
        values = {}
        for node in ast.parse((root / "src/calibrex/solvers" / name).read_bytes()).body:
            if isinstance(node, ast.Assign) and isinstance(node.value, ast.Constant):
                for target in node.targets:
                    if isinstance(target, ast.Name) and isinstance(node.value.value, str):
                        values[target.id] = node.value.value
        return values

    query = constants("cuda_lidar_map_adapter.py")["_SOURCE"]
    system = constants("cuda_scan_odometry_adapter.py")
    if "_INTERVAL_FUSED_SOURCE" in system:
        query += constants("cuda_spatial_interval_adapter.py")["_SOURCE"]
    return hashlib.sha256(query.encode()).hexdigest(), hashlib.sha256((
        system["_SYSTEM_SOURCE"] + system["_FUSED_SOURCE"] + system["_COVARIANCE_SOURCE"]
        + system.get("_NORMAL_SOURCE", "") + system.get("_INTERVAL_FUSED_SOURCE", "")
    ).encode()).hexdigest()


def _spatial_result(
    result: dict, hashes: tuple[str, str], prior_hashes: tuple[str, str], *, intervals: bool,
) -> dict:
    """Permit only source-bound pruning identity and actual-work metadata to differ."""
    _validate_backend_execution(result, "cuda")
    value = _without_run_metadata(result)
    assert isinstance(value, dict)
    suffix = ";spatial_pruning=fp32-intervals-v1"
    for name in ("initial_odometry_execution", "segment_odometry_execution"):
        for report in value["options"][name]:
            statistics = report["statistics"]
            for counter, total in (("gpu_interval_query_rows", "gpu_query_rows"),
                                   ("gpu_interval_system_calls", "gpu_fused_system_calls")):
                count = statistics.pop(counter, 0)
                if (type(count) is not int or
                        (not 0 < count <= statistics.get(total, 0) if intervals else count != 0)):
                    raise ValueError("interval spatial work does not match the declared path")
            identity = report["executed_backends"][0]
            if identity.endswith(suffix) != intervals:
                raise ValueError("interval spatial identity differs")
            if intervals:
                identity = identity.removesuffix(suffix)
            for label, actual, prior in zip(("query", "system"), hashes, prior_hashes, strict=True):
                token = label + "_sha256=" + actual
                if token not in identity:
                    raise ValueError("interval spatial kernel hash differs")
                identity = identity.replace(token, label + "_sha256=" + prior)
            report["executed_backends"][0] = identity
    return value


def _validate_spatial_pair(
    prior: dict, current: dict, left: list[dict], right: list[dict],
    prior_hashes: tuple[str, str], current_hashes: tuple[str, str],
) -> None:
    """Require exact calibration, complete fits and all previous work counters."""
    _validate_exact_pair(
        _spatial_result(prior, prior_hashes, prior_hashes, intervals=False),
        _spatial_result(current, current_hashes, prior_hashes, intervals=True), left, right,
    )


def _odometry_options(args: argparse.Namespace) -> dict[str, float | int]:
    return {
        "voxel_size_m": args.odometry_voxel_size_m,
        "local_map_scans": args.odometry_local_map_scans,
        "geometry_quantum_m": args.odometry_geometry_quantum_m,
        "motion_translation_quantum_m": args.odometry_motion_translation_quantum_m,
        "motion_rotation_quantum_rad": args.odometry_motion_rotation_quantum_rad,
    }


def _odometry_arguments(args: argparse.Namespace) -> list[str]:
    arguments = []
    for name, value in _odometry_options(args).items():
        arguments.extend(["--odometry-" + name.replace("_", "-"), str(value)])
    return arguments


def _validate_backend_execution(result: dict, backend: str) -> None:
    """Require real requested work in every initial and final odometry pass."""
    options = result["options"]
    groups = [options["initial_odometry_execution"], options["segment_odometry_execution"]]
    if options["odometry_backend"] != backend or any(not group for group in groups):
        raise ValueError("missing or incorrect requested odometry execution")
    for report in [r for group in groups for r in group]:
        if report["requested_backend"] != backend:
            raise ValueError("odometry pass requested another backend")
        identities = report["executed_backends"]
        if len(identities) != 1 or (
            identities[0] != "numpy_scipy" if backend == "cpu"
            else not identities[0].startswith("calibrex.cuda_scan_odometry/v0.1;")
        ):
            raise ValueError("odometry pass executed another backend")
        statistics = report["statistics"]
        if any(type(value) is not int or value < 0 for value in statistics.values()):
            raise ValueError("invalid odometry execution counters")
        if backend == "cuda" and any(
            statistics.get(name, 0) <= 0 for name in ("gpu_query_rows", "gpu_fused_system_calls")
        ):
            raise ValueError("CUDA pass did not execute spatial queries and systems")


def _time_controls(holder: Any) -> dict[str, float | int]:
    """Observe signed-control scoring, including its original baseline calls.

    This scope starts after holdout RMS evaluation; any states retained there
    are already prepared. Complete-worker time includes all preparation.
    """
    totals: dict[str, float | int] = {"calls": 0, "scores": 0, "seconds": 0.0}
    original = holder._controls

    def measured(chi2: Any, calibration: Any, threshold: float) -> Any:
        def score(*args: Any, **kwargs: Any) -> Any:
            totals["scores"] += 1
            return chi2(*args, **kwargs)

        begin = time.perf_counter()
        try:
            return original(score, calibration, threshold)
        finally:
            totals["calls"] += 1
            totals["seconds"] += time.perf_counter() - begin

    holder._controls = measured
    return totals


def _time_normal_preparation(holder: Any) -> dict[str, float | int]:
    """Time outermost preparation calls; nested cache completion is counted once.

    Count host time in synchronous work or async submission/completion; CPU work
    between those phases is excluded. This does not time overlapped device work.
    No additional GPU synchronization is inserted by this observer.
    """
    totals: dict[str, float | int] = {"calls": 0, "seconds": 0.0}
    depth = 0
    for name in ("_ensure", "_ensure_all", "_prime_normals", "_finish_normal_preparation"):
        if not hasattr(holder, name):
            continue
        original = getattr(holder, name)

        def measured(*args: Any, _original: Any = original, **kwargs: Any) -> Any:
            nonlocal depth
            depth += 1
            begin = time.perf_counter()
            try:
                return _original(*args, **kwargs)
            finally:
                depth -= 1
                if depth == 0:
                    totals["calls"] += 1
                    totals["seconds"] += time.perf_counter() - begin

        setattr(holder, name, measured)
    return totals


def _backend(comparison: str, variant: str) -> str:
    """Select actual execution for baseline/current source comparisons."""
    if comparison == "cuda-cuda" or (comparison == "cpu-cuda" and variant == "current"):
        return "cuda"
    return "cpu"


def _query_workers(args: argparse.Namespace) -> dict[str, int]:
    return {"prior": args.prior_query_workers, "current": args.current_query_workers}


def _validate_query_workers(result: dict, workers: int) -> None:
    if result["options"]["odometry_options"].get("query_workers") != workers:
        raise ValueError("worker did not use the declared CPU query workers")


def _worker_arguments(args: argparse.Namespace) -> list[str]:
    arguments = []
    if args.comparison in ("cpu-cpu", "cpu-source"):
        arguments += ["--prior-query-workers", str(args.prior_query_workers),
                      "--current-query-workers", str(args.current_query_workers)]
    if _plane_comparison(args):
        arguments += ["--prior-fit-plane-backend", args.prior_fit_plane_backend,
                      "--current-fit-plane-backend", args.current_fit_plane_backend]
    for variant, enabled in _resident_backends(args).items():
        if enabled:
            arguments.append(f"--{variant}-resident-odometry")
    return arguments


def _plane_backends(args: argparse.Namespace) -> dict[str, str]:
    return {"prior": args.prior_fit_plane_backend, "current": args.current_fit_plane_backend}


def _plane_comparison(args: argparse.Namespace) -> bool:
    return "cuda" in _plane_backends(args).values()


def _resident_backends(args: argparse.Namespace) -> dict[str, bool]:
    return {variant: bool(getattr(args, f"{variant}_resident_odometry", False))
            for variant in ("prior", "current")}


def _resident_hashes(root: Path) -> dict[str, str]:
    """Read frozen source constants without importing optional CUDA dependencies."""
    paths = {
        "resident": root / "src/calibrex/solvers/_cuda_resident_kernels.py",
        "native": root / "src/calibrex/solvers/cuda_scan_odometry_adapter.py",
    }
    constants = {}
    for path in paths.values():
        for node in ast.parse(path.read_bytes()).body:
            if isinstance(node, ast.Assign) and isinstance(node.value, ast.Constant):
                for target in node.targets:
                    if isinstance(target, ast.Name) and isinstance(node.value.value, str):
                        constants[target.id] = node.value.value
    sources = {
        "query_sha256": constants["_GRID_SOURCE"],
        "system_sha256": "".join(constants[name] for name in (
            "_SYSTEM_SOURCE", "_FUSED_GRID_SOURCE", "_COVARIANCE_SOURCE", "_NORMAL_SOURCE",
        )),
        "geometry_sha256": "".join(constants[name] for name in (
            "_VOXEL_SOURCE", "_GYRO_SOURCE", "_GEOMETRY_SOURCE",
        )),
    }
    return {key: hashlib.sha256(source.encode()).hexdigest() for key, source in sources.items()}


def _validate_resident_execution(result: dict, enabled: bool, hashes: dict | None) -> None:
    options = result["options"]
    if options.get("resident_odometry", False) != enabled:
        raise ValueError("resident geometry option differs")
    work = options.get("resident_geometry_execution")
    if not enabled:
        if work is not None:
            raise ValueError("ordinary odometry reported resident geometry work")
        return
    if hashes is None or not isinstance(work, dict) or any(
        type(value) is not int or value < 0 for value in work.values()
    ):
        raise ValueError("resident geometry counters are missing or invalid")
    if any(work.get(name, 0) <= 0 for name in (
        "source_captures", "source_cache_hits", "preprocess_calls", "preprocess_rows", "transforms",
    )) or not 0 < work.get("source_cache_bytes", 0) <= 512 * 1024**2:
        raise ValueError("resident geometry did not execute the declared work")
    for name in ("initial_odometry_execution", "segment_odometry_execution"):
        for report in options[name]:
            identity = dict(
                field.split("=", 1) for field in report["executed_backends"][0].split(";")[1:]
            )
            expected = {
                "fp64": "true", "query_fma": "false", "geometry_pipeline": "resident-v1", **hashes,
            }
            if any(identity.get(key) != value for key, value in expected.items()):
                raise ValueError("resident geometry kernel identity differs from frozen sources")
            statistics = report["statistics"]
            if statistics.get("gpu_grid_captures", 0) + statistics.get(
                "gpu_grid_native_fallbacks", 0,
            ) <= 0:
                raise ValueError("resident geometry did not capture any targets")


def _same_float_bits(left: object, right: object) -> bool:
    if type(left) is not type(right):
        return False
    if isinstance(left, dict):
        assert isinstance(right, dict)
        return left.keys() == right.keys() and all(
            _same_float_bits(left[key], right[key]) for key in left
        )
    if isinstance(left, list):
        assert isinstance(right, list)
        return len(left) == len(right) and all(
            _same_float_bits(a, b) for a, b in zip(left, right, strict=True)
        )
    if isinstance(left, float):
        assert isinstance(right, float)
        return struct.pack("<d", left) == struct.pack("<d", right)
    return left == right


def _validate_resident_pair(
    prior: dict, current: dict, left: list[dict], right: list[dict],
    native_hashes: tuple[str, str], resident_hashes: dict[str, str],
) -> None:
    """Keep every float bit, decision and common work counter after declared metadata."""
    _validate_backend_execution(prior, "cuda")
    _validate_backend_execution(current, "cuda")
    _validate_resident_execution(prior, False, None)
    _validate_resident_execution(current, True, resident_hashes)
    values = [_without_run_metadata(result) for result in (prior, current)]
    changed_counters = (
        "serialized_tree_captures", "public_node_tree_captures", "scalar_query_repairs",
        "gpu_grid_captures", "gpu_grid_native_fallbacks", "gpu_grid_brute_rows",
        "native_tree_captures", "sparse_native_normal_rows", "spatial_native_normal_rows",
        "gpu_parallel_fallback_dispatches",
    )
    for index, value in enumerate(values):
        assert isinstance(value, dict)
        value["options"].pop("resident_odometry", None)
        value["options"].pop("resident_geometry_execution", None)
        for name in ("initial_odometry_execution", "segment_odometry_execution"):
            for report in value["options"][name]:
                if index == 0:
                    identity = report["executed_backends"][0]
                    if any(f"{name}={digest}" not in identity for name, digest in zip(
                        ("query_sha256", "system_sha256"), native_hashes, strict=True,
                    )):
                        raise ValueError("ordinary spatial identity differs from frozen sources")
                report["executed_backends"] = ["<validated FP64 odometry execution>"]
                for counter in changed_counters:
                    report["statistics"].pop(counter, None)
    if not _same_float_bits(values[0], values[1]):
        raise ValueError("resident complete calibration/evidence or common work differs")
    fits = [[{key: value for key, value in fit.items() if key != "seconds"} for fit in records]
            for records in (left, right)]
    if not _same_float_bits(*fits):
        raise ValueError("resident complete fit float bits differ")


def _plane_kernel_hash(source: bytes) -> str:
    for node in ast.parse(source).body:
        if (isinstance(node, ast.Assign) and isinstance(node.value, ast.Constant)
                and any(isinstance(t, ast.Name) and t.id == "_SOURCE" for t in node.targets)):
            return hashlib.sha256(node.value.value.encode()).hexdigest()
    raise ValueError("CUDA plane kernel source is missing")


def _validate_plane_execution(
    result: dict, rounds: list[dict], backend: str, kernel_hash: str | None,
) -> None:
    options = result["options"]
    if options.get("fit_plane_backend", "numpy") != backend:
        raise ValueError("spline plane backend differs")
    reports = options.get("fit_execution", [])
    if backend == "cuda" and len(reports) != len(rounds):
        raise ValueError("missing spline plane execution reports")
    for index, record in enumerate(rounds):
        identity, stats = record.get("execution_backend"), record.get("execution_statistics")
        if backend == "numpy":
            if identity != "numpy" or stats != {}:
                raise ValueError("native spline planes executed another backend")
            continue
        expected = ("calibrex.cuda_bspline_planes/v0.1;fp64=true;fma=false;"
                    "normal=deterministic_banded;imu=numpy;covariance=numpy;"
                    f"sha256={kernel_hash}")
        if (identity != expected
                or reports[index] != {"executed_backend": identity, "statistics": stats}):
            raise ValueError("CUDA spline plane identity differs")
        names = {"gpu_plane_calls", "gpu_plane_rows", "gpu_plane_derivative_calls",
                 "gpu_plane_derivative_rows", "gpu_plane_normal_calls"}
        if not isinstance(stats, dict) or set(stats) != names or any(
            type(v) is not int or v < 0 for v in stats.values()
        ):
            raise ValueError("invalid CUDA spline plane work")
        if not (stats["gpu_plane_calls"] >= stats["gpu_plane_derivative_calls"] > 0
                and stats["gpu_plane_rows"] >= stats["gpu_plane_derivative_rows"] > 0
                and stats["gpu_plane_normal_calls"] == record["iterations"]
                and stats["gpu_plane_derivative_calls"] >= stats["gpu_plane_normal_calls"]):
            raise ValueError("CUDA spline plane work is inconsistent")
        count = result["rounds"][index]["plane_returns"]
        if (stats["gpu_plane_rows"] != count * stats["gpu_plane_calls"]
                or stats["gpu_plane_derivative_rows"]
                != count * stats["gpu_plane_derivative_calls"]):
            raise ValueError("CUDA spline plane rows differ from the measured workload")


def _validate_plane_fit_pair(left: list[dict], right: list[dict]) -> None:
    """Require every complete fit field, with explicit roundoff tolerances."""
    import numpy as np

    if len(left) != len(right):
        raise ValueError("spline fit round count differs")
    for a, b in zip(left, right, strict=True):
        for key in ("status", "iterations", "parameter_count", "residual_count"):
            if a[key] != b[key]:
                raise ValueError(f"spline fit {key} differs")
        for key in ("trajectory_rotations", "trajectory_positions", "initial_cost",
                    "final_cost", "residual_variance", "calibration_covariance"):
            covariance = key == "calibration_covariance"
            if not np.allclose(a[key], b[key], rtol=1e-5 if covariance else 1e-6,
                               atol=1e-11 if covariance else 1e-9):
                raise ValueError(f"spline fit {key} differs")
        if a["calibration"].keys() != b["calibration"].keys() or any(
            not np.allclose(a["calibration"][k], b["calibration"][k], rtol=1e-6, atol=1e-9)
            for k in a["calibration"]
        ):
            raise ValueError("spline fit calibration differs")


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--bag", default="data/public/glim_versatile/livox.bag")
    parser.add_argument("--output", required=True)
    parser.add_argument("--baseline-commit", default="b89e50bbd9e48a05a2acb6010ac46e50ece9220f")
    parser.add_argument("--repeats", type=int, default=3)
    parser.add_argument("--max-seconds", type=float, default=12.0)
    parser.add_argument("--max-rounds", type=int, default=4)
    parser.add_argument("--require-exact-results", action="store_true",
                        help="require every result/evidence and full fit field to match exactly")
    parser.add_argument("--require-exact-preparation-results", action="store_true",
                        help="require exact outputs/work, allowing only async scheduling metadata")
    parser.add_argument("--require-exact-spatial-results", action="store_true",
                        help="require exact outputs/work, allowing source-bound pruning metadata")
    parser.add_argument("--require-exact-resident-results", action="store_true",
                        help="require identical float bits and common work after resident metadata")
    parser.add_argument(
        "--comparison", choices=("source", "cpu-cpu", "cpu-source", "cpu-cuda", "cuda-cuda"),
        default="source",
    )
    parser.add_argument("--prior-query-workers", type=int, default=1)
    parser.add_argument("--current-query-workers", type=int, default=8)
    parser.add_argument("--prior-fit-plane-backend", choices=("numpy", "cuda"), default="numpy")
    parser.add_argument("--current-fit-plane-backend", choices=("numpy", "cuda"), default="numpy")
    parser.add_argument("--prior-resident-odometry", action="store_true",
                        help="use the resident geometry pipeline in the prior CUDA worker")
    parser.add_argument("--current-resident-odometry", action="store_true",
                        help="use the resident geometry pipeline in the current CUDA worker")
    parser.add_argument("--odometry-voxel-size-m", type=float)
    parser.add_argument("--odometry-local-map-scans", type=int)
    parser.add_argument("--odometry-geometry-quantum-m", type=float)
    parser.add_argument("--odometry-motion-translation-quantum-m", type=float)
    parser.add_argument("--odometry-motion-rotation-quantum-rad", type=float)
    parser.add_argument("--worker", action="store_true", help=argparse.SUPPRESS)
    parser.add_argument("--variant", choices=("prior", "current"), help=argparse.SUPPRESS)
    parser.add_argument("--source-root", help=argparse.SUPPRESS)
    parser.add_argument("--source-commit", help=argparse.SUPPRESS)
    args = parser.parse_args()
    if args.worker:
        _worker(args)
        return
    if args.repeats < 1 or args.max_rounds < 2 or args.max_seconds <= 0:
        parser.error("need positive repeats/seconds and at least two map rounds")
    if args.prior_query_workers < 1 or args.current_query_workers < 1:
        parser.error("query workers must be positive integers")
    if _plane_comparison(args) and (
        args.comparison not in ("cpu-cuda", "cuda-cuda") or args.require_exact_results
        or (args.comparison == "cpu-cuda" and args.prior_fit_plane_backend != "numpy")
    ):
        parser.error("CUDA plane comparison requires GPU workers and numeric output comparison")
    if any(_resident_backends(args).values()) and (
        args.comparison not in ("cpu-cuda", "cuda-cuda")
        or any(enabled and _backend(args.comparison, variant) != "cuda"
               for variant, enabled in _resident_backends(args).items())
        or any(getattr(args, name) is None or not (
            0 < getattr(args, name) < float("inf")
        ) for name in (
            "odometry_geometry_quantum_m", "odometry_motion_translation_quantum_m",
            "odometry_motion_rotation_quantum_rad",
        ))
    ):
        parser.error("resident geometry requires a CUDA worker and three explicit positive quanta")
    if args.require_exact_preparation_results and (
        args.comparison != "cuda-cuda" or args.require_exact_results
        or args.prior_fit_plane_backend != args.current_fit_plane_backend
    ):
        parser.error("exact preparation comparison requires cuda-cuda with the same plane backend")
    if args.require_exact_spatial_results and (
        args.comparison != "cuda-cuda" or args.require_exact_results
        or args.require_exact_preparation_results
        or args.prior_fit_plane_backend != args.current_fit_plane_backend
    ):
        parser.error("exact spatial comparison requires cuda-cuda with the same plane backend")
    if args.require_exact_resident_results and (
        args.comparison != "cuda-cuda" or _resident_backends(args) != {
            "prior": False, "current": True,
        } or args.prior_fit_plane_backend != args.current_fit_plane_backend
        or args.require_exact_results or args.require_exact_preparation_results
        or args.require_exact_spatial_results
    ):
        parser.error("exact resident comparison requires ordinary/resident CUDA with the same fit")
    if args.comparison != "source" and (
        any(value is None for value in _odometry_options(args).values())
        or args.odometry_voxel_size_m <= 0 or args.odometry_local_map_scans < 1
        or any(value < 0 for key, value in _odometry_options(args).items()
               if key not in ("voxel_size_m", "local_map_scans"))
    ):
        parser.error("CPU/CUDA comparisons require all five explicit valid odometry settings")
    out = Path(args.output).resolve()
    out.mkdir(parents=True, exist_ok=False)
    commit = subprocess.check_output(["git", "rev-parse", args.baseline_commit], text=True).strip()
    current_commit = subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip()
    if args.comparison in ("cpu-cpu", "cpu-cuda"):
        commit = current_commit
    subprocess.run(
        ["git", "diff", "--exit-code", "HEAD", "--", "src/calibrex"],
        check=True,
        stdout=subprocess.DEVNULL,
    )
    roots = {v: out / f"sources-{v}" for v in ("prior", "current")}
    _freeze(roots["prior"], None if args.comparison == "cpu-cpu" else commit)
    _freeze(roots["current"], None)
    launcher_digest = _digest(Path(__file__))
    bag = Path(args.bag).resolve()
    _write(
        out / "protocol.json",
        {
            "schema_version": "slac.bspline_lidar_imu_comparison/v0.1",
            "record_type": "protocol",
            "baseline_commit": commit,
            "current_commit": current_commit,
            "repeats": args.repeats,
            "max_seconds": args.max_seconds,
            "max_rounds": args.max_rounds,
            **({"exact_outputs_required": True} if args.require_exact_results else {}),
            **({"exact_preparation_outputs_required": True}
               if args.require_exact_preparation_results else {}),
            **({"exact_spatial_outputs_required": True}
               if args.require_exact_spatial_results else {}),
            **({"exact_resident_outputs_required": True}
               if args.require_exact_resident_results else {}),
            "bag_sha256": _digest(bag),
            "launcher_sha256": launcher_digest,
            "scope": (
                "fresh process through complete bag calibration, saved evidence "
                "and import verification"
            ),
            "quality_tolerances": {
                "rotation_deg": 0.001,
                "lever_arm_m": 1e-5,
                "clock_offset_s": 1e-7,
                "holdout_whitened_rms": 1e-5,
                "block_evidence_relative": 1e-4,
                "block_evidence_absolute": 1e-8,
                "control_chi2_relative": 1e-5,
                "control_chi2_absolute": 1e-5,
            },
            **({"comparison": args.comparison, "odometry_options": _odometry_options(args)}
               if args.comparison != "source" else {}),
            **({"isolated_cuda_caches": True}
               if args.comparison in ("cpu-cuda", "cuda-cuda") else {}),
            **({"cpu_query_workers": _query_workers(args)}
               if args.comparison in ("cpu-cpu", "cpu-source") else {}),
            **({"fit_plane_backends": _plane_backends(args)} if _plane_comparison(args) else {}),
            **({"resident_odometry": _resident_backends(args)}
               if any(_resident_backends(args).values()) else {}),
        },
    )
    attempts: list[dict] = []
    for repeat in range(args.repeats):
        order = ("prior", "current") if repeat % 2 == 0 else ("current", "prior")
        for variant in order:
            worker_out = out / f"{repeat:02d}-{variant}"
            worker_out.mkdir()
            env = os.environ.copy()
            env.update(
                dict.fromkeys(("OPENBLAS_NUM_THREADS", "OMP_NUM_THREADS", "MKL_NUM_THREADS"), "1")
            )
            env["PYTHONPATH"] = str(roots[variant] / "src") + os.pathsep + str(Path.cwd())
            env["PYTHONDONTWRITEBYTECODE"] = "1"
            if args.comparison in ("cpu-cuda", "cuda-cuda"):
                env["CUPY_CACHE_DIR"] = str(worker_out / "cupy-cache")
            command = [
                sys.executable,
                "-m",
                "tools.benchmark_bspline_lidar_imu",
                "--worker",
                "--variant",
                variant,
                "--source-root",
                str(roots[variant]),
                "--source-commit",
                commit if variant == "prior" else current_commit,
                "--bag",
                str(bag),
                "--max-seconds",
                str(args.max_seconds),
                "--max-rounds",
                str(args.max_rounds),
                "--output",
                str(worker_out),
                *(["--comparison", args.comparison, *_odometry_arguments(args)]
                  if args.comparison != "source" else []),
                *_worker_arguments(args),
            ]
            begin = time.perf_counter()
            with (worker_out / "worker.log").open("wb") as log:
                process = subprocess.run(
                    command, env=env, stdout=log, stderr=subprocess.STDOUT, check=False
                )
            attempt = {
                "repeat": repeat,
                "variant": variant,
                "exit_code": process.returncode,
                "whole_seconds": time.perf_counter() - begin,
                "worker": worker_out.name,
            }
            attempts.append(attempt)
            _write(out / "attempts.json", attempts)
            print(json.dumps(attempt), flush=True)
            if process.returncode != 0:
                raise RuntimeError(f"worker failed; retained {worker_out}")
    pairs = []
    for repeat in range(args.repeats):
        prior = json.loads((out / f"{repeat:02d}-prior/result.json").read_bytes())
        current = json.loads((out / f"{repeat:02d}-current/result.json").read_bytes())
        if args.require_exact_results:
            records = [json.loads((out / f"{repeat:02d}-{v}/measurement.json").read_bytes())
                       for v in ("prior", "current")]
            _validate_exact_pair(prior, current, records[0]["rounds"], records[1]["rounds"])
        if args.require_exact_preparation_results:
            records = [json.loads((out / f"{repeat:02d}-{v}/measurement.json").read_bytes())
                       for v in ("prior", "current")]
            _validate_preparation_pair(prior, current, records[0]["rounds"], records[1]["rounds"])
        if args.require_exact_spatial_results:
            records = [json.loads((out / f"{repeat:02d}-{v}/measurement.json").read_bytes())
                       for v in ("prior", "current")]
            _validate_spatial_pair(
                prior, current, records[0]["rounds"], records[1]["rounds"],
                _scan_spatial_hashes(roots["prior"]), _scan_spatial_hashes(roots["current"]),
            )
        if _plane_comparison(args):
            records = [json.loads((out / f"{repeat:02d}-{v}/measurement.json").read_bytes())
                       for v in ("prior", "current")]
            _validate_plane_fit_pair(records[0]["rounds"], records[1]["rounds"])
        if args.require_exact_resident_results:
            records = [json.loads((out / f"{repeat:02d}-{v}/measurement.json").read_bytes())
                       for v in ("prior", "current")]
            _validate_resident_pair(
                prior, current, records[0]["rounds"], records[1]["rounds"],
                _scan_spatial_hashes(roots["prior"]), _resident_hashes(roots["current"]),
            )
        pairs.append(_compare(prior, current))
    medians = {
        v: statistics.median(a["whole_seconds"] for a in attempts if a["variant"] == v)
        for v in roots
    }
    fit_medians = {
        v: statistics.median(
            json.loads((out / a["worker"] / "measurement.json").read_bytes())["fit_seconds"]
            for a in attempts
            if a["variant"] == v
        )
        for v in roots
    }
    for root in roots.values():
        _verify_sources(root)
    if _digest(Path(__file__)) != launcher_digest:
        raise RuntimeError("benchmark launcher changed during comparison")
    summary = {
        "schema_version": "slac.bspline_lidar_imu_comparison/v0.1",
        "record_type": "summary",
        "whole_median_seconds": medians,
        "whole_speedup": medians["prior"] / medians["current"],
        "fit_median_seconds": fit_medians,
        "fit_speedup": fit_medians["prior"] / fit_medians["current"],
        "quality": pairs,
        "quality_passed": all(p["passed"] for p in pairs),
        "provenance": {
            "protocol_sha256": _digest(out / "protocol.json"),
            "attempts_sha256": _digest(out / "attempts.json"),
        },
    }
    _write(out / "summary.json", summary)
    print(json.dumps(summary), flush=True)
    if not summary["quality_passed"]:
        raise RuntimeError("complete result quality comparison failed")


if __name__ == "__main__":
    main()
