"""Authenticated, staged over-the-air updates with rollback support."""

from __future__ import annotations

import fcntl
import hashlib
import hmac
import json
import os
import re
import secrets
import shlex
import shutil
import stat
import tempfile
import threading
import uuid
import zipfile
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from typing import Any, Callable, Mapping, Optional, Sequence
from urllib.parse import urlparse

from packaging.version import InvalidVersion, Version

from .http import UrllibTransport, json_body

RELEASE_VERSION_FILENAME = ".labcobot-version"
SOURCE_REVISION_FILENAME = ".labcobot-source-revision"
OTA_MANIFEST_SCHEMA = "supraworx.ota.manifest.v2"
OTA_REPORT_CONTRACT = "ota-status-v2"
_MINT_DEVICE_UID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:\-]{0,127}$")
_MINT_VERSION_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.\-]{0,99}$")
_MINT_SOURCE_REVISION_RE = re.compile(r"^[0-9a-f]{40}$")
_RELEASE_TREE_DIGEST_DOMAIN = b"labcobot-release-tree-v1\0"
_MAX_RELEASE_TREE_ENTRIES = 50_000
_MAX_RELEASE_TREE_BYTES = 4 * 1024 * 1024 * 1024
_MAX_RELEASE_RELATIVE_PATH_BYTES = 4096
OTA_ADMISSION_REVALIDATION_PHASES = (
    "before_network",
    "after_staging",
    "before_installing_state",
    "before_first_rename",
    "after_second_rename",
    "before_restart",
)


class OTAError(RuntimeError):
    """Raised when an update cannot be verified or applied."""


class OTAAdmissionChanged(OTAError):
    """Raised with one bounded reason when a veto-only OTA lease changes."""

    _REASONS = frozenset(
        (
            "camera_changed",
            "controller_changed",
            "fatal_changed",
            "lease_expired",
            "owner_changed",
            "safety_changed",
            "telemetry_changed",
            "updater_unsupported",
            "workspace_teach_changed",
        )
    )

    def __init__(self, reason: str = "telemetry_changed") -> None:
        self.reason = reason if reason in self._REASONS else "telemetry_changed"
        super().__init__("OTA admission lease changed")


def _revalidate_ota_admission(
    revalidator: Optional[Callable[[str], bool]],
    phase: str,
) -> None:
    """Run one bounded veto callback without exposing its failure details."""

    if phase not in OTA_ADMISSION_REVALIDATION_PHASES:
        raise OTAAdmissionChanged("updater_unsupported")
    if revalidator is None:
        return
    try:
        accepted = revalidator(phase)
    except OTAAdmissionChanged:
        raise
    except Exception as exc:
        raise OTAAdmissionChanged("telemetry_changed") from exc
    if accepted is not True:
        raise OTAAdmissionChanged("telemetry_changed")


@dataclass(frozen=True)
class OTAResult:
    """Result of an OTA poll/apply operation."""

    status: str
    current_version: str
    available_version: str = ""
    message: str = ""
    manifest_verification: str = "not_checked"
    archive_digest_verification: str = "not_checked"


class PrivilegedOTAClient:
    """Unprivileged request/status client for the root-owned OTA installer.

    The client has no publisher key, download URL input, archive path or
    release-switch primitive.  It can only submit one of three fixed actions
    and consume the bounded public status written by the privileged service.
    """

    REQUEST_SCHEMA = "labcobot.privileged-ota.request.v1"
    STATUS_SCHEMA = "labcobot.privileged-ota.status.v1"
    _REQUEST_ID = re.compile(r"[0-9a-f]{32}")
    _UPDATE_ID = re.compile(r"[0-9a-f]{64}")
    _STATUSES = {
        "busy",
        "error",
        "healthy",
        "pending_restart",
        "rolled_back",
        "up_to_date",
        "update_available",
    }
    _MANIFEST_EVIDENCE = {
        "not_checked",
        "not_available",
        "checking",
        "verified",
        "failed",
    }
    _ARCHIVE_EVIDENCE = {
        "not_checked",
        "not_required",
        "checking",
        "verified",
        "failed",
    }
    ota_admission_revalidation_phases = frozenset()

    def __init__(
        self,
        *,
        request_dir: Path,
        status_path: Path,
        current_version: str,
        response_timeout: float = 180.0,
        expected_status_owner_uid: int = 0,
        monotonic: Callable[[], float] = __import__("time").monotonic,
        sleep: Callable[[float], None] = __import__("time").sleep,
    ) -> None:
        if not request_dir.is_absolute() or not status_path.is_absolute():
            raise ValueError("Privileged OTA paths must be absolute")
        if ".." in request_dir.parts or ".." in status_path.parts:
            raise ValueError("Privileged OTA paths must be canonical")
        if not 1.0 <= float(response_timeout) <= 900.0:
            raise ValueError("Privileged OTA response timeout is invalid")
        self.request_dir = request_dir
        self.status_path = status_path
        self.current_version = str(current_version)
        self.response_timeout = float(response_timeout)
        self.expected_status_owner_uid = int(expected_status_owner_uid)
        self._monotonic = monotonic
        self._sleep = sleep
        self._verification_lock = threading.Lock()
        self._verification: Mapping[str, str] = {
            "manifest_verification": "not_checked",
            "archive_digest_verification": "not_checked",
        }

    @staticmethod
    def _canonical_request(value: Mapping[str, Any]) -> bytes:
        return (
            json.dumps(
                dict(value),
                allow_nan=False,
                ensure_ascii=False,
                separators=(",", ":"),
                sort_keys=True,
            )
            + "\n"
        ).encode("utf-8")

    def _read_status(self) -> Mapping[str, Any]:
        try:
            metadata = self.status_path.lstat()
            payload = self.status_path.read_bytes()
        except OSError as exc:
            raise OTAError("Privileged OTA status is unavailable") from exc
        if (
            stat.S_ISLNK(metadata.st_mode)
            or not stat.S_ISREG(metadata.st_mode)
            or metadata.st_uid != self.expected_status_owner_uid
            or metadata.st_mode & 0o222
            or not 1 <= len(payload) <= 4096
        ):
            raise OTAError("Privileged OTA status path is unsafe")
        try:
            value = json.loads(payload.decode("utf-8"))
        except (UnicodeError, json.JSONDecodeError) as exc:
            raise OTAError("Privileged OTA status is invalid") from exc
        if (
            not isinstance(value, Mapping)
            or payload != self._canonical_request(value)
            or value.get("schema") != self.STATUS_SCHEMA
            or set(value)
            not in (
                {
                    "archive_digest_verification",
                    "available_version",
                    "completed_at",
                    "current_version",
                    "manifest_verification",
                    "report_pending",
                    "request_id",
                    "schema",
                    "status",
                    "update_id",
                },
                {
                    "archive_digest_verification",
                    "available_version",
                    "completed_at",
                    "current_version",
                    "error_code",
                    "manifest_verification",
                    "report_pending",
                    "request_id",
                    "schema",
                    "status",
                    "update_id",
                },
            )
        ):
            raise OTAError("Privileged OTA status contract is invalid")
        status = value.get("status")
        request_id = value.get("request_id")
        update_id = value.get("update_id")
        manifest = value.get("manifest_verification")
        archive = value.get("archive_digest_verification")
        if (
            status not in self._STATUSES
            or not isinstance(request_id, str)
            or (request_id and self._REQUEST_ID.fullmatch(request_id) is None)
            or not isinstance(update_id, str)
            or (update_id and self._UPDATE_ID.fullmatch(update_id) is None)
            or manifest not in self._MANIFEST_EVIDENCE
            or archive not in self._ARCHIVE_EVIDENCE
            or not isinstance(value.get("report_pending"), bool)
            or isinstance(value.get("completed_at"), bool)
            or not isinstance(value.get("completed_at"), int)
        ):
            raise OTAError("Privileged OTA status values are invalid")
        return value

    def _submit(self, action: str, *, update_id: str = "") -> Mapping[str, Any]:
        if action not in {"check", "install", "commit", "report"}:
            raise OTAError("Privileged OTA action is invalid")
        if (action in {"commit", "report"}) != bool(update_id):
            raise OTAError("Privileged OTA update identity is invalid")
        if update_id and self._UPDATE_ID.fullmatch(update_id) is None:
            raise OTAError("Privileged OTA update identity is invalid")
        request_id = uuid.uuid4().hex
        request = {
            "action": action,
            "request_id": request_id,
            "schema": self.REQUEST_SCHEMA,
            "update_id": update_id,
        }
        self.request_dir.mkdir(parents=True, exist_ok=True)
        if self.request_dir.is_symlink() or not self.request_dir.is_dir():
            raise OTAError("Privileged OTA request directory is unsafe")
        descriptor, name = tempfile.mkstemp(
            prefix=".request-", dir=str(self.request_dir)
        )
        temporary = Path(name)
        target = self.request_dir / (request_id + ".json")
        try:
            with os.fdopen(descriptor, "wb") as handle:
                handle.write(self._canonical_request(request))
                handle.flush()
                os.fsync(handle.fileno())
            os.chmod(temporary, 0o600, follow_symlinks=False)
            os.link(temporary, target, follow_symlinks=False)
        finally:
            temporary.unlink(missing_ok=True)
        deadline = self._monotonic() + self.response_timeout
        while self._monotonic() < deadline:
            try:
                response = self._read_status()
            except OTAError:
                self._sleep(0.1)
                continue
            if response.get("request_id") == request_id:
                manifest = str(response["manifest_verification"])
                archive = str(response["archive_digest_verification"])
                with self._verification_lock:
                    self._verification = {
                        "manifest_verification": manifest,
                        "archive_digest_verification": archive,
                    }
                if response["status"] == "error":
                    raise OTAError("Privileged OTA installer rejected the request")
                return response
            self._sleep(0.1)
        raise OTAError("Privileged OTA installer did not answer in time")

    @staticmethod
    def _result(response: Mapping[str, Any]) -> OTAResult:
        return OTAResult(
            status=str(response["status"]),
            current_version=str(response["current_version"]),
            available_version=str(response["available_version"]),
            manifest_verification=str(response["manifest_verification"]),
            archive_digest_verification=str(response["archive_digest_verification"]),
        )

    def check_for_update(self) -> OTAResult:
        return self._result(self._submit("check"))

    def check_and_apply(
        self,
        *,
        restart: Optional[Callable[[], None]] = None,
        admission_revalidator: Optional[Callable[[str], bool]] = None,
    ) -> OTAResult:
        if admission_revalidator is not None:
            _revalidate_ota_admission(admission_revalidator, "before_network")
            # The v1 root request protocol cannot invoke the application-side
            # telemetry lease at its internal switch boundaries.  Never submit
            # a privileged install that could outlive its only validation.
            raise OTAAdmissionChanged("updater_unsupported")
        result = self._result(self._submit("install"))
        if result.status == "pending_restart" and restart is not None:
            restart()
        return result

    def verification_status(self) -> Mapping[str, str]:
        with self._verification_lock:
            return dict(self._verification)

    def startup_pending_update_id(self) -> Optional[str]:
        try:
            status = self._read_status()
        except OTAError:
            return None
        if status["status"] != "pending_restart":
            return None
        update_id = str(status["update_id"])
        return update_id if self._UPDATE_ID.fullmatch(update_id) else None

    def pending_boot_identity_status(
        self,
        *,
        expected_update_id: str,
    ) -> Mapping[str, Any]:
        """Return the root-owned pending identity used by the commit request."""

        if self._UPDATE_ID.fullmatch(expected_update_id) is None:
            raise OTAError("Privileged OTA update identity is invalid")
        status = self._read_status()
        current_update_id = str(status["update_id"])
        ready = bool(
            status["status"] == "pending_restart"
            and self._UPDATE_ID.fullmatch(current_update_id) is not None
            and hmac.compare_digest(current_update_id, expected_update_id)
        )
        return {
            "schema_version": 1,
            "status": "verified" if ready else "identity_changed",
            "ready": ready,
            "update_id_sha256": hashlib.sha256(
                expected_update_id.encode("ascii")
            ).hexdigest(),
            "target_version": str(status["available_version"]) if ready else "",
            "target_source_revision": "",
            "target_tree_sha256": "",
        }

    def mark_boot_success(
        self,
        *,
        expected_update_id: str,
        health_generation: Optional[int] = None,
    ) -> bool:
        _validated_health_generation(health_generation)
        response = self._submit("commit", update_id=expected_update_id)
        return response["status"] == "healthy"

    def health_report_pending(self) -> bool:
        try:
            status = self._read_status()
        except OTAError:
            return False
        return (
            status["status"]
            in {
                "error",
                "healthy",
                "rolled_back",
            }
            and status["report_pending"] is True
        )

    def report_boot_success(self) -> bool:
        status = self._read_status()
        if (
            status["status"]
            not in {
                "error",
                "healthy",
                "rolled_back",
            }
            or status["report_pending"] is not True
        ):
            return False
        update_id = str(status["update_id"])
        if self._UPDATE_ID.fullmatch(update_id) is None:
            raise OTAError("Privileged OTA health identity is invalid")
        response = self._submit("report", update_id=update_id)
        return (
            response["status"]
            in {
                "healthy",
                "rolled_back",
            }
            and response["report_pending"] is False
        )

    def rollback_pending(self) -> bool:
        """The unprivileged runtime has no rollback/switch authority."""

        return False


