"""Hash-pinned commissioning evidence and live guards for capture motion.

Loading this module never connects to or changes a robot.  A deployment must
load one reviewed evidence file by its exact SHA-256 digest, bind every
deployment identity supplied by configuration, and inject a read-only xArm
safety source.  The guarded runner then checks that evidence immediately
before arming and repeatedly until the commissioned motion deadline.

The concrete :class:`XArmCaptureSafetySource` is deliberately read-only.  It
uses the SDK's ``get_reduced_mode`` and ``get_reduced_states(is_radian=True)``
queries and never enables motion, changes reduced mode, or clears faults.
"""

from __future__ import annotations

import hashlib
import hmac
import ipaddress
import json
import math
import os
import re
import stat
import threading
import time
from contextlib import nullcontext
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Callable, Mapping, Optional, Protocol, Sequence, Tuple, Union

try:
    from .capture_motion import (
        CONTROLLER_DEVICE_TYPE,
        HARD_JOINT_LOWER_RADIANS,
        HARD_JOINT_UPPER_RADIANS,
        JOINT_COUNT,
        JOINT_LIMIT_POLICY,
        MOTION_DURATION_SECONDS,
        MotionSafetyError,
    )
except ImportError:  # Direct ``main/runtime.py`` execution on the Pi.
    from capture_motion import (  # type: ignore[no-redef]
        CONTROLLER_DEVICE_TYPE,
        HARD_JOINT_LOWER_RADIANS,
        HARD_JOINT_UPPER_RADIANS,
        JOINT_COUNT,
        JOINT_LIMIT_POLICY,
        MOTION_DURATION_SECONDS,
        MotionSafetyError,
    )


SAFETY_EVIDENCE_SCHEMA = "labcobot.capture-safety-evidence.v1"
ROBOT_MODEL = "lite6"
MAX_EVIDENCE_BYTES = 1024 * 1024
MAX_EVIDENCE_LIFETIME = timedelta(days=7)
MAX_WITNESS_AGE_AT_ISSUE = timedelta(hours=24)
MAX_CLOCK_SKEW = timedelta(minutes=5)
MAX_GUARD_INTERVAL_SECONDS = 0.05
MAX_REDUCED_TCP_SPEED_MM_PER_SECOND = 250.0
MAX_REDUCED_JOINT_SPEED_RADIANS_PER_SECOND = math.radians(60.0)

_SHA256 = re.compile(r"^[0-9a-f]{64}$")
_SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@+-]{0,127}$")
_MXID = re.compile(r"^[A-Za-z0-9_-]{8,128}$")
_TIMESTAMP = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$")
_VALIDATED_EVIDENCE = object()


class CaptureSafetyError(MotionSafetyError):
    """Base class for commissioning and runtime-guard failures."""


class SafetyEvidenceError(CaptureSafetyError):
    """The hash-pinned evidence artifact is missing, stale, or malformed."""


class RuntimeSafetyError(CaptureSafetyError):
    """Live controller state no longer matches commissioned evidence."""


@dataclass(frozen=True)
class CameraEvidence:
    mxid: str
    source_role: str


@dataclass(frozen=True)
class WitnessEvidence:
    witnessed_by: str
    witnessed_at: datetime


@dataclass(frozen=True)
class ControllerPhaseEvidence:
    state: int
    mode: int
    motor_enable_states: Tuple[bool, ...]
    motor_brake_states: Tuple[bool, ...]


@dataclass(frozen=True)
class ReducedModeEvidence:
    commissioned_by: str
    commissioned_at: datetime
    tcp_boundary_mm: Tuple[float, ...]
    max_tcp_speed_mm_per_second: float
    max_joint_speed_radians_per_second: float
    joint_ranges_radians: Tuple[float, ...]
    safety_boundary_enabled: bool
    collision_rebound_enabled: bool


@dataclass(frozen=True, init=False)
class CaptureSafetyEvidence:
    """Immutable commissioning evidence created only by the strict loader."""

    evidence_id: str
    evidence_sha256: str
    device_uid: str
    issued_at: datetime
    expires_at: datetime
    xarm_ip: str
    xarm_model: str
    axis_count: int
    controller_device_type: int
    tool_camera: CameraEvidence
    workspace_camera: CameraEvidence
    motion_profile_sha256: str
    workspace_lower_xyz_mm: Tuple[float, float, float]
    workspace_upper_xyz_mm: Tuple[float, float, float]
    joint_limit_policy: str
    emergency_stop_witness: WitnessEvidence
    camera_visibility_witness: WitnessEvidence
    reduced_mode: ReducedModeEvidence
    preflight_controller: ControllerPhaseEvidence
    active_controller: ControllerPhaseEvidence
    _validation_token: object

    @classmethod
    def _create(cls, **values: Any) -> "CaptureSafetyEvidence":
        instance = object.__new__(cls)
        for name, value in values.items():
            object.__setattr__(instance, name, value)
        object.__setattr__(instance, "_validation_token", _VALIDATED_EVIDENCE)
        return instance


@dataclass(frozen=True)
class ControllerSafetySnapshot:
    """One strict, read-only xArm safety snapshot in SDK radian units."""

    connected: bool
    state: int
    mode: int
    error_code: int
    warn_code: int
    device_type: int
    axis_count: int
    moving: bool
    motor_enable_states: Tuple[bool, ...]
    motor_brake_states: Tuple[bool, ...]
    reduced_mode_enabled: bool
    reduced_tcp_boundary_mm: Tuple[float, ...]
    reduced_max_tcp_speed_mm_per_second: float
    reduced_max_joint_speed_radians_per_second: float
    reduced_joint_ranges_radians: Tuple[float, ...]
    safety_boundary_enabled: bool
    collision_rebound_enabled: bool


class ControllerSafetySource(Protocol):
    def read_capture_safety_snapshot(self) -> ControllerSafetySnapshot:
        """Return a fresh controller snapshot without changing controller state."""


def _exact_keys(value: Any, expected: Sequence[str], name: str) -> Mapping[str, Any]:
    if not isinstance(value, Mapping) or not all(isinstance(key, str) for key in value):
        raise SafetyEvidenceError(
            "safety_evidence_invalid", "{} must be a JSON object".format(name)
        )
    if set(value) != set(expected):
        raise SafetyEvidenceError(
            "safety_evidence_invalid", "{} fields do not match schema".format(name)
        )
    return value


def _reject_duplicate_keys(pairs: Sequence[Tuple[str, Any]]) -> Mapping[str, Any]:
    result = {}
    for key, value in pairs:
        if key in result:
            raise SafetyEvidenceError(
                "safety_evidence_invalid", "Evidence contains duplicate JSON fields"
            )
        result[key] = value
    return result


