"""Root-pinned, site-specific safety commissioning for the Lite6 workcell.

This module records and validates commissioning evidence.  It deliberately
does not expose a controller command, a motion permit, or a way to turn an
assessment into motion authority.  A valid profile is one additional veto
gate; the canonical runtime motion-release gate remains authoritative.
"""

from __future__ import annotations

import hashlib
import hmac
import json
import math
import os
import re
import stat
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from types import MappingProxyType
from typing import Any, Mapping, Optional, Sequence, Tuple

SITE_SAFETY_SCHEMA = "labcobot.site-safety-commissioning.v1"
SITE_SAFETY_STATUS_SCHEMA = "labcobot.site-safety-status.v1"
_MAX_PROFILE_BYTES = 256 * 1024
_MAX_PROFILE_LIFETIME = timedelta(days=366)
_MAX_EVIDENCE_AGE_AT_APPROVAL = timedelta(days=30)
_SHA256 = re.compile(r"[0-9a-f]{64}")
_SAFE_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.:@+-]{0,127}")
_FIRMWARE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.+-]{0,63}")
_LITE6_JOINT_LOWER_RADIANS = tuple(
    math.radians(value) for value in (-360, -150, -3.5, -360, -124, -360)
)
_LITE6_JOINT_UPPER_RADIANS = tuple(
    math.radians(value) for value in (360, 150, 300, 360, 124, 360)
)

_TOP_LEVEL_KEYS = frozenset(
    {
        "schema",
        "site",
        "robot",
        "artifacts",
        "controller_readback",
        "swept_volume",
        "physical_safety",
        "cameras",
        "trajectories",
        "approval",
        "limitations",
    }
)
_ARTIFACT_KEYS = frozenset(
    {
        "safety_profile_sha256",
        "geometry_profile_sha256",
        "sampling_tool_profile_sha256",
        "tool_calibration_sha256",
        "workspace_calibration_sha256",
        "tray_fixture_sha256",
        "process_profile_sha256",
        "home_trajectory_sha256",
        "recovery_trajectory_sha256",
    }
)


class SiteSafetyError(ValueError):
    """A bounded validation failure with a stable operator-facing code."""

    def __init__(self, code: str, message: str) -> None:
        self.code = str(code)
        super().__init__(message)


@dataclass(frozen=True)
class ClearanceSegment:
    segment_id: str
    trajectory_sha256: str
    evidence_sha256: str
    minimum_clearance_mm: float


