"""Read-only Lite6 digital-twin snapshots for the local operator UI.

This module is deliberately not a robot-control surface.  It converts the
already bounded xArm telemetry snapshot into a deterministic visualization
model and combines it with optional, fail-closed dual-depth evidence.  Nothing
in this module opens a device, calls the xArm SDK, or authorizes motion.

The link chain is transcribed in millimetres from the Lite6 URDF dimensions
shipped with the project documentation.  It is suitable for visualization and
consistency checking only; it is not a safety-rated collision model.
"""

from __future__ import annotations

import copy
import hashlib
import json
import math
import re
import threading
import time
from typing import Any, Mapping, Optional, Sequence, Tuple

import numpy as np

DIGITAL_TWIN_SCHEMA_VERSION = 1
DIGITAL_TWIN_COORDINATE_FRAME = "xarm_base_mm"

_HEX_64 = re.compile(r"[0-9a-f]{64}\Z")
_MAX_SAMPLE_AGE_SECONDS = 86_400.0
_MAX_POSE_SEQUENCE = 10_000_000_000
_MODEL_TCP_TOLERANCE_MM = 30.0
_MAX_PLAN_POINTS = 32

# Lite6 documented mechanical ranges.  Boundary values are accepted; any
# telemetry or planned visualization sample outside one of these ranges is
# invalid rather than wrapped or clamped.
LITE6_JOINT_LIMITS_RAD = (
    (-2.0 * math.pi, 2.0 * math.pi),
    (-math.radians(150.0), math.radians(150.0)),
    (-math.radians(3.5), math.radians(300.0)),
    (-2.0 * math.pi, 2.0 * math.pi),
    (-math.radians(124.0), math.radians(124.0)),
    (-2.0 * math.pi, 2.0 * math.pi),
)

# The closed blocker vocabulary is shared with the frontend contract.  Unknown
# strings are discarded at the producer boundary rather than leaked to the UI.
DIGITAL_TWIN_BLOCKERS = (
    "no_xarm_telemetry",
    "stale_pose",
    "invalid_joint_vector",
    "model_tcp_mismatch",
    "tool_depth_unavailable",
    "workspace_depth_unavailable",
    "tool_depth_calibration_missing",
    "workspace_depth_calibration_missing",
    "dual_depth_stale",
    "dual_depth_skew",
    "tool_pose_skew",
    "depth_sequence_mismatch",
    "unknown_swept_volume",
    "camera_disagreement",
    "clearance_below_limit",
    "trajectory_unavailable",
    "motion_release_incomplete",
)
_DIGITAL_TWIN_BLOCKER_SET = frozenset(DIGITAL_TWIN_BLOCKERS)


def _finite_vector(
    value: Any,
    length: int,
    *,
    minimum: float = -1_000_000.0,
    maximum: float = 1_000_000.0,
) -> Optional[Tuple[float, ...]]:
    if not isinstance(value, (tuple, list)) or len(value) != length:
        return None
    result = []
    for item in value:
        if isinstance(item, bool):
            return None
        try:
            number = float(item)
        except (TypeError, ValueError, OverflowError):
            return None
        if not math.isfinite(number) or not minimum <= number <= maximum:
            return None
        result.append(number)
    return tuple(result)


def _bounded_number(
    value: Any,
    *,
    minimum: float,
    maximum: float,
) -> Optional[float]:
    if isinstance(value, bool):
        return None
    try:
        number = float(value)
    except (TypeError, ValueError, OverflowError):
        return None
    if not math.isfinite(number) or not minimum <= number <= maximum:
        return None
    return number


def _lite6_joint_vector(value: Any) -> Optional[Tuple[float, ...]]:
    joints = _finite_vector(value, 6, minimum=-1_000.0, maximum=1_000.0)
    if joints is None or any(
        not lower <= angle <= upper
        for angle, (lower, upper) in zip(joints, LITE6_JOINT_LIMITS_RAD)
    ):
        return None
    return joints


def _bounded_sequence(value: Any) -> int:
    if isinstance(value, bool) or not isinstance(value, int):
        return 0
    return min(max(value, 0), _MAX_POSE_SEQUENCE)


def _digest(value: Any) -> str:
    return value if isinstance(value, str) and _HEX_64.fullmatch(value) else ""