def _reject_nonstandard_constant(_value: str) -> None:
    raise SafetyEvidenceError(
        "safety_evidence_invalid", "Evidence contains a non-standard JSON number"
    )


def _parse_json_float(value: str) -> float:
    try:
        number = float(value)
    except (TypeError, ValueError, OverflowError) as exc:
        raise SafetyEvidenceError(
            "safety_evidence_invalid", "Evidence contains an invalid JSON number"
        ) from exc
    if not math.isfinite(number):
        raise SafetyEvidenceError(
            "safety_evidence_invalid", "Evidence contains a non-finite JSON number"
        )
    return number


def _read_regular_file(path: Path) -> bytes:
    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0)
    flags |= getattr(os, "O_NOFOLLOW", 0)
    descriptor: Optional[int] = None
    try:
        before = os.lstat(str(path))
        if stat.S_ISLNK(before.st_mode):
            raise SafetyEvidenceError(
                "safety_evidence_unsafe_path",
                "Safety evidence may not be a symbolic link",
            )
        descriptor = os.open(str(path), flags)
        info = os.fstat(descriptor)
        if not stat.S_ISREG(info.st_mode):
            raise SafetyEvidenceError(
                "safety_evidence_unsafe_path",
                "Safety evidence must be a regular file",
            )
        if info.st_size < 2 or info.st_size > MAX_EVIDENCE_BYTES:
            raise SafetyEvidenceError(
                "safety_evidence_size_invalid", "Safety evidence size is invalid"
            )
        remaining = MAX_EVIDENCE_BYTES + 1
        chunks = []
        while remaining > 0:
            chunk = os.read(descriptor, min(65536, remaining))
            if not chunk:
                break
            chunks.append(chunk)
            remaining -= len(chunk)
        payload = b"".join(chunks)
        if len(payload) != info.st_size or len(payload) > MAX_EVIDENCE_BYTES:
            raise SafetyEvidenceError(
                "safety_evidence_size_invalid",
                "Safety evidence changed while being read",
            )
        return payload
    except SafetyEvidenceError:
        raise
    except OSError as exc:
        raise SafetyEvidenceError(
            "safety_evidence_read_failed", "Safety evidence could not be read"
        ) from exc
    finally:
        if descriptor is not None:
            os.close(descriptor)


def _safe_id(value: Any, name: str) -> str:
    if not isinstance(value, str) or not _SAFE_ID.fullmatch(value):
        raise SafetyEvidenceError(
            "safety_evidence_identity_invalid", "{} is invalid".format(name)
        )
    return value


def _sha256(value: Any, name: str) -> str:
    if not isinstance(value, str) or not _SHA256.fullmatch(value):
        raise SafetyEvidenceError(
            "safety_evidence_identity_invalid",
            "{} must be 64 lowercase hexadecimal characters".format(name),
        )
    return value


def _finite_number(value: Any, name: str) -> float:
    if isinstance(value, bool):
        raise SafetyEvidenceError(
            "safety_evidence_invalid", "{} must be numeric".format(name)
        )
    try:
        number = float(value)
    except (TypeError, ValueError, OverflowError) as exc:
        raise SafetyEvidenceError(
            "safety_evidence_invalid", "{} must be numeric".format(name)
        ) from exc
    if not math.isfinite(number):
        raise SafetyEvidenceError(
            "safety_evidence_invalid", "{} must be finite".format(name)
        )
    return number


def _finite_vector(value: Any, length: int, name: str) -> Tuple[float, ...]:
    if not isinstance(value, (list, tuple)) or len(value) != length:
        raise SafetyEvidenceError(
            "safety_evidence_invalid",
            "{} must contain exactly {} values".format(name, length),
        )
    return tuple(
        _finite_number(item, "{}[{}]".format(name, index))
        for index, item in enumerate(value)
    )


def _bool_vector(value: Any, name: str) -> Tuple[bool, ...]:
    if not isinstance(value, (list, tuple)) or len(value) != JOINT_COUNT:
        raise SafetyEvidenceError(
            "safety_evidence_invalid",
            "{} must contain exactly six booleans".format(name),
        )
    if not all(type(item) is bool for item in value):
        raise SafetyEvidenceError(
            "safety_evidence_invalid", "{} must contain booleans".format(name)
        )
    return tuple(value)


def _timestamp(value: Any, name: str) -> datetime:
    if not isinstance(value, str) or not _TIMESTAMP.fullmatch(value):
        raise SafetyEvidenceError(
            "safety_evidence_time_invalid",
            "{} must be canonical UTC ISO-8601 seconds".format(name),
        )
    try:
        parsed = datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ")
    except ValueError as exc:
        raise SafetyEvidenceError(
            "safety_evidence_time_invalid", "{} is invalid".format(name)
        ) from exc
    return parsed.replace(tzinfo=timezone.utc)


def _utc_now(value: Optional[datetime]) -> datetime:
    result = datetime.now(timezone.utc) if value is None else value
    if not isinstance(result, datetime) or result.tzinfo is None:
        raise SafetyEvidenceError(
            "safety_evidence_time_invalid", "Current time must be timezone-aware"
        )
    try:
        return result.astimezone(timezone.utc)
    except (TypeError, ValueError, OverflowError) as exc:
        raise SafetyEvidenceError(
            "safety_evidence_time_invalid", "Current time is invalid"
        ) from exc


def _parse_witness(
    value: Any,
    *,
    name: str,
    assertion_fields: Sequence[str],
) -> WitnessEvidence:
    value = _exact_keys(
        value,
        ("witnessed_by", "witnessed_at") + tuple(assertion_fields),
        name,
    )
    for field in assertion_fields:
        if value[field] is not True:
            raise SafetyEvidenceError(
                "safety_evidence_witness_invalid",
                "{} assertion {} is not verified".format(name, field),
            )
    return WitnessEvidence(
        witnessed_by=_safe_id(value["witnessed_by"], "{} witness".format(name)),
        witnessed_at=_timestamp(value["witnessed_at"], "{} time".format(name)),
    )


def _parse_phase(value: Any, name: str) -> ControllerPhaseEvidence:
    value = _exact_keys(
        value,
        ("state", "mode", "motor_enable_states", "motor_brake_states"),
        name,
    )
    if type(value["state"]) is not int or type(value["mode"]) is not int:
        raise SafetyEvidenceError(
            "safety_evidence_controller_invalid",
            "{} state and mode must be integers".format(name),
        )
    return ControllerPhaseEvidence(
        state=value["state"],
        mode=value["mode"],
        motor_enable_states=_bool_vector(
            value["motor_enable_states"], "{} motor enables".format(name)
        ),
        motor_brake_states=_bool_vector(
            value["motor_brake_states"], "{} motor brakes".format(name)
        ),
    )