@dataclass(frozen=True)
class SiteSafetyCommissioningProfile:
    """Normalized immutable evidence and every binding used at runtime."""

    profile_sha256: str
    site_id: str
    device_uid: str
    commissioned_at: datetime
    valid_until: datetime
    robot_serial: str
    firmware_version: str
    artifact_bindings: Mapping[str, str]
    tool_camera_mxid: str
    workspace_camera_mxid: str
    required_clearance_mm: float
    minimum_clearance_mm: float
    segments: Tuple[ClearanceSegment, ...]
    controller_readback_sha256: str
    physical_estop_sha256: str
    software_stop_sha256: str
    guarding_evidence_sha256: str
    camera_evidence_sha256: str
    approval_evidence_sha256: str
    approved_by: str
    qualification: str
    approved_at: datetime
    source: Mapping[str, Any]

    @classmethod
    def from_document(
        cls,
        document: Mapping[str, Any],
        *,
        profile_sha256: str,
    ) -> "SiteSafetyCommissioningProfile":
        _digest(profile_sha256, "profile SHA-256")
        root = _exact_mapping(document, _TOP_LEVEL_KEYS, "site-safety profile")
        if root["schema"] != SITE_SAFETY_SCHEMA:
            raise SiteSafetyError("site_safety_schema_invalid", "Unsupported schema")

        site = _exact_mapping(
            root["site"],
            {"site_id", "device_uid", "commissioned_at", "valid_until"},
            "site",
        )
        site_id = _identifier(site["site_id"], "site ID")
        device_uid = _identifier(site["device_uid"], "device UID")
        commissioned_at = _timestamp(site["commissioned_at"], "commissioned_at")
        valid_until = _timestamp(site["valid_until"], "valid_until")
        if not commissioned_at < valid_until <= commissioned_at + _MAX_PROFILE_LIFETIME:
            raise SiteSafetyError(
                "site_safety_time_invalid",
                "Site evidence validity must be positive and no longer than 366 days",
            )

        robot = _exact_mapping(
            root["robot"],
            {"model", "axis_count", "device_type", "serial", "firmware"},
            "robot",
        )
        if (
            robot["model"] != "lite6"
            or type(robot["axis_count"]) is not int
            or robot["axis_count"] != 6
            or type(robot["device_type"]) is not int
            or robot["device_type"] != 9
        ):
            raise SiteSafetyError(
                "site_safety_robot_mismatch",
                "Commissioning is restricted to one Lite6 controller type 9",
            )
        robot_serial = _identifier(robot["serial"], "robot serial")
        firmware = _firmware(robot["firmware"])

        raw_artifacts = _exact_mapping(root["artifacts"], _ARTIFACT_KEYS, "artifacts")
        artifacts = MappingProxyType(
            {
                name: _digest(raw_artifacts[name], name)
                for name in sorted(_ARTIFACT_KEYS)
            }
        )

        controller = _validate_controller(
            root["controller_readback"],
            robot_serial=robot_serial,
            firmware=firmware,
            safety_profile_sha256=artifacts["safety_profile_sha256"],
            commissioned_at=commissioned_at,
        )
        segments, required_clearance, minimum_clearance = _validate_swept_volume(
            root["swept_volume"],
            artifacts=artifacts,
            commissioned_at=commissioned_at,
        )
        physical = _validate_physical_safety(
            root["physical_safety"], commissioned_at=commissioned_at
        )
        cameras = _validate_cameras(
            root["cameras"], artifacts=artifacts, commissioned_at=commissioned_at
        )
        _validate_trajectories(
            root["trajectories"],
            artifacts=artifacts,
            required_clearance_mm=required_clearance,
            commissioned_at=commissioned_at,
        )
        approval = _validate_approval(
            root["approval"],
            commissioned_at=commissioned_at,
            valid_until=valid_until,
        )
        limitations = _exact_mapping(
            root["limitations"],
            {
                "motion_authority",
                "site_values_not_mechanical_maxima",
                "changes_require_recommissioning",
            },
            "limitations",
        )
        if (
            limitations["motion_authority"] != "none"
            or limitations["site_values_not_mechanical_maxima"] is not True
            or limitations["changes_require_recommissioning"] is not True
        ):
            raise SiteSafetyError(
                "site_safety_authority_invalid",
                "Site evidence must remain non-authorizing and site-specific",
            )
        if approval["approved_at"] < max(
            commissioned_at,
            controller["observed_at"],
            physical["physical_tested_at"],
            physical["software_tested_at"],
            cameras["observed_at"],
        ):
            raise SiteSafetyError(
                "site_safety_approval_invalid",
                "Competent-person approval predates commissioning evidence",
            )

        return cls(
            profile_sha256=profile_sha256,
            site_id=site_id,
            device_uid=device_uid,
            commissioned_at=commissioned_at,
            valid_until=valid_until,
            robot_serial=robot_serial,
            firmware_version=firmware,
            artifact_bindings=artifacts,
            tool_camera_mxid=cameras["tool_mxid"],
            workspace_camera_mxid=cameras["workspace_mxid"],
            required_clearance_mm=required_clearance,
            minimum_clearance_mm=minimum_clearance,
            segments=segments,
            controller_readback_sha256=controller["evidence_sha256"],
            physical_estop_sha256=physical["physical_evidence_sha256"],
            software_stop_sha256=physical["software_evidence_sha256"],
            guarding_evidence_sha256=physical["guarding_evidence_sha256"],
            camera_evidence_sha256=cameras["evidence_sha256"],
            approval_evidence_sha256=approval["evidence_sha256"],
            approved_by=approval["approved_by"],
            qualification=approval["qualification"],
            approved_at=approval["approved_at"],
            source=_freeze_json(root),
        )


@dataclass(frozen=True)
class SiteSafetyRuntimeBindings:
    """Current immutable runtime identities compared with the site record."""

    device_uid: str = ""
    robot_serial: str = ""
    firmware_version: str = ""
    safety_profile_sha256: str = ""
    geometry_profile_sha256: str = ""
    sampling_tool_profile_sha256: str = ""
    tool_calibration_sha256: str = ""
    workspace_calibration_sha256: str = ""
    tool_camera_mxid: str = ""
    workspace_camera_mxid: str = ""
    tray_fixture_sha256: str = ""
    process_profile_sha256: str = ""
    home_trajectory_sha256: str = ""
    recovery_trajectory_sha256: str = ""
    required_clearance_mm: float = 0.0
    controller_profile_consistent: bool = False