def _rotation_rpy(roll: float, pitch: float, yaw: float) -> np.ndarray:
    cx, sx = math.cos(roll), math.sin(roll)
    cy, sy = math.cos(pitch), math.sin(pitch)
    cz, sz = math.cos(yaw), math.sin(yaw)
    rotation_x = np.asarray(
        ((1.0, 0.0, 0.0), (0.0, cx, -sx), (0.0, sx, cx)), dtype=float
    )
    rotation_y = np.asarray(
        ((cy, 0.0, sy), (0.0, 1.0, 0.0), (-sy, 0.0, cy)), dtype=float
    )
    rotation_z = np.asarray(
        ((cz, -sz, 0.0), (sz, cz, 0.0), (0.0, 0.0, 1.0)), dtype=float
    )
    return rotation_z.dot(rotation_y).dot(rotation_x)


def _axis_rotation(axis: Sequence[float], angle: float) -> np.ndarray:
    normalized = np.asarray(axis, dtype=float)
    normalized = normalized / np.linalg.norm(normalized)
    x, y, z = normalized
    cosine = math.cos(angle)
    sine = math.sin(angle)
    one_minus_cosine = 1.0 - cosine
    return np.asarray(
        (
            (
                cosine + x * x * one_minus_cosine,
                x * y * one_minus_cosine - z * sine,
                x * z * one_minus_cosine + y * sine,
            ),
            (
                y * x * one_minus_cosine + z * sine,
                cosine + y * y * one_minus_cosine,
                y * z * one_minus_cosine - x * sine,
            ),
            (
                z * x * one_minus_cosine - y * sine,
                z * y * one_minus_cosine + x * sine,
                cosine + z * z * one_minus_cosine,
            ),
        ),
        dtype=float,
    )


class Lite6VisualizationKinematics:
    """Deterministic seven-point Lite6 visualization chain.

    The first point is the robot base and the following six points are the child
    link origins after applying each revolute joint transform.  Consumers must
    never use these points to generate commands or to replace a commissioned
    collision model.
    """

    # Parent-to-child origins and axes from the Lite6 URDF, converted to mm.
    _JOINTS = (
        ((0.0, 0.0, 243.5), (0.0, 0.0, 0.0), (0.0, 0.0, 1.0)),
        (
            (0.0, 0.0, 0.0),
            (-1.724127, -1.570792, 0.153342),
            (0.000001, 0.0, 1.0),
        ),
        (
            (200.2, 0.0, 0.0),
            (3.141585, 0.0, 1.570801),
            (0.0, -0.000001, 1.0),
        ),
        ((87.0, -227.61, 0.0), (1.5708, 0.0, 0.0), (0.0, 0.0, 1.0)),
        ((0.0, 0.0, 0.0), (1.5708, 0.0, 0.0), (0.0, 0.0, 1.0)),
        ((0.0, 62.5, 0.0), (-1.5708, 0.0, 0.0), (0.0, 0.0, 1.0)),
    )

    _CANONICAL_MODEL = {
        "schema_version": 1,
        "model": "ufactory_lite6_visualization_chain",
        "units": "mm_rad",
        "joints": _JOINTS,
        "joint_limits_rad": LITE6_JOINT_LIMITS_RAD,
    }
    model_sha256 = hashlib.sha256(
        json.dumps(
            _CANONICAL_MODEL,
            allow_nan=False,
            separators=(",", ":"),
            sort_keys=True,
        ).encode("utf-8")
    ).hexdigest()

    def link_points_mm(
        self, joint_angles_rad: Sequence[float]
    ) -> Tuple[Tuple[float, ...], ...]:
        joints = _lite6_joint_vector(joint_angles_rad)
        if joints is None:
            raise ValueError(
                "Lite6 FK requires exactly six finite radian joints within "
                "documented mechanical limits"
            )
        transform = np.eye(4, dtype=float)
        points = [(0.0, 0.0, 0.0)]
        for angle, (origin_xyz, origin_rpy, axis) in zip(joints, self._JOINTS):
            origin = np.eye(4, dtype=float)
            origin[:3, :3] = _rotation_rpy(*origin_rpy)
            origin[:3, 3] = np.asarray(origin_xyz, dtype=float)
            rotation = np.eye(4, dtype=float)
            rotation[:3, :3] = _axis_rotation(axis, angle)
            transform = transform.dot(origin).dot(rotation)
            points.append(tuple(float(value) for value in transform[:3, 3]))
        if len(points) != 7 or not np.all(np.isfinite(np.asarray(points))):
            raise ValueError("Lite6 FK produced an invalid visualization chain")
        return tuple(points)


