"""Bounded HTTP MJPEG handoff for the two workcell camera streams."""

from __future__ import annotations

import hmac
import json
import logging
import math
import socket
import threading
import time
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Any, Callable, Mapping, Optional
from urllib.parse import urlsplit

from .cameras import CameraRole, CameraStream, encode_jpeg

LOGGER = logging.getLogger("labcobot.mjpeg")
STREAM_PATHS: Mapping[str, CameraRole] = {
    "/tool.mjpg": CameraRole.TOOL,
    "/workspace.mjpg": CameraRole.WORKSPACE,
}
SNAPSHOT_PATHS: Mapping[str, CameraRole] = {
    "/api/cameras/tool/snapshot.jpg": CameraRole.TOOL,
    "/api/cameras/workspace/snapshot.jpg": CameraRole.WORKSPACE,
}
SNAPSHOT_CLIENT_HEADER = "lvgl-frontend-v1"
SNAPSHOT_MAX_BYTES = 512 * 1024
OTA_CHECK_PATH = "/api/ota/check"
OTA_ACTION_HEADER = "signed-ota-check"
CAPTURE_START_PATH = "/api/capture-run/start"
CAPTURE_STOP_PATH = "/api/capture-run/stop"
CAPTURE_START_ACTION_HEADER = "capture-run-start"
CAPTURE_STOP_ACTION_HEADER = "capture-run-stop"
SOFTWARE_STOP_PATH = "/api/safety/stop"
SOFTWARE_STOP_ACTION_HEADER = "software-stop-all-v1"
MOTION_SEARCH_ARM_PATH = "/api/motion/search/arm"
MOTION_SEARCH_ARM_ACTION_HEADER = "dynamic-valve-search-arm-v1"
SAFETY_CONFIGURE_PATH = "/api/safety/configure"
SAFETY_CONFIGURE_ACTION_HEADER = "lite6-safety-boundary-configure-v1"
PETRI_START_PATH = "/api/petri/start"
PETRI_START_ACTION_HEADER = "commissioned-petri-start-v1"
REGISTRATION_VERIFY_PATH = "/api/registration/verify"
REGISTRATION_VERIFY_ACTION_HEADER = "device-registration-verify-v1"
RECOVERY_START_PATH = "/api/recovery/home"
RECOVERY_START_ACTION_HEADER = "reviewed-recovery-home-v1"
MANUAL_REVIEW_PATH = "/api/petri/review"
MANUAL_REVIEW_ACTION_HEADER = "petri-review-select-v1"
MANUAL_REVIEW_PREVIEW_PATH = "/api/petri/review.jpg"
WORKSPACE_TEACH_START_PATH = "/api/workspace-teach/start"
WORKSPACE_TEACH_CAPTURE_PATH = "/api/workspace-teach/capture"
WORKSPACE_TEACH_CANCEL_PATH = "/api/workspace-teach/cancel"
WORKSPACE_TEACH_FINALIZE_PATH = "/api/workspace-teach/finalize"
WORKSPACE_TEACH_START_ACTION_HEADER = "workspace-teach-start-v1"
WORKSPACE_TEACH_CAPTURE_ACTION_HEADER = "workspace-teach-capture-v1"
WORKSPACE_TEACH_CANCEL_ACTION_HEADER = "workspace-teach-cancel-v1"
WORKSPACE_TEACH_FINALIZE_ACTION_HEADER = "workspace-teach-finalize-v1"
TENANT_LOGIN_PATH = "/api/tenant-auth/login"
TENANT_LOGOUT_PATH = "/api/tenant-auth/logout"
TENANT_LOGIN_ACTION_HEADER = "tenant-mobile-login-v1"
TENANT_LOGOUT_ACTION_HEADER = "tenant-mobile-logout-v1"
HAND_TRAINING_START_PATH = "/api/hand-training/start"
HAND_TRAINING_TOOL_PATH = "/api/hand-training/tool"
HAND_TRAINING_END_PATH = "/api/hand-training/end"
HAND_TRAINING_START_ACTION_HEADER = "hand-training-start-v1"
HAND_TRAINING_TOOL_ACTION_HEADER = "hand-training-tool-v1"
HAND_TRAINING_END_ACTION_HEADER = "hand-training-end-v1"
_LOOPBACK_HOSTS = frozenset(("127.0.0.1", "::1", "localhost"))
_LOOPBACK_CLIENTS = frozenset(("127.0.0.1", "::1", "::ffff:127.0.0.1"))
_GENERAL_REQUEST_LIMIT = 16
_STOP_REQUEST_LIMIT = 2
_REQUEST_LINE_PEEK_BYTES = 256
_REQUEST_LINE_PEEK_TIMEOUT_SECONDS = 0.05
_GENERAL_REQUEST_LANE = "general"
_STOP_REQUEST_LANE = "stop"


def _reject_duplicate_object_pairs(pairs):
    value = {}
    for key, item in pairs:
        if key in value:
            raise ValueError("duplicate JSON key")
        value[key] = item
    return value


def _accepts_exact_jpeg(value: Optional[str]) -> bool:
    """Require the LVGL client to ask for exactly the JPEG media type."""

    return bool(value) and value.strip().lower() == "image/jpeg"


def _is_bounded_baseline_jpeg(payload: object) -> bool:
    """Validate the bounded structural markers needed by the LVGL decoder."""

    if not isinstance(payload, bytes):
        return False
    if len(payload) > SNAPSHOT_MAX_BYTES or len(payload) < 8:
        return False
    if not payload.startswith(b"\xff\xd8") or not payload.endswith(b"\xff\xd9"):
        return False

    # Walk marker segments up to the scan data.  A progressive SOF marker is
    # rejected explicitly, while a baseline SOF0 marker is required.  This
    # avoids treating marker-like bytes inside compressed scan data as image
    # structure.
    offset = 2
    saw_baseline = False
    while offset < len(payload) - 2:
        if payload[offset] != 0xFF:
            return False
        while offset < len(payload) and payload[offset] == 0xFF:
            offset += 1
        if offset >= len(payload):
            return False
        marker = payload[offset]
        offset += 1
        if marker == 0x00:
            return False
        if marker == 0xD9:
            return saw_baseline and offset == len(payload) - 1
        if marker == 0xC2:
            return False
        if marker == 0xDA:
            return saw_baseline
        if marker in range(0xD0, 0xD8) or marker in (0x01, 0xD8):
            continue
        if offset + 2 > len(payload) - 2:
            return False
        segment_length = int.from_bytes(payload[offset : offset + 2], "big")
        if segment_length < 2:
            return False
        segment_end = offset + segment_length
        if segment_end > len(payload) - 2:
            return False
        if marker == 0xC0:
            # SOF0 contains precision, dimensions, component count, and at
            # least one three-byte component descriptor.
            if segment_length < 11:
                return False
            saw_baseline = True
        offset = segment_end
    return saw_baseline


class _BoundedHTTPServer(ThreadingHTTPServer):
    daemon_threads = True
    allow_reuse_address = True

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
        # General health, preview, and action callbacks may block on a device or
        # integration.  Keep their capacity separate from the universal STOP
        # path so saturation cannot discard STOP before the HTTP route is even
        # parsed.  Both lanes are bounded; at most 18 request threads exist.
        self._general_request_slots = threading.BoundedSemaphore(_GENERAL_REQUEST_LIMIT)
        self._stop_request_slots = threading.BoundedSemaphore(_STOP_REQUEST_LIMIT)
        self._request_lanes: dict[int, str] = {}
        self._request_lanes_lock = threading.Lock()

    def get_request(self):
        request, address = super().get_request()
        request.settimeout(2.0)
        return request, address

    def process_request(self, request: Any, client_address: Any) -> None:
        request_kind = self._peek_request_kind(request)
        lane = self._acquire_request_lane(request_kind)
        if lane is None:
            self.shutdown_request(request)
            return
        with self._request_lanes_lock:
            self._request_lanes[id(request)] = lane
        try:
            super().process_request(request, client_address)
        except Exception:
            self._release_request_lane(request)
            raise

    def process_request_thread(self, request: Any, client_address: Any) -> None:
        try:
            super().process_request_thread(request, client_address)
        finally:
            self._release_request_lane(request)

    def request_lane(self, request: Any) -> str:
        """Return the bounded admission lane for one active request."""

        with self._request_lanes_lock:
            return self._request_lanes.get(id(request), "")

    def _peek_request_kind(self, request: Any) -> str:
        """Classify a complete request line without consuming socket bytes.

        A request line can arrive fragmented.  Unknown or incomplete requests
        always use ordinary capacity.  The reserved lane is never loaned: only
        a completely parsed, exact universal STOP request line may consume it.
        """

        recv = getattr(request, "recv", None)
        settimeout = getattr(request, "settimeout", None)
        gettimeout = getattr(request, "gettimeout", None)
        if not callable(recv) or not callable(settimeout) or not callable(gettimeout):
            return "unknown"
        previous_timeout = gettimeout()
        try:
            settimeout(_REQUEST_LINE_PEEK_TIMEOUT_SECONDS)
            prefix = recv(_REQUEST_LINE_PEEK_BYTES, socket.MSG_PEEK)
        except (BlockingIOError, InterruptedError, TimeoutError, OSError):
            return "unknown"
        finally:
            try:
                settimeout(previous_timeout)
            except OSError:
                pass
        if not isinstance(prefix, bytes):
            return "unknown"
        line_end = prefix.find(b"\r\n")
        if line_end < 0:
            return "unknown"
        request_line = prefix[:line_end]
        parts = request_line.split(b" ")
        if (
            len(parts) == 3
            and parts[0] == b"POST"
            and parts[1] == SOFTWARE_STOP_PATH.encode("ascii")
            and parts[2] in (b"HTTP/1.0", b"HTTP/1.1")
        ):
            return "stop"
        return "general"

    def _acquire_request_lane(self, request_kind: str) -> Optional[str]:
        if request_kind == "stop":
            if self._stop_request_slots.acquire(blocking=False):
                return _STOP_REQUEST_LANE
            if self._general_request_slots.acquire(blocking=False):
                return _GENERAL_REQUEST_LANE
            return None
        if self._general_request_slots.acquire(blocking=False):
            return _GENERAL_REQUEST_LANE
        return None

    def _release_request_lane(self, request: Any) -> None:
        with self._request_lanes_lock:
            lane = self._request_lanes.pop(id(request), "")
        if lane == _STOP_REQUEST_LANE:
            self._stop_request_slots.release()
        elif lane == _GENERAL_REQUEST_LANE:
            self._general_request_slots.release()