def _parse_reduced(value: Any) -> ReducedModeEvidence:
    value = _exact_keys(
        value,
        (
            "commissioned_by",
            "commissioned_at",
            "enabled",
            "tcp_boundary_mm",
            "max_tcp_speed_mm_per_second",
            "max_joint_speed_radians_per_second",
            "joint_ranges_radians",
            "safety_boundary_enabled",
            "collision_rebound_enabled",
        ),
        "reduced_mode",
    )
    if value["enabled"] is not True:
        raise SafetyEvidenceError(
            "safety_evidence_reduced_mode_invalid",
            "Reduced mode must be explicitly commissioned on",
        )
    if value["safety_boundary_enabled"] is not True:
        raise SafetyEvidenceError(
            "safety_evidence_reduced_mode_invalid",
            "The commissioned reduced safety boundary must be enabled",
        )
    if value["collision_rebound_enabled"] is not False:
        raise SafetyEvidenceError(
            "safety_evidence_reduced_mode_invalid",
            "Collision rebound must be disabled for commissioned capture",
        )
    boundary = _finite_vector(value["tcp_boundary_mm"], 6, "reduced TCP boundary")
    for maximum, minimum in zip(boundary[0::2], boundary[1::2]):
        if maximum <= minimum or abs(maximum) > 2000.0 or abs(minimum) > 2000.0:
            raise SafetyEvidenceError(
                "safety_evidence_reduced_mode_invalid",
                "Reduced TCP boundary is invalid",
            )
    tcp_speed = _finite_number(
        value["max_tcp_speed_mm_per_second"], "reduced TCP speed"
    )
    joint_speed = _finite_number(
        value["max_joint_speed_radians_per_second"], "reduced joint speed"
    )
    if not 0.0 < tcp_speed <= MAX_REDUCED_TCP_SPEED_MM_PER_SECOND:
        raise SafetyEvidenceError(
            "safety_evidence_reduced_mode_invalid",
            "Reduced TCP speed exceeds the capture safety ceiling",
        )
    if not 0.0 < joint_speed <= MAX_REDUCED_JOINT_SPEED_RADIANS_PER_SECOND:
        raise SafetyEvidenceError(
            "safety_evidence_reduced_mode_invalid",
            "Reduced joint speed exceeds the capture safety ceiling",
        )
    joint_ranges = _finite_vector(
        value["joint_ranges_radians"], 14, "reduced joint ranges"
    )
    for index in range(JOINT_COUNT):
        lower = joint_ranges[index * 2]
        upper = joint_ranges[index * 2 + 1]
        if (
            lower >= upper
            or lower < HARD_JOINT_LOWER_RADIANS[index]
            or upper > HARD_JOINT_UPPER_RADIANS[index]
        ):
            raise SafetyEvidenceError(
                "safety_evidence_reduced_mode_invalid",
                "Reduced joint ranges may only narrow Lite6 limits",
            )
    if joint_ranges[12:] != (0.0, 0.0):
        raise SafetyEvidenceError(
            "safety_evidence_reduced_mode_invalid",
            "The non-physical seventh Lite6 joint range must be zero",
        )
    return ReducedModeEvidence(
        commissioned_by=_safe_id(value["commissioned_by"], "reduced-mode commissioner"),
        commissioned_at=_timestamp(
            value["commissioned_at"], "reduced-mode commissioning time"
        ),
        tcp_boundary_mm=boundary,
        max_tcp_speed_mm_per_second=tcp_speed,
        max_joint_speed_radians_per_second=joint_speed,
        joint_ranges_radians=joint_ranges,
        safety_boundary_enabled=True,
        collision_rebound_enabled=False,
    )


def _parse_camera(value: Any, role: str) -> CameraEvidence:
    value = _exact_keys(value, ("mxid", "source_role"), "{} camera".format(role))
    mxid = value["mxid"]
    if not isinstance(mxid, str) or not _MXID.fullmatch(mxid):
        raise SafetyEvidenceError(
            "safety_evidence_camera_invalid",
            "{} camera MXID is invalid".format(role),
        )
    if value["source_role"] != role:
        raise SafetyEvidenceError(
            "safety_evidence_camera_invalid",
            "{} camera source role is invalid".format(role),
        )
    return CameraEvidence(mxid=mxid, source_role=role)