class SecureOTAUpdater:
    """Verify, stage and atomically swap a complete application release."""

    ota_admission_revalidation_phases = frozenset(OTA_ADMISSION_REVALIDATION_PHASES)

    def __init__(
        self,
        *,
        metadata_url: str,
        report_url: str = "",
        uid_device: str,
        api_secret: str,
        signing_key: bytes,
        install_root: Path,
        current_version: str,
        state_dir: Path,
        transport: Optional[UrllibTransport] = None,
        timeout: float = 30.0,
        max_archive_bytes: int = 256 * 1024 * 1024,
        max_uncompressed_bytes: int = 1024 * 1024 * 1024,
        max_archive_entries: int = 20_000,
        allowed_download_hosts: Sequence[str] = (),
        preserve_paths: Sequence[str] = ("config.json",),
        required_path: str = "main/app.py",
        required_paths: Sequence[str] = ("main/runtime.py",),
        allow_insecure_http: bool = False,
    ) -> None:
        if not uid_device or not api_secret:
            raise ValueError("Registered device uid and api_secret are required")
        if not signing_key:
            raise ValueError("OTA signing key is required")
        if not allow_insecure_http and not _is_https(metadata_url):
            raise ValueError("OTA metadata URL must use HTTPS")
        if report_url:
            if not allow_insecure_http and not _is_https(report_url):
                raise ValueError("OTA report URL must use HTTPS")
            if not _same_origin(metadata_url, report_url):
                raise ValueError("OTA report URL must share the metadata origin")
        self.metadata_url = metadata_url
        self.report_url = str(report_url)
        self.uid_device = uid_device
        self.api_secret = api_secret
        self.signing_key = bytes(signing_key)
        self.install_root = Path(install_root).resolve()
        self.current_version = current_version
        self.state_dir = Path(state_dir).resolve()
        self.transport = transport or UrllibTransport()
        self.timeout = float(timeout)
        self.max_archive_bytes = int(max_archive_bytes)
        self.max_uncompressed_bytes = int(max_uncompressed_bytes)
        self.max_archive_entries = int(max_archive_entries)
        self.allowed_download_hosts = {
            str(host).strip().lower()
            for host in allowed_download_hosts
            if str(host).strip()
        }
        metadata_host = (urlparse(metadata_url).hostname or "").lower()
        if metadata_host:
            self.allowed_download_hosts.add(metadata_host)
        self.preserve_paths = tuple(
            _safe_relative_path(item) for item in preserve_paths
        )
        self.required_paths = tuple(
            dict.fromkeys(
                _safe_relative_path(item) for item in (required_path, *required_paths)
            )
        )
        self.allow_insecure_http = allow_insecure_http
        self._verification_lock = threading.Lock()
        self._verification: Mapping[str, str] = {
            "manifest_verification": "not_checked",
            "archive_digest_verification": "not_checked",
        }
        self.state_dir.mkdir(parents=True, exist_ok=True)

    def check_and_apply(
        self,
        *,
        restart: Optional[Callable[[], None]] = None,
        admission_revalidator: Optional[Callable[[str], bool]] = None,
    ) -> OTAResult:
        """Poll metadata and apply a newer verified release, if available."""
        self._set_verification("checking", "not_checked")
        with (self.state_dir / "update.lock").open("a+b") as lock_handle:
            try:
                fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError:
                self._set_verification("not_checked", "not_checked")
                return OTAResult(
                    "busy",
                    self.current_version,
                    message="Another OTA operation is in progress",
                )

            _revalidate_ota_admission(admission_revalidator, "before_network")
            self.recover_interrupted_swap()
            manifest, remote_version, newer = self._verified_manifest()
            if manifest is None:
                return OTAResult(
                    "up_to_date",
                    self.current_version,
                    manifest_verification="not_available",
                    archive_digest_verification="not_required",
                )
            if not newer:
                self._set_verification("verified", "not_required")
                return OTAResult(
                    "up_to_date",
                    self.current_version,
                    remote_version,
                    manifest_verification="verified",
                    archive_digest_verification="not_required",
                )

            self._download_stage_and_swap(
                manifest,
                admission_revalidator=admission_revalidator,
            )
            result = OTAResult(
                "pending_restart",
                self.current_version,
                remote_version,
                manifest_verification="verified",
                archive_digest_verification="verified",
            )
            try:
                _revalidate_ota_admission(admission_revalidator, "before_restart")
            except OTAAdmissionChanged:
                self.rollback_pending(failure_reason_code="admission_changed")
                raise
            if restart is not None:
                restart()
            return result

    def check_for_update(self) -> OTAResult:
        """Verify update metadata without downloading, swapping, or restarting.

        This is the only operation suitable for a scheduled poller.  Applying a
        release remains an explicit operator action through
        :meth:`check_and_apply`; a cached metadata result never authorizes a
        later swap.
        """

        self._set_verification("checking", "not_checked")
        with (self.state_dir / "update.lock").open("a+b") as lock_handle:
            try:
                fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError:
                self._set_verification("not_checked", "not_checked")
                return OTAResult(
                    "busy",
                    self.current_version,
                    message="Another OTA operation is in progress",
                )
            state = self._read_state()
            if state.get("status") in {"installing", "pending_restart"}:
                self._set_verification("not_checked", "not_checked")
                return OTAResult(
                    "busy",
                    self.current_version,
                    message="An OTA release is pending recovery or restart",
                )
            manifest, remote_version, newer = self._verified_manifest()
            if manifest is None:
                return OTAResult(
                    "up_to_date",
                    self.current_version,
                    manifest_verification="not_available",
                    archive_digest_verification="not_required",
                )
            self._set_verification("verified", "not_required")
            return OTAResult(
                "update_available" if newer else "up_to_date",
                self.current_version,
                remote_version,
                manifest_verification="verified",
                archive_digest_verification="not_required",
            )

    def _verified_manifest(self) -> tuple[Optional[Mapping[str, Any]], str, bool]:
        """Return only authenticated metadata and its version comparison."""

        try:
            manifest = self._fetch_manifest()
        except Exception:
            self._set_verification("failed", "not_checked")
            raise
        if manifest is None:
            self._set_verification("not_available", "not_required")
            return None, "", False
        # Authenticate every manifest before using even its version field.
        # Otherwise an unsigned stale response can be accepted as up-to-date
        # and conceal a valid available release.
        try:
            self._verify_manifest(manifest)
        except Exception:
            self._set_verification("failed", "not_checked")
            raise
        self._set_verification("verified", "not_checked")
        remote_version = str(manifest["version"])
        try:
            newer = Version(remote_version) > Version(self.current_version)
        except InvalidVersion as exc:
            self._set_verification("failed", "not_checked")
            raise OTAError("OTA manifest contains an invalid version") from exc
        return manifest, remote_version, newer

    def verification_status(self) -> Mapping[str, str]:
        """Return a bounded, secret-free snapshot of the latest OTA evidence."""

        with self._verification_lock:
            return dict(self._verification)

    def _set_verification(self, manifest: str, archive: str) -> None:
        manifest_states = {
            "not_checked",
            "not_available",
            "checking",
            "verified",
            "failed",
        }
        archive_states = {
            "not_checked",
            "not_required",
            "checking",
            "verified",
            "failed",
        }
        safe_manifest = manifest if manifest in manifest_states else "failed"
        safe_archive = archive if archive in archive_states else "failed"
        with self._verification_lock:
            self._verification = {
                "manifest_verification": safe_manifest,
                "archive_digest_verification": safe_archive,
            }

    def startup_pending_update_id(self) -> Optional[str]:
        """Return the boot-attempted pending release identity without mutation.

        The root-owned boot supervisor records ``boot_attempted`` while holding
        the same lock.  A runtime may retain the returned identity for its
        health window, but must present it again to :meth:`mark_boot_success`.
        Identity-less legacy pending states remain rollback-only.
        """

        with (self.state_dir / "update.lock").open("a+b") as lock_handle:
            fcntl.flock(lock_handle.fileno(), fcntl.LOCK_SH)
            state_payload = self._read_state()
            if (
                state_payload.get("status") != "pending_restart"
                or state_payload.get("boot_attempted") is not True
            ):
                return None
            update_id = self._validated_update_id(state_payload, required=False)
            if update_id is None:
                return None
            self._validate_single_boot_attempt(state_payload)
            return update_id

    def pending_boot_identity_status(
        self,
        *,
        expected_update_id: str,
    ) -> Mapping[str, Any]:
        """Validate the pending state and both exact release trees read-only."""

        expected = self._validated_update_id_value(
            expected_update_id,
            "expected OTA update identity",
        )
        with (self.state_dir / "update.lock").open("a+b") as lock_handle:
            fcntl.flock(lock_handle.fileno(), fcntl.LOCK_SH)
            state_payload = self._read_state()
            if (
                state_payload.get("status") != "pending_restart"
                or state_payload.get("boot_attempted") is not True
            ):
                return {
                    "schema_version": 1,
                    "status": "not_pending",
                    "ready": False,
                    "update_id_sha256": hashlib.sha256(
                        expected.encode("ascii")
                    ).hexdigest(),
                    "target_version": "",
                    "target_source_revision": "",
                    "target_tree_sha256": "",
                }
            update_id = self._validated_update_id(state_payload, required=True)
            self._validate_single_boot_attempt(state_payload)
            if not hmac.compare_digest(update_id, expected):
                return {
                    "schema_version": 1,
                    "status": "identity_changed",
                    "ready": False,
                    "update_id_sha256": hashlib.sha256(
                        expected.encode("ascii")
                    ).hexdigest(),
                    "target_version": "",
                    "target_source_revision": "",
                    "target_tree_sha256": "",
                }
            release_identity = self._preserved_release_identity(
                state_payload,
                required=True,
            )
            backup = self._validated_backup_path(state_payload)
            self._assert_state_release_tree(backup, state_payload, "previous")
            self._assert_state_release_tree(
                self.install_root,
                state_payload,
                "target",
            )
            return {
                "schema_version": 1,
                "status": "verified",
                "ready": True,
                "update_id_sha256": hashlib.sha256(
                    update_id.encode("ascii")
                ).hexdigest(),
                "target_version": release_identity["target_version"],
                "target_source_revision": release_identity["target_source_revision"],
                "target_tree_sha256": release_identity["target_tree_sha256"],
            }

    def mark_boot_success(
        self,
        *,
        expected_update_id: str,
        health_generation: Optional[int] = None,
    ) -> bool:
        """Commit exactly the pending release observed at process startup."""

        expected = self._validated_update_id_value(
            expected_update_id,
            "expected OTA update identity",
        )
        generation = _validated_health_generation(health_generation)
        with (self.state_dir / "update.lock").open("a+b") as lock_handle:
            fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX)
            state_payload = self._read_state()
            if state_payload.get("status") != "pending_restart":
                return False
            if state_payload.get("boot_attempted") is not True:
                raise OTAError(
                    "OTA pending release has not been authorized for this boot"
                )
            update_id = self._validated_update_id(state_payload, required=True)
            self._validate_single_boot_attempt(state_payload)
            if not hmac.compare_digest(update_id, expected):
                raise OTAError("OTA pending release identity changed during startup")
            report_metadata = dict(
                self._validated_report_metadata(state_payload, required=True)
            )
            report_metadata["report_observed_at"] = (
                datetime.now(timezone.utc)
                .isoformat(timespec="seconds")
                .replace("+00:00", "Z")
            )
            backup = self._validated_backup_path(state_payload)
            if not backup.is_dir() or backup.is_symlink():
                raise OTAError("OTA rollback backup is missing or unsafe")
            release_identity = self._preserved_release_identity(
                state_payload,
                required=True,
            )
            self._assert_state_release_tree(backup, state_payload, "previous")
            self._assert_state_release_tree(
                self.install_root,
                state_payload,
                "target",
            )
            terminal_state = {
                "status": "healthy",
                "version": state_payload.get("target_version", ""),
                "source_revision": state_payload.get("target_source_revision", ""),
                "tree_sha256": state_payload.get("target_tree_sha256", ""),
                **release_identity,
                "boot_attempted": True,
                "boot_attempt_count": 1,
                "backup_path": str(backup),
                "update_id": update_id,
                **report_metadata,
                "report_pending": True,
            }
            if generation is not None:
                terminal_state["boot_health_generation"] = generation
            # Commit the desired terminal state before removing the only
            # rollback copy. If power is lost during removal, recovery can
            # safely finish cleanup instead of seeing ``pending_restart`` with
            # no backup.
            self._write_state(terminal_state)
            self._finish_healthy_terminal(terminal_state)
            return True

    def health_report_pending(self) -> bool:
        """Return whether a locally healthy release still needs acknowledgement.

        The report identity is validated while reading it so a corrupt durable
        state cannot trigger an unauditable network request after restart.
        """

        with (self.state_dir / "update.lock").open("a+b") as lock_handle:
            fcntl.flock(lock_handle.fileno(), fcntl.LOCK_SH)
            state_payload = self._read_state()
            if (
                state_payload.get("status") not in {"healthy", "rolled_back"}
                or state_payload.get("report_pending") is not True
            ):
                return False
            self._validated_update_id(state_payload, required=True)
            self._validated_report_metadata(state_payload, required=True)
            version = str(
                state_payload.get(
                    "target_version"
                    if state_payload.get("status") == "rolled_back"
                    else "version"
                )
                or ""
            )
            if not version or len(version) > 100:
                raise OTAError("OTA healthy state contains an invalid version")
            return True

    def report_boot_success(self) -> bool:
        """Report a locally committed healthy release to its device manager.

        The local boot commit and rollback-backup removal happen before this
        network acknowledgement.  A transport failure therefore leaves the
        same event identity durably pending for a later retry and never rolls
        back an otherwise healthy release.
        """

        if not self.report_url:
            raise OTAError("OTA health report URL is not configured")
        with (self.state_dir / "update.lock").open("a+b") as lock_handle:
            fcntl.flock(lock_handle.fileno(), fcntl.LOCK_SH)
            state_payload = self._read_state()
            if (
                state_payload.get("status") not in {"healthy", "rolled_back"}
                or state_payload.get("report_pending") is not True
            ):
                return False
            local_status = str(state_payload.get("status"))
            update_id = self._validated_update_id(state_payload, required=True)
            report_metadata = self._validated_report_metadata(
                state_payload,
                required=True,
            )
            version = str(
                state_payload.get(
                    "target_version" if local_status == "rolled_back" else "version"
                )
                or ""
            )
            if not version or len(version) > 100:
                raise OTAError("OTA healthy state contains an invalid version")

        archive_sha256 = report_metadata["archive_sha256"]
        observed_at = report_metadata["report_observed_at"]
        base_event_id = report_metadata["report_event_id"]
        lifecycle_statuses = (
            ("rolled_back",)
            if local_status == "rolled_back"
            else ("locally_healthy", "health_report_pending", "healthy")
        )
        for lifecycle_status in lifecycle_statuses:
            event_id = (
                base_event_id
                if lifecycle_status == "healthy"
                else str(
                    uuid.uuid5(
                        uuid.UUID(base_event_id),
                        "labcobot-ota-lifecycle:{}".format(lifecycle_status),
                    )
                )
            )
            self._send_lifecycle_report(
                ota_status=lifecycle_status,
                event_id=event_id,
                version=version,
                archive_sha256=archive_sha256,
                observed_at=observed_at,
                failure_reason_code=(
                    str(
                        state_payload.get("failure_reason_code") or "boot_health_failed"
                    )
                    if lifecycle_status == "rolled_back"
                    else ""
                ),
            )

        with (self.state_dir / "update.lock").open("a+b") as lock_handle:
            fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX)
            current = self._read_state()
            if current.get("status") != local_status:
                raise OTAError("OTA state changed while reporting lifecycle health")
            if not hmac.compare_digest(
                self._validated_update_id(current, required=True),
                update_id,
            ):
                raise OTAError("OTA update identity changed while reporting health")
            current_report = self._validated_report_metadata(current, required=True)
            if any(
                not hmac.compare_digest(current_report[key], report_metadata[key])
                for key in (
                    "archive_sha256",
                    "report_event_id",
                    "report_observed_at",
                )
            ):
                raise OTAError("OTA report identity changed while reporting health")
            completed = dict(current)
            completed["report_pending"] = False
            completed["report_status"] = "reported"
            self._write_state(completed)
        return True

    def _send_lifecycle_report(
        self,
        *,
        ota_status: str,
        event_id: str,
        version: str,
        archive_sha256: str,
        observed_at: str,
        failure_reason_code: str = "",
    ) -> None:
        """Send one exact idempotent lifecycle event to Store Connect."""

        payload = {
            "schema_version": "labcobot.timeseries.v1",
            "event_id": event_id,
            "series": "robot.metrics",
            "observed_at": observed_at,
            "device": {"uid_device": self.uid_device},
            "attributes": {"source": "supraworx-managed-ota"},
            "values": {
                "ota_update_status": ota_status,
                "software_version": version,
                "archive_sha256": archive_sha256,
                "failure_reason_code": failure_reason_code,
            },
        }
        headers = dict(self._auth_headers())
        headers.update(
            {
                "Accept": "application/json",
                "Content-Type": "application/json",
                "Idempotency-Key": event_id,
            }
        )
        response = self.transport.request(
            "POST",
            self.report_url,
            headers=headers,
            body=json_body(payload),
            timeout=self.timeout,
            max_response_bytes=64 * 1024,
            allow_redirects=False,
        )
        expected_status = "recorded" if response.status_code in {200, 201} else None
        if response.status_code == 409:
            expected_status = "duplicate"
        if expected_status is None:
            raise OTAError(
                "OTA lifecycle report returned HTTP {}".format(response.status_code)
            )
        self._validate_report_response(
            response.json(),
            expected_status=expected_status,
            event_id=event_id,
            version=version,
            archive_sha256=archive_sha256,
            ota_status=ota_status,
            failure_reason_code=failure_reason_code,
        )

    def rollback_pending(self, *, failure_reason_code: str = "") -> bool:
        """Restore the previous release after a failed health check."""
        safe_failure_reason = (
            failure_reason_code
            if failure_reason_code in {"", "admission_changed"}
            else ""
        )
        state_payload = self._read_state()
        if state_payload.get("status") != "pending_restart":
            return False
        update_identity = self._preserved_update_identity(state_payload)
        release_identity = self._preserved_release_identity(
            state_payload,
            required=True,
        )
        boot_identity = self._preserved_boot_attempt(state_payload)
        backup = self._validated_backup_path(state_payload)
        if not backup.is_dir() or backup.is_symlink():
            raise OTAError("OTA rollback backup is missing or unsafe")
        self._assert_state_release_tree(backup, state_payload, "previous")
        failed = self._expected_failed_path()
        if _path_exists(failed):
            if failed.is_symlink() or not failed.is_dir():
                raise OTAError("OTA failed-release path is unsafe")
        if _path_exists(self.install_root):
            if self.install_root.is_symlink() or not self.install_root.is_dir():
                raise OTAError("OTA install root is unsafe")
            if _path_exists(failed):
                raise OTAError("OTA rollback filesystem state is ambiguous")
        elif not _path_exists(failed):
            raise OTAError("OTA pending release is missing")

        terminal_state = {
            "status": "rolled_back",
            "version": state_payload.get("previous_version", ""),
            "source_revision": state_payload.get("previous_source_revision", ""),
            "tree_sha256": state_payload.get("previous_tree_sha256", ""),
            **release_identity,
            **boot_identity,
            "backup_path": str(backup),
            "failed_path": str(failed),
            **update_identity,
            "report_pending": False,
        }
        if safe_failure_reason:
            terminal_state["failure_reason_code"] = safe_failure_reason
        # Record the rollback intent durably before either directory rename.
        # Every resulting filesystem combination can then be completed after
        # a sudden power loss without guessing which tree is authoritative.
        self._write_state(terminal_state)
        self._finish_rolled_back_terminal(terminal_state)
        return True

    def recover_interrupted_swap(self) -> bool:
        """Recover every crash state around the two directory renames."""
        state_payload = self._read_state()
        status = state_payload.get("status")
        if status in {"installing", "pending_restart", "healthy", "rolled_back"}:
            update_identity = self._preserved_update_identity(state_payload)
            release_identity = self._preserved_release_identity(
                state_payload,
                required=False,
            )
        else:
            update_identity = {}
            release_identity = {}
        if status == "healthy" and state_payload.get("backup_path"):
            return self._finish_healthy_terminal(state_payload)
        if (
            status == "rolled_back"
            and state_payload.get("backup_path")
            and state_payload.get("failed_path")
        ):
            return self._finish_rolled_back_terminal(state_payload)
        if status != "installing":
            return False
        backup = self._validated_backup_path(state_payload)
        failed = self._expected_failed_path()
        release_identity = self._preserved_release_identity(
            state_payload,
            required=True,
        )
        install_exists = _path_exists(self.install_root)
        backup_exists = _path_exists(backup)
        failed_exists = _path_exists(failed)
        if install_exists and (
            self.install_root.is_symlink() or not self.install_root.is_dir()
        ):
            raise OTAError("OTA install root is unsafe")
        if backup_exists and (backup.is_symlink() or not backup.is_dir()):
            raise OTAError("OTA rollback backup is unsafe")
        if failed_exists:
            if failed.is_symlink() or not failed.is_dir():
                raise OTAError("OTA failed-release path is unsafe")
            raise OTAError("OTA installing state has an unexpected failed-release tree")

        if not install_exists and backup_exists:
            self._assert_state_release_tree(backup, state_payload, "previous")
            _fsync_directory(self.install_root.parent)
            terminal_state = {
                "status": "rolled_back",
                "version": state_payload.get("previous_version", ""),
                **release_identity,
                "backup_path": str(backup),
                "failed_path": str(self._expected_failed_path()),
                **update_identity,
                "report_pending": False,
            }
            self._write_state(terminal_state)
            self._finish_rolled_back_terminal(terminal_state)
            return True
        if install_exists and backup_exists:
            # The new root rename completed, but power was lost before the
            # pending-restart state write. Preserve the rollback boundary.
            self._assert_state_release_tree(backup, state_payload, "previous")
            self._assert_state_release_tree(
                self.install_root,
                state_payload,
                "target",
            )
            _fsync_directory(self.install_root.parent)
            self._write_state(
                {
                    "status": "pending_restart",
                    "previous_version": state_payload.get("previous_version", ""),
                    "target_version": state_payload.get("target_version", ""),
                    **release_identity,
                    "backup_path": str(backup),
                    **update_identity,
                }
            )
            return True
        if install_exists and not backup_exists:
            # The first rename never completed (or was already restored).
            self._assert_state_release_tree(
                self.install_root,
                state_payload,
                "previous",
            )
            _fsync_directory(self.install_root.parent)
            self._write_state(
                {
                    "status": "rolled_back",
                    "version": state_payload.get("previous_version", ""),
                    **release_identity,
                    **update_identity,
                    "report_pending": False,
                }
            )
            return True
        raise OTAError("OTA install root and rollback backup are both missing")

    def _finish_healthy_terminal(self, state_payload: Mapping[str, Any]) -> bool:
        """Finish removal of a backup after a durable healthy transition."""

        backup = self._validated_backup_path(state_payload)
        if not _path_exists(self.install_root):
            raise OTAError("OTA healthy install root is missing")
        if self.install_root.is_symlink() or not self.install_root.is_dir():
            raise OTAError("OTA install root is unsafe")
        release_identity = self._preserved_release_identity(
            state_payload,
            required=False,
        )
        if release_identity:
            self._assert_state_release_tree(
                self.install_root,
                state_payload,
                "target",
            )
        if not _path_exists(backup):
            return False
        if backup.is_symlink() or not backup.is_dir():
            raise OTAError("OTA rollback backup is unsafe")
        shutil.rmtree(backup)
        _fsync_directory(self.install_root.parent)
        return True

    def _finish_rolled_back_terminal(
        self,
        state_payload: Mapping[str, Any],
    ) -> bool:
        """Complete a durably requested rollback from any rename boundary."""

        backup = self._validated_backup_path(state_payload)
        failed = self._validated_failed_path(state_payload)
        changed = False

        install_exists = _path_exists(self.install_root)
        backup_exists = _path_exists(backup)
        failed_exists = _path_exists(failed)
        for path, exists, error in (
            (self.install_root, install_exists, "OTA install root is unsafe"),
            (backup, backup_exists, "OTA rollback backup is unsafe"),
            (failed, failed_exists, "OTA failed-release path is unsafe"),
        ):
            if exists and (path.is_symlink() or not path.is_dir()):
                raise OTAError(error)

        if backup_exists:
            release_identity = self._preserved_release_identity(
                state_payload,
                required=False,
            )
            if release_identity:
                self._assert_state_release_tree(backup, state_payload, "previous")
            if install_exists and failed_exists:
                raise OTAError("OTA rollback filesystem state is ambiguous")
            if install_exists:
                os.replace(str(self.install_root), str(failed))
                _fsync_directory(self.install_root.parent)
                install_exists = False
                failed_exists = True
                changed = True
            os.replace(str(backup), str(self.install_root))
            _fsync_directory(self.install_root.parent)
            install_exists = True
            backup_exists = False
            changed = True

        if not install_exists:
            raise OTAError("OTA rolled-back install root is missing")
        release_identity = self._preserved_release_identity(
            state_payload,
            required=False,
        )
        if release_identity:
            self._assert_state_release_tree(
                self.install_root,
                state_payload,
                "previous",
            )
        if failed_exists:
            shutil.rmtree(failed)
            _fsync_directory(self.install_root.parent)
            changed = True
        return changed

    def _fetch_manifest(self) -> Optional[Mapping[str, Any]]:
        response = self.transport.request(
            "GET",
            self.metadata_url,
            headers=self._auth_headers(),
            timeout=self.timeout,
            max_response_bytes=256 * 1024,
            allow_redirects=False,
        )
        if response.status_code == 204:
            return None
        if response.status_code != 200:
            raise OTAError(
                "OTA metadata request returned HTTP {}".format(response.status_code)
            )
        payload = response.json()
        if not isinstance(payload, Mapping):
            raise OTAError("OTA manifest must be a JSON object")
        return payload

    def _verify_manifest(self, manifest: Mapping[str, Any]) -> None:
        required = {
            "version",
            "download_url",
            "sha256",
            "size",
            "archive_root",
            "source_revision",
            "signature",
        }
        if set(manifest) != required or any(
            manifest.get(key) in (None, "") for key in required
        ):
            raise OTAError("OTA manifest is missing signed release metadata")
        archive_root = manifest["archive_root"]
        if not isinstance(archive_root, str):
            raise OTAError("OTA manifest has an invalid archive root")
        try:
            validated_archive_root = _safe_relative_path(archive_root).as_posix()
        except OTAError as exc:
            raise OTAError("OTA manifest has an invalid archive root") from exc
        if (
            archive_root != validated_archive_root
            or "\\" in archive_root
            or "\x00" in archive_root
        ):
            raise OTAError("OTA manifest has an invalid archive root")

        download_url_value = manifest["download_url"]
        if not isinstance(download_url_value, str):
            raise OTAError("OTA manifest has an invalid download URL")
        download_url = download_url_value
        if not self.allow_insecure_http and not _is_https(download_url):
            raise OTAError("OTA download URL must use HTTPS")
        download_host = (urlparse(download_url).hostname or "").lower()
        if download_host not in self.allowed_download_hosts:
            raise OTAError("OTA download host is not allow-listed")
        digest_value = manifest["sha256"]
        if not isinstance(digest_value, str):
            raise OTAError("OTA manifest has an invalid SHA-256 digest")
        digest = digest_value
        if len(digest) != 64 or any(char not in "0123456789abcdef" for char in digest):
            raise OTAError("OTA manifest has an invalid SHA-256 digest")
        declared_size = manifest["size"]
        if not isinstance(declared_size, int) or isinstance(declared_size, bool):
            raise OTAError("OTA manifest has an invalid archive size")
        if declared_size <= 0 or declared_size > self.max_archive_bytes:
            raise OTAError("OTA archive exceeds configured maximum size")
        source_revision = _validated_source_revision(manifest["source_revision"])
        version = manifest["version"]
        if not isinstance(version, str) or not version or len(version) > 100:
            raise OTAError("OTA manifest contains an invalid version")

        signed_fields = {
            "archive_root": archive_root,
            "device_uid": self.uid_device,
            "download_url": download_url,
            "schema": OTA_MANIFEST_SCHEMA,
            "sha256": digest,
            "size": declared_size,
            "source_revision": source_revision,
            "version": version,
        }
        expected = hmac.new(
            self.signing_key,
            json_body(signed_fields),
            hashlib.sha256,
        ).hexdigest()
        provided_value = manifest["signature"]
        if not isinstance(provided_value, str):
            raise OTAError("OTA manifest signature verification failed")
        provided = provided_value
        if (
            len(provided) != 76
            or not provided.startswith("hmac-sha256:")
            or any(character not in "0123456789abcdef" for character in provided[12:])
            or not hmac.compare_digest(expected, provided[12:])
        ):
            raise OTAError("OTA manifest signature verification failed")

    def _download_stage_and_swap(
        self,
        manifest: Mapping[str, Any],
        *,
        admission_revalidator: Optional[Callable[[str], bool]] = None,
    ) -> None:
        download_url = str(manifest["download_url"])
        self._set_verification("verified", "checking")
        try:
            response = self.transport.request(
                "GET",
                download_url,
                headers=self._archive_headers(
                    download_url,
                    version=str(manifest["version"]),
                    source_revision=str(manifest["source_revision"]),
                ),
                timeout=self.timeout,
                max_response_bytes=self.max_archive_bytes,
                allow_redirects=False,
            )
            if response.status_code != 200:
                raise OTAError(
                    "OTA archive request returned HTTP {}".format(response.status_code)
                )
            expected_size = int(manifest["size"])
            if len(response.body) != expected_size:
                raise OTAError("OTA archive size does not match signed manifest")
            digest = hashlib.sha256(response.body).hexdigest()
            if not hmac.compare_digest(digest, str(manifest["sha256"]).lower()):
                raise OTAError("OTA archive checksum verification failed")
        except Exception:
            self._set_verification("verified", "failed")
            raise
        self._set_verification("verified", "verified")

        parent = self.install_root.parent
        stage = Path(tempfile.mkdtemp(prefix=".labcobot-ota-", dir=str(parent)))
        backup = self._expected_backup_path()
        update_id = secrets.token_hex(32)
        report_metadata = {
            "archive_sha256": str(manifest["sha256"]).lower(),
            "report_event_id": str(uuid.uuid4()),
            "report_observed_at": datetime.now(timezone.utc)
            .isoformat(timespec="seconds")
            .replace("+00:00", "Z"),
        }
        try:
            archive = stage / "release.zip"
            archive.write_bytes(response.body)
            extracted = stage / "extracted"
            extracted.mkdir()
            self._safe_extract(archive, extracted)
            archive_root = str(manifest.get("archive_root") or "").strip("/")
            source = (
                extracted / _safe_relative_path(archive_root)
                if archive_root
                else extracted
            )
            if not source.is_dir() or any(
                not (source / required).is_file() for required in self.required_paths
            ):
                raise OTAError("OTA archive does not contain the required application")
            self._verify_release_source_revision(
                source,
                str(manifest["source_revision"]),
            )
            self._verify_dependency_compatibility(source)
            self._preserve_local_files(source)
            self._write_release_version(source, str(manifest["version"]))
            previous_identity = _release_identity(self.install_root)
            target_identity = _release_identity(source)
            if not hmac.compare_digest(
                previous_identity["version"],
                self.current_version,
            ):
                raise OTAError(
                    "Installed release version does not match OTA runtime state"
                )
            if not hmac.compare_digest(
                target_identity["version"],
                str(manifest["version"]),
            ) or not hmac.compare_digest(
                target_identity["source_revision"],
                str(manifest["source_revision"]),
            ):
                raise OTAError("Staged release identity does not match signed manifest")
            release_identity = {
                "previous_version": previous_identity["version"],
                "previous_source_revision": previous_identity["source_revision"],
                "previous_tree_sha256": previous_identity["tree_sha256"],
                "target_version": target_identity["version"],
                "target_source_revision": target_identity["source_revision"],
                "target_tree_sha256": target_identity["tree_sha256"],
            }
            if _path_exists(backup):
                raise OTAError(
                    "Previous OTA backup has not been committed or rolled back"
                )

            _revalidate_ota_admission(admission_revalidator, "after_staging")
            _revalidate_ota_admission(
                admission_revalidator,
                "before_installing_state",
            )

            self._write_state(
                {
                    "status": "installing",
                    **release_identity,
                    "backup_path": str(backup),
                    "update_id": update_id,
                    **report_metadata,
                    "report_pending": True,
                }
            )
            try:
                _revalidate_ota_admission(
                    admission_revalidator,
                    "before_first_rename",
                )
            except OTAAdmissionChanged:
                self.recover_interrupted_swap()
                raise
            os.replace(str(self.install_root), str(backup))
            _fsync_directory(parent)
            try:
                os.replace(str(source), str(self.install_root))
            except BaseException:
                terminal_state = {
                    "status": "rolled_back",
                    "version": self.current_version,
                    "source_revision": previous_identity["source_revision"],
                    "tree_sha256": previous_identity["tree_sha256"],
                    **release_identity,
                    "backup_path": str(backup),
                    "failed_path": str(self._expected_failed_path()),
                    "update_id": update_id,
                    **report_metadata,
                    "report_pending": False,
                }
                self._write_state(terminal_state)
                self._finish_rolled_back_terminal(terminal_state)
                raise
            # A power loss between the second rename and its durability barrier
            # must leave the durable ``installing`` record intact for the boot
            # supervisor.  Do not catch BaseException across this fsync.
            _fsync_directory(parent)
            try:
                _revalidate_ota_admission(
                    admission_revalidator,
                    "after_second_rename",
                )
            except OTAAdmissionChanged:
                terminal_state = {
                    "status": "rolled_back",
                    "version": self.current_version,
                    "source_revision": previous_identity["source_revision"],
                    "tree_sha256": previous_identity["tree_sha256"],
                    **release_identity,
                    "backup_path": str(backup),
                    "failed_path": str(self._expected_failed_path()),
                    "update_id": update_id,
                    **report_metadata,
                    "report_pending": False,
                    "failure_reason_code": "admission_changed",
                }
                self._write_state(terminal_state)
                self._finish_rolled_back_terminal(terminal_state)
                raise
            self._write_state(
                {
                    "status": "pending_restart",
                    **release_identity,
                    "backup_path": str(backup),
                    "update_id": update_id,
                    **report_metadata,
                    "report_pending": True,
                }
            )
        finally:
            if stage.exists():
                shutil.rmtree(stage)

    def _safe_extract(self, archive: Path, destination: Path) -> None:
        with zipfile.ZipFile(str(archive)) as zip_handle:
            members = zip_handle.infolist()
            if len(members) > self.max_archive_entries:
                raise OTAError("OTA archive contains too many entries")
            total_size = 0
            for member in members:
                relative = _safe_relative_path(member.filename)
                mode = member.external_attr >> 16
                if stat.S_ISLNK(mode):
                    raise OTAError("OTA archive contains a symbolic link")
                total_size += int(member.file_size)
                if total_size > self.max_uncompressed_bytes:
                    raise OTAError("OTA archive expands beyond configured maximum")
                if (
                    member.file_size > 1024 * 1024
                    and member.compress_size > 0
                    and member.file_size > member.compress_size * 250
                ):
                    raise OTAError("OTA archive has an unsafe compression ratio")
                target = destination / relative
                if member.is_dir():
                    target.mkdir(parents=True, exist_ok=True)
                    continue
                target.parent.mkdir(parents=True, exist_ok=True)
                with zip_handle.open(member) as source, target.open("xb") as output:
                    shutil.copyfileobj(source, output, length=1024 * 1024)

    def _preserve_local_files(self, new_root: Path) -> None:
        for relative in self.preserve_paths:
            source = self.install_root / relative
            target = new_root / relative
            if not source.is_file():
                continue
            target.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(str(source), str(target))

    def _verify_dependency_compatibility(self, new_root: Path) -> None:
        """Reject releases that cannot use the externally managed virtualenv.

        Raspberry Pi deployments keep the virtual environment outside the
        atomically replaced application directory. Updating dependency
        declarations without rebuilding that environment would make rollback
        unreliable and can prevent the new process from importing at all.
        """

        try:
            current_requirements = _requirement_declaration_files(self.install_root)
            candidate_requirements = _requirement_declaration_files(new_root)

            current_root = self.install_root / "requirements.txt"
            candidate_root = new_root / "requirements.txt"
            if _path_exists(current_root) or _path_exists(candidate_root):
                if not _path_exists(current_root) or not _path_exists(candidate_root):
                    raise ValueError("requirements.txt is missing")
                # Validate include syntax, containment, cycles, and symlinks in
                # addition to comparing every declaration. Gateway installs
                # requirements/gateway.txt directly, so following only the
                # root production include graph is not sufficient.
                _requirement_file_closure(self.install_root)
                _requirement_file_closure(new_root)
        except (OSError, UnicodeError, ValueError) as exc:
            raise OTAError(
                "OTA dependency changes require a reviewed installer deployment"
            ) from exc

        if current_requirements.keys() != candidate_requirements.keys() or any(
            not hmac.compare_digest(
                current_requirements[relative],
                candidate_requirements[relative],
            )
            for relative in current_requirements
        ):
            raise OTAError(
                "OTA dependency changes require a reviewed installer deployment"
            )

    def _verify_release_source_revision(
        self,
        new_root: Path,
        expected_revision: str,
    ) -> None:
        """Bind extracted bytes to the signed Git source identity."""

        revision = _validated_source_revision(expected_revision)
        marker = Path(new_root) / SOURCE_REVISION_FILENAME
        try:
            metadata = marker.lstat()
            contents = marker.read_bytes()
        except OSError as exc:
            raise OTAError(
                "OTA source revision marker is missing or unreadable"
            ) from exc
        if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode):
            raise OTAError("OTA source revision marker is not a regular file")
        expected = (revision + "\n").encode("ascii")
        if not hmac.compare_digest(contents, expected):
            raise OTAError("OTA source revision marker does not match manifest")

    def _write_release_version(self, new_root: Path, version: str) -> None:
        """Persist the signed manifest version outside preserved local config."""

        descriptor, temporary_name = tempfile.mkstemp(
            prefix=".release-version.",
            dir=str(new_root),
        )
        temporary = Path(temporary_name)
        marker = new_root / RELEASE_VERSION_FILENAME
        try:
            with os.fdopen(descriptor, "wb") as handle:
                handle.write((str(version) + "\n").encode("ascii"))
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(str(temporary), str(marker))
            _fsync_directory(new_root)
        finally:
            if temporary.exists():
                temporary.unlink()

    def _auth_headers(self) -> Mapping[str, str]:
        return {
            "Accept": "application/json",
            "User-Agent": "LabCobot-OTA/1",
            "X-Device-Id": self.uid_device,
            "X-Device-Secret": self.api_secret,
            "X-Supraworx-OTA-Manifest-Contract": OTA_MANIFEST_SCHEMA,
            "X-Supraworx-OTA-Report-Contract": OTA_REPORT_CONTRACT,
        }

    def _archive_headers(
        self,
        download_url: str,
        *,
        version: str = "",
        source_revision: str = "",
    ) -> Mapping[str, str]:
        headers = {
            "Accept": "application/octet-stream",
            "User-Agent": "LabCobot-OTA/1",
        }
        if not self._is_authenticated_mint_archive_url(
            download_url,
            version=version,
            source_revision=source_revision,
        ):
            # A signed manifest may still name a public allow-listed archive,
            # but device credentials are confined to the exact Mint endpoint.
            return headers
        headers.update(
            {
                "X-Device-Id": self.uid_device,
                "X-Device-Secret": self.api_secret,
                "X-Supraworx-OTA-Manifest-Contract": OTA_MANIFEST_SCHEMA,
            }
        )
        return headers

    def _is_authenticated_mint_archive_url(
        self,
        download_url: str,
        *,
        version: str,
        source_revision: str,
    ) -> bool:
        """Confine device authentication to one immutable Mint artifact URL."""

        device_uid = str(self.uid_device)
        version = str(version)
        source_revision = str(source_revision)
        if (
            _MINT_DEVICE_UID_RE.fullmatch(device_uid) is None
            or _MINT_VERSION_RE.fullmatch(version) is None
            or _MINT_SOURCE_REVISION_RE.fullmatch(source_revision) is None
        ):
            return False

        archive_host = _standard_https_host(download_url)
        if archive_host is None:
            return False
        configured_urls = [self.metadata_url]
        if self.report_url:
            configured_urls.append(self.report_url)
        if any(
            _standard_https_host(configured_url) != archive_host
            for configured_url in configured_urls
        ):
            return False

        raw_url = str(download_url)
        try:
            parsed = urlparse(raw_url)
        except ValueError:
            return False
        expected_path = (
            "/api-gateway/apis/iotmanage/devices/{}/ota-artifacts/" "{}/{}/archive/"
        ).format(device_uid, version, source_revision)
        return (
            parsed.path == expected_path
            and not parsed.params
            and not parsed.query
            and not parsed.fragment
        )

    def _expected_backup_path(self) -> Path:
        return self.install_root.parent / (
            ".{}.ota-previous".format(self.install_root.name)
        )

    def _expected_failed_path(self) -> Path:
        return self.install_root.parent / (
            ".{}.ota-failed".format(self.install_root.name)
        )

    @staticmethod
    def _validated_update_id_value(value: Any, name: str) -> str:
        if (
            not isinstance(value, str)
            or len(value) != 64
            or any(character not in "0123456789abcdef" for character in value)
        ):
            raise OTAError("{} is invalid".format(name))
        return value

    @staticmethod
    def _validate_single_boot_attempt(state_payload: Mapping[str, Any]) -> None:
        count = state_payload.get("boot_attempt_count")
        if isinstance(count, bool) or not isinstance(count, int) or count != 1:
            raise OTAError("OTA pending state has an invalid boot-attempt count")

    @classmethod
    def _preserved_boot_attempt(
        cls,
        state_payload: Mapping[str, Any],
    ) -> Mapping[str, Any]:
        attempted = state_payload.get("boot_attempted")
        count = state_payload.get("boot_attempt_count")
        if attempted is None and count is None:
            return {}
        if attempted is True:
            if count is None:
                # A pre-v2 supervisor used the same durable boolean but did
                # not record the diagnostic count. It is still exactly one
                # attempt and remains rollback-only, never health-committable.
                return {"boot_attempted": True, "boot_attempt_count": 1}
            cls._validate_single_boot_attempt(state_payload)
            return {"boot_attempted": True, "boot_attempt_count": 1}
        if attempted is False and count == 0 and not isinstance(count, bool):
            return {"boot_attempted": False, "boot_attempt_count": 0}
        raise OTAError("OTA state has an invalid boot-attempt audit")

    @classmethod
    def _validated_update_id(
        cls,
        state_payload: Mapping[str, Any],
        *,
        required: bool,
    ) -> Optional[str]:
        if "update_id" not in state_payload:
            if required:
                raise OTAError("OTA pending state is missing its update identity")
            return None
        value = state_payload["update_id"]
        return cls._validated_update_id_value(value, "OTA state update identity")

    @classmethod
    def _preserved_update_identity(
        cls,
        state_payload: Mapping[str, Any],
    ) -> Mapping[str, str]:
        """Validate and copy an identity without upgrading legacy state."""

        update_id = cls._validated_update_id(state_payload, required=False)
        if update_id is None:
            return {}
        result: dict[str, Any] = {"update_id": update_id}
        report_metadata = cls._validated_report_metadata(
            state_payload,
            required=False,
        )
        result.update(report_metadata)
        if report_metadata:
            result["report_pending"] = state_payload.get("report_pending") is True
        return result

    @classmethod
    def _preserved_release_identity(
        cls,
        state_payload: Mapping[str, Any],
        *,
        required: bool,
    ) -> Mapping[str, str]:
        """Validate and copy exact previous/target release identities."""

        keys = (
            "previous_version",
            "previous_source_revision",
            "previous_tree_sha256",
            "target_version",
            "target_source_revision",
            "target_tree_sha256",
        )
        present = [key in state_payload for key in keys]
        extended_present = any(present[index] for index in (1, 2, 4, 5))
        if not required and not extended_present:
            return {}
        if not all(present):
            raise OTAError("OTA state has an incomplete release identity")
        previous = cls._validated_release_identity(state_payload, "previous")
        target = cls._validated_release_identity(state_payload, "target")
        return {
            "previous_version": previous["version"],
            "previous_source_revision": previous["source_revision"],
            "previous_tree_sha256": previous["tree_sha256"],
            "target_version": target["version"],
            "target_source_revision": target["source_revision"],
            "target_tree_sha256": target["tree_sha256"],
        }

    @staticmethod
    def _validated_release_identity(
        state_payload: Mapping[str, Any],
        prefix: str,
    ) -> Mapping[str, str]:
        version = state_payload.get(prefix + "_version")
        source_revision = state_payload.get(prefix + "_source_revision")
        tree_sha256 = state_payload.get(prefix + "_tree_sha256")
        if (
            not isinstance(version, str)
            or not 1 <= len(version) <= 100
            or version != version.strip()
            or any(
                ord(character) < 32 or ord(character) == 127 for character in version
            )
        ):
            raise OTAError("OTA state {} version is invalid".format(prefix))
        if not _is_canonical_source_revision(source_revision):
            raise OTAError("OTA state {} source revision is invalid".format(prefix))
        if (
            not isinstance(tree_sha256, str)
            or len(tree_sha256) != 64
            or any(character not in "0123456789abcdef" for character in tree_sha256)
        ):
            raise OTAError("OTA state {} tree digest is invalid".format(prefix))
        return {
            "version": version,
            "source_revision": str(source_revision),
            "tree_sha256": tree_sha256,
        }

    @classmethod
    def _assert_state_release_tree(
        cls,
        root: Path,
        state_payload: Mapping[str, Any],
        prefix: str,
    ) -> None:
        identity = cls._validated_release_identity(state_payload, prefix)
        actual = _release_identity(root)
        for key in ("version", "source_revision", "tree_sha256"):
            if not hmac.compare_digest(actual[key], identity[key]):
                label = "tree digest" if key == "tree_sha256" else key.replace("_", " ")
                raise OTAError(
                    "OTA {} release {} does not match durable state".format(
                        prefix,
                        label,
                    )
                )

    @staticmethod
    def _validated_report_metadata(
        state_payload: Mapping[str, Any],
        *,
        required: bool,
    ) -> Mapping[str, str]:
        digest = state_payload.get("archive_sha256")
        event_id = state_payload.get("report_event_id")
        observed_at = state_payload.get("report_observed_at")
        if digest is None and event_id is None and observed_at is None and not required:
            return {}
        if (
            not isinstance(digest, str)
            or len(digest) != 64
            or any(character not in "0123456789abcdef" for character in digest)
        ):
            raise OTAError("OTA state archive digest is invalid")
        if not isinstance(event_id, str):
            raise OTAError("OTA state report identity is invalid")
        try:
            parsed_event = uuid.UUID(event_id)
        except ValueError as exc:
            raise OTAError("OTA state report identity is invalid") from exc
        if str(parsed_event) != event_id:
            raise OTAError("OTA state report identity is invalid")
        if not isinstance(observed_at, str):
            raise OTAError("OTA state report timestamp is invalid")
        try:
            parsed_timestamp = datetime.fromisoformat(
                observed_at.replace("Z", "+00:00")
            )
        except ValueError as exc:
            raise OTAError("OTA state report timestamp is invalid") from exc
        if parsed_timestamp.tzinfo is None:
            raise OTAError("OTA state report timestamp is invalid")
        return {
            "archive_sha256": digest,
            "report_event_id": event_id,
            "report_observed_at": observed_at,
        }

    @staticmethod
    def _validate_report_response(
        payload: Any,
        *,
        expected_status: str,
        event_id: str,
        version: str,
        archive_sha256: str,
        ota_status: str,
        failure_reason_code: str,
    ) -> None:
        expected_fields = {
            "status",
            "event_id",
            "ota_update_status",
            "software_version",
            "archive_sha256",
            "failure_reason_code",
            "lifecycle_state",
            "target_matches",
            "deployment_id",
        }
        if (
            not isinstance(payload, Mapping)
            or set(payload) != expected_fields
            or payload.get("status") != expected_status
            or payload.get("event_id") != event_id
            or payload.get("ota_update_status") != ota_status
            or payload.get("software_version") != version
            or payload.get("archive_sha256") != archive_sha256
            or payload.get("failure_reason_code") != failure_reason_code
            or payload.get("lifecycle_state")
            != ("deployment_healthy" if ota_status == "healthy" else ota_status)
            or payload.get("target_matches") is not True
            or not isinstance(payload.get("deployment_id"), int)
            or isinstance(payload.get("deployment_id"), bool)
            or payload.get("deployment_id", 0) < 1
        ):
            raise OTAError("OTA lifecycle report response is invalid")

    def _validated_backup_path(self, state_payload: Mapping[str, Any]) -> Path:
        expected = self._expected_backup_path()
        recorded = str(state_payload.get("backup_path") or "")
        if not recorded or Path(recorded) != expected:
            raise OTAError("OTA state contains an invalid rollback path")
        return expected

    def _validated_failed_path(self, state_payload: Mapping[str, Any]) -> Path:
        expected = self._expected_failed_path()
        recorded = str(state_payload.get("failed_path") or "")
        if not recorded or Path(recorded) != expected:
            raise OTAError("OTA state contains an invalid failed-release path")
        return expected

    def _read_state(self) -> Mapping[str, Any]:
        path = self.state_dir / "state.json"
        try:
            payload = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            return {}
        return payload if isinstance(payload, Mapping) else {}

    def _write_state(self, payload: Mapping[str, Any]) -> None:
        path = self.state_dir / "state.json"
        descriptor, temporary_name = tempfile.mkstemp(
            prefix=".state.json.",
            dir=str(self.state_dir),
        )
        temporary = Path(temporary_name)
        try:
            with os.fdopen(descriptor, "wb") as handle:
                handle.write(json_body(payload))
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(str(temporary), str(path))
            _fsync_directory(self.state_dir)
        finally:
            if temporary.exists():
                temporary.unlink()