class MjpegServer:
    """Serve latest-frame MJPEG streams with bounded concurrent clients."""

    def __init__(
        self,
        streams: Mapping[CameraRole, CameraStream],
        *,
        host: str = "127.0.0.1",
        port: int = 8765,
        token: str = "",
        max_clients: int = 4,
        encoder: Optional[Callable[[object], bytes]] = None,
        status_provider: Optional[Callable[[], Mapping[str, Any]]] = None,
        update_requester: Optional[Callable[[], bool]] = None,
        capture_starter: Optional[Callable[[Optional[str]], Mapping[str, Any]]] = None,
        capture_stopper: Optional[Callable[[], str]] = None,
        software_stopper: Optional[Callable[[], str]] = None,
        motion_search_requester: Optional[
            Callable[[Optional[str]], Mapping[str, Any]]
        ] = None,
        safety_configure_requester: Optional[
            Callable[[Optional[str]], Mapping[str, Any]]
        ] = None,
        petri_starter: Optional[Callable[[Optional[str]], Mapping[str, Any]]] = None,
        registration_verifier: Optional[Callable[[], Mapping[str, Any]]] = None,
        recovery_starter: Optional[Callable[[Optional[str]], Mapping[str, Any]]] = None,
        manual_reviewer: Optional[Callable[[Optional[str]], Mapping[str, Any]]] = None,
        manual_review_previewer: Optional[Callable[[], bytes]] = None,
        workspace_teach_starter: Optional[Callable[[], Mapping[str, Any]]] = None,
        workspace_teach_capturer: Optional[Callable[[], Mapping[str, Any]]] = None,
        workspace_teach_canceller: Optional[Callable[[], Mapping[str, Any]]] = None,
        workspace_teach_finalizer: Optional[
            Callable[[Optional[str]], Mapping[str, Any]]
        ] = None,
        tenant_auth_login: Optional[
            Callable[[str, str, str], Mapping[str, Any]]
        ] = None,
        tenant_auth_logout: Optional[Callable[[], Mapping[str, Any]]] = None,
        hand_training_starter: Optional[Callable[[], Mapping[str, Any]]] = None,
        hand_training_tool: Optional[Callable[[bool], Mapping[str, Any]]] = None,
        hand_training_ender: Optional[Callable[[], Mapping[str, Any]]] = None,
        frame_timeout_seconds: float = 5.0,
        server_factory: Callable[[tuple, type], Any] = _BoundedHTTPServer,
    ) -> None:
        if max_clients < 1:
            raise ValueError("max_clients must be positive")
        if port < 0 or port > 65535:
            raise ValueError("port must be between 0 and 65535")
        if host not in _LOOPBACK_HOSTS:
            raise ValueError(
                "loopback MJPEG is required; use an authenticated TLS proxy "
                "or SSH tunnel for remote viewing"
            )
        missing = [
            role.value
            for role in (CameraRole.TOOL, CameraRole.WORKSPACE)
            if role not in streams
        ]
        if missing:
            raise ValueError(
                "MJPEG server is missing camera roles: {}".format(", ".join(missing))
            )
        self.streams = dict(streams)
        self.host = host
        self.port = int(port)
        self.token = str(token)
        self.max_clients = int(max_clients)
        self.encoder = encoder or encode_jpeg
        self.status_provider = status_provider
        self.update_requester = update_requester
        self.capture_starter = capture_starter
        self.capture_stopper = capture_stopper
        self.software_stopper = software_stopper
        self.motion_search_requester = motion_search_requester
        self.safety_configure_requester = safety_configure_requester
        self.petri_starter = petri_starter
        self.registration_verifier = registration_verifier
        self.recovery_starter = recovery_starter
        self.manual_reviewer = manual_reviewer
        self.manual_review_previewer = manual_review_previewer
        self.workspace_teach_starter = workspace_teach_starter
        self.workspace_teach_capturer = workspace_teach_capturer
        self.workspace_teach_canceller = workspace_teach_canceller
        self.workspace_teach_finalizer = workspace_teach_finalizer
        self.tenant_auth_login = tenant_auth_login
        self.tenant_auth_logout = tenant_auth_logout
        self.hand_training_starter = hand_training_starter
        self.hand_training_tool = hand_training_tool
        self.hand_training_ender = hand_training_ender
        self._workspace_teach_finalize_lock = threading.Lock()
        self._workspace_teach_finalize_challenge = ""
        self._workspace_teach_finalize_proposal_sha256 = ""
        self._workspace_teach_finalize_expires_at = 0.0
        self.frame_timeout_seconds = float(frame_timeout_seconds)
        self.server_factory = server_factory
        self._server: Optional[_BoundedHTTPServer] = None
        self._thread: Optional[threading.Thread] = None
        self._slots = threading.BoundedSemaphore(self.max_clients)

    @property
    def address(self):
        server = self._server
        return server.server_address if server is not None else None

    def start(self) -> None:
        if self._thread is not None and self._thread.is_alive():
            return
        parent = self

        class Handler(BaseHTTPRequestHandler):
            server_version = "LabCobot-MJPEG/1"

            def do_GET(self) -> None:  # noqa: N802 - BaseHTTPRequestHandler API
                if self._uses_reserved_stop_lane():
                    self._send_json(HTTPStatus.SERVICE_UNAVAILABLE, {"status": "busy"})
                    return
                parsed = urlsplit(self.path)
                if parsed.path == "/healthz":
                    self._send_health()
                    return
                if parsed.path == MANUAL_REVIEW_PREVIEW_PATH:
                    self._send_manual_review_preview(parsed.query)
                    return
                snapshot_role = SNAPSHOT_PATHS.get(parsed.path)
                if snapshot_role is not None:
                    self._send_camera_snapshot(snapshot_role, parsed.query)
                    return
                role = STREAM_PATHS.get(parsed.path)
                if role is None:
                    self.send_error(HTTPStatus.NOT_FOUND)
                    return
                if not parent._authorized(
                    self.headers.get("Authorization"), parsed.query
                ):
                    self.send_response(HTTPStatus.UNAUTHORIZED)
                    self.send_header("WWW-Authenticate", "Bearer")
                    self.send_header("Content-Length", "0")
                    self.end_headers()
                    return
                if not parent._slots.acquire(blocking=False):
                    self.send_response(HTTPStatus.SERVICE_UNAVAILABLE)
                    self.send_header("Retry-After", "1")
                    self.send_header("Content-Length", "0")
                    self.end_headers()
                    return
                try:
                    self.send_response(HTTPStatus.OK)
                    self.send_header(
                        "Content-Type",
                        "multipart/x-mixed-replace; boundary=frame",
                    )
                    self.send_header("Cache-Control", "no-store")
                    self.send_header("X-Content-Type-Options", "nosniff")
                    self.end_headers()
                    chunks = parent.streams[role].mjpeg_chunks(
                        encoder=parent.encoder,
                        timeout_s=parent.frame_timeout_seconds,
                    )
                    for chunk in chunks:
                        self.wfile.write(chunk)
                        self.wfile.flush()
                except (BrokenPipeError, ConnectionResetError):
                    return
                except Exception:
                    LOGGER.warning("MJPEG stream %s stopped", role.value, exc_info=True)
                finally:
                    parent._slots.release()

            def _send_camera_snapshot(
                self,
                role: CameraRole,
                query: str,
            ) -> None:
                # This endpoint is a deliberately narrow handoff from the
                # local camera process to the fullscreen LVGL process.  It is
                # never a LAN/browser-facing alternate stream URL.
                client_host = str(self.client_address[0]).lower()
                if query or client_host not in _LOOPBACK_CLIENTS:
                    self._send_empty(HTTPStatus.NOT_FOUND)
                    return
                if self.headers.get("Origin") or self.headers.get("Sec-Fetch-Site"):
                    self._send_json(
                        HTTPStatus.FORBIDDEN,
                        {"status": "forbidden"},
                    )
                    return
                if self.headers.get("X-LabCobot-Client") != SNAPSHOT_CLIENT_HEADER:
                    self._send_json(
                        HTTPStatus.FORBIDDEN,
                        {"status": "forbidden"},
                    )
                    return
                if not _accepts_exact_jpeg(self.headers.get("Accept")):
                    self._send_json(
                        HTTPStatus.NOT_ACCEPTABLE,
                        {"status": "not_acceptable"},
                    )
                    return
                if not parent._slots.acquire(blocking=False):
                    self._send_json(
                        HTTPStatus.SERVICE_UNAVAILABLE,
                        {"status": "busy"},
                    )
                    return

                try:
                    frame = parent.streams[role].latest(max_age_s=0.5)
                    if frame is None:
                        self._send_json(
                            HTTPStatus.SERVICE_UNAVAILABLE,
                            {"status": "unavailable"},
                        )
                        return
                    jpeg = parent.encoder(frame.image)
                    if not _is_bounded_baseline_jpeg(jpeg):
                        LOGGER.warning(
                            "Camera snapshot encoder returned invalid JPEG for %s",
                            role.value,
                        )
                        self._send_json(
                            HTTPStatus.SERVICE_UNAVAILABLE,
                            {"status": "unavailable"},
                        )
                        return

                    self.send_response(HTTPStatus.OK)
                    self.send_header("Content-Type", "image/jpeg")
                    self.send_header("Cache-Control", "no-store")
                    self.send_header("X-Content-Type-Options", "nosniff")
                    self.send_header("Content-Length", str(len(jpeg)))
                    self.end_headers()
                    self.wfile.write(jpeg)
                except (BrokenPipeError, ConnectionResetError):
                    return
                except Exception:
                    LOGGER.warning(
                        "Camera snapshot unavailable for %s",
                        role.value,
                        exc_info=True,
                    )
                    self._send_json(
                        HTTPStatus.SERVICE_UNAVAILABLE,
                        {"status": "unavailable"},
                    )
                finally:
                    parent._slots.release()

            def _send_manual_review_preview(self, query: str) -> None:
                """Serve only the image bound to the active review challenge."""

                client_host = str(self.client_address[0]).lower()
                host_headers = self.headers.get_all("Host") or []
                if (
                    query
                    or client_host not in _LOOPBACK_CLIENTS
                    or len(host_headers) != 1
                    or not parent._is_loopback_host_header(host_headers[0])
                ):
                    self._send_empty(HTTPStatus.NOT_FOUND)
                    return
                if self.headers.get("Origin") or self.headers.get("Sec-Fetch-Site"):
                    self._send_json(HTTPStatus.FORBIDDEN, {"status": "forbidden"})
                    return
                if self.headers.get("X-LabCobot-Client") != SNAPSHOT_CLIENT_HEADER:
                    self._send_json(HTTPStatus.FORBIDDEN, {"status": "forbidden"})
                    return
                if not _accepts_exact_jpeg(self.headers.get("Accept")):
                    self._send_json(
                        HTTPStatus.NOT_ACCEPTABLE,
                        {"status": "not_acceptable"},
                    )
                    return
                previewer = parent.manual_review_previewer
                if previewer is None:
                    self._send_json(
                        HTTPStatus.SERVICE_UNAVAILABLE,
                        {"status": "unavailable"},
                    )
                    return
                if not parent._slots.acquire(blocking=False):
                    self._send_json(HTTPStatus.SERVICE_UNAVAILABLE, {"status": "busy"})
                    return
                try:
                    jpeg = previewer()
                    if not _is_bounded_baseline_jpeg(jpeg):
                        raise ValueError("manual-review preview is not baseline JPEG")
                    self.send_response(HTTPStatus.OK)
                    self.send_header("Content-Type", "image/jpeg")
                    self.send_header("Cache-Control", "no-store")
                    self.send_header("X-Content-Type-Options", "nosniff")
                    self.send_header("Content-Length", str(len(jpeg)))
                    self.end_headers()
                    self.wfile.write(jpeg)
                except (BrokenPipeError, ConnectionResetError):
                    return
                except Exception:
                    LOGGER.warning("Manual-review preview unavailable", exc_info=True)
                    self._send_json(
                        HTTPStatus.SERVICE_UNAVAILABLE,
                        {"status": "unavailable"},
                    )
                finally:
                    parent._slots.release()

            def do_POST(self) -> None:  # noqa: N802 - BaseHTTPRequestHandler API
                parsed = urlsplit(self.path)
                if self._uses_reserved_stop_lane() and (
                    parsed.query or parsed.path != SOFTWARE_STOP_PATH
                ):
                    self._send_json(HTTPStatus.SERVICE_UNAVAILABLE, {"status": "busy"})
                    return
                if parsed.query:
                    self.send_error(HTTPStatus.NOT_FOUND)
                    return
                if parsed.path == OTA_CHECK_PATH:
                    self._request_update_check()
                    return
                if parsed.path == CAPTURE_START_PATH:
                    self._request_capture_action(start=True)
                    return
                if parsed.path == CAPTURE_STOP_PATH:
                    self._request_capture_action(start=False)
                    return
                if parsed.path == SOFTWARE_STOP_PATH:
                    self._request_software_stop()
                    return
                if parsed.path == MOTION_SEARCH_ARM_PATH:
                    self._request_motion_search()
                    return
                if parsed.path == SAFETY_CONFIGURE_PATH:
                    self._request_safety_configuration()
                    return
                if parsed.path == PETRI_START_PATH:
                    self._request_operator_confirmation(
                        parent.petri_starter,
                        PETRI_START_ACTION_HEADER,
                    )
                    return
                if parsed.path == RECOVERY_START_PATH:
                    self._request_operator_confirmation(
                        parent.recovery_starter,
                        RECOVERY_START_ACTION_HEADER,
                    )
                    return
                if parsed.path == REGISTRATION_VERIFY_PATH:
                    self._request_registration_verification()
                    return
                if parsed.path == MANUAL_REVIEW_PATH:
                    self._request_manual_review()
                    return
                if parsed.path == WORKSPACE_TEACH_START_PATH:
                    self._request_workspace_teach_action("start")
                    return
                if parsed.path == WORKSPACE_TEACH_CAPTURE_PATH:
                    self._request_workspace_teach_action("capture")
                    return
                if parsed.path == WORKSPACE_TEACH_CANCEL_PATH:
                    self._request_workspace_teach_action("cancel")
                    return
                if parsed.path == WORKSPACE_TEACH_FINALIZE_PATH:
                    self._request_workspace_teach_finalize()
                    return
                if parsed.path == TENANT_LOGIN_PATH:
                    self._request_tenant_login()
                    return
                if parsed.path == TENANT_LOGOUT_PATH:
                    self._request_tenant_logout()
                    return
                if parsed.path == HAND_TRAINING_START_PATH:
                    self._request_hand_training_action("start")
                    return
                if parsed.path == HAND_TRAINING_TOOL_PATH:
                    self._request_hand_training_action("tool")
                    return
                if parsed.path == HAND_TRAINING_END_PATH:
                    self._request_hand_training_action("end")
                    return
                self.send_error(HTTPStatus.NOT_FOUND)

            def _request_hand_training_action(self, action: str) -> None:
                """Expose only the three bounded recording actions on loopback."""

                if action == "tool":
                    payload = self._read_local_action(
                        HAND_TRAINING_TOOL_ACTION_HEADER,
                        allowed_keys=frozenset(("active",)),
                    )
                    callback = parent.hand_training_tool
                else:
                    payload = self._read_local_action(
                        HAND_TRAINING_START_ACTION_HEADER
                        if action == "start"
                        else HAND_TRAINING_END_ACTION_HEADER,
                        allowed_keys=frozenset(),
                    )
                    callback = (
                        parent.hand_training_starter
                        if action == "start"
                        else parent.hand_training_ender
                    )
                if payload is None:
                    return
                if action == "tool" and type(payload.get("active")) is not bool:
                    self._send_json(
                        HTTPStatus.BAD_REQUEST, {"status": "invalid_request"}
                    )
                    return
                if callback is None:
                    self._send_json(
                        HTTPStatus.SERVICE_UNAVAILABLE, {"status": "unavailable"}
                    )
                    return
                try:
                    raw = (
                        callback(payload["active"])
                        if action == "tool"
                        else callback()
                    )
                except Exception:
                    LOGGER.warning("Hand-training action unavailable", exc_info=True)
                    raw = {"status": "error"}
                value = dict(raw) if isinstance(raw, Mapping) else {"status": "error"}
                outcome = str(value.get("status") or "error")
                status_codes = {
                    "accepted": HTTPStatus.ACCEPTED,
                    "idle": HTTPStatus.OK,
                    "busy": HTTPStatus.CONFLICT,
                    "unsafe": HTTPStatus.CONFLICT,
                    "login_required": HTTPStatus.UNAUTHORIZED,
                    "invalid_request": HTTPStatus.BAD_REQUEST,
                    "unavailable": HTTPStatus.SERVICE_UNAVAILABLE,
                    "error": HTTPStatus.SERVICE_UNAVAILABLE,
                }
                response_status = status_codes.get(
                    outcome, HTTPStatus.SERVICE_UNAVAILABLE
                )
                response = {"status": outcome if outcome in status_codes else "error"}
                if action == "tool" and outcome == "accepted":
                    response["tool_on"] = value.get("tool_on") is True
                self._send_json(response_status, response)

            def _request_tenant_login(self) -> None:
                """Accept credentials only on the private loopback action edge."""

                payload = self._read_local_action(
                    TENANT_LOGIN_ACTION_HEADER,
                    allowed_keys=frozenset(("tenant_url", "username", "password")),
                    maximum_bytes=1024,
                )
                if payload is None:
                    return
                tenant_url = payload.get("tenant_url")
                username = payload.get("username")
                password = payload.get("password")
                if (
                    not isinstance(tenant_url, str)
                    or not 1 <= len(tenant_url) <= 512
                    or not isinstance(username, str)
                    or not 1 <= len(username) <= 254
                    or not isinstance(password, str)
                    or not 1 <= len(password) <= 128
                ):
                    self._send_json(
                        HTTPStatus.BAD_REQUEST,
                        {"status": "invalid_request"},
                    )
                    return
                callback = parent.tenant_auth_login
                if callback is None:
                    self._send_json(
                        HTTPStatus.SERVICE_UNAVAILABLE,
                        {"status": "unavailable"},
                    )
                    return
                try:
                    raw = callback(tenant_url, username, password)
                except Exception:
                    LOGGER.warning("Tenant mobile login unavailable", exc_info=True)
                    raw = {"status": "tenant_unavailable"}
                value = dict(raw) if isinstance(raw, Mapping) else {}
                outcome = value.get("status")
                if outcome == "authenticated":
                    expected = {
                        "status",
                        "tenant_url",
                        "tenant",
                        "username",
                        "display_name",
                        "access_expires_in",
                        "session_persisted",
                    }
                    if "device_registration" in value:
                        expected.add("device_registration")
                    registration = value.get("device_registration")
                    expires = value.get("access_expires_in")
                    if (
                        set(value) != expected
                        or not isinstance(value.get("tenant_url"), str)
                        or not isinstance(value.get("tenant"), str)
                        or not isinstance(value.get("username"), str)
                        or not isinstance(value.get("display_name"), str)
                        or isinstance(expires, bool)
                        or not isinstance(expires, int)
                        or not 0 <= expires <= 31_536_000
                        or not isinstance(value.get("session_persisted"), bool)
                        or (
                            registration is not None
                            and (
                                not isinstance(registration, Mapping)
                                or set(registration)
                                != {
                                    "uid_device",
                                    "token",
                                    "created",
                                    "email_sent",
                                    "email_recipient",
                                }
                                or not isinstance(registration.get("uid_device"), str)
                                or not isinstance(registration.get("token"), str)
                                or not 1 <= len(registration.get("token")) <= 128
                                or type(registration.get("created")) is not bool
                                or type(registration.get("email_sent")) is not bool
                                or not isinstance(registration.get("email_recipient"), str)
                            )
                        )
                    ):
                        self._send_json(
                            HTTPStatus.SERVICE_UNAVAILABLE,
                            {"status": "invalid_response"},
                        )
                        return
                    self._send_json(HTTPStatus.OK, value)
                    return
                statuses = {
                    "invalid_tenant_url": HTTPStatus.BAD_REQUEST,
                    "insecure_tenant_url": HTTPStatus.BAD_REQUEST,
                    "missing_credentials": HTTPStatus.BAD_REQUEST,
                    "invalid_credentials": HTTPStatus.UNAUTHORIZED,
                    "tenant_change_requires_logout": HTTPStatus.CONFLICT,
                    "login_rejected": HTTPStatus.FORBIDDEN,
                    "invalid_response": HTTPStatus.SERVICE_UNAVAILABLE,
                    "tenant_unavailable": HTTPStatus.SERVICE_UNAVAILABLE,
                    "unavailable": HTTPStatus.SERVICE_UNAVAILABLE,
                }
                status = statuses.get(str(outcome or ""))
                if status is None:
                    status = HTTPStatus.SERVICE_UNAVAILABLE
                    outcome = "tenant_unavailable"
                self._send_json(status, {"status": outcome})

            def _request_tenant_logout(self) -> None:
                if not self._read_empty_local_action(TENANT_LOGOUT_ACTION_HEADER):
                    return
                callback = parent.tenant_auth_logout
                if callback is None:
                    self._send_json(
                        HTTPStatus.SERVICE_UNAVAILABLE,
                        {"status": "unavailable"},
                    )
                    return
                try:
                    raw = callback()
                except Exception:
                    LOGGER.warning("Tenant mobile logout unavailable", exc_info=True)
                    raw = {"status": "tenant_unavailable"}
                value = dict(raw) if isinstance(raw, Mapping) else {}
                if value == {"status": "logged_out"}:
                    self._send_json(HTTPStatus.OK, value)
                    return
                self._send_json(
                    HTTPStatus.SERVICE_UNAVAILABLE,
                    {"status": "tenant_unavailable"},
                )

            def _uses_reserved_stop_lane(self) -> bool:
                lane_reader = getattr(self.server, "request_lane", None)
                return bool(
                    callable(lane_reader)
                    and lane_reader(self.request) == _STOP_REQUEST_LANE
                )

            def _request_update_check(self) -> None:
                if not self._read_empty_local_action(OTA_ACTION_HEADER):
                    return
                requester = parent.update_requester
                if requester is None:
                    self._send_json(
                        HTTPStatus.SERVICE_UNAVAILABLE,
                        {"status": "unavailable"},
                    )
                    return
                try:
                    accepted = bool(requester())
                except Exception:
                    LOGGER.exception("Local secure OTA request failed")
                    self._send_json(
                        HTTPStatus.INTERNAL_SERVER_ERROR,
                        {"status": "error"},
                    )
                    return
                self._send_json(
                    HTTPStatus.ACCEPTED if accepted else HTTPStatus.CONFLICT,
                    {"status": "accepted" if accepted else "busy"},
                )

            def _request_capture_action(self, *, start: bool) -> None:
                expected_header = (
                    CAPTURE_START_ACTION_HEADER if start else CAPTURE_STOP_ACTION_HEADER
                )
                if start:
                    payload = self._read_local_action_variants(
                        expected_header,
                        allowed_key_sets=frozenset(
                            (frozenset(), frozenset(("challenge",)))
                        ),
                    )
                else:
                    payload = self._read_local_action(
                        expected_header,
                        allowed_keys=frozenset(),
                    )
                if payload is None:
                    return
                challenge = payload.get("challenge")
                if challenge is not None and (
                    not isinstance(challenge, str)
                    or len(challenge) != 32
                    or any(value not in "0123456789abcdef" for value in challenge)
                ):
                    self._send_json(
                        HTTPStatus.BAD_REQUEST,
                        {"status": "invalid_request"},
                    )
                    return
                callback = parent.capture_starter if start else parent.capture_stopper
                if callback is None:
                    self._send_json(
                        HTTPStatus.SERVICE_UNAVAILABLE,
                        {"status": "unavailable"},
                    )
                    return
                try:
                    raw = callback(challenge) if start else callback()
                except Exception:
                    LOGGER.exception("Local capture-run action failed")
                    self._send_json(
                        HTTPStatus.INTERNAL_SERVER_ERROR,
                        {"status": "error"},
                    )
                    return
                value = dict(raw) if start and isinstance(raw, Mapping) else {}
                outcome = str(value.get("status") if start else raw or "error")
                if outcome in {"accepted", "confirm"}:
                    status = HTTPStatus.ACCEPTED
                elif outcome in {"busy", "unsafe", "idle"}:
                    status = HTTPStatus.CONFLICT
                elif outcome == "unavailable":
                    status = HTTPStatus.SERVICE_UNAVAILABLE
                else:
                    outcome = "error"
                    status = HTTPStatus.INTERNAL_SERVER_ERROR
                response = {"status": outcome}
                if outcome == "confirm":
                    token = value.get("challenge")
                    expires = value.get("expires_in_seconds")
                    if (
                        not isinstance(token, str)
                        or len(token) != 32
                        or any(
                            character not in "0123456789abcdef" for character in token
                        )
                        or isinstance(expires, bool)
                        or not isinstance(expires, int)
                        or not 1 <= expires <= 5
                    ):
                        self._send_json(
                            HTTPStatus.INTERNAL_SERVER_ERROR,
                            {"status": "error"},
                        )
                        return
                    response.update({"challenge": token, "expires_in_seconds": expires})
                self._send_json(status, response)

            def _request_software_stop(self) -> None:
                payload = self._read_local_action(
                    SOFTWARE_STOP_ACTION_HEADER,
                    allowed_keys=frozenset(),
                )
                if payload is None:
                    return
                callback = parent.software_stopper
                if callback is None:
                    self._send_json(
                        HTTPStatus.SERVICE_UNAVAILABLE,
                        {"status": "unavailable"},
                    )
                    return
                try:
                    outcome = str(callback() or "error")
                except Exception:
                    LOGGER.exception("Universal local software STOP failed")
                    outcome = "error"
                statuses = {
                    "accepted": HTTPStatus.ACCEPTED,
                    "idle": HTTPStatus.OK,
                    "unavailable": HTTPStatus.SERVICE_UNAVAILABLE,
                    "error": HTTPStatus.INTERNAL_SERVER_ERROR,
                }
                status = statuses.get(outcome)
                if status is None:
                    outcome = "error"
                    status = HTTPStatus.INTERNAL_SERVER_ERROR
                self._send_json(status, {"status": outcome})

            def _request_registration_verification(self) -> None:
                payload = self._read_local_action(
                    REGISTRATION_VERIFY_ACTION_HEADER,
                    allowed_keys=frozenset(),
                )
                if payload is None:
                    return
                requester = parent.registration_verifier
                if requester is None:
                    self._send_json(
                        HTTPStatus.SERVICE_UNAVAILABLE,
                        {"status": "unavailable"},
                    )
                    return
                try:
                    raw = requester()
                except Exception:
                    LOGGER.exception("Local registration verification failed")
                    raw = {"status": "error"}
                value = dict(raw) if isinstance(raw, Mapping) else {}
                outcome = str(value.get("status") or "error")
                statuses = {
                    "verified": HTTPStatus.OK,
                    "incomplete": HTTPStatus.CONFLICT,
                    "busy": HTTPStatus.CONFLICT,
                    "unavailable": HTTPStatus.SERVICE_UNAVAILABLE,
                    "error": HTTPStatus.INTERNAL_SERVER_ERROR,
                }
                status = statuses.get(outcome, HTTPStatus.INTERNAL_SERVER_ERROR)
                if outcome not in statuses:
                    outcome = "error"
                self._send_json(status, {"status": outcome})

            def _request_manual_review(self) -> None:
                """Select one bounded pending observation through two taps."""

                payload = self._read_local_action_variants(
                    MANUAL_REVIEW_ACTION_HEADER,
                    allowed_key_sets=frozenset(
                        (frozenset(), frozenset(("challenge",)))
                    ),
                )
                if payload is None:
                    return
                challenge = payload.get("challenge")
                if challenge is not None and (
                    not isinstance(challenge, str)
                    or len(challenge) != 32
                    or any(
                        character not in "0123456789abcdef" for character in challenge
                    )
                ):
                    self._send_json(
                        HTTPStatus.BAD_REQUEST,
                        {"status": "invalid_request"},
                    )
                    return
                requester = parent.manual_reviewer
                if requester is None:
                    self._send_json(
                        HTTPStatus.SERVICE_UNAVAILABLE,
                        {"status": "unavailable"},
                    )
                    return
                try:
                    raw = requester(challenge)
                except Exception:
                    LOGGER.exception("Local manual-review request failed")
                    raw = {"status": "error"}
                value = dict(raw) if isinstance(raw, Mapping) else {}
                outcome = str(value.get("status") or "error")
                statuses = {
                    "confirm": HTTPStatus.ACCEPTED,
                    "accepted": HTTPStatus.ACCEPTED,
                    "confirmation_required": HTTPStatus.CONFLICT,
                    "busy": HTTPStatus.CONFLICT,
                    "unavailable": HTTPStatus.SERVICE_UNAVAILABLE,
                    "error": HTTPStatus.INTERNAL_SERVER_ERROR,
                }
                status = statuses.get(outcome)
                if status is None:
                    self._send_json(
                        HTTPStatus.INTERNAL_SERVER_ERROR,
                        {"status": "error"},
                    )
                    return
                response = {"status": outcome}
                if outcome == "confirm":
                    token = value.get("challenge")
                    expires = value.get("expires_in_seconds")
                    candidate = value.get("candidate")
                    if not self._valid_review_confirmation(
                        token,
                        expires,
                        candidate,
                    ):
                        self._send_json(
                            HTTPStatus.INTERNAL_SERVER_ERROR,
                            {"status": "error"},
                        )
                        return
                    response.update(
                        {
                            "challenge": token,
                            "expires_in_seconds": expires,
                            "candidate": dict(candidate),
                        }
                    )
                elif outcome == "accepted":
                    observation_id = value.get("observation_id")
                    if not self._safe_registry_id(observation_id):
                        self._send_json(
                            HTTPStatus.INTERNAL_SERVER_ERROR,
                            {"status": "error"},
                        )
                        return
                    response["observation_id"] = observation_id
                elif set(value) != {"status"}:
                    self._send_json(
                        HTTPStatus.INTERNAL_SERVER_ERROR,
                        {"status": "error"},
                    )
                    return
                self._send_json(status, response)

            @staticmethod
            def _safe_registry_id(value: Any) -> bool:
                return bool(
                    isinstance(value, str)
                    and 3 <= len(value) <= 64
                    and value[0] in "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
                    and all(
                        character in "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._-"
                        for character in value
                    )
                )

            @classmethod
            def _valid_review_confirmation(
                cls,
                token: Any,
                expires: Any,
                candidate: Any,
            ) -> bool:
                if (
                    not isinstance(token, str)
                    or len(token) != 32
                    or any(character not in "0123456789abcdef" for character in token)
                    or isinstance(expires, bool)
                    or not isinstance(expires, int)
                    or not 1 <= expires <= 30
                    or not isinstance(candidate, Mapping)
                    or set(candidate)
                    != {
                        "observation_id",
                        "plate_id",
                        "cluster_id",
                        "captured_at",
                        "confidence",
                        "area_px2",
                    }
                ):
                    return False
                if not all(
                    cls._safe_registry_id(candidate.get(name))
                    for name in ("observation_id", "plate_id", "cluster_id")
                ):
                    return False
                captured_at = candidate.get("captured_at")
                if (
                    not isinstance(captured_at, str)
                    or not 1 <= len(captured_at) <= 64
                    or any(ord(character) < 32 for character in captured_at)
                ):
                    return False
                for name in ("confidence", "area_px2"):
                    number = candidate.get(name)
                    if (
                        isinstance(number, bool)
                        or not isinstance(number, (int, float))
                        or not math.isfinite(float(number))
                        or float(number) < 0.0
                    ):
                        return False
                return float(candidate["confidence"]) <= 1.0

            def _request_operator_confirmation(
                self,
                requester: Optional[Callable[[Optional[str]], Mapping[str, Any]]],
                action_header: str,
            ) -> None:
                payload = self._read_local_action_variants(
                    action_header,
                    allowed_key_sets=frozenset(
                        (frozenset(), frozenset(("challenge",)))
                    ),
                )
                if payload is None:
                    return
                challenge = payload.get("challenge")
                if challenge is not None and (
                    not isinstance(challenge, str)
                    or len(challenge) != 32
                    or any(value not in "0123456789abcdef" for value in challenge)
                ):
                    self._send_json(
                        HTTPStatus.BAD_REQUEST,
                        {"status": "invalid_request"},
                    )
                    return
                if requester is None:
                    self._send_json(
                        HTTPStatus.SERVICE_UNAVAILABLE,
                        {"status": "unavailable"},
                    )
                    return
                try:
                    raw = requester(challenge)
                except Exception:
                    LOGGER.exception("Local commissioned operation request failed")
                    raw = {"status": "error"}
                value = dict(raw) if isinstance(raw, Mapping) else {}
                outcome = str(value.get("status") or "error")
                statuses = {
                    "confirm": HTTPStatus.ACCEPTED,
                    "accepted": HTTPStatus.ACCEPTED,
                    "confirmation_required": HTTPStatus.CONFLICT,
                    "busy": HTTPStatus.CONFLICT,
                    "unsafe": HTTPStatus.CONFLICT,
                    "unavailable": HTTPStatus.SERVICE_UNAVAILABLE,
                    "error": HTTPStatus.INTERNAL_SERVER_ERROR,
                }
                status = statuses.get(outcome)
                if status is None:
                    outcome = "error"
                    status = HTTPStatus.INTERNAL_SERVER_ERROR
                response = {"status": outcome}
                if outcome == "confirm":
                    token = value.get("challenge")
                    expires = value.get("expires_in_seconds")
                    if (
                        not isinstance(token, str)
                        or len(token) != 32
                        or any(value not in "0123456789abcdef" for value in token)
                        or isinstance(expires, bool)
                        or not isinstance(expires, int)
                        or not 1 <= expires <= 5
                    ):
                        self._send_json(
                            HTTPStatus.INTERNAL_SERVER_ERROR,
                            {"status": "error"},
                        )
                        return
                    response.update({"challenge": token, "expires_in_seconds": expires})
                self._send_json(status, response)

            def _request_motion_search(self) -> None:
                payload = self._read_local_action_variants(
                    MOTION_SEARCH_ARM_ACTION_HEADER,
                    allowed_key_sets=frozenset(
                        (frozenset(), frozenset(("challenge",)))
                    ),
                )
                if payload is None:
                    return
                challenge = payload.get("challenge")
                if challenge is not None and (
                    not isinstance(challenge, str)
                    or len(challenge) != 32
                    or any(value not in "0123456789abcdef" for value in challenge)
                ):
                    self._send_json(
                        HTTPStatus.BAD_REQUEST,
                        {"status": "invalid_request"},
                    )
                    return
                requester = parent.motion_search_requester
                if requester is None:
                    self._send_json(
                        HTTPStatus.SERVICE_UNAVAILABLE,
                        {"status": "unavailable"},
                    )
                    return
                try:
                    raw = requester(challenge)
                except Exception:
                    LOGGER.exception("Reviewed local motion-search request failed")
                    raw = {"status": "error"}
                value = dict(raw) if isinstance(raw, Mapping) else {}
                outcome = str(value.get("status") or "error")
                statuses = {
                    "confirm": HTTPStatus.ACCEPTED,
                    "accepted": HTTPStatus.ACCEPTED,
                    "confirmation_required": HTTPStatus.CONFLICT,
                    "busy": HTTPStatus.CONFLICT,
                    "unsafe": HTTPStatus.CONFLICT,
                    "unavailable": HTTPStatus.SERVICE_UNAVAILABLE,
                    "error": HTTPStatus.INTERNAL_SERVER_ERROR,
                }
                status = statuses.get(outcome)
                if status is None:
                    outcome = "error"
                    status = HTTPStatus.INTERNAL_SERVER_ERROR
                response = {"status": outcome}
                if outcome == "confirm":
                    token = value.get("challenge")
                    expires = value.get("expires_in_seconds")
                    if (
                        not isinstance(token, str)
                        or len(token) != 32
                        or any(
                            character not in "0123456789abcdef" for character in token
                        )
                        or isinstance(expires, bool)
                        or not isinstance(expires, int)
                        or not 1 <= expires <= 5
                    ):
                        status = HTTPStatus.INTERNAL_SERVER_ERROR
                        response = {"status": "error"}
                    else:
                        response.update(
                            {
                                "challenge": token,
                                "expires_in_seconds": expires,
                            }
                        )
                self._send_json(status, response)

            def _request_safety_configuration(self) -> None:
                payload = self._read_local_action_variants(
                    SAFETY_CONFIGURE_ACTION_HEADER,
                    allowed_key_sets=frozenset(
                        (frozenset(), frozenset(("challenge",)))
                    ),
                )
                if payload is None:
                    return
                challenge = payload.get("challenge")
                if challenge is not None and (
                    not isinstance(challenge, str)
                    or len(challenge) != 32
                    or any(value not in "0123456789abcdef" for value in challenge)
                ):
                    self._send_json(
                        HTTPStatus.BAD_REQUEST,
                        {"status": "invalid_request"},
                    )
                    return
                requester = parent.safety_configure_requester
                if requester is None:
                    self._send_json(
                        HTTPStatus.SERVICE_UNAVAILABLE,
                        {"status": "unavailable"},
                    )
                    return
                try:
                    raw = requester(challenge)
                except Exception:
                    LOGGER.exception("Lite6 safety configuration request failed")
                    raw = {"status": "error"}
                value = dict(raw) if isinstance(raw, Mapping) else {}
                outcome = str(value.get("status") or "error")
                statuses = {
                    "confirm": HTTPStatus.ACCEPTED,
                    "configured": HTTPStatus.OK,
                    "aborted": HTTPStatus.CONFLICT,
                    "confirmation_required": HTTPStatus.CONFLICT,
                    "busy": HTTPStatus.CONFLICT,
                    "unsafe": HTTPStatus.CONFLICT,
                    "unavailable": HTTPStatus.SERVICE_UNAVAILABLE,
                    "error": HTTPStatus.INTERNAL_SERVER_ERROR,
                }
                status = statuses.get(outcome, HTTPStatus.INTERNAL_SERVER_ERROR)
                response = {"status": outcome if outcome in statuses else "error"}
                digest = value.get("profile_sha256")
                if (
                    isinstance(digest, str)
                    and len(digest) == 64
                    and all(character in "0123456789abcdef" for character in digest)
                ):
                    response["profile_sha256"] = digest
                if outcome == "confirm":
                    token = value.get("challenge")
                    expires = value.get("expires_in_seconds")
                    if (
                        not isinstance(token, str)
                        or len(token) != 32
                        or any(
                            character not in "0123456789abcdef" for character in token
                        )
                        or isinstance(expires, bool)
                        or not isinstance(expires, int)
                        or not 1 <= expires <= 5
                        or "profile_sha256" not in response
                    ):
                        status = HTTPStatus.INTERNAL_SERVER_ERROR
                        response = {"status": "error"}
                    else:
                        response.update(
                            {
                                "challenge": token,
                                "expires_in_seconds": expires,
                            }
                        )
                self._send_json(status, response)

            def _request_workspace_teach_action(self, action: str) -> None:
                """Dispatch one coordinate-free, read-only teach action."""

                contracts = {
                    "start": (
                        WORKSPACE_TEACH_START_ACTION_HEADER,
                        parent.workspace_teach_starter,
                        "accepted",
                        HTTPStatus.ACCEPTED,
                    ),
                    "capture": (
                        WORKSPACE_TEACH_CAPTURE_ACTION_HEADER,
                        parent.workspace_teach_capturer,
                        "accepted",
                        HTTPStatus.ACCEPTED,
                    ),
                    "cancel": (
                        WORKSPACE_TEACH_CANCEL_ACTION_HEADER,
                        parent.workspace_teach_canceller,
                        "cancelled",
                        HTTPStatus.OK,
                    ),
                }
                contract = contracts.get(action)
                if contract is None:
                    self._send_json(
                        HTTPStatus.INTERNAL_SERVER_ERROR,
                        {"status": "error"},
                    )
                    return
                expected_header, callback, success, success_status = contract
                if not self._read_empty_local_action(expected_header):
                    return
                if callback is None:
                    self._send_json(
                        HTTPStatus.SERVICE_UNAVAILABLE,
                        {"status": "unavailable"},
                    )
                    return
                try:
                    raw = callback()
                except Exception:
                    LOGGER.exception("Local workspace-teach %s failed", action)
                    self._send_json(
                        HTTPStatus.INTERNAL_SERVER_ERROR,
                        {"status": "error"},
                    )
                    return
                value = dict(raw) if isinstance(raw, Mapping) else {}
                outcome = value.get("status")
                if set(value) != {"status"} or not isinstance(outcome, str):
                    self._send_json(
                        HTTPStatus.INTERNAL_SERVER_ERROR,
                        {"status": "error"},
                    )
                    return
                if outcome == success:
                    self._send_json(success_status, {"status": success})
                    return
                failure_statuses = {
                    "confirmation_required": HTTPStatus.CONFLICT,
                    "busy": HTTPStatus.CONFLICT,
                    "unsafe": HTTPStatus.CONFLICT,
                    "unavailable": HTTPStatus.SERVICE_UNAVAILABLE,
                    "error": HTTPStatus.INTERNAL_SERVER_ERROR,
                }
                status = failure_statuses.get(outcome)
                if status is None:
                    self._send_json(
                        HTTPStatus.INTERNAL_SERVER_ERROR,
                        {"status": "error"},
                    )
                    return
                self._send_json(status, {"status": outcome})

            def _request_workspace_teach_finalize(self) -> None:
                """Validate both steps of proposal finalization at HTTP edge."""

                payload = self._read_local_action_variants(
                    WORKSPACE_TEACH_FINALIZE_ACTION_HEADER,
                    allowed_key_sets=frozenset(
                        (frozenset(), frozenset(("challenge",)))
                    ),
                )
                if payload is None:
                    return
                challenge = payload.get("challenge")
                if challenge is not None and not self._lower_hex(challenge, 32):
                    self._send_json(
                        HTTPStatus.BAD_REQUEST,
                        {"status": "invalid_request"},
                    )
                    return
                callback = parent.workspace_teach_finalizer
                if callback is None:
                    self._send_json(
                        HTTPStatus.SERVICE_UNAVAILABLE,
                        {"status": "unavailable"},
                    )
                    return

                # This short-lived edge binding is defense in depth only. The
                # callback remains authoritative and must independently bind
                # and consume the session, proposal digest, and STOP generation.
                expected_digest = ""
                if challenge is not None:
                    with parent._workspace_teach_finalize_lock:
                        valid_confirmation = bool(
                            parent._workspace_teach_finalize_challenge
                            and parent._workspace_teach_finalize_expires_at
                            >= time.monotonic()
                            and hmac.compare_digest(
                                parent._workspace_teach_finalize_challenge,
                                challenge,
                            )
                        )
                        if valid_confirmation:
                            expected_digest = (
                                parent._workspace_teach_finalize_proposal_sha256
                            )
                        parent._clear_workspace_teach_finalize_confirmation_locked()
                    if not valid_confirmation:
                        self._send_json(
                            HTTPStatus.CONFLICT,
                            {"status": "confirmation_required"},
                        )
                        return
                else:
                    with parent._workspace_teach_finalize_lock:
                        parent._clear_workspace_teach_finalize_confirmation_locked()

                try:
                    raw = callback(challenge)
                except Exception:
                    LOGGER.exception("Local workspace-teach finalize failed")
                    self._send_json(
                        HTTPStatus.INTERNAL_SERVER_ERROR,
                        {"status": "error"},
                    )
                    return
                value = dict(raw) if isinstance(raw, Mapping) else {}
                outcome = value.get("status")
                failure_statuses = {
                    "confirmation_required": HTTPStatus.CONFLICT,
                    "busy": HTTPStatus.CONFLICT,
                    "unsafe": HTTPStatus.CONFLICT,
                    "unavailable": HTTPStatus.SERVICE_UNAVAILABLE,
                    "error": HTTPStatus.INTERNAL_SERVER_ERROR,
                }
                if isinstance(outcome, str) and outcome in failure_statuses:
                    if set(value) != {"status"}:
                        self._send_json(
                            HTTPStatus.INTERNAL_SERVER_ERROR,
                            {"status": "error"},
                        )
                    else:
                        self._send_json(
                            failure_statuses[outcome],
                            {"status": outcome},
                        )
                    return

                digest = value.get("proposal_sha256")
                if challenge is None:
                    token = value.get("challenge")
                    expires = value.get("expires_in_seconds")
                    if (
                        set(value)
                        != {
                            "status",
                            "challenge",
                            "expires_in_seconds",
                            "proposal_sha256",
                        }
                        or outcome != "confirm"
                        or not self._lower_hex(token, 32)
                        or type(expires) is not int
                        or expires != 5
                        or not self._lower_hex(digest, 64)
                    ):
                        self._send_json(
                            HTTPStatus.INTERNAL_SERVER_ERROR,
                            {"status": "error"},
                        )
                        return
                    with parent._workspace_teach_finalize_lock:
                        parent._workspace_teach_finalize_challenge = token
                        parent._workspace_teach_finalize_proposal_sha256 = digest
                        parent._workspace_teach_finalize_expires_at = (
                            time.monotonic() + 5.0
                        )
                    self._send_json(
                        HTTPStatus.ACCEPTED,
                        {
                            "status": "confirm",
                            "challenge": token,
                            "expires_in_seconds": 5,
                            "proposal_sha256": digest,
                        },
                    )
                    return

                if (
                    set(value) != {"status", "proposal_sha256"}
                    or outcome != "proposed"
                    or not self._lower_hex(digest, 64)
                    or not hmac.compare_digest(digest, expected_digest)
                ):
                    self._send_json(
                        HTTPStatus.INTERNAL_SERVER_ERROR,
                        {"status": "error"},
                    )
                    return
                self._send_json(
                    HTTPStatus.OK,
                    {"status": "proposed", "proposal_sha256": digest},
                )

            @staticmethod
            def _lower_hex(value: Any, length: int) -> bool:
                return bool(
                    isinstance(value, str)
                    and len(value) == length
                    and all(character in "0123456789abcdef" for character in value)
                )

            def _read_empty_local_action(self, expected_header: str) -> bool:
                return (
                    self._read_local_action(
                        expected_header,
                        allowed_keys=frozenset(),
                    )
                    is not None
                )

            def _read_local_action(
                self,
                expected_header: str,
                *,
                allowed_keys: frozenset,
                maximum_bytes: int = 64,
            ) -> Optional[Mapping[str, Any]]:
                return self._read_local_action_variants(
                    expected_header,
                    allowed_key_sets=frozenset((allowed_keys,)),
                    maximum_bytes=maximum_bytes,
                )

            def _read_local_action_variants(
                self,
                expected_header: str,
                *,
                allowed_key_sets: frozenset,
                maximum_bytes: int = 64,
            ) -> Optional[Mapping[str, Any]]:
                # State-changing endpoints are deliberately local-only. A LAN
                # MJPEG listener may still serve authenticated streams, but it
                # never exposes robot, dataset, or OTA control.
                client_host = str(self.client_address[0]).lower()
                if client_host not in _LOOPBACK_CLIENTS:
                    self.send_error(HTTPStatus.NOT_FOUND)
                    return None
                host_headers = self.headers.get_all("Host") or []
                if len(host_headers) != 1 or not parent._is_loopback_host_header(
                    host_headers[0]
                ):
                    self.send_error(HTTPStatus.NOT_FOUND)
                    return None
                if self.headers.get_all("Origin") or self.headers.get_all(
                    "Sec-Fetch-Site"
                ):
                    self._send_json(
                        HTTPStatus.FORBIDDEN,
                        {"status": "forbidden"},
                    )
                    return None
                client_headers = self.headers.get_all("X-LabCobot-Client") or []
                if client_headers != [SNAPSHOT_CLIENT_HEADER]:
                    self._send_json(
                        HTTPStatus.FORBIDDEN,
                        {"status": "forbidden"},
                    )
                    return None
                action_headers = self.headers.get_all("X-LabCobot-Action") or []
                if action_headers != [expected_header]:
                    self._send_json(
                        HTTPStatus.FORBIDDEN,
                        {"status": "forbidden"},
                    )
                    return None
                content_types = self.headers.get_all("Content-Type") or []
                if (
                    len(content_types) != 1
                    or content_types[0].strip().lower() != "application/json"
                    or self.headers.get_all("Transfer-Encoding")
                ):
                    self._send_json(
                        HTTPStatus.UNSUPPORTED_MEDIA_TYPE,
                        {"status": "invalid_request"},
                    )
                    return None
                content_lengths = self.headers.get_all("Content-Length") or []
                if len(content_lengths) != 1:
                    self._send_json(
                        HTTPStatus.BAD_REQUEST,
                        {"status": "invalid_request"},
                    )
                    return None
                try:
                    content_length = int(content_lengths[0])
                except (TypeError, ValueError):
                    content_length = -1
                if content_length < 0 or content_length > maximum_bytes:
                    self._send_json(
                        HTTPStatus.REQUEST_ENTITY_TOO_LARGE,
                        {"status": "invalid_request"},
                    )
                    return None
                try:
                    request_payload = json.loads(
                        self.rfile.read(content_length).decode("utf-8"),
                        object_pairs_hook=_reject_duplicate_object_pairs,
                    )
                except (UnicodeError, ValueError):
                    self._send_json(
                        HTTPStatus.BAD_REQUEST,
                        {"status": "invalid_request"},
                    )
                    return None
                if (
                    not isinstance(request_payload, dict)
                    or frozenset(request_payload) not in allowed_key_sets
                ):
                    self._send_json(
                        HTTPStatus.BAD_REQUEST,
                        {"status": "invalid_request"},
                    )
                    return None
                return request_payload

            def _send_json(self, status: HTTPStatus, value: Mapping[str, Any]) -> None:
                payload = json.dumps(
                    dict(value),
                    allow_nan=False,
                    separators=(",", ":"),
                    sort_keys=True,
                ).encode("utf-8")
                self.send_response(status)
                self.send_header("Content-Type", "application/json")
                self.send_header("Cache-Control", "no-store")
                self.send_header("X-Content-Type-Options", "nosniff")
                self.send_header("Content-Length", str(len(payload)))
                self.end_headers()
                self.wfile.write(payload)

            def _send_empty(self, status: HTTPStatus) -> None:
                self.send_response(status)
                self.send_header("Cache-Control", "no-store")
                self.send_header("X-Content-Type-Options", "nosniff")
                self.send_header("Content-Length", "0")
                self.end_headers()

            def _send_health(self) -> None:
                snapshot = parent.health_snapshot()
                payload = json.dumps(
                    snapshot,
                    allow_nan=False,
                    separators=(",", ":"),
                    sort_keys=True,
                ).encode("utf-8")
                status = (
                    HTTPStatus.OK
                    if snapshot["ready"]
                    else HTTPStatus.SERVICE_UNAVAILABLE
                )
                self.send_response(status)
                self.send_header("Content-Type", "application/json")
                self.send_header("Cache-Control", "no-store")
                self.send_header("Content-Length", str(len(payload)))
                self.end_headers()
                self.wfile.write(payload)

            def log_message(self, fmt: str, *args) -> None:
                LOGGER.debug("MJPEG " + fmt, *args)

        server = self.server_factory((self.host, self.port), Handler)
        self._server = server
        self._thread = threading.Thread(
            target=server.serve_forever,
            kwargs={"poll_interval": 0.2},
            name="labcobot-mjpeg",
            daemon=True,
        )
        self._thread.start()

    def stop(self, join_timeout_seconds: float = 2.0) -> None:
        server, self._server = self._server, None
        if server is not None:
            server.shutdown()
            server.server_close()
        thread, self._thread = self._thread, None
        if thread is not None and thread is not threading.current_thread():
            thread.join(max(0.0, join_timeout_seconds))

    def is_alive(self) -> bool:
        """Return whether the configured local HTTP owner is still serving."""

        thread = self._thread
        return bool(thread is not None and thread.is_alive())

    def health_snapshot(self) -> Mapping[str, Any]:
        """Return freshness-aware camera and optional workcell health."""

        cameras = {}
        for role in (CameraRole.TOOL, CameraRole.WORKSPACE):
            stream = self.streams[role]
            try:
                stats = stream.stats()
                state = getattr(stats.state, "value", str(stats.state))
                fresh = (
                    stream.latest(max_age_s=max(0.1, self.frame_timeout_seconds * 2.0))
                    is not None
                )
                cameras[role.value] = {
                    "state": state,
                    "fresh": fresh,
                    "frames_captured": int(stats.frames_captured),
                    "reconnects": int(stats.reconnects),
                    "consecutive_errors": int(stats.consecutive_errors),
                }
            except Exception as exc:
                cameras[role.value] = {
                    "state": "degraded",
                    "fresh": False,
                    "reason": type(exc).__name__,
                }
        cameras_ready = all(
            camera.get("state") == "running" and camera.get("fresh") is True
            for camera in cameras.values()
        )
        runtime = {}
        runtime_ready = True
        if self.status_provider is not None:
            try:
                value = self.status_provider()
                if not isinstance(value, Mapping):
                    raise TypeError("status provider did not return a mapping")
                runtime = dict(value)
                runtime_ready = runtime.get("ready") is True
            except Exception as exc:
                runtime = {
                    "status": "degraded",
                    "ready": False,
                    "problems": ["status_provider_{}".format(type(exc).__name__)],
                }
                runtime_ready = False
        ready = cameras_ready and runtime_ready
        return {
            "status": "ok" if ready else "degraded",
            "ready": ready,
            "cameras": cameras,
            "runtime": runtime,
        }

    def set_status_provider(
        self,
        provider: Optional[Callable[[], Mapping[str, Any]]],
    ) -> None:
        """Attach the secret-free workcell snapshot after lifecycle assembly."""

        self.status_provider = provider

    def set_update_requester(
        self,
        requester: Optional[Callable[[], bool]],
    ) -> None:
        """Attach the loopback-only signed-OTA check trigger."""

        self.update_requester = requester

    def set_capture_controls(
        self,
        starter: Optional[Callable[[Optional[str]], Mapping[str, Any]]],
        stopper: Optional[Callable[[], str]],
    ) -> None:
        """Attach loopback-only, fail-closed capture-run controls."""

        self.capture_starter = starter
        self.capture_stopper = stopper

    def set_software_stop(
        self,
        stopper: Optional[Callable[[], str]],
    ) -> None:
        """Attach the universal loopback-only software STOP callback."""

        self.software_stopper = stopper

    def set_motion_search_control(
        self,
        requester: Optional[Callable[[Optional[str]], Mapping[str, Any]]],
    ) -> None:
        """Attach only the reviewed, two-step dynamic-valve search seam."""

        self.motion_search_requester = requester

    def set_safety_configuration_control(
        self,
        requester: Optional[Callable[[Optional[str]], Mapping[str, Any]]],
    ) -> None:
        """Bind only the local, two-step Lite6 safety configurator."""

        self.safety_configure_requester = requester

    def set_operator_controls(
        self,
        *,
        petri_starter: Optional[Callable[[Optional[str]], Mapping[str, Any]]] = None,
        registration_verifier: Optional[Callable[[], Mapping[str, Any]]] = None,
        recovery_starter: Optional[Callable[[Optional[str]], Mapping[str, Any]]] = None,
        manual_reviewer: Optional[Callable[[Optional[str]], Mapping[str, Any]]] = None,
        manual_review_previewer: Optional[Callable[[], bytes]] = None,
    ) -> None:
        """Attach the remaining bounded operator workflows."""

        self.petri_starter = petri_starter
        self.registration_verifier = registration_verifier
        self.recovery_starter = recovery_starter
        self.manual_reviewer = manual_reviewer
        self.manual_review_previewer = manual_review_previewer

    def set_workspace_teach_controls(
        self,
        *,
        starter: Optional[Callable[[], Mapping[str, Any]]] = None,
        capturer: Optional[Callable[[], Mapping[str, Any]]] = None,
        canceller: Optional[Callable[[], Mapping[str, Any]]] = None,
        finalizer: Optional[Callable[[Optional[str]], Mapping[str, Any]]] = None,
    ) -> None:
        """Attach coordinate-free, read-only workspace-teach callbacks."""

        self.workspace_teach_starter = starter
        self.workspace_teach_capturer = capturer
        self.workspace_teach_canceller = canceller
        self.workspace_teach_finalizer = finalizer
        with self._workspace_teach_finalize_lock:
            self._clear_workspace_teach_finalize_confirmation_locked()

    def set_tenant_auth_controls(
        self,
        *,
        login: Optional[Callable[[str, str, str], Mapping[str, Any]]] = None,
        logout: Optional[Callable[[], Mapping[str, Any]]] = None,
    ) -> None:
        """Attach secret-containing login only to the private local action edge."""

        self.tenant_auth_login = login
        self.tenant_auth_logout = logout

    def set_hand_training_controls(
        self,
        *,
        starter: Optional[Callable[[], Mapping[str, Any]]] = None,
        tool: Optional[Callable[[bool], Mapping[str, Any]]] = None,
        ender: Optional[Callable[[], Mapping[str, Any]]] = None,
    ) -> None:
        """Attach the local start, typed-tool, and end recording callbacks."""

        self.hand_training_starter = starter
        self.hand_training_tool = tool
        self.hand_training_ender = ender

    def _clear_workspace_teach_finalize_confirmation_locked(self) -> None:
        """Clear the short-lived HTTP-edge proposal confirmation binding."""

        self._workspace_teach_finalize_challenge = ""
        self._workspace_teach_finalize_proposal_sha256 = ""
        self._workspace_teach_finalize_expires_at = 0.0

    def _authorized(self, authorization: Optional[str], query: str) -> bool:
        del query
        if not self.token:
            return True
        supplied = ""
        if authorization:
            scheme, _, value = authorization.partition(" ")
            if scheme.lower() == "bearer":
                supplied = value.strip()
        return hmac.compare_digest(self.token, supplied)

    @staticmethod
    def _is_loopback_host_header(value: Any) -> bool:
        """Accept only an unambiguous loopback HTTP Host authority."""

        if not isinstance(value, str) or not value or len(value) > 128:
            return False
        authority = value.strip().lower()
        if authority.startswith("["):
            closing = authority.find("]")
            if closing < 0:
                return False
            host = authority[1:closing]
            suffix = authority[closing + 1 :]
            if suffix and (
                not suffix.startswith(":")
                or not suffix[1:].isdigit()
                or not 1 <= int(suffix[1:]) <= 65535
            ):
                return False
        else:
            host = authority
            if ":" in authority:
                host, port = authority.rsplit(":", 1)
                if not port.isdigit() or not 1 <= int(port) <= 65535:
                    return False
        return host in _LOOPBACK_HOSTS