def load_capture_safety_evidence(
    path: Union[os.PathLike, str],
    *,
    expected_sha256: str,
    device_uid: str,
    xarm_ip: str,
    xarm_model: str,
    tool_camera_mxid: str,
    workspace_camera_mxid: str,
    motion_profile_sha256: str,
    workspace_lower_xyz_mm: Sequence[float],
    workspace_upper_xyz_mm: Sequence[float],
    joint_limit_policy: str,
    now: Optional[datetime] = None,
) -> CaptureSafetyEvidence:
    """Load strict evidence and bind it to the complete runtime configuration."""

    expected_digest = _sha256(expected_sha256, "expected evidence SHA-256")
    expected_profile = _sha256(motion_profile_sha256, "motion profile SHA-256")
    configured_device_uid = _safe_id(device_uid, "configured device UID")
    try:
        configured_ip = ipaddress.ip_address(xarm_ip)
    except (TypeError, ValueError) as exc:
        raise SafetyEvidenceError(
            "safety_evidence_identity_invalid", "Configured xArm IP is invalid"
        ) from exc
    if configured_ip.version != 4 or str(configured_ip) != xarm_ip:
        raise SafetyEvidenceError(
            "safety_evidence_identity_invalid",
            "Configured xArm IP must be canonical IPv4",
        )
    if xarm_model != ROBOT_MODEL:
        raise SafetyEvidenceError(
            "safety_evidence_identity_invalid",
            "Configured xArm model must be lite6",
        )
    for mxid, role in (
        (tool_camera_mxid, "tool"),
        (workspace_camera_mxid, "workspace"),
    ):
        if not isinstance(mxid, str) or not _MXID.fullmatch(mxid):
            raise SafetyEvidenceError(
                "safety_evidence_camera_invalid",
                "Configured {} camera MXID is invalid".format(role),
            )
    if tool_camera_mxid == workspace_camera_mxid:
        raise SafetyEvidenceError(
            "safety_evidence_camera_invalid",
            "Configured camera MXIDs must be distinct",
        )
    expected_lower = _finite_vector(
        workspace_lower_xyz_mm, 3, "configured workspace lower"
    )
    expected_upper = _finite_vector(
        workspace_upper_xyz_mm, 3, "configured workspace upper"
    )
    if any(low >= high for low, high in zip(expected_lower, expected_upper)) or any(
        abs(value) > 2000.0 for value in expected_lower + expected_upper
    ):
        raise SafetyEvidenceError(
            "safety_evidence_workspace_invalid",
            "Configured workspace bounds are invalid",
        )
    current_time = _utc_now(now)
    payload = _read_regular_file(Path(path))
    actual_digest = hashlib.sha256(payload).hexdigest()
    if not hmac.compare_digest(actual_digest, expected_digest):
        raise SafetyEvidenceError(
            "safety_evidence_hash_mismatch",
            "Safety evidence does not match its approved SHA-256",
        )
    try:
        value = json.loads(
            payload.decode("utf-8"),
            object_pairs_hook=_reject_duplicate_keys,
            parse_constant=_reject_nonstandard_constant,
            parse_float=_parse_json_float,
        )
    except SafetyEvidenceError:
        raise
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise SafetyEvidenceError(
            "safety_evidence_invalid", "Safety evidence is not strict UTF-8 JSON"
        ) from exc
    value = _exact_keys(
        value,
        (
            "schema",
            "evidence_id",
            "device_uid",
            "issued_at",
            "expires_at",
            "xarm",
            "cameras",
            "motion_profile_sha256",
            "workspace_lower_xyz_mm",
            "workspace_upper_xyz_mm",
            "joint_limit_policy",
            "emergency_stop_witness",
            "camera_visibility_witness",
            "reduced_mode",
            "controller_phases",
        ),
        "evidence",
    )
    if value["schema"] != SAFETY_EVIDENCE_SCHEMA:
        raise SafetyEvidenceError(
            "safety_evidence_schema_invalid", "Safety evidence schema is unsupported"
        )
    evidence_id = _safe_id(value["evidence_id"], "evidence id")
    issued_at = _timestamp(value["issued_at"], "evidence issue time")
    expires_at = _timestamp(value["expires_at"], "evidence expiry time")
    if issued_at > current_time + MAX_CLOCK_SKEW:
        raise SafetyEvidenceError(
            "safety_evidence_not_yet_valid", "Safety evidence issue time is in future"
        )
    if expires_at <= current_time:
        raise SafetyEvidenceError(
            "safety_evidence_expired", "Safety evidence has expired"
        )
    if expires_at <= issued_at or expires_at - issued_at > MAX_EVIDENCE_LIFETIME:
        raise SafetyEvidenceError(
            "safety_evidence_time_invalid",
            "Safety evidence lifetime must be positive and at most seven days",
        )

    xarm = _exact_keys(
        value["xarm"],
        ("ip", "model", "axis_count", "controller_device_type"),
        "xarm",
    )
    try:
        parsed_ip = ipaddress.ip_address(xarm["ip"])
    except (TypeError, ValueError) as exc:
        raise SafetyEvidenceError(
            "safety_evidence_identity_invalid", "xArm IP is invalid"
        ) from exc
    if parsed_ip.version != 4 or str(parsed_ip) != xarm["ip"]:
        raise SafetyEvidenceError(
            "safety_evidence_identity_invalid", "xArm IP must be canonical IPv4"
        )
    if (
        xarm["model"] != ROBOT_MODEL
        or type(xarm["axis_count"]) is not int
        or xarm["axis_count"] != JOINT_COUNT
        or type(xarm["controller_device_type"]) is not int
        or xarm["controller_device_type"] != CONTROLLER_DEVICE_TYPE
    ):
        raise SafetyEvidenceError(
            "safety_evidence_identity_invalid",
            "Evidence does not identify a six-axis Lite6 controller type 9",
        )

    cameras = _exact_keys(value["cameras"], ("tool", "workspace"), "cameras")
    tool_camera = _parse_camera(cameras["tool"], "tool")
    workspace_camera = _parse_camera(cameras["workspace"], "workspace")
    if tool_camera.mxid == workspace_camera.mxid:
        raise SafetyEvidenceError(
            "safety_evidence_camera_invalid", "Camera MXIDs must be distinct"
        )
    profile_sha256 = _sha256(
        value["motion_profile_sha256"], "evidence motion profile SHA-256"
    )
    workspace_lower = _finite_vector(
        value["workspace_lower_xyz_mm"], 3, "evidence workspace lower"
    )
    workspace_upper = _finite_vector(
        value["workspace_upper_xyz_mm"], 3, "evidence workspace upper"
    )
    if any(low >= high for low, high in zip(workspace_lower, workspace_upper)) or any(
        abs(item) > 2000.0 for item in workspace_lower + workspace_upper
    ):
        raise SafetyEvidenceError(
            "safety_evidence_workspace_invalid", "Evidence workspace is invalid"
        )

    emergency = _parse_witness(
        value["emergency_stop_witness"],
        name="emergency-stop witness",
        assertion_fields=("physical_estop_pressed", "motion_inhibited"),
    )
    visibility = _parse_witness(
        value["camera_visibility_witness"],
        name="camera-visibility witness",
        assertion_fields=(
            "full_workcell_visible",
            "entire_robot_envelope_visible",
            "both_oak_views_verified",
        ),
    )
    reduced = _parse_reduced(value["reduced_mode"])
    phases = _exact_keys(
        value["controller_phases"], ("preflight", "active"), "controller phases"
    )
    preflight = _parse_phase(phases["preflight"], "preflight controller")
    active = _parse_phase(phases["active"], "active controller")
    if (
        preflight.state != 4
        or preflight.mode != 0
        or preflight.motor_enable_states != (False,) * JOINT_COUNT
        or preflight.motor_brake_states != (False,) * JOINT_COUNT
        or active.state != 0
        or active.mode != 1
        or active.motor_enable_states != (True,) * JOINT_COUNT
        or active.motor_brake_states != (True,) * JOINT_COUNT
    ):
        raise SafetyEvidenceError(
            "safety_evidence_controller_invalid",
            "Controller phases do not match the fixed locked/servo-stream policy",
        )
    for witness_time in (
        emergency.witnessed_at,
        visibility.witnessed_at,
        reduced.commissioned_at,
    ):
        if (
            witness_time > issued_at + MAX_CLOCK_SKEW
            or issued_at - witness_time > MAX_WITNESS_AGE_AT_ISSUE
        ):
            raise SafetyEvidenceError(
                "safety_evidence_witness_invalid",
                "Commissioning witnesses must be within 24 hours of evidence issue",
            )

    if (
        _safe_id(value["device_uid"], "evidence device UID") != configured_device_uid
        or xarm["ip"] != xarm_ip
        or xarm["model"] != xarm_model
        or tool_camera.mxid != tool_camera_mxid
        or workspace_camera.mxid != workspace_camera_mxid
        or profile_sha256 != expected_profile
        or workspace_lower != expected_lower
        or workspace_upper != expected_upper
        or value["joint_limit_policy"] != joint_limit_policy
        or joint_limit_policy != JOINT_LIMIT_POLICY
    ):
        raise SafetyEvidenceError(
            "safety_evidence_binding_mismatch",
            "Safety evidence does not match the complete runtime configuration",
        )
    for axis in range(3):
        reduced_max = reduced.tcp_boundary_mm[axis * 2]
        reduced_min = reduced.tcp_boundary_mm[axis * 2 + 1]
        if reduced_min < workspace_lower[axis] or reduced_max > workspace_upper[axis]:
            raise SafetyEvidenceError(
                "safety_evidence_reduced_mode_invalid",
                "Reduced TCP boundary may not exceed commissioned workspace",
            )

    return CaptureSafetyEvidence._create(
        evidence_id=evidence_id,
        evidence_sha256=actual_digest,
        device_uid=configured_device_uid,
        issued_at=issued_at,
        expires_at=expires_at,
        xarm_ip=xarm_ip,
        xarm_model=xarm_model,
        axis_count=JOINT_COUNT,
        controller_device_type=CONTROLLER_DEVICE_TYPE,
        tool_camera=tool_camera,
        workspace_camera=workspace_camera,
        motion_profile_sha256=profile_sha256,
        workspace_lower_xyz_mm=workspace_lower,
        workspace_upper_xyz_mm=workspace_upper,
        joint_limit_policy=joint_limit_policy,
        emergency_stop_witness=emergency,
        camera_visibility_witness=visibility,
        reduced_mode=reduced,
        preflight_controller=preflight,
        active_controller=active,
    )