def _release_identity(root: Path) -> Mapping[str, str]:
    release = Path(root)
    version = _read_release_marker(release, RELEASE_VERSION_FILENAME, 100)
    source_revision = _read_release_marker(
        release,
        SOURCE_REVISION_FILENAME,
        64,
    )
    if not _is_canonical_source_revision(source_revision):
        raise OTAError("OTA release source revision marker is invalid")
    return {
        "version": version,
        "source_revision": source_revision,
        "tree_sha256": _release_tree_sha256(release),
    }


def _read_release_marker(root: Path, filename: str, maximum: int) -> str:
    marker = Path(root) / filename
    try:
        metadata = marker.lstat()
        contents = marker.read_bytes()
    except OSError as exc:
        raise OTAError("OTA release identity marker is missing") from exc
    if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode):
        raise OTAError("OTA release identity marker is unsafe")
    if len(contents) > maximum + 1 or not contents.endswith(b"\n"):
        raise OTAError("OTA release identity marker is invalid")
    try:
        value = contents[:-1].decode("ascii")
    except UnicodeDecodeError as exc:
        raise OTAError("OTA release identity marker is invalid") from exc
    if not value or "\n" in value or "\r" in value:
        raise OTAError("OTA release identity marker is invalid")
    return value


def _release_tree_sha256(root: Path) -> str:
    release = Path(root)
    try:
        root_metadata = release.lstat()
    except OSError as exc:
        raise OTAError("OTA release tree is missing or unreadable") from exc
    if stat.S_ISLNK(root_metadata.st_mode) or not stat.S_ISDIR(root_metadata.st_mode):
        raise OTAError("OTA release tree is unsafe")
    try:
        entries = sorted(
            release.rglob("*"),
            key=lambda entry: entry.relative_to(release).as_posix(),
        )
    except OSError as exc:
        raise OTAError("OTA release tree cannot be enumerated") from exc
    if len(entries) > _MAX_RELEASE_TREE_ENTRIES:
        raise OTAError("OTA release tree contains too many entries")

    digest = hashlib.sha256()
    digest.update(_RELEASE_TREE_DIGEST_DOMAIN)
    total_bytes = 0
    for entry in entries:
        relative = entry.relative_to(release).as_posix().encode("utf-8")
        if not relative or len(relative) > _MAX_RELEASE_RELATIVE_PATH_BYTES:
            raise OTAError("OTA release tree contains an unsafe path")
        try:
            metadata = entry.lstat()
        except OSError as exc:
            raise OTAError("OTA release tree entry is unreadable") from exc
        mode = stat.S_IMODE(metadata.st_mode)
        if stat.S_ISDIR(metadata.st_mode):
            _update_release_tree_digest(digest, b"D", relative, mode, 0)
            continue
        if not stat.S_ISREG(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode):
            raise OTAError("OTA release tree contains a non-regular entry")
        total_bytes += int(metadata.st_size)
        if total_bytes > _MAX_RELEASE_TREE_BYTES:
            raise OTAError("OTA release tree is too large")
        _update_release_tree_digest(
            digest,
            b"F",
            relative,
            mode,
            int(metadata.st_size),
        )
        try:
            with entry.open("rb") as handle:
                while True:
                    chunk = handle.read(1024 * 1024)
                    if not chunk:
                        break
                    digest.update(chunk)
        except OSError as exc:
            raise OTAError("OTA release tree file is unreadable") from exc
    return digest.hexdigest()


