"""Fail-closed, read-only Lite6 workspace-teach acquisition.

The operator, UFACTORY controller, and site procedure own free-drive mode.  This
module never opens an SDK connection and has no motor-enable, mode/state,
error-clear, boundary-write, or motion-authorisation surface.  It only accepts
the already bounded runtime telemetry, guides six Cartesian extrema, and emits
an immutable *proposal* for separate qualified review.

The proposal is intentionally not a safety permit.  A TCP axis-aligned box
does not model robot links, tooling, cameras, cables, swept volume, guarding,
or stopping distance.
"""

from __future__ import annotations

import hashlib
import json
import math
import os
import re
import secrets
import stat
import tempfile
import threading
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, List, Mapping, Optional, Sequence, Tuple

WORKSPACE_TEACH_SCHEMA_VERSION = 1
WORKSPACE_TEACH_ARTIFACT_SCHEMA = "labcobot.workspace-teach-proposal.v1"
WORKSPACE_TEACH_COORDINATE_FRAME = "xarm_base_mm"
WORKSPACE_TEACH_REQUIRED_MODE = 2

WORKSPACE_TEACH_STEPS = (
    "x_min",
    "x_max",
    "y_min",
    "y_max",
    "z_min",
    "z_max",
)

WORKSPACE_TEACH_BLOCKERS = (
    "session_not_started",
    "session_finished",
    "telemetry_unavailable",
    "telemetry_stale",
    "xarm_disconnected",
    "xarm_fault",
    "wrong_robot",
    "teach_mode_not_active",
    "controller_not_ready",
    "invalid_joint_vector",
    "invalid_tcp_pose",
    "pose_outside_hard_envelope",
    "robot_moving",
    "pose_sequence_invalid",
    "pose_reused",
    "pose_unstable",
    "insufficient_span",
    "session_invalidated",
)

_BLOCKER_SET = frozenset(WORKSPACE_TEACH_BLOCKERS)
_HEX_64 = re.compile(r"[0-9a-f]{64}\Z")
_HEX_32 = re.compile(r"[0-9a-f]{32}\Z")
_MAX_POSE_SEQUENCE = 10_000_000_000
_MAX_ARTIFACT_BYTES = 128 * 1024
_MAX_PROPOSAL_ROOT_ENTRIES = 4096

# Same conservative Lite6 joint envelope used by the reviewed safety module.
_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)
)

_STEP_GUIDANCE: Mapping[str, Mapping[str, str]] = {
    "x_min": {
        "axis": "x",
        "side": "min",
        "title": "LINKE X-GRENZE",
        "instruction": "TCP von Hand zur kleinsten erlaubten X-Position fuehren.",
    },
    "x_max": {
        "axis": "x",
        "side": "max",
        "title": "RECHTE X-GRENZE",
        "instruction": "TCP von Hand zur groessten erlaubten X-Position fuehren.",
    },
    "y_min": {
        "axis": "y",
        "side": "min",
        "title": "VORDERE Y-GRENZE",
        "instruction": "TCP von Hand zur kleinsten erlaubten Y-Position fuehren.",
    },
    "y_max": {
        "axis": "y",
        "side": "max",
        "title": "HINTERE Y-GRENZE",
        "instruction": "TCP von Hand zur groessten erlaubten Y-Position fuehren.",
    },
    "z_min": {
        "axis": "z",
        "side": "min",
        "title": "UNTERE Z-GRENZE",
        "instruction": "TCP von Hand zur kleinsten erlaubten Z-Position fuehren.",
    },
    "z_max": {
        "axis": "z",
        "side": "max",
        "title": "OBERE Z-GRENZE",
        "instruction": "TCP von Hand zur groessten erlaubten Z-Position fuehren.",
    },
}


class WorkspaceTeachError(ValueError):
    """Invalid configuration, transition, telemetry, or proposal data."""


def _finite_vector(value: Any, length: int, name: str) -> Tuple[float, ...]:
    if not isinstance(value, (tuple, list)) or len(value) != length:
        raise WorkspaceTeachError(
            "{} must contain exactly {} finite values".format(name, length)
        )
    result = []
    for item in value:
        if isinstance(item, bool):
            raise WorkspaceTeachError(
                "{} must contain exactly {} finite values".format(name, length)
            )
        try:
            number = float(item)
        except (TypeError, ValueError, OverflowError) as exc:
            raise WorkspaceTeachError(
                "{} must contain exactly {} finite values".format(name, length)
            ) from exc
        if not math.isfinite(number):
            raise WorkspaceTeachError(
                "{} must contain exactly {} finite values".format(name, length)
            )
        result.append(number)
    return tuple(result)


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


def _utc_text(value: datetime) -> str:
    if not isinstance(value, datetime):
        raise WorkspaceTeachError("UTC clock returned an invalid value")
    if value.tzinfo is None or value.utcoffset() is None:
        raise WorkspaceTeachError("UTC clock must return a timezone-aware value")
    normalized = value.astimezone(timezone.utc)
    return normalized.isoformat(timespec="milliseconds").replace("+00:00", "Z")


def _canonical_json(document: Mapping[str, Any]) -> bytes:
    try:
        return json.dumps(
            document,
            allow_nan=False,
            ensure_ascii=True,
            separators=(",", ":"),
            sort_keys=True,
        ).encode("utf-8")
    except (TypeError, ValueError, OverflowError) as exc:
        raise WorkspaceTeachError("workspace proposal is not canonical JSON") from exc