class SiteSafetyCommissioningGate:
    """Evaluate current bindings and live readbacks without granting motion."""

    def __init__(
        self,
        profile: SiteSafetyCommissioningProfile,
        bindings: SiteSafetyRuntimeBindings,
        *,
        clock: Any = None,
    ) -> None:
        if not isinstance(profile, SiteSafetyCommissioningProfile):
            raise TypeError("site-safety profile is invalid")
        if not isinstance(bindings, SiteSafetyRuntimeBindings):
            raise TypeError("site-safety runtime bindings are invalid")
        self.profile = profile
        self.bindings = bindings
        self.clock = clock or (lambda: datetime.now(timezone.utc))

    def status(
        self,
        *,
        controller: Optional[Mapping[str, Any]] = None,
        cameras: Optional[Mapping[str, Any]] = None,
        sampling_tool: Optional[Mapping[str, Any]] = None,
    ) -> Mapping[str, Any]:
        profile = self.profile
        blockers = list(self._binding_blockers())
        try:
            now = _aware_utc(self.clock(), "current time")
        except Exception:
            now = None
            blockers.append("site_safety_clock_invalid")
        if (
            now is not None
            and not profile.commissioned_at <= now <= profile.valid_until
        ):
            blockers.append("site_safety_evidence_expired")
        controller_value = controller if isinstance(controller, Mapping) else {}
        if (
            controller_value.get("ready") is not True
            or controller_value.get("controller_binding_verified") is not True
        ):
            blockers.append("site_safety_controller_unverified")
        camera_value = cameras if isinstance(cameras, Mapping) else {}
        for role, expected in (
            ("tool", profile.tool_camera_mxid),
            ("workspace", profile.workspace_camera_mxid),
        ):
            current = camera_value.get(role)
            if not isinstance(current, Mapping) or (
                current.get("state") != "running"
                or current.get("fresh") is not True
                or str(
                    current.get("mxid")
                    or current.get("source_id")
                    or current.get("device_id")
                    or ""
                )
                != expected
            ):
                blockers.append("site_safety_camera_unverified")
                break
        tool_value = sampling_tool if isinstance(sampling_tool, Mapping) else {}
        if self.bindings.sampling_tool_profile_sha256 and (
            tool_value.get("ready") is not True
            or not hmac.compare_digest(
                str(tool_value.get("profile_sha256") or ""),
                profile.artifact_bindings["sampling_tool_profile_sha256"],
            )
        ):
            blockers.append("site_safety_tool_unverified")
        blockers = tuple(dict.fromkeys(blockers))
        return {
            "schema": SITE_SAFETY_STATUS_SCHEMA,
            "configured": True,
            "ready": not blockers,
            "motion_authority": "none",
            "profile_sha256": profile.profile_sha256,
            "site_id": profile.site_id,
            "device_uid": profile.device_uid,
            "valid_until": _format_timestamp(profile.valid_until),
            "required_clearance_mm": profile.required_clearance_mm,
            "verified_minimum_clearance_mm": profile.minimum_clearance_mm,
            "verified_segment_count": len(profile.segments),
            "blockers": list(blockers),
        }

    def _binding_blockers(self) -> Tuple[str, ...]:
        expected = self.profile.artifact_bindings
        values = (
            (self.bindings.device_uid, self.profile.device_uid),
            (self.bindings.robot_serial, self.profile.robot_serial),
            (self.bindings.firmware_version, self.profile.firmware_version),
            (self.bindings.safety_profile_sha256, expected["safety_profile_sha256"]),
            (
                self.bindings.geometry_profile_sha256,
                expected["geometry_profile_sha256"],
            ),
            (
                self.bindings.sampling_tool_profile_sha256,
                expected["sampling_tool_profile_sha256"],
            ),
            (
                self.bindings.tool_calibration_sha256,
                expected["tool_calibration_sha256"],
            ),
            (
                self.bindings.workspace_calibration_sha256,
                expected["workspace_calibration_sha256"],
            ),
            (self.bindings.tool_camera_mxid, self.profile.tool_camera_mxid),
            (self.bindings.workspace_camera_mxid, self.profile.workspace_camera_mxid),
            (
                self.bindings.tray_fixture_sha256,
                expected["tray_fixture_sha256"],
            ),
            (
                self.bindings.process_profile_sha256,
                expected["process_profile_sha256"],
            ),
            (
                self.bindings.home_trajectory_sha256,
                expected["home_trajectory_sha256"],
            ),
            (
                self.bindings.recovery_trajectory_sha256,
                expected["recovery_trajectory_sha256"],
            ),
        )
        if any(
            not left or not right or not hmac.compare_digest(str(left), str(right))
            for left, right in values
        ):
            return ("site_safety_binding_mismatch",)
        try:
            clearance_matches = math.isclose(
                float(self.bindings.required_clearance_mm),
                self.profile.required_clearance_mm,
                rel_tol=0.0,
                abs_tol=1e-6,
            )
        except (TypeError, ValueError, OverflowError):
            clearance_matches = False
        if (
            self.bindings.controller_profile_consistent is not True
            or not clearance_matches
        ):
            return ("site_safety_binding_mismatch",)
        return ()


def unavailable_site_safety_status() -> Mapping[str, Any]:
    return {
        "schema": SITE_SAFETY_STATUS_SCHEMA,
        "configured": False,
        "ready": False,
        "motion_authority": "none",
        "profile_sha256": "",
        "site_id": "",
        "device_uid": "",
        "valid_until": "",
        "required_clearance_mm": None,
        "verified_minimum_clearance_mm": None,
        "verified_segment_count": 0,
        "blockers": ["site_safety_not_configured"],
    }