def _update_release_tree_digest(
    digest: Any,
    kind: bytes,
    relative: bytes,
    mode: int,
    size: int,
) -> None:
    digest.update(kind)
    digest.update(len(relative).to_bytes(4, "big"))
    digest.update(relative)
    digest.update(mode.to_bytes(4, "big"))
    digest.update(size.to_bytes(8, "big"))


def _safe_relative_path(value: str) -> Path:
    posix_path = PurePosixPath(str(value).replace("\\", "/"))
    if posix_path.is_absolute() or not posix_path.parts:
        raise OTAError("OTA path must be relative")
    if any(part in {"", ".", ".."} for part in posix_path.parts):
        raise OTAError("OTA path contains an unsafe segment")
    return Path(*posix_path.parts)


def _validated_source_revision(value: Any) -> str:
    if not _is_canonical_source_revision(value):
        raise OTAError("OTA manifest has an invalid source revision")
    return str(value)


def _is_canonical_source_revision(value: Any) -> bool:
    return (
        isinstance(value, str)
        and len(value) in {40, 64}
        and all(character in "0123456789abcdef" for character in value)
    )


def _requirement_file_closure(root: Path) -> Mapping[str, bytes]:
    """Return every local requirement file reachable from requirements.txt.

    Requirement includes are resolved relative to the file containing the
    directive, matching pip's behavior. Every path must stay beneath ``root``
    and every path component must be a real directory or regular file rather
    than a symbolic link.
    """

    root = Path(root)
    files: dict[str, bytes] = {}
    visiting: set[str] = set()

    def visit(relative: Path) -> None:
        key = relative.as_posix()
        if key in visiting:
            raise ValueError("requirement include cycle")
        if key in files:
            return

        contents = _read_requirement_file(root, relative)
        visiting.add(key)
        try:
            for include in _requirement_includes(contents):
                visit(_requirement_include_path(relative.parent, include))
        finally:
            visiting.remove(key)
        files[key] = contents

    visit(Path("requirements.txt"))
    return files