__all__ = [
    "MjpegServer",
    "MANUAL_REVIEW_ACTION_HEADER",
    "MANUAL_REVIEW_PATH",
    "MANUAL_REVIEW_PREVIEW_PATH",
    "CAPTURE_START_ACTION_HEADER",
    "CAPTURE_START_PATH",
    "CAPTURE_STOP_ACTION_HEADER",
    "CAPTURE_STOP_PATH",
    "MOTION_SEARCH_ARM_ACTION_HEADER",
    "MOTION_SEARCH_ARM_PATH",
    "OTA_ACTION_HEADER",
    "OTA_CHECK_PATH",
    "PETRI_START_ACTION_HEADER",
    "PETRI_START_PATH",
    "RECOVERY_START_ACTION_HEADER",
    "RECOVERY_START_PATH",
    "REGISTRATION_VERIFY_ACTION_HEADER",
    "REGISTRATION_VERIFY_PATH",
    "SAFETY_CONFIGURE_ACTION_HEADER",
    "SAFETY_CONFIGURE_PATH",
    "SNAPSHOT_CLIENT_HEADER",
    "SNAPSHOT_MAX_BYTES",
    "SNAPSHOT_PATHS",
    "SOFTWARE_STOP_ACTION_HEADER",
    "SOFTWARE_STOP_PATH",
    "STREAM_PATHS",
    "WORKSPACE_TEACH_CANCEL_ACTION_HEADER",
    "WORKSPACE_TEACH_CANCEL_PATH",
    "WORKSPACE_TEACH_CAPTURE_ACTION_HEADER",
    "WORKSPACE_TEACH_CAPTURE_PATH",
    "WORKSPACE_TEACH_FINALIZE_ACTION_HEADER",
    "WORKSPACE_TEACH_FINALIZE_PATH",
    "WORKSPACE_TEACH_START_ACTION_HEADER",
    "WORKSPACE_TEACH_START_PATH",
]