def load_site_safety_commissioning_profile(
    path: str,
    *,
    expected_sha256: str,
) -> SiteSafetyCommissioningProfile:
    """Load one bounded root-owned regular profile below ``/etc/labcobot``."""

    digest = _digest(expected_sha256, "site-safety profile SHA-256")
    source = Path(path)
    root = Path("/etc/labcobot")
    if not source.is_absolute() or ".." in source.parts:
        raise SiteSafetyError("site_safety_path_invalid", "Profile path is unsafe")
    try:
        relative = source.relative_to(root)
    except ValueError as exc:
        raise SiteSafetyError(
            "site_safety_path_invalid", "Profile must be below /etc/labcobot"
        ) from exc
    if not relative.parts:
        raise SiteSafetyError("site_safety_path_invalid", "Profile path is unsafe")
    try:
        before = source.lstat()
    except OSError as exc:
        raise SiteSafetyError(
            "site_safety_profile_unavailable", "Profile could not be inspected"
        ) from exc
    if (
        stat.S_ISLNK(before.st_mode)
        or not stat.S_ISREG(before.st_mode)
        or before.st_uid != 0
        or bool(before.st_mode & (stat.S_IWGRP | stat.S_IWOTH))
        or before.st_nlink != 1
    ):
        raise SiteSafetyError(
            "site_safety_profile_unprotected",
            "Profile must be a single-link, root-owned protected regular file",
        )
    descriptor = -1
    try:
        descriptor = os.open(
            str(source),
            os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0),
        )
        opened = os.fstat(descriptor)
        if (
            opened.st_dev != before.st_dev
            or opened.st_ino != before.st_ino
            or opened.st_nlink != 1
            or not stat.S_ISREG(opened.st_mode)
        ):
            raise SiteSafetyError(
                "site_safety_profile_changed", "Profile changed while opening"
            )
        chunks = []
        remaining = _MAX_PROFILE_BYTES + 1
        while remaining:
            chunk = os.read(descriptor, min(64 * 1024, remaining))
            if not chunk:
                break
            chunks.append(chunk)
            remaining -= len(chunk)
        payload = b"".join(chunks)
    except SiteSafetyError:
        raise
    except OSError as exc:
        raise SiteSafetyError(
            "site_safety_profile_unavailable", "Profile could not be read"
        ) from exc
    finally:
        if descriptor >= 0:
            os.close(descriptor)
    if not payload or len(payload) > _MAX_PROFILE_BYTES:
        raise SiteSafetyError("site_safety_profile_invalid", "Profile size is invalid")
    actual = hashlib.sha256(payload).hexdigest()
    if not hmac.compare_digest(actual, digest):
        raise SiteSafetyError(
            "site_safety_hash_mismatch", "Profile SHA-256 does not match"
        )
    document = _strict_json(payload)
    return SiteSafetyCommissioningProfile.from_document(document, profile_sha256=actual)


def site_profile_matches_controller_profile(
    site_profile: SiteSafetyCommissioningProfile,
    controller_profile: Any,
) -> bool:
    """Compare the archived controller readback with the pinned live profile."""

    if not isinstance(site_profile, SiteSafetyCommissioningProfile):
        return False
    controller = site_profile.source.get("controller_readback")
    if not isinstance(controller, Mapping):
        return False
    expected_firmware = str(
        getattr(controller_profile, "expected_firmware_version", "") or ""
    )
    if expected_firmware[:1] in {"v", "V"}:
        expected_firmware = expected_firmware[1:]
    try:
        return bool(
            hmac.compare_digest(
                str(getattr(controller_profile, "profile_sha256", "")),
                site_profile.artifact_bindings["safety_profile_sha256"],
            )
            and str(getattr(controller_profile, "expected_robot_sn", ""))
            == site_profile.robot_serial
            and expected_firmware == site_profile.firmware_version
            and _vectors_equal(
                controller["tcp_boundary_mm"], controller_profile.tcp_boundary_mm
            )
            and _vectors_equal(
                controller["joint_ranges_radians"],
                controller_profile.joint_ranges_radians,
            )
            and math.isclose(
                float(controller["max_tcp_speed_mm_per_second"]),
                float(controller_profile.max_tcp_speed_mm_per_second),
                rel_tol=0.0,
                abs_tol=1e-6,
            )
            and math.isclose(
                float(controller["max_joint_speed_radians_per_second"]),
                float(controller_profile.max_joint_speed_radians_per_second),
                rel_tol=0.0,
                abs_tol=1e-6,
            )
            and _vectors_equal(
                controller["tcp_offset_xyz_mm_rpy_degrees"],
                controller_profile.expected_tcp_offset_xyz_mm_rpy_degrees,
            )
            and math.isclose(
                float(controller["payload_kg"]),
                float(controller_profile.expected_tcp_load_kg),
                rel_tol=0.0,
                abs_tol=1e-6,
            )
            and _vectors_equal(
                controller["center_of_gravity_mm"],
                controller_profile.expected_tcp_center_of_gravity_mm,
            )
            and int(controller["collision_sensitivity"])
            >= int(controller_profile.minimum_collision_sensitivity)
        )
    except (AttributeError, TypeError, ValueError, OverflowError):
        return False