@dataclass(frozen=True)
class _PoseSample:
    pose_sequence: int
    observed_at_s: float
    sample_age_seconds: float
    joint_angles_rad: Tuple[float, ...]
    tcp_pose_mm_rad: Tuple[float, ...]


@dataclass(frozen=True)
class WorkspaceTeachPoint:
    """One backend-captured extrema sample."""

    step: str
    pose_sequence: int
    captured_at_utc: str
    joint_angles_rad: Tuple[float, ...]
    tcp_pose_mm_rad: Tuple[float, ...]

    def to_dict(self) -> Mapping[str, Any]:
        return {
            "step": self.step,
            "pose_sequence": self.pose_sequence,
            "captured_at_utc": self.captured_at_utc,
            "joint_angles_rad": list(self.joint_angles_rad),
            "tcp_pose_mm_rad": list(self.tcp_pose_mm_rad),
        }


@dataclass(frozen=True)
class WorkspaceTeachProposal:
    """Canonical review proposal and its exact content digest."""

    document: Mapping[str, Any]
    canonical_bytes: bytes
    sha256: str

    @classmethod
    def build(cls, document: Mapping[str, Any]) -> "WorkspaceTeachProposal":
        copied = json.loads(_canonical_json(document).decode("utf-8"))
        _validate_proposal_document(copied)
        canonical = _canonical_json(copied)
        return cls(
            document=copied,
            canonical_bytes=canonical,
            sha256=hashlib.sha256(canonical).hexdigest(),
        )

    def to_dict(self) -> Mapping[str, Any]:
        return json.loads(self.canonical_bytes.decode("utf-8"))