def _sdk_result(result: Any, name: str) -> Any:
    if (
        not isinstance(result, (tuple, list))
        or len(result) != 2
        or type(result[0]) is not int
        or result[0] != 0
    ):
        raise RuntimeSafetyError(
            "runtime_safety_unavailable", "xArm {} query failed".format(name)
        )
    return result[1]


def _runtime_int(value: Any, name: str) -> int:
    if type(value) is not int:
        raise RuntimeSafetyError(
            "runtime_safety_state_invalid", "xArm {} is invalid".format(name)
        )
    return value


def _runtime_bool(value: Any, name: str, *, allow_integer: bool = False) -> bool:
    if type(value) is bool:
        return value
    if allow_integer and type(value) is int and value in (0, 1):
        return bool(value)
    raise RuntimeSafetyError(
        "runtime_safety_state_invalid", "xArm {} is invalid".format(name)
    )


def _runtime_number(value: Any, name: str) -> float:
    if isinstance(value, bool):
        raise RuntimeSafetyError(
            "runtime_safety_state_invalid", "xArm {} is invalid".format(name)
        )
    try:
        number = float(value)
    except (TypeError, ValueError, OverflowError) as exc:
        raise RuntimeSafetyError(
            "runtime_safety_state_invalid", "xArm {} is invalid".format(name)
        ) from exc
    if not math.isfinite(number):
        raise RuntimeSafetyError(
            "runtime_safety_state_invalid", "xArm {} is invalid".format(name)
        )
    return number


def _runtime_vector(value: Any, length: int, name: str) -> Tuple[float, ...]:
    if not isinstance(value, (list, tuple)) or len(value) != length:
        raise RuntimeSafetyError(
            "runtime_safety_state_invalid",
            "xArm {} must contain exactly {} values".format(name, length),
        )
    return tuple(
        _runtime_number(item, "{}[{}]".format(name, index))
        for index, item in enumerate(value)
    )


def _runtime_motor_vector(value: Any, name: str) -> Tuple[bool, ...]:
    if not isinstance(value, (list, tuple)) or len(value) < JOINT_COUNT:
        raise RuntimeSafetyError(
            "runtime_safety_state_invalid", "xArm {} is incomplete".format(name)
        )
    return tuple(
        _runtime_bool(item, "{}[{}]".format(name, index), allow_integer=True)
        for index, item in enumerate(value[:JOINT_COUNT])
    )