def _requirement_declaration_files(root: Path) -> Mapping[str, bytes]:
    """Return every dependency declaration shipped by a deployment tree.

    The production and gateway installers select different requirement entry
    points. Compare all root ``requirements*.txt`` files and every recursive
    ``requirements/**/*.txt`` file so an OTA cannot silently change the
    externally managed virtual environment's declaration through an entry
    point that is not reachable from the root production file.
    """

    root = Path(root)
    relatives = []
    for entry in root.iterdir():
        if entry.name.startswith("requirements") and entry.name.endswith(".txt"):
            relatives.append(Path(entry.name))

    requirements_dir = root / "requirements"
    if _path_exists(requirements_dir):
        metadata = requirements_dir.lstat()
        if stat.S_ISLNK(metadata.st_mode):
            raise ValueError("symbolic link in requirement path")
        if not stat.S_ISDIR(metadata.st_mode):
            raise ValueError("requirements path is not a directory")
        for current, directory_names, file_names in os.walk(
            str(requirements_dir),
            topdown=True,
            followlinks=False,
        ):
            directory_names.sort()
            file_names.sort()
            current_path = Path(current)
            for directory_name in directory_names:
                directory = current_path / directory_name
                directory_metadata = directory.lstat()
                if stat.S_ISLNK(directory_metadata.st_mode):
                    raise ValueError("symbolic link in requirement path")
                if not stat.S_ISDIR(directory_metadata.st_mode):
                    raise ValueError("requirement parent is not a directory")
            for file_name in file_names:
                if not file_name.endswith(".txt"):
                    continue
                relatives.append((current_path / file_name).relative_to(root))

    return {
        relative.as_posix(): _read_requirement_file(root, relative)
        for relative in sorted(set(relatives), key=lambda item: item.as_posix())
    }