def _validate_controller(
    value: Any,
    *,
    robot_serial: str,
    firmware: str,
    safety_profile_sha256: str,
    commissioned_at: datetime,
) -> Mapping[str, Any]:
    controller = _exact_mapping(
        value,
        {
            "evidence_sha256",
            "observed_at",
            "robot_serial",
            "firmware",
            "safety_profile_sha256",
            "reduced_mode_enabled",
            "safety_boundary_enabled",
            "collision_rebound_disabled",
            "collision_sensitivity",
            "tcp_boundary_mm",
            "joint_ranges_radians",
            "max_tcp_speed_mm_per_second",
            "max_joint_speed_radians_per_second",
            "tcp_offset_xyz_mm_rpy_degrees",
            "payload_kg",
            "center_of_gravity_mm",
        },
        "controller readback",
    )
    observed_at = _timestamp(controller["observed_at"], "controller observed_at")
    _recent_evidence(observed_at, commissioned_at, "controller readback")
    if (
        controller["robot_serial"] != robot_serial
        or _firmware(controller["firmware"]) != firmware
        or controller["safety_profile_sha256"] != safety_profile_sha256
        or controller["reduced_mode_enabled"] is not True
        or controller["safety_boundary_enabled"] is not True
        or controller["collision_rebound_disabled"] is not True
        or type(controller["collision_sensitivity"]) is not int
        or not 1 <= controller["collision_sensitivity"] <= 5
    ):
        raise SiteSafetyError(
            "site_safety_controller_invalid", "Controller readback is inconsistent"
        )
    boundary = _vector(controller["tcp_boundary_mm"], 6, "TCP boundary")
    if any(boundary[index * 2] <= boundary[index * 2 + 1] for index in range(3)):
        raise SiteSafetyError(
            "site_safety_controller_invalid", "TCP boundary is invalid"
        )
    ranges = _vector(controller["joint_ranges_radians"], 14, "joint ranges")
    if any(
        ranges[index * 2] >= ranges[index * 2 + 1]
        or ranges[index * 2] <= _LITE6_JOINT_LOWER_RADIANS[index]
        or ranges[index * 2 + 1] >= _LITE6_JOINT_UPPER_RADIANS[index]
        for index in range(6)
    ) or ranges[12:] != (0.0, 0.0):
        raise SiteSafetyError(
            "site_safety_controller_invalid",
            "Joint ranges must be conservative site limits",
        )
    tcp_speed = _positive(controller["max_tcp_speed_mm_per_second"], "TCP speed", 250.0)
    joint_speed = _positive(
        controller["max_joint_speed_radians_per_second"], "joint speed", math.pi / 3
    )
    if math.isclose(tcp_speed, 250.0) or math.isclose(joint_speed, math.pi / 3):
        raise SiteSafetyError(
            "site_safety_controller_invalid",
            "Site speed limits must be below loader ceilings",
        )
    tcp_offset = _vector(controller["tcp_offset_xyz_mm_rpy_degrees"], 6, "TCP offset")
    if any(abs(item) > 1000.0 for item in tcp_offset[:3]) or any(
        abs(item) > 360.0 for item in tcp_offset[3:]
    ):
        raise SiteSafetyError("site_safety_controller_invalid", "TCP offset is invalid")
    _nonnegative(controller["payload_kg"], "payload", 10.0)
    center = _vector(controller["center_of_gravity_mm"], 3, "center of gravity")
    if any(abs(item) > 1000.0 for item in center):
        raise SiteSafetyError(
            "site_safety_controller_invalid", "Center of gravity is invalid"
        )
    return {
        "evidence_sha256": _digest(controller["evidence_sha256"], "readback evidence"),
        "observed_at": observed_at,
    }