class XArmCaptureSafetySource:
    """Read the exact safety seam exposed by xarm-python-sdk 1.18.x.

    ``lock`` should be the same lock used by ``XArmMotionAdapter`` so a safety
    snapshot cannot interleave SDK requests with a motion command.
    """

    def __init__(self, controller: Any, *, lock: Any = None) -> None:
        required = (
            "get_state",
            "get_err_warn_code",
            "get_is_moving",
            "get_reduced_mode",
            "get_reduced_states",
        )
        if any(not callable(getattr(controller, name, None)) for name in required):
            raise RuntimeSafetyError(
                "runtime_safety_source_invalid",
                "xArm safety source is missing required read-only queries",
            )
        if lock is not None and not (
            callable(getattr(lock, "__enter__", None))
            and callable(getattr(lock, "__exit__", None))
        ):
            raise RuntimeSafetyError(
                "runtime_safety_source_invalid", "xArm safety lock is invalid"
            )
        self.controller = controller
        self.lock = lock

    def read_capture_safety_snapshot(self) -> ControllerSafetySnapshot:
        context = self.lock if self.lock is not None else nullcontext()
        try:
            with context:
                connected = getattr(self.controller, "connected", None)
                state = _sdk_result(self.controller.get_state(), "state")
                mode = getattr(self.controller, "mode", None)
                errors = _sdk_result(
                    self.controller.get_err_warn_code(), "error/warning"
                )
                moving_result = self.controller.get_is_moving()
                if isinstance(moving_result, (tuple, list)) and len(moving_result) == 2:
                    moving_result = _sdk_result(moving_result, "moving state")
                reduced_mode = _sdk_result(
                    self.controller.get_reduced_mode(), "reduced mode"
                )
                reduced_states = _sdk_result(
                    self.controller.get_reduced_states(is_radian=True),
                    "reduced states",
                )
                device_type = getattr(self.controller, "device_type", None)
                axis_count = getattr(self.controller, "axis", None)
                motor_enable = getattr(self.controller, "motor_enable_states", None)
                motor_brake = getattr(self.controller, "motor_brake_states", None)
        except RuntimeSafetyError:
            raise
        except Exception as exc:
            raise RuntimeSafetyError(
                "runtime_safety_unavailable", "xArm safety snapshot failed"
            ) from exc

        if not isinstance(errors, (list, tuple)) or len(errors) != 2:
            raise RuntimeSafetyError(
                "runtime_safety_state_invalid", "xArm error/warning state is invalid"
            )
        if not isinstance(reduced_states, (list, tuple)) or len(reduced_states) != 7:
            raise RuntimeSafetyError(
                "runtime_safety_state_invalid",
                "xArm reduced-state response lacks required limits",
            )
        mode_from_states = _runtime_bool(
            reduced_states[0], "reduced-state mode", allow_integer=True
        )
        mode_from_query = _runtime_bool(
            reduced_mode, "reduced mode", allow_integer=True
        )
        if mode_from_states != mode_from_query:
            raise RuntimeSafetyError(
                "runtime_safety_state_invalid",
                "xArm reduced-mode queries disagree",
            )
        return ControllerSafetySnapshot(
            connected=_runtime_bool(connected, "connected state"),
            state=_runtime_int(state, "state"),
            mode=_runtime_int(mode, "mode"),
            error_code=_runtime_int(errors[0], "error code"),
            warn_code=_runtime_int(errors[1], "warning code"),
            device_type=_runtime_int(device_type, "device type"),
            axis_count=_runtime_int(axis_count, "axis count"),
            moving=_runtime_bool(moving_result, "moving state"),
            motor_enable_states=_runtime_motor_vector(
                motor_enable, "motor-enable states"
            ),
            motor_brake_states=_runtime_motor_vector(motor_brake, "motor-brake states"),
            reduced_mode_enabled=mode_from_query,
            reduced_tcp_boundary_mm=_runtime_vector(
                reduced_states[1], 6, "reduced TCP boundary"
            ),
            reduced_max_tcp_speed_mm_per_second=_runtime_number(
                reduced_states[2], "reduced TCP speed"
            ),
            reduced_max_joint_speed_radians_per_second=_runtime_number(
                reduced_states[3], "reduced joint speed"
            ),
            reduced_joint_ranges_radians=_runtime_vector(
                reduced_states[4], 14, "reduced joint ranges"
            ),
            safety_boundary_enabled=_runtime_bool(
                reduced_states[5], "safety-boundary state", allow_integer=True
            ),
            collision_rebound_enabled=_runtime_bool(
                reduced_states[6], "collision-rebound state", allow_integer=True
            ),
        )