def _read_requirement_file(root: Path, relative: Path) -> bytes:
    current = root
    parts = relative.parts
    if not parts:
        raise ValueError("empty requirement path")
    for index, part in enumerate(parts):
        current = current / part
        metadata = current.lstat()
        if stat.S_ISLNK(metadata.st_mode):
            raise ValueError("symbolic link in requirement path")
        if index < len(parts) - 1:
            if not stat.S_ISDIR(metadata.st_mode):
                raise ValueError("requirement parent is not a directory")
        elif not stat.S_ISREG(metadata.st_mode):
            raise ValueError("requirement path is not a regular file")
    return current.read_bytes()


def _requirement_includes(contents: bytes) -> Sequence[str]:
    try:
        text = contents.decode("utf-8")
    except UnicodeDecodeError as exc:
        raise UnicodeError("requirement file is not UTF-8") from exc

    includes = []
    for raw_line in text.splitlines():
        try:
            tokens = shlex.split(raw_line, comments=True, posix=True)
        except ValueError as exc:
            raise ValueError("invalid requirement include syntax") from exc
        if not tokens:
            continue
        first = tokens[0]
        include = ""
        if first in {"-r", "--requirement"}:
            if len(tokens) != 2:
                raise ValueError("invalid requirement include syntax")
            include = tokens[1]
        elif first.startswith("--requirement="):
            if len(tokens) != 1:
                raise ValueError("invalid requirement include syntax")
            include = first.split("=", 1)[1]
        elif first.startswith("-r") and first != "-r":
            if len(tokens) != 1:
                raise ValueError("invalid requirement include syntax")
            include = first[2:]
        if include:
            includes.append(include)
        elif first in {"-r", "--requirement", "--requirement="}:
            raise ValueError("requirement include path is empty")
    return tuple(includes)