def _empty_camera(
    source_id: str = "", calibration_sha256: str = ""
) -> Mapping[str, Any]:
    return {
        "source_id": source_id,
        "frame_sequence": 0,
        "age_seconds": None,
        "depth_ready": False,
        "calibration_sha256": _digest(calibration_sha256),
    }


def _empty_scene(required_clearance_mm: float) -> Mapping[str, Any]:
    return {
        "scene_sha256": "",
        "pair_skew_seconds": None,
        "coverage_ratio": 0.0,
        "nearest_obstacle_mm": None,
        "current_clearance_mm": None,
        "planned_min_clearance_mm": None,
        "required_clearance_mm": required_clearance_mm,
        "swept_volume_clear": False,
    }


def _empty_plan(*, configured: bool) -> Mapping[str, Any]:
    return {
        "state": "none" if configured else "unavailable",
        "plan_sha256": "",
        "waypoint_count": 0,
        "tcp_path_mm": [],
    }


class DigitalTwinSnapshotProvider:
    """Build the exact bounded ``runtime.digital_twin`` wire contract."""

    def __init__(
        self,
        *,
        required_clearance_mm: float = 0.0,
        kinematics: Optional[Lite6VisualizationKinematics] = None,
        tcp_match_tolerance_mm: float = _MODEL_TCP_TOLERANCE_MM,
        geometry_profile_sha256: str = "",
        clearance_evidence_ttl_seconds: float = 0.5,
        monotonic: Any = time.monotonic,
    ) -> None:
        required = _bounded_number(
            required_clearance_mm,
            minimum=0.0,
            maximum=10_000.0,
        )
        tolerance = _bounded_number(
            tcp_match_tolerance_mm,
            minimum=0.001,
            maximum=1_000.0,
        )
        evidence_ttl = _bounded_number(
            clearance_evidence_ttl_seconds,
            minimum=0.001,
            maximum=60.0,
        )
        if required is None:
            raise ValueError("required_clearance_mm must be finite and bounded")
        if tolerance is None:
            raise ValueError("tcp_match_tolerance_mm must be positive and bounded")
        if evidence_ttl is None or not callable(monotonic):
            raise ValueError("clearance evidence TTL/clock is invalid")
        self.required_clearance_mm = required
        self.kinematics = kinematics or Lite6VisualizationKinematics()
        self.tcp_match_tolerance_mm = tolerance
        self.geometry_profile_sha256 = _digest(geometry_profile_sha256)
        self.clearance_evidence_ttl_seconds = evidence_ttl
        self._monotonic = monotonic
        self._lock = threading.Lock()
        self._clearance_evidence: Optional[Mapping[str, Any]] = None

    def record_clearance_evidence(self, evidence: Optional[Mapping[str, Any]]) -> None:
        """Atomically retain only a copied preflight result for later rendering."""

        try:
            copied = (
                copy.deepcopy(dict(evidence)) if isinstance(evidence, Mapping) else None
            )
        except Exception:
            copied = None
        with self._lock:
            # Invalid new evidence clears a previously positive result instead
            # of leaving a stale point-in-time assessment visible.
            self._clearance_evidence = copied

    def snapshot(
        self,
        *,
        configured: bool,
        xarm_status: Mapping[str, Any],
        pose_sequence: int,
        workspace_lower_xyz_mm: Sequence[float] = (),
        workspace_upper_xyz_mm: Sequence[float] = (),
        camera_status: Optional[Mapping[str, Any]] = None,
        tool_calibration_sha256: str = "",
        workspace_calibration_sha256: str = "",
        motion_release_valid: bool = False,
    ) -> Mapping[str, Any]:
        """Return one exact, secret-free snapshot with no motion authority."""

        configured = bool(configured)
        xarm = xarm_status if isinstance(xarm_status, Mapping) else {}
        metrics = xarm.get("metrics")
        metrics = metrics if isinstance(metrics, Mapping) else {}
        age = _bounded_number(
            xarm.get("sample_age_seconds"),
            minimum=0.0,
            maximum=_MAX_SAMPLE_AGE_SECONDS,
        )
        if age is None:
            age = _MAX_SAMPLE_AGE_SECONDS

        joints = _lite6_joint_vector(metrics.get("joint_angles_rad"))
        tcp_position = _finite_vector(metrics.get("tcp_position_mm"), 3)
        tcp_orientation = _finite_vector(
            metrics.get("tcp_orientation_rad"),
            3,
            minimum=-1_000.0,
            maximum=1_000.0,
        )
        pose_fresh = bool(
            configured
            and xarm.get("available") is True
            and xarm.get("fresh") is True
            and joints is not None
            and tcp_position is not None
            and tcp_orientation is not None
        )

        zero_joints = (0.0,) * 6
        zero_points = ((0.0, 0.0, 0.0),) * 7
        link_points = zero_points
        if joints is not None:
            try:
                link_points = self.kinematics.link_points_mm(joints)
            except Exception:
                joints = None
                pose_fresh = False
                link_points = zero_points

        tcp_pose = (
            tuple(tcp_position) + tuple(tcp_orientation)
            if tcp_position is not None and tcp_orientation is not None
            else (0.0,) * 6
        )
        model_verified = False
        if pose_fresh and tcp_position is not None:
            mismatch = float(
                np.linalg.norm(
                    np.asarray(link_points[-1], dtype=float)
                    - np.asarray(tcp_position, dtype=float)
                )
            )
            model_verified = mismatch <= self.tcp_match_tolerance_mm

        lower, upper = self._workspace_bounds(
            workspace_lower_xyz_mm,
            workspace_upper_xyz_mm,
            metrics,
        )
        current_camera_status = (
            camera_status if isinstance(camera_status, Mapping) else {}
        )
        cameras = self._camera_snapshot(
            current_camera_status,
            tool_calibration_sha256=tool_calibration_sha256,
            workspace_calibration_sha256=workspace_calibration_sha256,
        )
        with self._lock:
            evidence = (
                copy.deepcopy(dict(self._clearance_evidence))
                if isinstance(self._clearance_evidence, Mapping)
                else {}
            )
        try:
            scene, plan, evidence_gates, evidence_blockers, evidence_cameras = (
                self._normalize_clearance_evidence(evidence, configured=configured)
            )
            binding_blockers, invalid_camera_roles = self._clearance_binding_failures(
                evidence,
                pose_sequence=pose_sequence,
                joint_angles_rad=joints,
                current_camera_status=current_camera_status,
                current_camera_snapshot=cameras,
                evidence_cameras=evidence_cameras,
            )
        except Exception:
            scene = dict(_empty_scene(self.required_clearance_mm))
            plan = dict(_empty_plan(configured=configured))
            if evidence:
                plan["state"] = "failed"
            evidence_gates = {
                "pair_synchronized": False,
                "calibrations_verified": False,
                "coverage_verified": False,
                "trajectory_valid": False,
                "clearance_verified": False,
            }
            evidence_blockers = ("unknown_swept_volume",)
            evidence_cameras = {}
            binding_blockers = ("dual_depth_stale",)
            invalid_camera_roles = ("tool", "workspace")
        if binding_blockers:
            for name in evidence_gates:
                evidence_gates[name] = False
            scene["swept_volume_clear"] = False
            if evidence:
                plan["state"] = "failed"
            evidence_blockers = tuple(evidence_blockers) + tuple(binding_blockers)
            for role in invalid_camera_roles:
                if role in evidence_cameras:
                    evidence_cameras[role]["depth_ready"] = False
        for role in ("tool", "workspace"):
            if evidence_cameras.get(role):
                cameras[role] = evidence_cameras[role]

        calibrations_verified = bool(
            _digest(cameras["tool"]["calibration_sha256"])
            and _digest(cameras["workspace"]["calibration_sha256"])
            and evidence_gates["calibrations_verified"]
        )
        # This visualization/preflight feature has no motion-release authority.
        # Retain the argument for wire-compatible callers, but never trust it.
        _ = motion_release_valid
        release_valid = False
        gates = {
            "pose_fresh": pose_fresh,
            "pair_synchronized": evidence_gates["pair_synchronized"],
            "calibrations_verified": calibrations_verified,
            "model_verified": model_verified,
            "coverage_verified": evidence_gates["coverage_verified"],
            "trajectory_valid": evidence_gates["trajectory_valid"],
            "clearance_verified": evidence_gates["clearance_verified"],
            "motion_release_valid": release_valid,
            "admission_ready": False,
            "blockers": [],
        }
        gates["admission_ready"] = all(
            gates[name]
            for name in (
                "pose_fresh",
                "pair_synchronized",
                "calibrations_verified",
                "model_verified",
                "coverage_verified",
                "trajectory_valid",
                "clearance_verified",
                "motion_release_valid",
            )
        )

        blockers = set(evidence_blockers)
        if not configured or xarm.get("available") is not True:
            blockers.add("no_xarm_telemetry")
        if not pose_fresh:
            blockers.add("stale_pose")
        if joints is None:
            blockers.add("invalid_joint_vector")
        if pose_fresh and not model_verified:
            blockers.add("model_tcp_mismatch")
        for role in ("tool", "workspace"):
            if cameras[role]["depth_ready"] is not True:
                blockers.add("{}_depth_unavailable".format(role))
            if not cameras[role]["calibration_sha256"]:
                blockers.add("{}_depth_calibration_missing".format(role))
        if not gates["trajectory_valid"]:
            blockers.add("trajectory_unavailable")
        if not release_valid:
            blockers.add("motion_release_incomplete")
        gates["blockers"] = [
            blocker for blocker in DIGITAL_TWIN_BLOCKERS if blocker in blockers
        ]

        if not configured:
            state = "unavailable"
        elif gates["admission_ready"]:
            state = "ready"
        else:
            state = "blocked"
        return {
            "schema_version": DIGITAL_TWIN_SCHEMA_VERSION,
            "configured": configured,
            "state": state,
            "coordinate_frame": DIGITAL_TWIN_COORDINATE_FRAME,
            "model_sha256": self.kinematics.model_sha256,
            "sample_age_seconds": age,
            "pose_sequence": _bounded_sequence(pose_sequence),
            "joint_angles_rad": list(joints or zero_joints),
            "link_points_mm": [list(point) for point in link_points],
            "tcp_pose_mm_rad": list(tcp_pose),
            "workspace_bounds_mm": {"lower": list(lower), "upper": list(upper)},
            "cameras": cameras,
            "scene": scene,
            "plan": plan,
            "gates": gates,
        }

    @staticmethod
    def _workspace_bounds(
        lower_value: Sequence[float],
        upper_value: Sequence[float],
        metrics: Mapping[str, Any],
    ) -> Tuple[Tuple[float, ...], Tuple[float, ...]]:
        lower = _finite_vector(lower_value, 3)
        upper = _finite_vector(upper_value, 3)
        if (
            lower is None
            or upper is None
            or any(low >= high for low, high in zip(lower, upper))
        ):
            safety = metrics.get("safety")
            safety = safety if isinstance(safety, Mapping) else {}
            lower = _finite_vector(safety.get("tcp_boundary_min_mm"), 3)
            upper = _finite_vector(safety.get("tcp_boundary_max_mm"), 3)
        if (
            lower is None
            or upper is None
            or any(low >= high for low, high in zip(lower, upper))
        ):
            # Manufacturer mechanical envelope, used only as a visual scale.
            lower = (-440.0, -440.0, -165.0)
            upper = (440.0, 440.0, 683.5)
        return lower, upper

    @staticmethod
    def _camera_snapshot(
        camera_status: Mapping[str, Any],
        *,
        tool_calibration_sha256: str,
        workspace_calibration_sha256: str,
    ) -> dict:
        result = {}
        calibration_digests = {
            "tool": _digest(tool_calibration_sha256),
            "workspace": _digest(workspace_calibration_sha256),
        }
        for role in ("tool", "workspace"):
            raw = camera_status.get(role)
            raw = raw if isinstance(raw, Mapping) else {}
            source_id = raw.get("source_id")
            if not isinstance(source_id, str) or len(source_id) > 255:
                source_id = ""
            sequence = _bounded_sequence(raw.get("sequence"))
            age = _bounded_number(
                raw.get("sample_age_seconds"),
                minimum=0.0,
                maximum=_MAX_SAMPLE_AGE_SECONDS,
            )
            value = dict(_empty_camera(source_id, calibration_digests[role]))
            value["frame_sequence"] = sequence
            value["age_seconds"] = age
            result[role] = value
        return result

    def _clearance_binding_failures(
        self,
        evidence: Mapping[str, Any],
        *,
        pose_sequence: Any,
        joint_angles_rad: Optional[Sequence[float]],
        current_camera_status: Mapping[str, Any],
        current_camera_snapshot: Mapping[str, Mapping[str, Any]],
        evidence_cameras: Mapping[str, Mapping[str, Any]],
    ) -> Tuple[Tuple[str, ...], Tuple[str, ...]]:
        """Veto stored evidence unless it is bound to current cached inputs."""

        if not evidence:
            return (), ()
        blockers = set()
        invalid_roles = set()
        binding = evidence.get("_binding")
        if not isinstance(binding, Mapping) or binding.get("schema_version") != 1:
            return (
                "dual_depth_stale",
                "tool_pose_skew",
                "depth_sequence_mismatch",
            ), ("tool", "workspace")

        evaluated_at = _bounded_number(
            binding.get("evaluated_at_monotonic_s"),
            minimum=0.0,
            maximum=100_000_000_000.0,
        )
        try:
            now = float(self._monotonic())
        except Exception:
            now = float("nan")
        if (
            evaluated_at is None
            or not math.isfinite(now)
            or not 0.0 <= now - evaluated_at <= self.clearance_evidence_ttl_seconds
        ):
            blockers.add("dual_depth_stale")

        bound_pose_sequence = binding.get("pose_sequence")
        bound_joints = _lite6_joint_vector(binding.get("joint_angles_rad"))
        current_pose_sequence = _bounded_sequence(pose_sequence)
        joints_match = bool(
            bound_joints is not None
            and joint_angles_rad is not None
            and all(
                math.isclose(before, after, rel_tol=0.0, abs_tol=1e-9)
                for before, after in zip(bound_joints, joint_angles_rad)
            )
        )
        if (
            isinstance(bound_pose_sequence, bool)
            or not isinstance(bound_pose_sequence, int)
            or bound_pose_sequence <= 0
            or bound_pose_sequence != current_pose_sequence
            or not joints_match
        ):
            blockers.add("tool_pose_skew")

        if _digest(binding.get("model_sha256")) != self.kinematics.model_sha256:
            blockers.add("trajectory_unavailable")
        bound_geometry = _digest(binding.get("geometry_profile_sha256"))
        if not bound_geometry or (
            self.geometry_profile_sha256
            and bound_geometry != self.geometry_profile_sha256
        ):
            blockers.add("trajectory_unavailable")

        for role in ("tool", "workspace"):
            current = current_camera_status.get(role)
            current = current if isinstance(current, Mapping) else {}
            evidence_camera = evidence_cameras.get(role)
            evidence_camera = (
                evidence_camera if isinstance(evidence_camera, Mapping) else {}
            )
            binding_sequence = binding.get("{}_frame_sequence".format(role))
            binding_generation = binding.get("{}_camera_generation".format(role))
            binding_mxid = binding.get("{}_camera_mxid".format(role))
            current_sequence = current.get("sequence")
            current_generation = current.get("reconnects", 0)
            current_mxid = current.get("mxid", "")
            evidence_sequence = evidence_camera.get("frame_sequence")
            current_age = _bounded_number(
                current.get("sample_age_seconds"),
                minimum=0.0,
                maximum=_MAX_SAMPLE_AGE_SECONDS,
            )
            healthy = bool(
                current.get("state") == "running"
                and current.get("fresh") is True
                and current_age is not None
                and current_age <= self.clearance_evidence_ttl_seconds
                and current.get("source_id") == evidence_camera.get("source_id")
            )
            sequences_match = bool(
                not isinstance(binding_sequence, bool)
                and isinstance(binding_sequence, int)
                and binding_sequence > 0
                and not isinstance(current_sequence, bool)
                and isinstance(current_sequence, int)
                and current_sequence == binding_sequence == evidence_sequence
            )
            if not healthy:
                blockers.add("{}_depth_unavailable".format(role))
                invalid_roles.add(role)
            if not sequences_match:
                blockers.add("depth_sequence_mismatch")
                invalid_roles.add(role)
            generation_matches = bool(
                not isinstance(binding_generation, bool)
                and isinstance(binding_generation, int)
                and binding_generation >= 0
                and not isinstance(current_generation, bool)
                and isinstance(current_generation, int)
                and current_generation == binding_generation
            )
            mxid_matches = bool(
                isinstance(binding_mxid, str)
                and isinstance(current_mxid, str)
                and current_mxid == binding_mxid
            )
            if not generation_matches or not mxid_matches:
                blockers.add("depth_sequence_mismatch")
                invalid_roles.add(role)

            configured_camera = current_camera_snapshot.get(role)
            configured_camera = (
                configured_camera if isinstance(configured_camera, Mapping) else {}
            )
            configured_digest = _digest(configured_camera.get("calibration_sha256"))
            evidence_digest = _digest(evidence_camera.get("calibration_sha256"))
            bound_digest = _digest(binding.get("{}_calibration_sha256".format(role)))
            if not configured_digest or not (
                configured_digest == evidence_digest == bound_digest
            ):
                blockers.add("{}_depth_calibration_missing".format(role))
                invalid_roles.add(role)

        return (
            tuple(blocker for blocker in DIGITAL_TWIN_BLOCKERS if blocker in blockers),
            tuple(role for role in ("tool", "workspace") if role in invalid_roles),
        )

    def _normalize_clearance_evidence(
        self,
        evidence: Mapping[str, Any],
        *,
        configured: bool,
    ) -> Tuple[dict, dict, dict, Tuple[str, ...], dict]:
        scene = dict(_empty_scene(self.required_clearance_mm))
        plan = dict(_empty_plan(configured=configured))
        gates = {
            "pair_synchronized": False,
            "calibrations_verified": False,
            "coverage_verified": False,
            "trajectory_valid": False,
            "clearance_verified": False,
        }
        blocker_set = set()
        cameras = {}
        if not evidence:
            return scene, plan, gates, (), cameras

        raw_scene = evidence.get("scene")
        raw_scene = raw_scene if isinstance(raw_scene, Mapping) else {}
        scene_sha256 = _digest(raw_scene.get("scene_sha256"))
        pair_skew = _bounded_number(
            raw_scene.get("pair_skew_seconds"), minimum=0.0, maximum=60.0
        )
        coverage = _bounded_number(
            raw_scene.get("coverage_ratio"), minimum=0.0, maximum=1.0
        )
        evidence_required = _bounded_number(
            raw_scene.get("required_clearance_mm"), minimum=0.0, maximum=10_000.0
        )
        required_matches = evidence_required == self.required_clearance_mm
        swept_volume_clear = raw_scene.get("swept_volume_clear") is True
        scene.update(
            {
                "scene_sha256": scene_sha256,
                "pair_skew_seconds": pair_skew,
                "coverage_ratio": coverage if coverage is not None else 0.0,
                # The configured requirement is authoritative.  Evidence may
                # prove compliance with that exact value, but it may never
                # replace or weaken it in the UI contract.
                "required_clearance_mm": self.required_clearance_mm,
                "swept_volume_clear": bool(swept_volume_clear and required_matches),
            }
        )
        for name in (
            "nearest_obstacle_mm",
            "current_clearance_mm",
            "planned_min_clearance_mm",
        ):
            scene[name] = _bounded_number(
                raw_scene.get(name), minimum=0.0, maximum=1_000_000.0
            )

        raw_plan = evidence.get("plan")
        raw_plan = raw_plan if isinstance(raw_plan, Mapping) else {}
        state = raw_plan.get("state")
        if state not in {
            "unavailable",
            "none",
            "planned",
            "validated",
            "active",
            "failed",
        }:
            state = "failed"
        path = raw_plan.get("tcp_path_mm")
        path = path if isinstance(path, (tuple, list)) else ()
        parsed_path = []
        path_valid = bool(path)
        for point in path[:_MAX_PLAN_POINTS]:
            vector = _finite_vector(point, 3)
            if vector is None:
                parsed_path = []
                path_valid = False
                state = "failed"
                break
            parsed_path.append(list(vector))
        raw_waypoint_count = raw_plan.get("waypoint_count")
        waypoint_count_valid = not (
            isinstance(raw_waypoint_count, bool)
            or not isinstance(raw_waypoint_count, int)
            or not 0 <= raw_waypoint_count <= 1_000_000
        )
        waypoint_count = raw_waypoint_count if waypoint_count_valid else 0
        plan_sha256 = _digest(raw_plan.get("plan_sha256"))
        trajectory_structure_valid = bool(
            state in {"validated", "active"}
            and plan_sha256
            and path_valid
            and len(parsed_path) > 0
            and waypoint_count_valid
            and waypoint_count >= len(parsed_path)
        )
        if state in {"validated", "active"} and not trajectory_structure_valid:
            state = "failed"
        if not required_matches:
            state = "failed"
        plan.update(
            {
                "state": state,
                "plan_sha256": plan_sha256,
                "waypoint_count": waypoint_count,
                "tcp_path_mm": parsed_path,
            }
        )

        raw_gates = evidence.get("gates")
        raw_gates = raw_gates if isinstance(raw_gates, Mapping) else {}
        raw_blockers = raw_gates.get("blockers")
        if isinstance(raw_blockers, (tuple, list)):
            blocker_set.update(
                value
                for value in raw_blockers
                if isinstance(value, str) and value in _DIGITAL_TWIN_BLOCKER_SET
            )

        raw_cameras = evidence.get("cameras")
        raw_cameras = raw_cameras if isinstance(raw_cameras, Mapping) else {}
        for role in ("tool", "workspace"):
            raw = raw_cameras.get(role)
            if not isinstance(raw, Mapping):
                continue
            source_id = raw.get("source_id")
            if not isinstance(source_id, str) or len(source_id) > 255:
                source_id = ""
            value = dict(
                _empty_camera(source_id, _digest(raw.get("calibration_sha256")))
            )
            value["frame_sequence"] = _bounded_sequence(raw.get("frame_sequence"))
            value["age_seconds"] = _bounded_number(
                raw.get("age_seconds"),
                minimum=0.0,
                maximum=_MAX_SAMPLE_AGE_SECONDS,
            )
            value["depth_ready"] = raw.get("depth_ready") is True
            cameras[role] = value

        # Producer booleans remain vetoes, but can no longer manufacture a
        # positive gate without the bounded evidence that gate represents.
        claimed = {name: raw_gates.get(name) is True for name in gates}
        both_cameras_ready = all(
            role in cameras and cameras[role]["depth_ready"] is True
            for role in ("tool", "workspace")
        )
        both_calibrations_valid = all(
            role in cameras and bool(_digest(cameras[role]["calibration_sha256"]))
            for role in ("tool", "workspace")
        )
        gates["pair_synchronized"] = bool(
            claimed["pair_synchronized"]
            and both_cameras_ready
            and pair_skew is not None
        )
        gates["calibrations_verified"] = bool(
            claimed["calibrations_verified"] and both_calibrations_valid
        )
        gates["coverage_verified"] = bool(
            claimed["coverage_verified"] and coverage == 1.0 and scene_sha256
        )
        gates["trajectory_valid"] = bool(
            claimed["trajectory_valid"]
            and trajectory_structure_valid
            and required_matches
        )
        gates["clearance_verified"] = bool(
            claimed["clearance_verified"]
            and scene["swept_volume_clear"] is True
            and required_matches
            and gates["pair_synchronized"]
            and gates["calibrations_verified"]
            and gates["coverage_verified"]
            and gates["trajectory_valid"]
            and scene["current_clearance_mm"] is not None
            and scene["current_clearance_mm"] >= self.required_clearance_mm
            and scene["planned_min_clearance_mm"] is not None
            and scene["planned_min_clearance_mm"] >= self.required_clearance_mm
        )

        # Contradictory positive claims are rendered as a failed plan.  A
        # producer-side negative claim is never upgraded by this consumer.
        if (claimed["trajectory_valid"] and not gates["trajectory_valid"]) or (
            claimed["clearance_verified"] and not gates["clearance_verified"]
        ):
            plan["state"] = "failed"
            gates["trajectory_valid"] = False
            gates["clearance_verified"] = False
            scene["swept_volume_clear"] = False

        if both_cameras_ready and not gates["pair_synchronized"]:
            blocker_set.add("dual_depth_skew")
        if not gates["coverage_verified"]:
            blocker_set.add("unknown_swept_volume")
        if claimed["clearance_verified"] and not gates["clearance_verified"]:
            blocker_set.add("unknown_swept_volume")
        if not required_matches:
            blocker_set.update(("trajectory_unavailable", "unknown_swept_volume"))
        if (
            scene["current_clearance_mm"] is not None
            and scene["current_clearance_mm"] < self.required_clearance_mm
        ) or (
            scene["planned_min_clearance_mm"] is not None
            and scene["planned_min_clearance_mm"] < self.required_clearance_mm
        ):
            blocker_set.add("clearance_below_limit")

        blockers = tuple(
            blocker for blocker in DIGITAL_TWIN_BLOCKERS if blocker in blocker_set
        )
        return scene, plan, gates, blockers, cameras


__all__ = [
    "DIGITAL_TWIN_BLOCKERS",
    "DIGITAL_TWIN_COORDINATE_FRAME",
    "DIGITAL_TWIN_SCHEMA_VERSION",
    "LITE6_JOINT_LIMITS_RAD",
    "DigitalTwinSnapshotProvider",
    "Lite6VisualizationKinematics",
]