class CaptureSafetyRuntimeGuard:
    """Callable runtime guard for one validated commissioning artifact."""

    def __init__(
        self,
        evidence: CaptureSafetyEvidence,
        source: ControllerSafetySource,
        *,
        wall_clock: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
    ) -> None:
        if not isinstance(evidence, CaptureSafetyEvidence) or (
            evidence._validation_token is not _VALIDATED_EVIDENCE
        ):
            raise SafetyEvidenceError(
                "safety_evidence_not_validated",
                "Runtime guard requires hash-validated safety evidence",
            )
        if not callable(getattr(source, "read_capture_safety_snapshot", None)):
            raise RuntimeSafetyError(
                "runtime_safety_source_invalid", "Controller safety source is invalid"
            )
        if not callable(wall_clock):
            raise RuntimeSafetyError(
                "runtime_safety_source_invalid", "Runtime wall clock is invalid"
            )
        self.evidence = evidence
        self.source = source
        self.wall_clock = wall_clock

    def __call__(self) -> ControllerSafetySnapshot:
        return self.ensure_active()

    def _current_time(self) -> datetime:
        try:
            current = self.wall_clock()
        except Exception as exc:
            raise RuntimeSafetyError(
                "runtime_safety_clock_failed", "Runtime wall clock failed"
            ) from exc
        if not isinstance(current, datetime) or current.tzinfo is None:
            raise RuntimeSafetyError(
                "runtime_safety_clock_failed", "Runtime wall clock is invalid"
            )
        try:
            return current.astimezone(timezone.utc)
        except (TypeError, ValueError, OverflowError) as exc:
            raise RuntimeSafetyError(
                "runtime_safety_clock_failed", "Runtime wall clock is invalid"
            ) from exc

    def _snapshot(self) -> ControllerSafetySnapshot:
        current_time = self._current_time()
        if current_time + MAX_CLOCK_SKEW < self.evidence.issued_at:
            raise RuntimeSafetyError(
                "runtime_safety_clock_failed",
                "Runtime clock precedes commissioning evidence",
            )
        if current_time >= self.evidence.expires_at:
            raise RuntimeSafetyError(
                "runtime_safety_evidence_expired",
                "Commissioning evidence expired before motion completed",
            )
        try:
            snapshot = self.source.read_capture_safety_snapshot()
        except RuntimeSafetyError:
            raise
        except Exception as exc:
            raise RuntimeSafetyError(
                "runtime_safety_unavailable", "Controller safety state is unavailable"
            ) from exc
        if not isinstance(snapshot, ControllerSafetySnapshot):
            raise RuntimeSafetyError(
                "runtime_safety_state_invalid", "Controller safety state is invalid"
            )
        scalar_bools = (
            snapshot.connected,
            snapshot.moving,
            snapshot.reduced_mode_enabled,
            snapshot.safety_boundary_enabled,
            snapshot.collision_rebound_enabled,
        )
        scalar_ints = (
            snapshot.state,
            snapshot.mode,
            snapshot.error_code,
            snapshot.warn_code,
            snapshot.device_type,
            snapshot.axis_count,
        )
        if not all(type(value) is bool for value in scalar_bools) or not all(
            type(value) is int for value in scalar_ints
        ):
            raise RuntimeSafetyError(
                "runtime_safety_state_invalid",
                "Controller safety scalar types are invalid",
            )
        if (
            not isinstance(snapshot.motor_enable_states, tuple)
            or not isinstance(snapshot.motor_brake_states, tuple)
            or not isinstance(snapshot.reduced_tcp_boundary_mm, tuple)
            or not isinstance(snapshot.reduced_joint_ranges_radians, tuple)
            or len(snapshot.motor_enable_states) != JOINT_COUNT
            or len(snapshot.motor_brake_states) != JOINT_COUNT
            or not all(
                type(value) is bool
                for value in (
                    snapshot.motor_enable_states + snapshot.motor_brake_states
                )
            )
            or len(snapshot.reduced_tcp_boundary_mm) != 6
            or len(snapshot.reduced_joint_ranges_radians) != 14
        ):
            raise RuntimeSafetyError(
                "runtime_safety_state_invalid",
                "Controller safety vectors are invalid",
            )
        numeric_values = (
            snapshot.reduced_tcp_boundary_mm
            + snapshot.reduced_joint_ranges_radians
            + (
                snapshot.reduced_max_tcp_speed_mm_per_second,
                snapshot.reduced_max_joint_speed_radians_per_second,
            )
        )
        if any(
            isinstance(value, bool)
            or not isinstance(value, (int, float))
            or not math.isfinite(float(value))
            for value in numeric_values
        ):
            raise RuntimeSafetyError(
                "runtime_safety_state_invalid",
                "Controller reduced-mode values are invalid",
            )
        evidence = self.evidence
        reduced = evidence.reduced_mode
        if snapshot.connected is not True:
            raise RuntimeSafetyError(
                "runtime_safety_disconnected", "xArm is not connected"
            )
        if (
            snapshot.axis_count != evidence.axis_count
            or snapshot.device_type != evidence.controller_device_type
        ):
            raise RuntimeSafetyError(
                "runtime_safety_identity_mismatch",
                "Live xArm identity does not match commissioning evidence",
            )
        if snapshot.error_code != 0 or snapshot.warn_code != 0:
            raise RuntimeSafetyError(
                "runtime_safety_controller_fault",
                "xArm reports an error or warning during commissioned motion",
            )
        if snapshot.reduced_mode_enabled is not True:
            raise RuntimeSafetyError(
                "runtime_safety_reduced_mode_off", "xArm reduced mode is not active"
            )
        expected_reduced = (
            reduced.tcp_boundary_mm,
            reduced.max_tcp_speed_mm_per_second,
            reduced.max_joint_speed_radians_per_second,
            reduced.joint_ranges_radians,
            reduced.safety_boundary_enabled,
            reduced.collision_rebound_enabled,
        )
        actual_reduced = (
            snapshot.reduced_tcp_boundary_mm,
            snapshot.reduced_max_tcp_speed_mm_per_second,
            snapshot.reduced_max_joint_speed_radians_per_second,
            snapshot.reduced_joint_ranges_radians,
            snapshot.safety_boundary_enabled,
            snapshot.collision_rebound_enabled,
        )
        if actual_reduced != expected_reduced:
            raise RuntimeSafetyError(
                "runtime_safety_reduced_config_mismatch",
                "Live reduced-mode limits differ from commissioning evidence",
            )
        return snapshot

    @staticmethod
    def _matches_phase(
        snapshot: ControllerSafetySnapshot,
        expected: ControllerPhaseEvidence,
    ) -> bool:
        return (
            snapshot.state == expected.state
            and snapshot.mode == expected.mode
            and snapshot.motor_enable_states == expected.motor_enable_states
            and snapshot.motor_brake_states == expected.motor_brake_states
        )

    def ensure_preflight(self) -> ControllerSafetySnapshot:
        snapshot = self._snapshot()
        if snapshot.moving is not False or not self._matches_phase(
            snapshot, self.evidence.preflight_controller
        ):
            raise RuntimeSafetyError(
                "runtime_safety_preflight_mismatch",
                "xArm is not in the commissioned locked preflight state",
            )
        return snapshot

    def ensure_transition(self) -> ControllerSafetySnapshot:
        snapshot = self._snapshot()
        if snapshot.moving is not False or not (
            self._matches_phase(snapshot, self.evidence.preflight_controller)
            or self._matches_phase(snapshot, self.evidence.active_controller)
        ):
            raise RuntimeSafetyError(
                "runtime_safety_transition_mismatch",
                "xArm left the commissioned arming transition states",
            )
        return snapshot

    def ensure_active(self) -> ControllerSafetySnapshot:
        snapshot = self._snapshot()
        if type(snapshot.moving) is not bool or not self._matches_phase(
            snapshot, self.evidence.active_controller
        ):
            raise RuntimeSafetyError(
                "runtime_safety_active_mismatch",
                "xArm left the commissioned active servo-stream state",
            )
        return snapshot


class _FailureBox:
    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._failure: Optional[RuntimeSafetyError] = None

    def set(
        self, failure: RuntimeSafetyError, *, replace_existing: bool = False
    ) -> None:
        with self._lock:
            if self._failure is None or replace_existing:
                self._failure = failure

    def get(self) -> Optional[RuntimeSafetyError]:
        with self._lock:
            return self._failure


class SafetyGuardedMotionSession:
    """Session facade that preserves the live guard failure as the run fault."""

    def __init__(
        self,
        session: Any,
        runner: Any,
        guard: CaptureSafetyRuntimeGuard,
        failure_box: _FailureBox,
        *,
        deadline_monotonic_s: float,
        monitor_interval_seconds: float,
        monotonic_clock: Callable[[], float],
    ) -> None:
        self.session = session
        self.runner = runner
        self.guard = guard
        self.failure_box = failure_box
        self.deadline_monotonic_s = deadline_monotonic_s
        self.monitor_interval_seconds = monitor_interval_seconds
        self.monotonic_clock = monotonic_clock
        self._monitor_stop = threading.Event()
        self._monitor = threading.Thread(
            target=self._monitor_loop,
            name="capture-safety-runtime-guard",
            daemon=True,
        )
        self._monitor.start()

    def _monitor_loop(self) -> None:
        while not self._monitor_stop.is_set():
            try:
                if float(self.monotonic_clock()) >= self.deadline_monotonic_s:
                    return
                active = self.session.is_active()
                if type(active) is not bool or not active:
                    return
                self.guard.ensure_active()
            except RuntimeSafetyError as exc:
                self.failure_box.set(exc)
                try:
                    self.runner.halt()
                except Exception:
                    self.failure_box.set(
                        RuntimeSafetyError(
                            "runtime_safety_halt_failed",
                            "Runtime safety mismatch could not halt xArm",
                        ),
                        replace_existing=True,
                    )
                return
            except Exception:
                self.failure_box.set(
                    RuntimeSafetyError(
                        "runtime_safety_monitor_failed",
                        "Runtime safety monitor failed",
                    )
                )
                try:
                    self.runner.halt()
                except Exception:
                    self.failure_box.set(
                        RuntimeSafetyError(
                            "runtime_safety_halt_failed",
                            "Runtime safety monitor could not halt xArm",
                        ),
                        replace_existing=True,
                    )
                return
            self._monitor_stop.wait(self.monitor_interval_seconds)

    def is_active(self) -> bool:
        active = self.session.is_active()
        if type(active) is not bool:
            raise RuntimeSafetyError(
                "runtime_safety_session_invalid", "Motion session state is invalid"
            )
        return active

    def wait(self, timeout_seconds: float) -> Any:
        try:
            result = self.session.wait(timeout_seconds)
        except Exception as exc:
            failure = self.failure_box.get()
            if failure is not None:
                raise failure from exc
            raise
        finally:
            self._monitor_stop.set()
            if threading.current_thread() is not self._monitor:
                self._monitor.join(MAX_GUARD_INTERVAL_SECONDS * 2.0)
        failure = self.failure_box.get()
        if failure is not None:
            raise failure
        return result

    def request_halt(self) -> None:
        self._monitor_stop.set()
        callback = getattr(self.session, "request_halt", None)
        if not callable(callback):
            raise RuntimeSafetyError(
                "runtime_safety_session_invalid", "Motion session cannot be halted"
            )
        callback()