def _validate_swept_volume(
    value: Any,
    *,
    artifacts: Mapping[str, str],
    commissioned_at: datetime,
) -> Tuple[Tuple[ClearanceSegment, ...], float, float]:
    swept = _exact_mapping(
        value,
        {
            "evidence_sha256",
            "observed_at",
            "geometry_profile_sha256",
            "tray_fixture_sha256",
            "required_clearance_mm",
            "full_tray_minimum_clearance_mm",
            "full_tray_loaded",
            "reduced_speed_dry_run",
            "segments",
        },
        "swept volume",
    )
    if (
        swept["geometry_profile_sha256"] != artifacts["geometry_profile_sha256"]
        or swept["tray_fixture_sha256"] != artifacts["tray_fixture_sha256"]
        or swept["full_tray_loaded"] is not True
        or swept["reduced_speed_dry_run"] is not True
    ):
        raise SiteSafetyError(
            "site_safety_clearance_invalid",
            "Swept-volume artifacts or dry-run conditions do not match",
        )
    _recent_evidence(
        _timestamp(swept["observed_at"], "swept-volume observed_at"),
        commissioned_at,
        "swept-volume evidence",
    )
    required = _positive(swept["required_clearance_mm"], "required clearance", 1000.0)
    minimum = _positive(
        swept["full_tray_minimum_clearance_mm"], "minimum clearance", 10_000.0
    )
    raw_segments = swept["segments"]
    if not isinstance(raw_segments, list) or not 1 <= len(raw_segments) <= 512:
        raise SiteSafetyError(
            "site_safety_clearance_invalid", "Every dry-run segment is required"
        )
    result = []
    identifiers = set()
    for raw in raw_segments:
        item = _exact_mapping(
            raw,
            {
                "segment_id",
                "trajectory_sha256",
                "evidence_sha256",
                "minimum_clearance_mm",
                "links_clear",
                "tool_clear",
                "cables_clear",
                "carried_object_clear",
                "tray_fixture_clear",
                "camera_visibility_clear",
            },
            "clearance segment",
        )
        segment_id = _identifier(item["segment_id"], "segment ID")
        if segment_id in identifiers:
            raise SiteSafetyError(
                "site_safety_clearance_invalid", "Segment IDs must be unique"
            )
        identifiers.add(segment_id)
        for gate in (
            "links_clear",
            "tool_clear",
            "cables_clear",
            "carried_object_clear",
            "tray_fixture_clear",
            "camera_visibility_clear",
        ):
            if item[gate] is not True:
                raise SiteSafetyError(
                    "site_safety_clearance_invalid",
                    "Every swept-volume segment must be conservatively clear",
                )
        clearance = _positive(
            item["minimum_clearance_mm"], "segment clearance", 10_000.0
        )
        if clearance < required:
            raise SiteSafetyError(
                "site_safety_clearance_invalid", "A segment is below site clearance"
            )
        result.append(
            ClearanceSegment(
                segment_id=segment_id,
                trajectory_sha256=_digest(
                    item["trajectory_sha256"], "segment trajectory"
                ),
                evidence_sha256=_digest(item["evidence_sha256"], "segment evidence"),
                minimum_clearance_mm=clearance,
            )
        )
    calculated = min(segment.minimum_clearance_mm for segment in result)
    if not math.isclose(minimum, calculated, rel_tol=0.0, abs_tol=1e-6):
        raise SiteSafetyError(
            "site_safety_clearance_invalid",
            "Full-tray minimum must equal the minimum reviewed segment",
        )
    _digest(swept["evidence_sha256"], "swept-volume evidence")
    return tuple(result), required, minimum


def _validate_physical_safety(
    value: Any, *, commissioned_at: datetime
) -> Mapping[str, Any]:
    physical = _exact_mapping(
        value,
        {"guarding", "physical_estop", "software_stop"},
        "physical safety",
    )
    guarding = _exact_mapping(
        physical["guarding"],
        {"evidence_sha256", "verified_at", "verified_by", "guarding_complete"},
        "guarding",
    )
    if guarding["guarding_complete"] is not True:
        raise SiteSafetyError("site_safety_guarding_invalid", "Guarding is incomplete")
    guarding_at = _timestamp(guarding["verified_at"], "guarding verified_at")
    _recent_evidence(guarding_at, commissioned_at, "guarding evidence")
    _identifier(guarding["verified_by"], "guarding verifier")

    physical_stop = _exact_mapping(
        physical["physical_estop"],
        {
            "evidence_sha256",
            "tested_at",
            "performed_by",
            "witnessed_by",
            "motion_inhibited",
            "manual_reset_required",
            "no_automatic_restart",
        },
        "physical emergency stop",
    )
    performer = _identifier(physical_stop["performed_by"], "E-stop performer")
    witness = _identifier(physical_stop["witnessed_by"], "E-stop witness")
    if performer == witness or any(
        physical_stop[name] is not True
        for name in (
            "motion_inhibited",
            "manual_reset_required",
            "no_automatic_restart",
        )
    ):
        raise SiteSafetyError(
            "site_safety_estop_invalid", "Physical E-stop evidence is invalid"
        )
    physical_at = _timestamp(physical_stop["tested_at"], "physical E-stop tested_at")
    _recent_evidence(physical_at, commissioned_at, "physical E-stop evidence")

    software = _exact_mapping(
        physical["software_stop"],
        {
            "evidence_sha256",
            "tested_at",
            "measured_by",
            "latency_ms",
            "stopping_distance_mm",
            "independent_from_physical_estop",
        },
        "software STOP",
    )
    if software["independent_from_physical_estop"] is not True:
        raise SiteSafetyError(
            "site_safety_software_stop_invalid", "Software STOP must be separate"
        )
    software_at = _timestamp(software["tested_at"], "software STOP tested_at")
    _recent_evidence(software_at, commissioned_at, "software STOP evidence")
    _identifier(software["measured_by"], "software STOP measurer")
    _positive(software["latency_ms"], "software STOP latency", 60_000.0)
    _nonnegative(
        software["stopping_distance_mm"], "software stopping distance", 10_000.0
    )
    physical_digest = _digest(
        physical_stop["evidence_sha256"], "physical E-stop evidence"
    )
    software_digest = _digest(software["evidence_sha256"], "software STOP evidence")
    if hmac.compare_digest(physical_digest, software_digest):
        raise SiteSafetyError(
            "site_safety_stop_evidence_invalid", "STOP evidence must be independent"
        )
    return {
        "guarding_evidence_sha256": _digest(
            guarding["evidence_sha256"], "guarding evidence"
        ),
        "physical_evidence_sha256": physical_digest,
        "software_evidence_sha256": software_digest,
        "physical_tested_at": physical_at,
        "software_tested_at": software_at,
    }