def _requirement_include_path(parent: Path, value: str) -> Path:
    raw = str(value)
    if not raw or "\\" in raw or "\x00" in raw:
        raise ValueError("unsafe requirement include path")
    posix_path = PurePosixPath(raw)
    raw_parts = raw.split("/")
    if (
        posix_path.is_absolute()
        or not posix_path.parts
        or any(part in {"", ".", ".."} for part in raw_parts)
        or "://" in raw
    ):
        raise ValueError("unsafe requirement include path")
    return parent.joinpath(*posix_path.parts)


def _is_https(url: str) -> bool:
    return urlparse(str(url)).scheme.lower() == "https"


def _standard_https_host(url: str) -> Optional[str]:
    """Return a credential-free HTTPS host only when its port is exactly 443."""

    raw_url = str(url)
    if (
        raw_url != raw_url.strip()
        or any(character.isspace() for character in raw_url)
        or any(ord(character) < 32 or ord(character) == 127 for character in raw_url)
    ):
        return None
    try:
        parsed = urlparse(raw_url)
        host = (parsed.hostname or "").lower()
        port = parsed.port
    except ValueError:
        return None
    if (
        parsed.scheme.lower() != "https"
        or not host
        or parsed.username is not None
        or parsed.password is not None
        or port not in {None, 443}
    ):
        return None
    return host


def _same_origin(first: str, second: str) -> bool:
    """Return whether two HTTP(S) URLs share scheme, host, and effective port."""

    def origin(value: str) -> tuple[str, str, Optional[int]]:
        parsed = urlparse(str(value))
        scheme = parsed.scheme.lower()
        host = (parsed.hostname or "").lower()
        try:
            port = parsed.port
        except ValueError:
            return ("", "", None)
        if port is None:
            port = 443 if scheme == "https" else 80 if scheme == "http" else None
        return (scheme, host, port)

    first_origin = origin(first)
    return bool(first_origin[0] and first_origin[1]) and first_origin == origin(second)


def _path_exists(path: Path) -> bool:
    """Return true for regular paths and broken symbolic links."""

    return os.path.lexists(str(path))


def _validated_health_generation(value: Any) -> Optional[int]:
    if value is None:
        return None
    if isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= 10**9:
        raise OTAError("OTA boot-health generation is invalid")
    return value


def _fsync_directory(path: Path) -> None:
    """Persist directory-entry changes across a sudden power loss."""

    flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
    descriptor = os.open(str(path), flags)
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)