class WorkspaceTeachRecorder:
    """Guide six extrema using only fresh, stable, read-only telemetry.

    ``ingest`` must be called for every new cached xArm pose sample.  ``capture``
    deliberately accepts no coordinate argument, so a frontend cannot inject a
    box; it records the latest backend-validated stable sample for the expected
    step only.
    """

    def __init__(
        self,
        *,
        hard_lower_xyz_mm: Sequence[float],
        hard_upper_xyz_mm: Sequence[float],
        inset_xyz_mm: Sequence[float],
        minimum_span_xyz_mm: Sequence[float],
        robot_binding_sha256: str,
        commissioning_permit_sha256: str,
        maximum_sample_age_seconds: float = 0.25,
        stability_duration_seconds: float = 0.5,
        stability_minimum_samples: int = 3,
        tcp_stability_tolerance_mm: float = 1.0,
        joint_stability_tolerance_rad: float = 0.01,
        monotonic: Callable[[], float] = time.monotonic,
        utcnow: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
        token_hex: Callable[[int], str] = secrets.token_hex,
    ) -> None:
        self.hard_lower_xyz_mm = _finite_vector(
            hard_lower_xyz_mm, 3, "hard lower workspace"
        )
        self.hard_upper_xyz_mm = _finite_vector(
            hard_upper_xyz_mm, 3, "hard upper workspace"
        )
        self.inset_xyz_mm = _finite_vector(inset_xyz_mm, 3, "workspace inset")
        self.minimum_span_xyz_mm = _finite_vector(
            minimum_span_xyz_mm, 3, "minimum workspace span"
        )
        if any(
            lower >= upper
            for lower, upper in zip(self.hard_lower_xyz_mm, self.hard_upper_xyz_mm)
        ):
            raise WorkspaceTeachError("hard workspace envelope is invalid")
        if any(value <= 0.0 for value in self.inset_xyz_mm):
            raise WorkspaceTeachError("workspace inset must be positive")
        if any(value <= 0.0 for value in self.minimum_span_xyz_mm):
            raise WorkspaceTeachError("minimum workspace span must be positive")
        if any(
            minimum + 2.0 * inset > upper - lower
            for minimum, inset, lower, upper in zip(
                self.minimum_span_xyz_mm,
                self.inset_xyz_mm,
                self.hard_lower_xyz_mm,
                self.hard_upper_xyz_mm,
            )
        ):
            raise WorkspaceTeachError(
                "hard workspace cannot contain the minimum inset proposal"
            )
        self.robot_binding_sha256 = _digest(robot_binding_sha256, "robot binding")
        self.commissioning_permit_sha256 = _digest(
            commissioning_permit_sha256, "commissioning permit"
        )
        self.maximum_sample_age_seconds = self._positive_bounded(
            maximum_sample_age_seconds,
            "maximum sample age",
            maximum=3.0,
        )
        self.stability_duration_seconds = self._positive_bounded(
            stability_duration_seconds,
            "stability duration",
            maximum=10.0,
        )
        if (
            isinstance(stability_minimum_samples, bool)
            or not isinstance(stability_minimum_samples, int)
            or not 2 <= stability_minimum_samples <= 100
        ):
            raise WorkspaceTeachError(
                "stability minimum samples must be between 2 and 100"
            )
        self.stability_minimum_samples = stability_minimum_samples
        self.tcp_stability_tolerance_mm = self._positive_bounded(
            tcp_stability_tolerance_mm,
            "TCP stability tolerance",
            maximum=10.0,
        )
        self.joint_stability_tolerance_rad = self._positive_bounded(
            joint_stability_tolerance_rad,
            "joint stability tolerance",
            maximum=0.2,
        )
        self._monotonic = monotonic
        self._utcnow = utcnow
        self._token_hex = token_hex
        self._lock = threading.RLock()
        self._state = "idle"
        self._session_id = ""
        self._points: List[WorkspaceTeachPoint] = []
        self._stable_samples: List[_PoseSample] = []
        self._last_pose_sequence = -1
        self._latest_sample: Optional[_PoseSample] = None
        self._blockers: Tuple[str, ...] = ("session_not_started",)
        self._proposal: Optional[WorkspaceTeachProposal] = None

    @staticmethod
    def _positive_bounded(value: Any, name: str, *, maximum: float) -> float:
        if isinstance(value, bool):
            raise WorkspaceTeachError("{} is invalid".format(name))
        try:
            number = float(value)
        except (TypeError, ValueError, OverflowError) as exc:
            raise WorkspaceTeachError("{} is invalid".format(name)) from exc
        if not math.isfinite(number) or not 0.0 < number <= maximum:
            raise WorkspaceTeachError("{} is invalid".format(name))
        return number

    @property
    def state(self) -> str:
        return self._state

    @property
    def session_id(self) -> str:
        return self._session_id

    def begin(self) -> Mapping[str, Any]:
        """Start a new acquisition without changing the controller."""

        with self._lock:
            if self._state not in {
                "idle",
                "cancelled",
                "invalidated",
                "proposed",
            }:
                raise WorkspaceTeachError(
                    "an active workspace-teach session already exists"
                )
            token = self._token_hex(16)
            if not isinstance(token, str) or _HEX_32.fullmatch(token) is None:
                raise WorkspaceTeachError(
                    "session token generator returned invalid data"
                )
            self._session_id = token
            self._state = "awaiting_teach"
            self._points = []
            self._stable_samples = []
            self._last_pose_sequence = -1
            self._latest_sample = None
            self._blockers = ("teach_mode_not_active",)
            self._proposal = None
            return self.snapshot()

    def cancel(self) -> Mapping[str, Any]:
        """Cancel acquisition locally; this is not a controller STOP."""

        with self._lock:
            if self._state == "invalidated":
                return self.snapshot()
            if self._state not in {
                "idle",
                "proposed",
                "cancelled",
                "invalidated",
            }:
                self._state = "cancelled"
            self._stable_samples = []
            self._latest_sample = None
            self._blockers = ("session_finished",)
            return self.snapshot()

    def invalidate(self) -> Mapping[str, Any]:
        """Irreversibly invalidate the active session after STOP/fault/restart."""

        with self._lock:
            if self._state not in {"idle", "proposed", "cancelled", "invalidated"}:
                self._state = "invalidated"
                self._stable_samples = []
                self._latest_sample = None
                self._proposal = None
                self._blockers = ("session_invalidated",)
            return self.snapshot()

    def ingest(
        self,
        xarm_status: Mapping[str, Any],
        *,
        pose_sequence: int,
    ) -> Mapping[str, Any]:
        """Accept one new normalized telemetry sample and update stability."""

        with self._lock:
            if self._state in {
                "idle",
                "cancelled",
                "invalidated",
                "review",
                "proposed",
            }:
                self._blockers = (
                    (
                        "session_not_started"
                        if self._state == "idle"
                        else "session_finished"
                    ),
                )
                return self.snapshot()
            blocker, sample = self._validate_telemetry(
                xarm_status,
                pose_sequence=pose_sequence,
            )
            if blocker is not None:
                self._stable_samples = []
                self._latest_sample = None
                self._blockers = (blocker,)
                if (
                    blocker
                    in {
                        "telemetry_unavailable",
                        "telemetry_stale",
                        "xarm_disconnected",
                        "xarm_fault",
                        "wrong_robot",
                        "teach_mode_not_active",
                        "controller_not_ready",
                        "invalid_joint_vector",
                        "invalid_tcp_pose",
                        "pose_outside_hard_envelope",
                    }
                    and self._points
                ):
                    self._state = "invalidated"
                    self._blockers = ("session_invalidated", blocker)
                else:
                    self._state = "awaiting_teach"
                return self.snapshot()
            assert sample is not None

            if sample.pose_sequence <= self._last_pose_sequence:
                self._blockers = (
                    (
                        "pose_reused"
                        if sample.pose_sequence == self._last_pose_sequence
                        else "pose_sequence_invalid"
                    ),
                )
                if sample.pose_sequence < self._last_pose_sequence and self._points:
                    self._state = "invalidated"
                    self._blockers = (
                        "session_invalidated",
                        "pose_sequence_invalid",
                    )
                return self.snapshot()
            self._last_pose_sequence = sample.pose_sequence

            if xarm_status.get("moving") is not False:
                self._stable_samples = []
                self._latest_sample = sample
                self._state = "positioning"
                self._blockers = ("robot_moving",)
                return self.snapshot()

            if not self._stable_samples or self._sample_matches_stable_anchor(sample):
                self._stable_samples.append(sample)
            else:
                self._stable_samples = [sample]
            if len(self._stable_samples) > 100:
                self._stable_samples = self._stable_samples[-100:]
            self._latest_sample = sample

            stable_duration = (
                sample.observed_at_s - self._stable_samples[0].observed_at_s
                if self._stable_samples
                else 0.0
            )
            if (
                len(self._stable_samples) >= self.stability_minimum_samples
                and stable_duration >= self.stability_duration_seconds
            ):
                self._state = "ready_to_capture"
                self._blockers = ()
            else:
                self._state = "positioning"
                self._blockers = ("pose_unstable",)
            return self.snapshot()

    def capture(self) -> Mapping[str, Any]:
        """Capture the expected extrema from backend telemetry only."""

        with self._lock:
            if self._state != "ready_to_capture" or self._latest_sample is None:
                raise WorkspaceTeachError(
                    "no stable backend TCP pose is ready to capture"
                )
            now = self._monotonic()
            try:
                capture_age = float(now) - self._latest_sample.observed_at_s
            except (TypeError, ValueError, OverflowError) as exc:
                raise WorkspaceTeachError("backend TCP pose age is invalid") from exc
            if (
                not math.isfinite(capture_age)
                or capture_age < 0.0
                or capture_age > self.maximum_sample_age_seconds
            ):
                self._stable_samples = []
                self._latest_sample = None
                self._state = "awaiting_teach"
                self._blockers = ("telemetry_stale",)
                raise WorkspaceTeachError("backend TCP pose became stale")
            if len(self._points) >= len(WORKSPACE_TEACH_STEPS):
                raise WorkspaceTeachError("all workspace extrema are already captured")
            step = WORKSPACE_TEACH_STEPS[len(self._points)]
            sample = self._latest_sample
            point = WorkspaceTeachPoint(
                step=step,
                pose_sequence=sample.pose_sequence,
                captured_at_utc=_utc_text(self._utcnow()),
                joint_angles_rad=sample.joint_angles_rad,
                tcp_pose_mm_rad=sample.tcp_pose_mm_rad,
            )
            self._points.append(point)
            self._stable_samples = []
            self._latest_sample = None
            if len(self._points) == len(WORKSPACE_TEACH_STEPS):
                try:
                    self._proposal = self._build_proposal()
                except WorkspaceTeachError:
                    self._state = "invalidated"
                    self._blockers = (
                        "session_invalidated",
                        "insufficient_span",
                    )
                else:
                    self._state = "review"
                    self._blockers = ()
            else:
                self._state = "positioning"
                self._blockers = ("pose_unstable",)
            return self.snapshot()

    def finalize(self) -> WorkspaceTeachProposal:
        """Reject the legacy non-durable finalization bypass."""

        raise WorkspaceTeachError(
            "durable proposal persistence is required before finalization"
        )

    def review_proposal(self) -> WorkspaceTeachProposal:
        """Return the immutable review proposal without changing state."""

        with self._lock:
            if self._state != "review" or self._proposal is None:
                raise WorkspaceTeachError("workspace proposal is not ready for review")
            return self._proposal

    def _mark_persisted(
        self,
        *,
        session_id: str,
        proposal_sha256: str,
        still_valid: Callable[[], bool],
    ) -> WorkspaceTeachProposal:
        """Expose ``proposed`` only after the caller proves durable storage.

        ``review_proposal`` and this transition form a two-phase seam so the
        append-only file write can run without holding recorder, workcell,
        motion, or operation-transition locks.  The caller must hold the same
        short linearization lock used to publish its STOP fence while this
        method evaluates ``still_valid`` and changes state.  Therefore a STOP
        that linearizes first makes this transition fail; a final mark that
        linearizes first is already a completed durable proposal.
        """

        with self._lock:
            if not callable(still_valid):
                raise WorkspaceTeachError("workspace proposal fence is unavailable")
            if self._state != "review" or self._proposal is None:
                raise WorkspaceTeachError("workspace proposal is not ready for review")
            if (
                not isinstance(session_id, str)
                or not secrets.compare_digest(self._session_id, session_id)
                or not isinstance(proposal_sha256, str)
                or not secrets.compare_digest(
                    self._proposal.sha256,
                    proposal_sha256,
                )
            ):
                raise WorkspaceTeachError("workspace proposal identity changed")
            try:
                fence_valid = still_valid() is True
            except Exception as exc:
                raise WorkspaceTeachError(
                    "workspace proposal fence validation failed"
                ) from exc
            if not fence_valid:
                raise WorkspaceTeachError(
                    "workspace proposal was invalidated before finalization"
                )
            self._state = "proposed"
            self._blockers = ("session_finished",)
            return self._proposal

    def snapshot(self) -> Mapping[str, Any]:
        """Return the bounded frontend contract; no raw controller object leaks."""

        with self._lock:
            current_step = (
                WORKSPACE_TEACH_STEPS[len(self._points)]
                if len(self._points) < len(WORKSPACE_TEACH_STEPS)
                and self._state
                not in {"idle", "cancelled", "invalidated", "review", "proposed"}
                else ""
            )
            guidance = dict(_STEP_GUIDANCE[current_step]) if current_step else {}
            current_tcp = (
                list(self._latest_sample.tcp_pose_mm_rad[:3])
                if self._latest_sample is not None
                else []
            )
            raw_lower, raw_upper = self._partial_bounds()
            blockers = [value for value in self._blockers if value in _BLOCKER_SET]
            return {
                "schema_version": WORKSPACE_TEACH_SCHEMA_VERSION,
                "configured": True,
                "state": self._state,
                "coordinate_frame": WORKSPACE_TEACH_COORDINATE_FRAME,
                "required_controller_mode": WORKSPACE_TEACH_REQUIRED_MODE,
                "session_id": self._session_id,
                "current_step": current_step,
                "completed_steps": len(self._points),
                "total_steps": len(WORKSPACE_TEACH_STEPS),
                "guidance": guidance,
                "pose_ready": self._state == "ready_to_capture",
                "current_tcp_position_mm": current_tcp,
                "hard_envelope_mm": {
                    "lower": list(self.hard_lower_xyz_mm),
                    "upper": list(self.hard_upper_xyz_mm),
                },
                "captured_bounds_mm": {"lower": raw_lower, "upper": raw_upper},
                "proposal_sha256": (
                    self._proposal.sha256 if self._proposal is not None else ""
                ),
                "blockers": blockers,
                "safety_notice": (
                    "TCP-BOX IST NUR EIN PRUEFVORSCHLAG; KEINE MOTION-FREIGABE"
                ),
            }

    def _validate_telemetry(
        self,
        xarm_status: Mapping[str, Any],
        *,
        pose_sequence: int,
    ) -> Tuple[Optional[str], Optional[_PoseSample]]:
        if (
            not isinstance(xarm_status, Mapping)
            or xarm_status.get("available") is not True
        ):
            return "telemetry_unavailable", None
        if xarm_status.get("connected") is not True:
            return "xarm_disconnected", None
        if (
            xarm_status.get("healthy") is not True
            or xarm_status.get("error_code") != 0
            or xarm_status.get("warning_code") != 0
        ):
            return "xarm_fault", None
        if xarm_status.get("axis_count") != 6 or xarm_status.get("device_type") != 9:
            return "wrong_robot", None
        if xarm_status.get("mode") != WORKSPACE_TEACH_REQUIRED_MODE:
            return "teach_mode_not_active", None
        if xarm_status.get("state") != 0:
            return "controller_not_ready", None
        age = xarm_status.get("sample_age_seconds")
        if isinstance(age, bool):
            return "telemetry_stale", None
        try:
            age_number = float(age)
        except (TypeError, ValueError, OverflowError):
            return "telemetry_stale", None
        if (
            xarm_status.get("fresh") is not True
            or not math.isfinite(age_number)
            or not 0.0 <= age_number <= self.maximum_sample_age_seconds
        ):
            return "telemetry_stale", None
        if (
            isinstance(pose_sequence, bool)
            or not isinstance(pose_sequence, int)
            or not 0 <= pose_sequence <= _MAX_POSE_SEQUENCE
        ):
            return "pose_sequence_invalid", None
        metrics = xarm_status.get("metrics")
        if not isinstance(metrics, Mapping):
            return "invalid_joint_vector", None
        try:
            joints = _finite_vector(metrics.get("joint_angles_rad"), 6, "joint angles")
        except WorkspaceTeachError:
            return "invalid_joint_vector", None
        if any(
            joint < lower or joint > upper
            for joint, lower, upper in zip(
                joints,
                _LITE6_JOINT_LOWER_RADIANS,
                _LITE6_JOINT_UPPER_RADIANS,
            )
        ):
            return "invalid_joint_vector", None
        try:
            position = _finite_vector(metrics.get("tcp_position_mm"), 3, "TCP position")
            orientation = _finite_vector(
                metrics.get("tcp_orientation_rad"), 3, "TCP orientation"
            )
        except WorkspaceTeachError:
            return "invalid_tcp_pose", None
        if any(
            value < lower or value > upper
            for value, lower, upper in zip(
                position,
                self.hard_lower_xyz_mm,
                self.hard_upper_xyz_mm,
            )
        ):
            return "pose_outside_hard_envelope", None
        now = self._monotonic()
        if isinstance(now, bool):
            return "telemetry_unavailable", None
        try:
            observed_at = float(now)
        except (TypeError, ValueError, OverflowError):
            return "telemetry_unavailable", None
        if not math.isfinite(observed_at):
            return "telemetry_unavailable", None
        return (
            None,
            _PoseSample(
                pose_sequence=pose_sequence,
                observed_at_s=observed_at,
                sample_age_seconds=age_number,
                joint_angles_rad=joints,
                tcp_pose_mm_rad=position + orientation,
            ),
        )

    def _sample_matches_stable_anchor(self, sample: _PoseSample) -> bool:
        anchor = self._stable_samples[0]
        tcp_delta = math.sqrt(
            sum(
                (current - original) ** 2
                for current, original in zip(
                    sample.tcp_pose_mm_rad[:3], anchor.tcp_pose_mm_rad[:3]
                )
            )
        )
        joint_delta = max(
            abs(current - original)
            for current, original in zip(
                sample.joint_angles_rad, anchor.joint_angles_rad
            )
        )
        return bool(
            tcp_delta <= self.tcp_stability_tolerance_mm
            and joint_delta <= self.joint_stability_tolerance_rad
        )

    def _partial_bounds(self) -> Tuple[List[Optional[float]], List[Optional[float]]]:
        extrema = {
            point.step: point.tcp_pose_mm_rad["xyz".index(point.step[0])]
            for point in self._points
        }
        lower = [extrema.get("{}_min".format(axis)) for axis in "xyz"]
        upper = [extrema.get("{}_max".format(axis)) for axis in "xyz"]
        return lower, upper

    def _build_proposal(self) -> WorkspaceTeachProposal:
        if len(self._points) != len(WORKSPACE_TEACH_STEPS):
            raise WorkspaceTeachError("six workspace extrema are required")
        raw_lower_values, raw_upper_values = self._partial_bounds()
        if any(value is None for value in raw_lower_values + raw_upper_values):
            raise WorkspaceTeachError("workspace extrema are incomplete")
        raw_lower = tuple(
            float(value) for value in raw_lower_values if value is not None
        )
        raw_upper = tuple(
            float(value) for value in raw_upper_values if value is not None
        )
        proposed_lower = tuple(
            value + inset for value, inset in zip(raw_lower, self.inset_xyz_mm)
        )
        proposed_upper = tuple(
            value - inset for value, inset in zip(raw_upper, self.inset_xyz_mm)
        )
        if any(
            upper - lower < minimum
            for lower, upper, minimum in zip(
                proposed_lower,
                proposed_upper,
                self.minimum_span_xyz_mm,
            )
        ):
            raise WorkspaceTeachError("workspace proposal span is too small")
        if any(
            proposed < hard
            for proposed, hard in zip(proposed_lower, self.hard_lower_xyz_mm)
        ) or any(
            proposed > hard
            for proposed, hard in zip(proposed_upper, self.hard_upper_xyz_mm)
        ):
            raise WorkspaceTeachError("workspace proposal exceeds hard envelope")
        document = {
            "schema_version": WORKSPACE_TEACH_SCHEMA_VERSION,
            "schema": WORKSPACE_TEACH_ARTIFACT_SCHEMA,
            "status": "review_proposal",
            "coordinate_frame": WORKSPACE_TEACH_COORDINATE_FRAME,
            "session_id": self._session_id,
            "created_at_utc": _utc_text(self._utcnow()),
            "robot": {"model": "lite6", "axis_count": 6, "device_type": 9},
            "bindings": {
                "robot_sha256": self.robot_binding_sha256,
                "commissioning_permit_sha256": self.commissioning_permit_sha256,
            },
            "acquisition": {
                "source": "bounded_xarm_read_only_telemetry",
                "required_controller_mode": WORKSPACE_TEACH_REQUIRED_MODE,
                "maximum_sample_age_seconds": self.maximum_sample_age_seconds,
                "stability_duration_seconds": self.stability_duration_seconds,
                "stability_minimum_samples": self.stability_minimum_samples,
                "tcp_stability_tolerance_mm": self.tcp_stability_tolerance_mm,
                "joint_stability_tolerance_rad": (self.joint_stability_tolerance_rad),
            },
            "hard_envelope_mm": {
                "lower": list(self.hard_lower_xyz_mm),
                "upper": list(self.hard_upper_xyz_mm),
            },
            "raw_extrema_mm": {
                "lower": list(raw_lower),
                "upper": list(raw_upper),
            },
            "inset_xyz_mm": list(self.inset_xyz_mm),
            "minimum_span_xyz_mm": list(self.minimum_span_xyz_mm),
            "proposed_bounds_mm": {
                "lower": list(proposed_lower),
                "upper": list(proposed_upper),
            },
            "samples": [point.to_dict() for point in self._points],
            "limitations": [
                "tcp_axis_aligned_box_only",
                "links_tool_cameras_cables_not_modelled",
                "swept_volume_not_verified",
                "not_a_motion_release",
                "requires_separate_qualified_review_and_promotion",
            ],
        }
        return WorkspaceTeachProposal.build(document)