def _validate_cameras(
    value: Any, *, artifacts: Mapping[str, str], commissioned_at: datetime
) -> Mapping[str, Any]:
    cameras = _exact_mapping(
        value,
        {
            "evidence_sha256",
            "observed_at",
            "tool",
            "workspace",
            "full_tray_visible",
            "complete_swept_volume_visible",
            "minimum_coverage_ratio",
        },
        "cameras",
    )
    identities = {}
    for role, digest_name in (
        ("tool", "tool_calibration_sha256"),
        ("workspace", "workspace_calibration_sha256"),
    ):
        camera = _exact_mapping(
            cameras[role], {"mxid", "calibration_sha256"}, "{} camera".format(role)
        )
        identities[role] = _identifier(camera["mxid"], "{} camera MXID".format(role))
        if camera["calibration_sha256"] != artifacts[digest_name]:
            raise SiteSafetyError(
                "site_safety_camera_invalid", "Camera calibration binding changed"
            )
    if identities["tool"] == identities["workspace"] or (
        cameras["full_tray_visible"] is not True
        or cameras["complete_swept_volume_visible"] is not True
    ):
        raise SiteSafetyError(
            "site_safety_camera_invalid", "Dual-camera visibility is incomplete"
        )
    coverage = _positive(cameras["minimum_coverage_ratio"], "camera coverage", 1.0)
    if coverage > 1.0:
        raise SiteSafetyError("site_safety_camera_invalid", "Coverage exceeds one")
    observed_at = _timestamp(cameras["observed_at"], "camera observed_at")
    _recent_evidence(observed_at, commissioned_at, "camera evidence")
    return {
        "evidence_sha256": _digest(cameras["evidence_sha256"], "camera evidence"),
        "observed_at": observed_at,
        "tool_mxid": identities["tool"],
        "workspace_mxid": identities["workspace"],
    }


def _validate_trajectories(
    value: Any,
    *,
    artifacts: Mapping[str, str],
    required_clearance_mm: float,
    commissioned_at: datetime,
) -> None:
    trajectories = _exact_mapping(value, {"home", "recovery"}, "trajectories")
    for name, digest_name in (
        ("home", "home_trajectory_sha256"),
        ("recovery", "recovery_trajectory_sha256"),
    ):
        item = _exact_mapping(
            trajectories[name],
            {
                "trajectory_sha256",
                "evidence_sha256",
                "reviewed_at",
                "reviewed_by",
                "supervised_with_dummy_material",
                "minimum_clearance_mm",
            },
            "{} trajectory".format(name),
        )
        if (
            item["trajectory_sha256"] != artifacts[digest_name]
            or item["supervised_with_dummy_material"] is not True
            or _positive(
                item["minimum_clearance_mm"], "{} clearance".format(name), 10_000.0
            )
            < required_clearance_mm
        ):
            raise SiteSafetyError(
                "site_safety_trajectory_invalid",
                "{} trajectory evidence is invalid".format(name),
            )
        _digest(item["evidence_sha256"], "{} evidence".format(name))
        _identifier(item["reviewed_by"], "{} reviewer".format(name))
        _recent_evidence(
            _timestamp(item["reviewed_at"], "{} reviewed_at".format(name)),
            commissioned_at,
            "{} trajectory".format(name),
        )


def _validate_approval(
    value: Any, *, commissioned_at: datetime, valid_until: datetime
) -> Mapping[str, Any]:
    approval = _exact_mapping(
        value,
        {
            "approved_by",
            "qualification",
            "approved_at",
            "evidence_sha256",
            "competent_person",
            "site_acceptance",
            "production_release_approved",
        },
        "approval",
    )
    if any(
        approval[name] is not True
        for name in (
            "competent_person",
            "site_acceptance",
            "production_release_approved",
        )
    ):
        raise SiteSafetyError(
            "site_safety_approval_invalid", "Competent-person approval is missing"
        )
    approved_at = _timestamp(approval["approved_at"], "approved_at")
    if not commissioned_at <= approved_at < valid_until:
        raise SiteSafetyError(
            "site_safety_approval_invalid", "Approval time is outside validity"
        )
    qualification = str(approval["qualification"] or "").strip()
    if not 3 <= len(qualification) <= 160 or any(
        ord(item) < 32 for item in qualification
    ):
        raise SiteSafetyError(
            "site_safety_approval_invalid", "Approver qualification is invalid"
        )
    return {
        "approved_by": _identifier(approval["approved_by"], "approver"),
        "qualification": qualification,
        "approved_at": approved_at,
        "evidence_sha256": _digest(approval["evidence_sha256"], "approval evidence"),
    }