class SafetyGuardedMotionRunner:
    """Decorate a commissioned runner with hash-pinned live safety evidence."""

    commissioned = True

    def __init__(
        self,
        runner: Any,
        guard: CaptureSafetyRuntimeGuard,
        *,
        monitor_interval_seconds: float = MAX_GUARD_INTERVAL_SECONDS,
        monotonic_clock: Optional[Callable[[], float]] = None,
    ) -> None:
        if getattr(runner, "commissioned", None) is not True:
            raise RuntimeSafetyError(
                "runtime_safety_runner_invalid",
                "Safety guard requires a commissioned motion runner",
            )
        if not all(callable(getattr(runner, name, None)) for name in ("start", "halt")):
            raise RuntimeSafetyError(
                "runtime_safety_runner_invalid", "Motion runner seam is incomplete"
            )
        if not isinstance(guard, CaptureSafetyRuntimeGuard):
            raise RuntimeSafetyError(
                "runtime_safety_runner_invalid", "Capture safety guard is invalid"
            )
        if (
            getattr(runner, "profile_sha256", None)
            != guard.evidence.motion_profile_sha256
            or getattr(runner, "duration_seconds", None) != MOTION_DURATION_SECONDS
        ):
            raise RuntimeSafetyError(
                "runtime_safety_runner_mismatch",
                "Motion runner does not match safety evidence",
            )
        if isinstance(monitor_interval_seconds, bool):
            raise RuntimeSafetyError(
                "runtime_safety_runner_invalid", "Guard interval is invalid"
            )
        try:
            interval = float(monitor_interval_seconds)
        except (TypeError, ValueError, OverflowError) as exc:
            raise RuntimeSafetyError(
                "runtime_safety_runner_invalid", "Guard interval is invalid"
            ) from exc
        if (
            not math.isfinite(interval)
            or not 0.0 < interval <= MAX_GUARD_INTERVAL_SECONDS
        ):
            raise RuntimeSafetyError(
                "runtime_safety_runner_invalid", "Guard interval is invalid"
            )
        inferred_clock = getattr(runner, "clock", None)
        clock = monotonic_clock or inferred_clock or time.monotonic
        if not callable(clock):
            raise RuntimeSafetyError(
                "runtime_safety_runner_invalid", "Monotonic clock is invalid"
            )
        self.runner = runner
        self.guard = guard
        self.profile_sha256 = guard.evidence.motion_profile_sha256
        self.duration_seconds = MOTION_DURATION_SECONDS
        self.monitor_interval_seconds = interval
        self.monotonic_clock = clock

    def start(
        self,
        *,
        start_at_monotonic_s: float,
        duration_seconds: float,
        stop_requested: Callable[[], bool],
    ) -> SafetyGuardedMotionSession:
        failure_box = _FailureBox()
        start_complete = threading.Event()

        try:
            self.guard.ensure_preflight()
        except Exception:
            try:
                self.runner.halt()
            except Exception as halt_exc:
                raise RuntimeSafetyError(
                    "runtime_safety_halt_failed",
                    "Safety preflight failure could not halt xArm",
                ) from halt_exc
            raise

        def guarded_stop_requested() -> bool:
            try:
                requested = stop_requested()
            except Exception:
                raise
            if type(requested) is not bool:
                return requested  # Underlying runner validates its callback contract.
            if requested:
                return True
            # During the underlying runner's arming sequence this callback is
            # the only guard and therefore validates every transition point.
            # Once start returns, the independent bounded monitor below owns
            # live polling; avoiding duplicate SDK snapshots here prevents the
            # runner's 100 Hz stop checks from starving its command stream.
            if start_complete.is_set():
                return False
            try:
                adapter = getattr(self.runner, "motion_adapter", None)
                if getattr(adapter, "prepared", None) is True:
                    self.guard.ensure_active()
                else:
                    self.guard.ensure_transition()
            except RuntimeSafetyError as exc:
                failure_box.set(exc)
                return True
            return False

        try:
            session = self.runner.start(
                start_at_monotonic_s=start_at_monotonic_s,
                duration_seconds=duration_seconds,
                stop_requested=guarded_stop_requested,
            )
            self.guard.ensure_active()
            start_complete.set()
        except Exception as exc:
            failure = failure_box.get()
            if failure is not None:
                raise failure from exc
            try:
                self.runner.halt()
            except Exception as halt_exc:
                raise RuntimeSafetyError(
                    "runtime_safety_halt_failed",
                    "Safety preflight failure could not halt xArm",
                ) from halt_exc
            if isinstance(exc, CaptureSafetyError):
                raise
            raise
        return SafetyGuardedMotionSession(
            session,
            self.runner,
            self.guard,
            failure_box,
            deadline_monotonic_s=float(start_at_monotonic_s) + MOTION_DURATION_SECONDS,
            monitor_interval_seconds=self.monitor_interval_seconds,
            monotonic_clock=self.monotonic_clock,
        )

    def halt(self) -> None:
        self.runner.halt()


__all__ = [
    "CaptureSafetyError",
    "CaptureSafetyEvidence",
    "CaptureSafetyRuntimeGuard",
    "ControllerPhaseEvidence",
    "ControllerSafetySnapshot",
    "ControllerSafetySource",
    "MAX_GUARD_INTERVAL_SECONDS",
    "ReducedModeEvidence",
    "RuntimeSafetyError",
    "SAFETY_EVIDENCE_SCHEMA",
    "SafetyEvidenceError",
    "SafetyGuardedMotionRunner",
    "SafetyGuardedMotionSession",
    "XArmCaptureSafetySource",
    "load_capture_safety_evidence",
]