class WorkspaceTeachProposalStore:
    """Append-only state storage for untrusted, review-only proposals."""

    def __init__(self, root: Path) -> None:
        self.root = Path(root)

    def save(self, proposal: WorkspaceTeachProposal) -> Path:
        if not isinstance(proposal, WorkspaceTeachProposal):
            raise WorkspaceTeachError("workspace proposal type is invalid")
        if (
            not proposal.canonical_bytes
            or len(proposal.canonical_bytes) > _MAX_ARTIFACT_BYTES
        ):
            raise WorkspaceTeachError("workspace proposal size is invalid")
        session_id = proposal.document.get("session_id")
        if not isinstance(session_id, str) or _HEX_32.fullmatch(session_id) is None:
            raise WorkspaceTeachError("workspace proposal session is invalid")
        root = self._verified_root()
        destination = root / "{}.json".format(session_id)
        descriptor, temporary_name = tempfile.mkstemp(
            prefix=".workspace-teach-",
            suffix=".tmp",
            dir=str(root),
        )
        temporary = Path(temporary_name)
        try:
            os.fchmod(descriptor, 0o600)
            with os.fdopen(descriptor, "wb") as output:
                output.write(proposal.canonical_bytes)
                output.flush()
                os.fsync(output.fileno())
            try:
                os.link(str(temporary), str(destination), follow_symlinks=False)
            except FileExistsError:
                # A prior fsync result may have been lost to the caller after
                # the immutable link was already created.  Treat only the
                # exact canonical proposal as an idempotent retry; a different
                # file at the same session path remains a hard failure.
                existing = self.load(
                    session_id,
                    expected_sha256=proposal.sha256,
                )
                if existing.canonical_bytes != proposal.canonical_bytes:
                    raise WorkspaceTeachError(
                        "workspace proposal already exists with different content"
                    )
            else:
                # Remove the staging name before the directory fsync.  The
                # destination is already an immutable hard link to fully
                # fsynced bytes; persisting the transient second link would
                # leave a power-loss recovery ambiguity (st_nlink == 2).
                temporary.unlink()
            self._fsync_directory(root)
        finally:
            try:
                temporary.unlink()
            except FileNotFoundError:
                pass
        return destination

    def load(self, session_id: str, *, expected_sha256: str) -> WorkspaceTeachProposal:
        if not isinstance(session_id, str) or _HEX_32.fullmatch(session_id) is None:
            raise WorkspaceTeachError("workspace proposal session is invalid")
        expected = _digest(expected_sha256, "workspace proposal")
        root = self._verified_root()
        path = root / "{}.json".format(session_id)
        flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0)
        flags |= getattr(os, "O_NOFOLLOW", 0)
        try:
            descriptor = os.open(str(path), flags)
        except OSError as exc:
            raise WorkspaceTeachError("workspace proposal cannot be opened") from exc
        try:
            raw = os.read(descriptor, _MAX_ARTIFACT_BYTES + 1)
            metadata = os.fstat(descriptor)
            if metadata.st_nlink > 1:
                self._recover_interrupted_staging_link(root, metadata)
                metadata = os.fstat(descriptor)
        finally:
            os.close(descriptor)
        if len(raw) > _MAX_ARTIFACT_BYTES or not raw:
            raise WorkspaceTeachError("workspace proposal size is invalid")
        if (
            not stat.S_ISREG(metadata.st_mode)
            or metadata.st_uid != os.geteuid()
            or stat.S_IMODE(metadata.st_mode) != 0o600
            or metadata.st_nlink != 1
            or metadata.st_size != len(raw)
        ):
            raise WorkspaceTeachError("workspace proposal file is invalid")
        actual = hashlib.sha256(raw).hexdigest()
        if not secrets.compare_digest(actual, expected):
            raise WorkspaceTeachError("workspace proposal SHA-256 does not match")
        try:
            document = json.loads(raw.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise WorkspaceTeachError("workspace proposal JSON is invalid") from exc
        proposal = WorkspaceTeachProposal.build(document)
        if proposal.canonical_bytes != raw or proposal.sha256 != actual:
            raise WorkspaceTeachError("workspace proposal is not canonical")
        return proposal

    def _recover_interrupted_staging_link(
        self, root: Path, metadata: os.stat_result
    ) -> None:
        """Remove only a same-inode staging link left by an interrupted save.

        A crash can occur after the no-overwrite hard link is created but
        before its private temporary name is removed.  Such a link is not a
        second proposal and may be recovered deterministically.  Any other
        hard link remains a hard failure in ``load``.
        """

        matched = 0
        try:
            with os.scandir(str(root)) as entries:
                for index, entry in enumerate(entries, start=1):
                    if index > _MAX_PROPOSAL_ROOT_ENTRIES:
                        raise WorkspaceTeachError(
                            "workspace proposal root contains too many entries"
                        )
                    if not (
                        entry.name.startswith(".workspace-teach-")
                        and entry.name.endswith(".tmp")
                    ):
                        continue
                    entry_metadata = entry.stat(follow_symlinks=False)
                    if (
                        stat.S_ISREG(entry_metadata.st_mode)
                        and entry_metadata.st_dev == metadata.st_dev
                        and entry_metadata.st_ino == metadata.st_ino
                        and entry_metadata.st_uid == os.geteuid()
                        and stat.S_IMODE(entry_metadata.st_mode) == 0o600
                    ):
                        os.unlink(entry.path)
                        matched += 1
        except WorkspaceTeachError:
            raise
        except OSError as exc:
            raise WorkspaceTeachError(
                "workspace proposal interrupted save cannot be recovered"
            ) from exc
        if matched:
            self._fsync_directory(root)

    def _verified_root(self) -> Path:
        try:
            metadata = self.root.lstat()
            resolved = self.root.resolve(strict=True)
        except OSError as exc:
            raise WorkspaceTeachError("workspace proposal root is unavailable") from exc
        if (
            self.root.is_symlink()
            or not self.root.is_dir()
            or resolved != self.root.absolute()
            or not stat.S_ISDIR(metadata.st_mode)
            or metadata.st_uid != os.geteuid()
            or stat.S_IMODE(metadata.st_mode) & 0o077
        ):
            raise WorkspaceTeachError("workspace proposal root is unsafe")
        return resolved

    @staticmethod
    def _fsync_directory(root: Path) -> None:
        flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
        descriptor = os.open(str(root), flags)
        try:
            os.fsync(descriptor)
        finally:
            os.close(descriptor)


def _validate_proposal_document(document: Any) -> None:
    expected_keys = {
        "schema_version",
        "schema",
        "status",
        "coordinate_frame",
        "session_id",
        "created_at_utc",
        "robot",
        "bindings",
        "acquisition",
        "hard_envelope_mm",
        "raw_extrema_mm",
        "inset_xyz_mm",
        "minimum_span_xyz_mm",
        "proposed_bounds_mm",
        "samples",
        "limitations",
    }
    if not isinstance(document, dict) or set(document) != expected_keys:
        raise WorkspaceTeachError("workspace proposal schema is invalid")
    if (
        document.get("schema_version") != WORKSPACE_TEACH_SCHEMA_VERSION
        or document.get("schema") != WORKSPACE_TEACH_ARTIFACT_SCHEMA
        or document.get("status") != "review_proposal"
        or document.get("coordinate_frame") != WORKSPACE_TEACH_COORDINATE_FRAME
        or not isinstance(document.get("session_id"), str)
        or _HEX_32.fullmatch(document["session_id"]) is None
        or not isinstance(document.get("created_at_utc"), str)
        or len(document["created_at_utc"]) > 64
    ):
        raise WorkspaceTeachError("workspace proposal identity is invalid")
    robot = document.get("robot")
    if robot != {"model": "lite6", "axis_count": 6, "device_type": 9}:
        raise WorkspaceTeachError("workspace proposal robot identity is invalid")
    bindings = document.get("bindings")
    if not isinstance(bindings, dict) or set(bindings) != {
        "robot_sha256",
        "commissioning_permit_sha256",
    }:
        raise WorkspaceTeachError("workspace proposal bindings are invalid")
    _digest(bindings["robot_sha256"], "robot binding")
    _digest(bindings["commissioning_permit_sha256"], "commissioning permit")
    for name in (
        "hard_envelope_mm",
        "raw_extrema_mm",
        "proposed_bounds_mm",
    ):
        bounds = document.get(name)
        if not isinstance(bounds, dict) or set(bounds) != {"lower", "upper"}:
            raise WorkspaceTeachError("{} is invalid".format(name))
        lower = _finite_vector(bounds["lower"], 3, "{} lower".format(name))
        upper = _finite_vector(bounds["upper"], 3, "{} upper".format(name))
        if any(low >= high for low, high in zip(lower, upper)):
            raise WorkspaceTeachError("{} is invalid".format(name))
    inset = _finite_vector(document.get("inset_xyz_mm"), 3, "workspace inset")
    minimum = _finite_vector(
        document.get("minimum_span_xyz_mm"), 3, "minimum workspace span"
    )
    if any(value <= 0.0 for value in inset + minimum):
        raise WorkspaceTeachError("workspace proposal margins are invalid")
    hard = document["hard_envelope_mm"]
    proposed = document["proposed_bounds_mm"]
    if any(
        value < hard_value
        for value, hard_value in zip(proposed["lower"], hard["lower"])
    ) or any(
        value > hard_value
        for value, hard_value in zip(proposed["upper"], hard["upper"])
    ):
        raise WorkspaceTeachError("workspace proposal exceeds hard envelope")
    if any(
        upper - lower < span
        for lower, upper, span in zip(proposed["lower"], proposed["upper"], minimum)
    ):
        raise WorkspaceTeachError("workspace proposal span is invalid")
    samples = document.get("samples")
    if not isinstance(samples, list) or len(samples) != len(WORKSPACE_TEACH_STEPS):
        raise WorkspaceTeachError("workspace proposal samples are invalid")
    seen_sequences = set()
    for expected_step, sample in zip(WORKSPACE_TEACH_STEPS, samples):
        if not isinstance(sample, dict) or set(sample) != {
            "step",
            "pose_sequence",
            "captured_at_utc",
            "joint_angles_rad",
            "tcp_pose_mm_rad",
        }:
            raise WorkspaceTeachError("workspace proposal sample is invalid")
        sequence = sample.get("pose_sequence")
        if (
            sample.get("step") != expected_step
            or isinstance(sequence, bool)
            or not isinstance(sequence, int)
            or not 0 <= sequence <= _MAX_POSE_SEQUENCE
            or sequence in seen_sequences
            or not isinstance(sample.get("captured_at_utc"), str)
            or len(sample["captured_at_utc"]) > 64
        ):
            raise WorkspaceTeachError("workspace proposal sample identity is invalid")
        seen_sequences.add(sequence)
        _finite_vector(sample.get("joint_angles_rad"), 6, "sample joints")
        _finite_vector(sample.get("tcp_pose_mm_rad"), 6, "sample TCP pose")
    acquisition = document.get("acquisition")
    if not isinstance(acquisition, dict) or set(acquisition) != {
        "source",
        "required_controller_mode",
        "maximum_sample_age_seconds",
        "stability_duration_seconds",
        "stability_minimum_samples",
        "tcp_stability_tolerance_mm",
        "joint_stability_tolerance_rad",
    }:
        raise WorkspaceTeachError("workspace proposal acquisition is invalid")
    if (
        acquisition.get("source") != "bounded_xarm_read_only_telemetry"
        or acquisition.get("required_controller_mode") != WORKSPACE_TEACH_REQUIRED_MODE
    ):
        raise WorkspaceTeachError("workspace proposal acquisition is invalid")
    limitations = document.get("limitations")
    required_limitations = {
        "tcp_axis_aligned_box_only",
        "links_tool_cameras_cables_not_modelled",
        "swept_volume_not_verified",
        "not_a_motion_release",
        "requires_separate_qualified_review_and_promotion",
    }
    if not isinstance(limitations, list) or set(limitations) != required_limitations:
        raise WorkspaceTeachError("workspace proposal limitations are invalid")


__all__ = [
    "WORKSPACE_TEACH_ARTIFACT_SCHEMA",
    "WORKSPACE_TEACH_BLOCKERS",
    "WORKSPACE_TEACH_COORDINATE_FRAME",
    "WORKSPACE_TEACH_REQUIRED_MODE",
    "WORKSPACE_TEACH_SCHEMA_VERSION",
    "WORKSPACE_TEACH_STEPS",
    "WorkspaceTeachError",
    "WorkspaceTeachPoint",
    "WorkspaceTeachProposal",
    "WorkspaceTeachProposalStore",
    "WorkspaceTeachRecorder",
]