def _strict_json(payload: bytes) -> Mapping[str, Any]:
    try:
        value = json.loads(
            payload.decode("utf-8"),
            object_pairs_hook=_unique_object,
            parse_constant=_reject_constant,
        )
    except (UnicodeError, json.JSONDecodeError, SiteSafetyError) as exc:
        if isinstance(exc, SiteSafetyError):
            raise
        raise SiteSafetyError(
            "site_safety_json_invalid", "Profile is not strict UTF-8 JSON"
        ) from exc
    if not isinstance(value, Mapping):
        raise SiteSafetyError("site_safety_json_invalid", "Profile must be an object")
    return value


def _unique_object(pairs: Sequence[Tuple[str, Any]]) -> Mapping[str, Any]:
    result = {}
    for key, value in pairs:
        if key in result:
            raise SiteSafetyError("site_safety_json_invalid", "Duplicate JSON key")
        result[key] = value
    return result


def _reject_constant(_value: str) -> None:
    raise SiteSafetyError("site_safety_json_invalid", "Non-finite JSON number")


def _exact_mapping(value: Any, keys: Sequence[str], name: str) -> Mapping[str, Any]:
    expected = frozenset(keys)
    if not isinstance(value, Mapping) or frozenset(value) != expected:
        raise SiteSafetyError(
            "site_safety_schema_invalid", "{} fields do not match schema".format(name)
        )
    return value


def _digest(value: Any, name: str) -> str:
    if not isinstance(value, str) or _SHA256.fullmatch(value) is None:
        raise SiteSafetyError(
            "site_safety_digest_invalid", "{} is not a SHA-256".format(name)
        )
    return value


def _identifier(value: Any, name: str) -> str:
    if not isinstance(value, str) or _SAFE_ID.fullmatch(value) is None:
        raise SiteSafetyError(
            "site_safety_identity_invalid", "{} is invalid".format(name)
        )
    return value


def _firmware(value: Any) -> str:
    if not isinstance(value, str) or _FIRMWARE.fullmatch(value) is None:
        raise SiteSafetyError("site_safety_identity_invalid", "Firmware is invalid")
    return value[1:] if value[:1] in {"v", "V"} else value


def _number(value: Any, name: str) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise SiteSafetyError(
            "site_safety_number_invalid", "{} is not numeric".format(name)
        )
    result = float(value)
    if not math.isfinite(result):
        raise SiteSafetyError(
            "site_safety_number_invalid", "{} is not finite".format(name)
        )
    return result


def _positive(value: Any, name: str, maximum: float) -> float:
    result = _number(value, name)
    if not 0.0 < result <= maximum:
        raise SiteSafetyError(
            "site_safety_number_invalid", "{} is out of range".format(name)
        )
    return result


def _nonnegative(value: Any, name: str, maximum: float) -> float:
    result = _number(value, name)
    if not 0.0 <= result <= maximum:
        raise SiteSafetyError(
            "site_safety_number_invalid", "{} is out of range".format(name)
        )
    return result


def _vector(value: Any, length: int, name: str) -> Tuple[float, ...]:
    if not isinstance(value, list) or len(value) != length:
        raise SiteSafetyError(
            "site_safety_number_invalid",
            "{} must contain {} values".format(name, length),
        )
    return tuple(_number(item, name) for item in value)


def _vectors_equal(left: Any, right: Any) -> bool:
    try:
        left_values = tuple(float(item) for item in left)
        right_values = tuple(float(item) for item in right)
    except (TypeError, ValueError, OverflowError):
        return False
    return len(left_values) == len(right_values) and all(
        math.isclose(first, second, rel_tol=0.0, abs_tol=1e-6)
        for first, second in zip(left_values, right_values)
    )


def _timestamp(value: Any, name: str) -> datetime:
    if not isinstance(value, str) or not value.endswith("Z"):
        raise SiteSafetyError(
            "site_safety_time_invalid", "{} must be canonical UTC".format(name)
        )
    try:
        parsed = datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ").replace(
            tzinfo=timezone.utc
        )
    except ValueError as exc:
        raise SiteSafetyError(
            "site_safety_time_invalid", "{} is invalid".format(name)
        ) from exc
    if _format_timestamp(parsed) != value:
        raise SiteSafetyError(
            "site_safety_time_invalid", "{} is not canonical".format(name)
        )
    return parsed


def _aware_utc(value: Any, name: str) -> datetime:
    if not isinstance(value, datetime) or value.tzinfo is None:
        raise SiteSafetyError(
            "site_safety_time_invalid", "{} must be timezone aware".format(name)
        )
    return value.astimezone(timezone.utc)


def _format_timestamp(value: datetime) -> str:
    return value.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _recent_evidence(observed: datetime, commissioned: datetime, name: str) -> None:
    if not commissioned - _MAX_EVIDENCE_AGE_AT_APPROVAL <= observed <= commissioned:
        raise SiteSafetyError(
            "site_safety_time_invalid", "{} is stale or future-dated".format(name)
        )


def _freeze_json(value: Any) -> Any:
    if isinstance(value, Mapping):
        return MappingProxyType(
            {key: _freeze_json(item) for key, item in value.items()}
        )
    if isinstance(value, list):
        return tuple(_freeze_json(item) for item in value)
    return value
