#!/usr/bin/env python3
"""
Privileged root helper for Synex Snapshots.

Only explicitly supported actions are allowed.
Arbitrary command execution is never accepted.
"""

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timedelta
import calendar
import contextlib
import fcntl
import hashlib
import io
import json
import os
from pathlib import Path
import re
import shutil
import stat
import subprocess
import sys
import tempfile
import uuid


BTRFS_COMMAND = Path("/usr/bin/btrfs")
FINDMNT_COMMAND = Path("/usr/bin/findmnt")
MOUNT_COMMAND = Path("/usr/bin/mount")
UMOUNT_COMMAND = Path("/usr/bin/umount")
SYSTEMCTL_COMMAND = Path("/usr/bin/systemctl")
TAR_COMMAND = Path("/usr/bin/tar")
ZSTD_COMMAND = Path("/usr/bin/zstd")

FSTAB_PATH = Path("/etc/fstab")

SNAPSHOT_SUBVOLUME = "@snapshots"
SNAPSHOT_MOUNTPOINT = Path("/.snapshots")

TOP_LEVEL_SUBVOLUME_ID = 5

SNAPSHOT_PURPOSES = {
    "manual",
    "auto",
    "pre-restore",
}

RESTORE_INTERNAL_PREFIX = ".synex-snapshots-restore-"
RESTORE_STATE_PATH = SNAPSHOT_MOUNTPOINT / ".restore-state.json"
BOOT_ID_PATH = Path("/proc/sys/kernel/random/boot_id")
HELPER_LOCK_PATH = Path("/run/synex-snapshots-root-helper.lock")

AUTOMATION_CONFIG_DIR = Path("/etc/synex-snapshots")
AUTOMATION_CONFIG_PATH = AUTOMATION_CONFIG_DIR / "automation.conf"
AUTOMATION_FREQUENCIES = {
    "hourly",
    "daily",
    "weekly",
    "monthly",
}
AUTOMATION_DEFAULT_ENABLED = False
AUTOMATION_DEFAULT_FREQUENCY = "daily"
AUTOMATION_DEFAULT_RETENTION = 5
AUTOMATION_MAX_RETENTION = 999


_SUBVOLUME_LIST_LINE_RE = re.compile(
    r"^ID\s+(?P<id>\d+)\s+"
    r"gen\s+\d+\s+"
    r"parent\s+(?P<parent>\d+)\s+"
    r"top level\s+\d+\s+"
    r"parent_uuid\s+(?P<parent_uuid>\S+)\s+"
    r"uuid\s+(?P<uuid>\S+)\s+"
    r"path\s+(?P<path>.+)$"
)


ALLOWED_ACTIONS = {
    "scan-subvolumes",
    "configure-snapshot-storage",
    "create-single-snapshot",
    "create-full-snapshot",
    "delete-snapshot-set",
    "preflight-restore",
    "restore-snapshot-set",
    "restore-status",
    "finalize-restore",
    "set-automation-config",
    "automation-run",
}


class HelperError(RuntimeError):
    """
    Controlled error raised by the privileged helper.
    """


@dataclass(frozen=True)
class RootFilesystemInfo:
    """
    Information about the running Btrfs root filesystem.
    """

    source: str
    device: str
    uuid: str
    fsroot: str


@dataclass(frozen=True)
class RestoreContext:
    """
    Restore identity derived from the running root and persisted layout.

    mode is either ``normal`` or ``snapshot-boot``. In snapshot-boot
    mode the running root is historical, while canonical_root_name is
    the writable top-level subvolume configured for /.
    """

    mode: str
    running_root_name: str
    canonical_root_name: str
    booted_set_name: str | None = None


@dataclass(frozen=True)
class FstabEntry:
    """
    Parsed active /etc/fstab entry.
    """

    source: str
    target: str
    fstype: str
    options: str
    dump: str
    passno: str


@dataclass(frozen=True)
class MountState:
    """
    State of /.snapshots in the running mount table.
    """

    mounted: bool
    correct: bool
    target: str | None = None
    fstype: str | None = None
    uuid: str | None = None
    fsroot: str | None = None


@dataclass(frozen=True)
class SubvolumeRecord:
    """
    Btrfs subvolume metadata required by snapshot operations.
    """

    subvolume_id: int
    parent_id: int
    name: str
    uuid: str
    parent_uuid: str | None = None


@dataclass(frozen=True)
class BootFilesystemInfo:
    """
    Runtime information about /boot coverage.

    separate=False means /boot is part of the root filesystem and is
    therefore covered by the root Btrfs snapshot itself.
    """

    separate: bool
    source: str | None = None
    fstype: str | None = None
    uuid: str | None = None
    mountpoint: str = "/boot"


@dataclass(frozen=True)
class AutomationConfig:
    """Validated automatic snapshot configuration."""

    enabled: bool
    frequency: str
    retention: int


@dataclass(frozen=True)
class AutomaticSnapshotRecord:
    """Minimal metadata required by automation scheduling/retention."""

    set_id: str
    set_name: str
    created_at: datetime


def print_error(message: str) -> None:
    """
    Print a diagnostic error to stderr.
    """

    print(message, file=sys.stderr)


def require_root() -> None:
    """
    Require effective root privileges.
    """

    if os.geteuid() != 0:
        raise HelperError(
            "Synex Snapshots root helper must run as root."
        )


def build_environment() -> dict[str, str]:
    """
    Return a stable environment for command parsing.
    """

    environment = os.environ.copy()

    environment["LC_ALL"] = "C"
    environment["LANG"] = "C"

    return environment


def run_command(
    command: list[str],
    *,
    check: bool = True,
) -> subprocess.CompletedProcess[str]:
    """
    Execute a command without invoking a shell.
    """

    result = subprocess.run(
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        env=build_environment(),
        check=False,
    )

    if check and result.returncode != 0:
        message = (
            result.stderr.strip()
            or result.stdout.strip()
            or (
                f"{command[0]} failed with "
                f"exit code {result.returncode}."
            )
        )

        raise HelperError(message)

    return result


def ensure_required_commands() -> None:
    """
    Ensure commands required by the helper exist.
    """

    required = (
        BTRFS_COMMAND,
        FINDMNT_COMMAND,
        MOUNT_COMMAND,
        UMOUNT_COMMAND,
        SYSTEMCTL_COMMAND,
    )

    for command in required:
        if not command.exists():
            raise HelperError(
                f"Required command is not available: {command}"
            )


def daemon_reload() -> None:
    """
    Reload the systemd manager configuration.

    This is required after changing /etc/fstab so that systemd mount
    units generated from fstab reflect the current configuration.
    """

    run_command(
        [
            str(SYSTEMCTL_COMMAND),
            "daemon-reload",
        ]
    )


def normalize_fsroot(fsroot: str) -> str:
    """
    Normalize a Btrfs FSROOT value.

    Examples:
        /@          -> @
        /@snapshots -> @snapshots
    """

    return fsroot.strip().lstrip("/")


def strip_source_subvolume(source: str) -> str:
    """
    Strip findmnt's Btrfs subvolume suffix.

    Example:
        /dev/vda3[/@] -> /dev/vda3
    """

    return source.split("[", 1)[0]


def get_root_filesystem_info() -> RootFilesystemInfo:
    """
    Detect the Btrfs filesystem containing /.
    """

    result = run_command(
        [
            str(FINDMNT_COMMAND),
            "--raw",
            "--noheadings",
            "--target",
            "/",
            "--output",
            "SOURCE,FSTYPE,UUID,FSROOT",
        ]
    )

    line = result.stdout.strip()

    if not line:
        raise HelperError(
            "Unable to detect the root filesystem."
        )

    fields = line.split()

    if len(fields) != 4:
        raise HelperError(
            "Unexpected findmnt output for the root filesystem."
        )

    source, fstype, filesystem_uuid, fsroot = fields

    if fstype.lower() != "btrfs":
        raise HelperError(
            "The root filesystem is not Btrfs."
        )

    if not filesystem_uuid or filesystem_uuid == "-":
        raise HelperError(
            "Unable to determine the Btrfs filesystem UUID."
        )

    device = strip_source_subvolume(source)

    if not device:
        raise HelperError(
            "Unable to determine the Btrfs root device."
        )

    return RootFilesystemInfo(
        source=source,
        device=device,
        uuid=filesystem_uuid,
        fsroot=fsroot,
    )


def get_boot_filesystem_info() -> BootFilesystemInfo:
    """
    Detect whether /boot is part of root or mounted separately.

    A configured but currently unmounted /boot is treated as an error.
    This prevents a root-containing snapshot set from silently archiving
    the empty mountpoint directory instead of the real boot filesystem.
    """

    boot_path = Path("/boot")

    if boot_path.is_symlink():
        raise HelperError(
            "/boot is a symbolic link. Boot coverage is ambiguous."
        )

    if not boot_path.exists() or not boot_path.is_dir():
        raise HelperError(
            "/boot does not exist as a directory."
        )

    fstab_entries = parse_fstab(
        read_fstab()
    )

    boot_fstab_entries = [
        entry
        for entry in fstab_entries
        if entry.target == "/boot"
    ]

    if len(boot_fstab_entries) > 1:
        raise HelperError(
            "Multiple /boot entries were found in /etc/fstab."
        )

    result = run_command(
        [
            str(FINDMNT_COMMAND),
            "--raw",
            "--noheadings",
            "--mountpoint",
            "/boot",
            "--output",
            "SOURCE,FSTYPE,UUID,TARGET",
        ],
        check=False,
    )

    line = result.stdout.strip()

    if not line:
        if boot_fstab_entries:
            raise HelperError(
                "A separate /boot filesystem is configured in "
                "/etc/fstab but is not mounted."
            )

        return BootFilesystemInfo(
            separate=False,
        )

    fields = line.split()

    if len(fields) != 4:
        raise HelperError(
            "Unexpected findmnt output for /boot."
        )

    source, fstype, filesystem_uuid, target = fields

    if target != "/boot":
        raise HelperError(
            "The detected /boot mount has an unexpected target."
        )

    if not filesystem_uuid or filesystem_uuid == "-":
        raise HelperError(
            "Unable to determine the UUID of the separate /boot "
            "filesystem."
        )

    if boot_fstab_entries:
        persisted = boot_fstab_entries[0]

        if persisted.fstype.lower() != fstype.lower():
            raise HelperError(
                "The mounted /boot filesystem type does not match "
                "/etc/fstab."
            )

        if persisted.source.startswith("UUID="):
            persisted_uuid = persisted.source.split("=", 1)[1]

            if persisted_uuid != filesystem_uuid:
                raise HelperError(
                    "The mounted /boot filesystem UUID does not match "
                    "/etc/fstab."
                )

    return BootFilesystemInfo(
        separate=True,
        source=source,
        fstype=fstype.lower(),
        uuid=filesystem_uuid,
    )


def snapshot_set_contains_root(
    sources: list[SubvolumeRecord],
    *,
    root_source_name: str | None = None,
) -> bool:
    """
    Return whether the requested snapshot set contains the logical root.

    Normal snapshot creation keeps the existing runtime identity check
    against the subvolume mounted at /. Snapshot-boot pre-restore creation
    may instead provide the canonical writable root name resolved from the
    persisted system layout.
    """

    if root_source_name is not None:
        normalized_root_name = normalize_fsroot(
            root_source_name
        )

        if not is_safe_top_level_source_path(
            normalized_root_name
        ):
            raise HelperError(
                "The canonical root subvolume name is not a supported "
                "direct top-level Btrfs path."
            )

        return any(
            source.name == normalized_root_name
            for source in sources
        )

    root_source = read_subvolume_record(
        Path("/")
    )

    return any(
        source.subvolume_id == root_source.subvolume_id
        and source.uuid == root_source.uuid
        for source in sources
    )


def ensure_boot_archive_commands() -> None:
    """
    Ensure commands required for an external /boot archive exist.
    """

    for command in (
        TAR_COMMAND,
        ZSTD_COMMAND,
    ):
        if not command.exists():
            raise HelperError(
                f"Required command is not available: {command}"
            )


def sha256_file(path: Path) -> str:
    """
    Return the SHA-256 checksum of one regular file.
    """

    digest = hashlib.sha256()

    try:
        with path.open("rb") as handle:
            while True:
                chunk = handle.read(1024 * 1024)

                if not chunk:
                    break

                digest.update(chunk)
    except OSError as exc:
        raise HelperError(
            f"Unable to checksum {path}: {exc}"
        ) from exc

    return digest.hexdigest()


def create_boot_archive(
    pending_directory: Path,
    boot: BootFilesystemInfo,
) -> tuple[dict[str, object], Path]:
    """
    Create and validate the archive for a separate ext4 /boot.

    Nested filesystems such as /boot/efi are deliberately excluded by
    --one-file-system. The archive is part of the same pending snapshot
    transaction and is never published independently.
    """

    if not boot.separate:
        raise HelperError(
            "Internal error: boot archive requested for non-separate /boot."
        )

    if boot.fstype != "ext4":
        raise HelperError(
            "A separate /boot filesystem is supported only when it "
            "uses ext4."
        )

    if not boot.source or not boot.uuid:
        raise HelperError(
            "Incomplete metadata for the separate /boot filesystem."
        )

    ensure_boot_archive_commands()

    archives_directory = (
        pending_directory
        / "archives"
    )
    archives_directory.mkdir(
        mode=0o755,
        parents=False,
        exist_ok=False,
    )
    archives_directory.chmod(0o755)

    archive_path = (
        archives_directory
        / "boot.tar.zst"
    )

    run_command(
        [
            str(TAR_COMMAND),
            "--create",
            f"--use-compress-program={ZSTD_COMMAND}",
            "--acls",
            "--xattrs",
            "--xattrs-include=*",
            "--numeric-owner",
            "--one-file-system",
            "--file",
            str(archive_path),
            "--directory",
            "/boot",
            ".",
        ]
    )

    if archive_path.is_symlink() or not archive_path.is_file():
        raise HelperError(
            "The /boot archive was not created as a regular file."
        )

    archive_path.chmod(0o644)

    try:
        archive_size = archive_path.stat().st_size
    except OSError as exc:
        raise HelperError(
            f"Unable to stat the /boot archive: {exc}"
        ) from exc

    if archive_size <= 0:
        raise HelperError(
            "The /boot archive is empty."
        )

    try:
        archive_fd = os.open(
            str(archive_path),
            os.O_RDONLY,
        )

        try:
            os.fsync(archive_fd)
        finally:
            os.close(archive_fd)
    except OSError as exc:
        raise HelperError(
            f"Unable to flush the /boot archive: {exc}"
        ) from exc

    archive_sha256 = sha256_file(
        archive_path
    )

    fsync_directory(
        archives_directory
    )

    metadata: dict[str, object] = {
        "mode": "archive",
        "mountpoint": "/boot",
        "source": boot.source,
        "fstype": boot.fstype,
        "uuid": boot.uuid,
        "archive": {
            "path": "archives/boot.tar.zst",
            "compression": "zstd",
            "size_bytes": archive_size,
            "sha256": archive_sha256,
        },
    }

    return metadata, archive_path


def plan_boot_coverage(
    sources: list[SubvolumeRecord],
    *,
    root_source_name: str | None = None,
) -> BootFilesystemInfo | None:
    """
    Validate the /boot coverage plan before creating snapshot members.

    None means the set does not contain the logical root subvolume.
    Normal creation identifies root from the running mount.
    Snapshot-boot pre-restore creation may identify the canonical writable
    root by name. A non-separate result means /boot is already inside root.
    A separate result has already been validated as supported ext4.
    """

    if not snapshot_set_contains_root(
        sources,
        root_source_name=root_source_name,
    ):
        return None

    boot = get_boot_filesystem_info()

    if boot.separate:
        if boot.fstype != "ext4":
            raise HelperError(
                "A separate /boot filesystem is supported only when it "
                "uses ext4."
            )

        if not boot.source or not boot.uuid:
            raise HelperError(
                "Incomplete metadata for the separate /boot filesystem."
            )

        ensure_boot_archive_commands()

    return boot


def materialize_boot_coverage(
    pending_directory: Path,
    boot: BootFilesystemInfo | None,
) -> tuple[dict[str, object] | None, list[Path]]:
    """
    Materialize the already validated /boot coverage plan.
    """

    if boot is None:
        return None, []

    if not boot.separate:
        return (
            {
                "mode": "included-in-root",
                "mountpoint": "/boot",
            },
            [],
        )

    metadata, archive_path = (
        create_boot_archive(
            pending_directory,
            boot,
        )
    )

    return metadata, [archive_path]


def read_fstab() -> str:
    """
    Read /etc/fstab.
    """

    try:
        return FSTAB_PATH.read_text(
            encoding="utf-8"
        )
    except OSError as exc:
        raise HelperError(
            f"Unable to read {FSTAB_PATH}: {exc}"
        ) from exc


def parse_fstab(content: str) -> list[FstabEntry]:
    """
    Parse active /etc/fstab entries.

    Comments and blank lines are ignored.
    """

    entries: list[FstabEntry] = []

    for raw_line in content.splitlines():
        stripped = raw_line.strip()

        if not stripped:
            continue

        if stripped.startswith("#"):
            continue

        fields = stripped.split()

        if len(fields) < 4:
            continue

        entries.append(
            FstabEntry(
                source=fields[0],
                target=fields[1],
                fstype=fields[2],
                options=fields[3],
                dump=(
                    fields[4]
                    if len(fields) >= 5
                    else "0"
                ),
                passno=(
                    fields[5]
                    if len(fields) >= 6
                    else "0"
                ),
            )
        )

    return entries


def get_root_fstab_entry(
    entries: list[FstabEntry],
) -> FstabEntry | None:
    """
    Return the persisted Btrfs root entry from /etc/fstab.

    A single root entry is expected. If no Btrfs root entry exists,
    None is returned so callers can use a safe runtime fallback.
    """

    root_entries = [
        entry
        for entry in entries
        if entry.target == "/"
        and entry.fstype.lower() == "btrfs"
    ]

    if len(root_entries) > 1:
        raise HelperError(
            "Multiple Btrfs root entries were found in /etc/fstab."
        )

    if not root_entries:
        return None

    return root_entries[0]


def _get_fstab_subvolume_name(
    entry: FstabEntry,
) -> str | None:
    """
    Return the named Btrfs subvolume selected by one fstab entry.

    Multiple subvol= selectors are rejected because the persisted
    canonical layout would be ambiguous for restore.
    """

    values: list[str] = []

    for option in entry.options.split(","):
        option = option.strip()

        if not option.startswith("subvol="):
            continue

        value = normalize_fsroot(
            option.split("=", 1)[1]
        )

        if value:
            values.append(value)

    if len(values) > 1:
        raise HelperError(
            f"Multiple subvol= selectors were found for {entry.target}."
        )

    if not values:
        return None

    return values[0]


def get_restore_context(
    root: RootFilesystemInfo,
) -> RestoreContext:
    """
    Resolve whether restore is running normally or from a Synex snapshot.

    Normal restore keeps the existing runtime behavior. Snapshot-boot
    restore is accepted only when the running FSROOT has the exact Synex
    snapshot layout and /etc/fstab identifies the canonical root
    subvolume on the same Btrfs filesystem.
    """

    running_root_name = normalize_fsroot(
        root.fsroot
    )

    # Normal boot:
    #
    #   / -> one direct writable top-level Btrfs subvolume.
    #
    # Keep the existing restore semantics unchanged.
    if is_safe_top_level_source_path(
        running_root_name
    ):
        return RestoreContext(
            mode="normal",
            running_root_name=running_root_name,
            canonical_root_name=running_root_name,
        )

    # Snapshot boot must have exactly:
    #
    # @snapshots/<set>/subvolumes/<canonical-root>
    parts = running_root_name.split("/")

    if not (
        len(parts) == 4
        and parts[0] == SNAPSHOT_SUBVOLUME
        and parts[2] == "subvolumes"
    ):
        raise HelperError(
            "The running root is neither a supported canonical top-level "
            "subvolume nor a Synex snapshot root."
        )

    booted_set_name = parts[1]
    booted_root_member = parts[3]

    if (
        not booted_set_name
        or booted_set_name in {".", ".."}
        or "\x00" in booted_set_name
        or Path(booted_set_name).name != booted_set_name
    ):
        raise HelperError(
            "The running snapshot set name is invalid."
        )

    # During Snapshot Boot the running root cannot tell us the
    # canonical target name, because FSROOT points into @snapshots.
    #
    # /etc/fstab remains the authoritative description of the
    # installed system layout.
    entries = parse_fstab(
        read_fstab()
    )

    root_entry = get_root_fstab_entry(
        entries
    )

    if root_entry is None:
        raise HelperError(
            "Snapshot-boot restore requires a persisted Btrfs root entry "
            "in /etc/fstab."
        )

    if not source_matches_root(
        root_entry.source,
        root,
    ):
        raise HelperError(
            "The persisted root entry does not refer to the running "
            "Btrfs filesystem."
        )

    canonical_root_name = (
        _get_fstab_subvolume_name(
            root_entry
        )
    )

    if canonical_root_name is None:
        raise HelperError(
            "Snapshot-boot restore requires the persisted root entry to "
            "use a named subvol= selector."
        )

    if not is_safe_top_level_source_path(
        canonical_root_name
    ):
        raise HelperError(
            "The persisted root subvolume is not a supported direct "
            "top-level Btrfs subvolume."
        )

    # The member booted by GRUB must actually represent the canonical
    # root defined by the installed system.
    if booted_root_member != canonical_root_name:
        raise HelperError(
            "The booted snapshot root member does not match the "
            "persisted canonical root subvolume."
        )

    return RestoreContext(
        mode="snapshot-boot",
        running_root_name=running_root_name,
        canonical_root_name=canonical_root_name,
        booted_set_name=booted_set_name,
    )


def get_persisted_source_mountpoints(
    root: RootFilesystemInfo,
    top_level_mount: Path,
    sources: list[SubvolumeRecord],
    *,
    required_root_name: str,
) -> dict[str, str | None]:
    """
    Resolve canonical mountpoints for top-level restore sources.

    During Snapshot Boot the visible /etc/fstab belongs to the historical
    root. The pre-restore snapshot must instead describe the current
    canonical system that is about to be replaced.

    Therefore this function reads etc/fstab directly from the canonical
    writable root subvolume through the private Btrfs top-level mount.

    Operational top-level subvolumes without a persisted mountpoint remain
    mapped to None. This preserves Full semantics without inventing mount
    roles for arbitrary additional subvolumes.
    """

    root_name = normalize_fsroot(
        required_root_name
    )

    if not is_safe_top_level_source_path(
        root_name
    ):
        raise HelperError(
            "The canonical root subvolume name is not a supported "
            "direct top-level Btrfs path."
        )

    source_names = {
        source.name
        for source in sources
    }

    if root_name not in source_names:
        raise HelperError(
            "The canonical root subvolume is not present in the "
            "pre-restore source set."
        )

    canonical_root_path = (
        top_level_mount
        / root_name
    )

    if (
        canonical_root_path.is_symlink()
        or not canonical_root_path.exists()
    ):
        raise HelperError(
            "The canonical root subvolume is not available through the "
            "Btrfs top-level mount."
        )

    canonical_root_record = read_subvolume_record(
        canonical_root_path
    )

    if (
        canonical_root_record.parent_id != TOP_LEVEL_SUBVOLUME_ID
        or canonical_root_record.name != root_name
        or subvolume_is_read_only(
            canonical_root_path
        )
    ):
        raise HelperError(
            "The canonical root restore target is not a valid writable "
            "top-level Btrfs subvolume."
        )

    canonical_fstab_path = (
        canonical_root_path
        / "etc"
        / "fstab"
    )

    if (
        canonical_fstab_path.is_symlink()
        or not canonical_fstab_path.is_file()
    ):
        raise HelperError(
            "The canonical system does not contain a safe regular "
            "/etc/fstab file."
        )

    try:
        canonical_fstab_content = (
            canonical_fstab_path.read_text(
                encoding="utf-8"
            )
        )
    except (OSError, UnicodeError) as exc:
        raise HelperError(
            f"Unable to read the canonical system /etc/fstab: {exc}"
        ) from exc

    entries = parse_fstab(
        canonical_fstab_content
    )

    mountpoints: dict[str, str | None] = {
        source.name: None
        for source in sources
    }

    seen_subvolumes: set[str] = set()
    target_owners: dict[str, str] = {}

    for entry in entries:
        if entry.fstype.lower() != "btrfs":
            continue

        if not source_matches_root(
            entry.source,
            root,
        ):
            continue

        if entry.target == "/.snapshots":
            continue

        subvolume_name = _get_fstab_subvolume_name(
            entry
        )

        if subvolume_name is None:
            continue

        if subvolume_name not in source_names:
            continue

        if subvolume_name in seen_subvolumes:
            raise HelperError(
                f"The canonical subvolume {subvolume_name} has multiple "
                "persisted Btrfs mount entries."
            )

        target = entry.target

        if (
            not isinstance(target, str)
            or not target.startswith("/")
            or "\x00" in target
        ):
            raise HelperError(
                "A persisted Btrfs mountpoint required for restore is "
                "invalid."
            )

        previous_owner = target_owners.get(
            target
        )

        if (
            previous_owner is not None
            and previous_owner != subvolume_name
        ):
            raise HelperError(
                f"The persisted mountpoint {target} is assigned to "
                "multiple canonical Btrfs subvolumes."
            )

        mountpoints[subvolume_name] = target
        seen_subvolumes.add(
            subvolume_name
        )
        target_owners[target] = subvolume_name

    if mountpoints.get(root_name) != "/":
        raise HelperError(
            "The canonical root subvolume could not be resolved to / "
            "from its persisted /etc/fstab."
        )

    return mountpoints



def get_snapshot_boot_pre_restore_mountpoints(
    root: RootFilesystemInfo,
    top_level_mount: Path,
    sources: list[SubvolumeRecord],
    *,
    required_root_name: str,
) -> dict[str, str | None]:
    """
    Resolve mountpoint metadata for a Snapshot Boot pre-restore set.

    Normal case:
        When the canonical /etc/fstab still exists, preserve the exact
        previous behavior and resolve canonical mountpoints from it.

    Destructive-recovery fallback:
        When the canonical /etc/fstab no longer exists, recovery must not be
        blocked. Snapshot Boot has already established the canonical root
        identity, so record that root as mounted at / and preserve every
        other operational top-level subvolume with an unknown mountpoint.

    Unsafe or malformed existing fstab files are deliberately NOT bypassed:
    if the path exists (or is a symlink), the original strict validation is
    used unchanged.
    """

    root_name = normalize_fsroot(
        required_root_name
    )

    canonical_fstab_path = (
        top_level_mount
        / root_name
        / "etc"
        / "fstab"
    )

    if (
        canonical_fstab_path.exists()
        or canonical_fstab_path.is_symlink()
    ):
        return get_persisted_source_mountpoints(
            root,
            top_level_mount,
            sources,
            required_root_name=root_name,
        )

    if not is_safe_top_level_source_path(
        root_name
    ):
        raise HelperError(
            "The canonical root subvolume name is not a supported "
            "direct top-level Btrfs path."
        )

    source_names = {
        source.name
        for source in sources
    }

    if root_name not in source_names:
        raise HelperError(
            "The canonical root subvolume is not present in the "
            "pre-restore source set."
        )

    mountpoints: dict[str, str | None] = {
        source.name: None
        for source in sources
    }
    mountpoints[root_name] = "/"

    return mountpoints

def get_snapshot_fstab_source(
    entries: list[FstabEntry],
    root: RootFilesystemInfo,
) -> str:
    """
    Return the source that should be persisted for /.snapshots.

    When /etc/fstab already contains a Btrfs root entry, its source is
    preserved exactly.

    Examples:
        UUID=...
        /dev/mapper/luks-...

    If no persisted Btrfs root entry exists, fall back to the detected
    Btrfs filesystem UUID.
    """

    root_entry = get_root_fstab_entry(entries)

    if root_entry is not None:
        return root_entry.source

    return f"UUID={root.uuid}"


def derive_snapshot_mount_options(
    entries: list[FstabEntry],
) -> str:
    """
    Derive /.snapshots Btrfs options from the persisted root entry.

    The root subvolume selector is removed and replaced by
    subvol=/@snapshots.
    """

    root_entry = get_root_fstab_entry(entries)

    if root_entry is None:
        return "subvol=/@snapshots,defaults"

    root_options = root_entry.options.split(",")

    filtered_options: list[str] = []

    for option in root_options:
        option = option.strip()

        if not option:
            continue

        if option.startswith("subvol="):
            continue

        if option.startswith("subvolid="):
            continue

        if option not in filtered_options:
            filtered_options.append(option)

    return ",".join(
        ["subvol=/@snapshots", *filtered_options]
    )


def source_matches_root(
    source: str,
    root: RootFilesystemInfo,
) -> bool:
    """
    Return whether an fstab source clearly refers to the root Btrfs FS.
    """

    if source == f"UUID={root.uuid}":
        return True

    if source == root.device:
        return True

    if source.startswith("/dev/"):
        try:
            return (
                os.path.realpath(source)
                == os.path.realpath(root.device)
            )
        except OSError:
            return False

    return False


def options_select_snapshot_subvolume(
    options: str,
) -> bool:
    """
    Return whether Btrfs mount options select @snapshots.
    """

    for option in options.split(","):
        option = option.strip()

        if not option.startswith("subvol="):
            continue

        value = option.split("=", 1)[1]

        return normalize_fsroot(value) == SNAPSHOT_SUBVOLUME

    return False


def validate_fstab_state(
    entries: list[FstabEntry],
    root: RootFilesystemInfo,
) -> bool:
    """
    Validate the /.snapshots fstab state.

    Returns:
        True when a valid entry already exists.
        False when no entry exists.

    Raises:
        HelperError when a conflicting entry exists.
    """

    snapshot_entries = [
        entry
        for entry in entries
        if entry.target == str(SNAPSHOT_MOUNTPOINT)
    ]

    if len(snapshot_entries) > 1:
        raise HelperError(
            "Multiple /.snapshots entries were found in /etc/fstab."
        )

    if not snapshot_entries:
        return False

    entry = snapshot_entries[0]

    if entry.fstype.lower() != "btrfs":
        raise HelperError(
            "/etc/fstab contains a conflicting /.snapshots "
            "entry with a non-Btrfs filesystem."
        )

    if not source_matches_root(
        entry.source,
        root,
    ):
        raise HelperError(
            "/etc/fstab contains a /.snapshots entry "
            "for a different filesystem."
        )

    if not options_select_snapshot_subvolume(
        entry.options
    ):
        raise HelperError(
            "/etc/fstab contains a /.snapshots entry "
            "that does not use @snapshots."
        )

    return True


def get_snapshot_mount_state(
    root: RootFilesystemInfo,
) -> MountState:
    """
    Inspect the current runtime mount state of /.snapshots.
    """

    result = run_command(
        [
            str(FINDMNT_COMMAND),
            "--raw",
            "--noheadings",
            "--target",
            str(SNAPSHOT_MOUNTPOINT),
            "--output",
            "TARGET,FSTYPE,UUID,FSROOT",
        ],
        check=False,
    )

    line = result.stdout.strip()

    if not line:
        return MountState(
            mounted=False,
            correct=False,
        )

    fields = line.split()

    if len(fields) != 4:
        raise HelperError(
            "Unexpected findmnt output for /.snapshots."
        )

    target, fstype, filesystem_uuid, fsroot = fields

    if target != str(SNAPSHOT_MOUNTPOINT):
        return MountState(
            mounted=False,
            correct=False,
        )

    correct = (
        fstype.lower() == "btrfs"
        and filesystem_uuid == root.uuid
        and normalize_fsroot(fsroot)
        == SNAPSHOT_SUBVOLUME
    )

    return MountState(
        mounted=True,
        correct=correct,
        target=target,
        fstype=fstype,
        uuid=filesystem_uuid,
        fsroot=fsroot,
    )


def validate_mountpoint(
    mount_state: MountState,
) -> bool:
    """
    Validate the /.snapshots path.

    Returns:
        True when the directory must be created.
        False when it already exists.

    Existing user data is never hidden by mounting over it.
    """

    path = SNAPSHOT_MOUNTPOINT

    if path.is_symlink():
        raise HelperError(
            "/.snapshots is a symbolic link. "
            "Synex Snapshots will not modify it."
        )

    if mount_state.mounted:
        if not mount_state.correct:
            raise HelperError(
                "/.snapshots is already mounted with "
                "an incompatible filesystem."
            )

        return False

    if not path.exists():
        return True

    if not path.is_dir():
        raise HelperError(
            "/.snapshots exists but is not a directory."
        )

    try:
        has_contents = any(path.iterdir())
    except OSError as exc:
        raise HelperError(
            f"Unable to inspect /.snapshots: {exc}"
        ) from exc

    if has_contents:
        raise HelperError(
            "/.snapshots contains existing data. "
            "Synex Snapshots will not mount over it."
        )

    return False


def mount_top_level(
    root: RootFilesystemInfo,
) -> Path:
    """
    Mount Btrfs top-level ID 5 into a private temporary directory.
    """

    temporary_mount = Path(
        tempfile.mkdtemp(
            prefix="synex-snapshots-top-",
            dir="/run",
        )
    )

    try:
        run_command(
            [
                str(MOUNT_COMMAND),
                "-t",
                "btrfs",
                "-o",
                f"subvolid={TOP_LEVEL_SUBVOLUME_ID}",
                root.device,
                str(temporary_mount),
            ]
        )
    except Exception:
        temporary_mount.rmdir()
        raise

    return temporary_mount


def unmount_top_level(
    temporary_mount: Path,
) -> None:
    """
    Unmount and remove the temporary top-level mount.
    """

    try:
        run_command(
            [
                str(UMOUNT_COMMAND),
                str(temporary_mount),
            ]
        )
    finally:
        try:
            temporary_mount.rmdir()
        except OSError:
            pass


def validate_snapshot_subvolume(
    top_level_mount: Path,
) -> bool:
    """
    Validate @snapshots.

    Returns:
        True when @snapshots must be created.
        False when a valid dedicated subvolume already exists.
    """

    snapshot_path = (
        top_level_mount
        / SNAPSHOT_SUBVOLUME
    )

    if not snapshot_path.exists():
        return True

    if snapshot_path.is_symlink():
        raise HelperError(
            "@snapshots exists as a symbolic link."
        )

    result = run_command(
        [
            str(BTRFS_COMMAND),
            "subvolume",
            "show",
            str(snapshot_path),
        ],
        check=False,
    )

    if result.returncode != 0:
        raise HelperError(
            "@snapshots exists but is not a Btrfs subvolume."
        )

    parent_uuid: str | None = None
    parent_id: int | None = None

    for raw_line in result.stdout.splitlines():
        line = raw_line.strip()

        if line.startswith("Parent UUID:"):
            parent_uuid = (
                line.split(":", 1)[1].strip()
            )

        elif line.startswith("Parent ID:"):
            value = line.split(":", 1)[1].strip()

            try:
                parent_id = int(value)
            except ValueError:
                raise HelperError(
                    "Unable to determine the parent ID "
                    "of @snapshots."
                )

    if parent_id != TOP_LEVEL_SUBVOLUME_ID:
        raise HelperError(
            "@snapshots is not a direct child "
            "of the Btrfs top-level."
        )

    if parent_uuid not in (None, "-"):
        raise HelperError(
            "@snapshots is a snapshot of another subvolume. "
            "A dedicated storage subvolume is required."
        )

    return False


def atomic_write_fstab(
    content: str,
) -> None:
    """
    Replace /etc/fstab atomically while preserving ownership and mode.
    """

    try:
        original_stat = FSTAB_PATH.stat()
    except OSError as exc:
        raise HelperError(
            f"Unable to stat {FSTAB_PATH}: {exc}"
        ) from exc

    temporary_fd: int | None = None
    temporary_name: str | None = None

    try:
        temporary_fd, temporary_name = tempfile.mkstemp(
            prefix=".fstab.synex-snapshots.",
            dir=str(FSTAB_PATH.parent),
        )

        os.fchmod(
            temporary_fd,
            stat.S_IMODE(original_stat.st_mode),
        )

        os.fchown(
            temporary_fd,
            original_stat.st_uid,
            original_stat.st_gid,
        )

        with os.fdopen(
            temporary_fd,
            "w",
            encoding="utf-8",
            closefd=True,
        ) as handle:
            temporary_fd = None

            handle.write(content)
            handle.flush()
            os.fsync(handle.fileno())

        os.replace(
            temporary_name,
            FSTAB_PATH,
        )

        temporary_name = None

        directory_fd = os.open(
            str(FSTAB_PATH.parent),
            os.O_RDONLY | os.O_DIRECTORY,
        )

        try:
            os.fsync(directory_fd)
        finally:
            os.close(directory_fd)

    finally:
        if temporary_fd is not None:
            os.close(temporary_fd)

        if temporary_name is not None:
            try:
                os.unlink(temporary_name)
            except FileNotFoundError:
                pass


def build_fstab_with_snapshot_entry(
    original_content: str,
    source: str,
    mount_options: str,
) -> str:
    """
    Append the canonical /.snapshots entry to /etc/fstab.

    The source is inherited from the persisted Btrfs root entry whenever
    available. This preserves the existing system configuration style,
    including /dev/mapper paths used by encrypted installations.
    """

    content = original_content

    if content and not content.endswith("\n"):
        content += "\n"

    content += (
        f"{source} "
        f"{SNAPSHOT_MOUNTPOINT} "
        f"btrfs "
        f"{mount_options} "
        "0 0\n"
    )

    return content


def create_snapshot_subvolume(
    top_level_mount: Path,
) -> None:
    """
    Create the dedicated @snapshots Btrfs subvolume.
    """

    run_command(
        [
            str(BTRFS_COMMAND),
            "subvolume",
            "create",
            str(
                top_level_mount
                / SNAPSHOT_SUBVOLUME
            ),
        ]
    )


def snapshot_subvolume_is_empty(
    top_level_mount: Path,
) -> bool:
    """
    Return True when a newly created @snapshots contains no data.
    """

    snapshot_path = (
        top_level_mount
        / SNAPSHOT_SUBVOLUME
    )

    try:
        return not any(snapshot_path.iterdir())
    except OSError:
        return False


def delete_created_snapshot_subvolume(
    top_level_mount: Path,
) -> None:
    """
    Roll back a newly created empty @snapshots subvolume.

    User data is never deleted during rollback.
    """

    snapshot_path = (
        top_level_mount
        / SNAPSHOT_SUBVOLUME
    )

    if not snapshot_path.exists():
        return

    if not snapshot_subvolume_is_empty(
        top_level_mount
    ):
        raise HelperError(
            "Rollback stopped because the newly created "
            "@snapshots subvolume is no longer empty."
        )

    run_command(
        [
            str(BTRFS_COMMAND),
            "subvolume",
            "delete",
            str(snapshot_path),
        ]
    )


def read_subvolume_record(
    lookup_path: Path,
    *,
    root_id: int | None = None,
) -> SubvolumeRecord:
    """
    Read Btrfs metadata for one subvolume.

    When root_id is provided, btrfs resolves that subvolume ID inside
    the filesystem containing lookup_path.
    """

    command = [
        str(BTRFS_COMMAND),
        "subvolume",
        "show",
    ]

    if root_id is not None:
        command.extend(
            [
                "-r",
                str(root_id),
            ]
        )

    command.append(
        str(lookup_path)
    )

    result = run_command(command)

    fields: dict[str, str] = {}

    for raw_line in result.stdout.splitlines():
        line = raw_line.strip()

        if ":" not in line:
            continue

        key, value = line.split(":", 1)
        fields[key.strip()] = value.strip()

    try:
        subvolume_id = int(
            fields["Subvolume ID"]
        )
        parent_id = int(
            fields["Parent ID"]
        )
        name = fields["Name"]
        subvolume_uuid = fields["UUID"]
    except (KeyError, ValueError) as exc:
        raise HelperError(
            "Unable to read complete Btrfs subvolume metadata."
        ) from exc

    parent_uuid_value = fields.get(
        "Parent UUID",
        "-",
    )

    parent_uuid = (
        None
        if parent_uuid_value == "-"
        else parent_uuid_value
    )

    return SubvolumeRecord(
        subvolume_id=subvolume_id,
        parent_id=parent_id,
        name=name,
        uuid=subvolume_uuid,
        parent_uuid=parent_uuid,
    )


def is_safe_top_level_source_path(path_text: str) -> bool:
    """
    Return whether a Btrfs path is one direct child of top-level ID 5.

    ``btrfs subvolume list -p`` reports the ID of the containing
    subvolume, not ordinary directory depth. A subvolume stored under a
    normal directory can therefore have parent ID 5 while its path still
    contains ``/``. Synex Snapshots only manages one-component paths
    directly below the Btrfs top-level.
    """

    return (
        bool(path_text)
        and path_text not in {".", ".."}
        and "/" not in path_text
        and "\x00" not in path_text
        and Path(path_text).name == path_text
    )


def get_subvolume_relative_path(
    top_level_mount: Path,
    subvolume_id: int,
) -> str:
    """
    Return the path of one subvolume relative to Btrfs top-level ID 5.

    This is intentionally resolved from ``btrfs subvolume list`` rather
    than from ``btrfs subvolume show`` because ``show`` exposes only the
    leaf Name field. The relative path is required to distinguish a real
    top-level source such as ``@home`` from a subvolume stored below an
    ordinary directory such as ``timeshift-btrfs/snapshots/...``.
    """

    result = run_command(
        [
            str(BTRFS_COMMAND),
            "subvolume",
            "list",
            "-p",
            "-u",
            "-q",
            str(top_level_mount),
        ]
    )

    matched_path: str | None = None

    for raw_line in result.stdout.splitlines():
        line = raw_line.strip()

        if not line:
            continue

        match = _SUBVOLUME_LIST_LINE_RE.match(
            line
        )

        if match is None:
            raise HelperError(
                "Unexpected btrfs subvolume list output: "
                f"{line}"
            )

        if int(match.group("id")) != subvolume_id:
            continue

        if matched_path is not None:
            raise HelperError(
                "Duplicate Btrfs subvolume ID detected while resolving "
                "the snapshot source."
            )

        matched_path = match.group("path")

    if matched_path is None:
        raise HelperError(
            "The requested Btrfs subvolume could not be resolved by ID."
        )

    return matched_path


def validate_snapshot_source(
    top_level_mount: Path,
    subvolume_id: int,
) -> SubvolumeRecord:
    """
    Resolve and validate a source for a single snapshot.

    Synex Snapshots manages direct children of Btrfs top-level ID 5.
    Nested subvolumes, @snapshots and Synex internal restore staging
    subvolumes are rejected. A non-empty Parent UUID is valid here: after
    a restore, an operational writable subvolume legitimately keeps the
    restored snapshot as its Btrfs parent.
    """

    source = read_subvolume_record(
        top_level_mount,
        root_id=subvolume_id,
    )

    if source.subvolume_id != subvolume_id:
        raise HelperError(
            "The resolved Btrfs subvolume ID does not match "
            "the requested source."
        )

    if source.parent_id != TOP_LEVEL_SUBVOLUME_ID:
        raise HelperError(
            "Nested Btrfs subvolumes are not managed "
            "by Synex Snapshots."
        )

    relative_path = get_subvolume_relative_path(
        top_level_mount,
        subvolume_id,
    )

    if (
        not is_safe_top_level_source_path(relative_path)
        or relative_path != source.name
    ):
        raise HelperError(
            "Only direct top-level Btrfs subvolumes are managed as "
            "snapshot sources."
        )

    if source.name == SNAPSHOT_SUBVOLUME:
        raise HelperError(
            "The @snapshots storage subvolume cannot be "
            "used as a snapshot source."
        )

    if source.name.startswith(RESTORE_INTERNAL_PREFIX):
        raise HelperError(
            "Synex Snapshots internal restore staging subvolumes cannot "
            "be used as snapshot sources."
        )

    if (
        not source.name
        or source.name in {".", ".."}
        or "/" in source.name
        or "\x00" in source.name
        or Path(source.name).name != source.name
    ):
        raise HelperError(
            "The source subvolume name cannot be represented "
            "safely inside a snapshot set."
        )

    if subvolume_is_read_only(
        top_level_mount / source.name
    ):
        raise HelperError(
            "A read-only top-level Btrfs subvolume is not an operational "
            "snapshot source."
        )

    return source


def list_operational_snapshot_sources(
    top_level_mount: Path,
) -> list[SubvolumeRecord]:
    """
    Return all operational top-level Btrfs subvolumes for Full scope.

    Full includes operational direct children of Btrfs top-level ID 5.
    @snapshots, nested subvolumes and Synex internal restore staging
    subvolumes are excluded. Parent UUID is lineage metadata and does not
    determine whether a top-level writable subvolume is operational.
    """

    result = run_command(
        [
            str(BTRFS_COMMAND),
            "subvolume",
            "list",
            "-p",
            "-u",
            "-q",
            str(top_level_mount),
        ]
    )

    sources: list[SubvolumeRecord] = []
    seen_ids: set[int] = set()
    seen_names: set[str] = set()

    for raw_line in result.stdout.splitlines():
        line = raw_line.strip()

        if not line:
            continue

        match = _SUBVOLUME_LIST_LINE_RE.match(
            line
        )

        if match is None:
            raise HelperError(
                "Unexpected btrfs subvolume list output: "
                f"{line}"
            )

        parent_id = int(
            match.group("parent")
        )

        if parent_id != TOP_LEVEL_SUBVOLUME_ID:
            continue

        subvolume_id = int(
            match.group("id")
        )
        name = match.group("path")
        if name == SNAPSHOT_SUBVOLUME:
            continue

        if name.startswith(RESTORE_INTERNAL_PREFIX):
            continue

        if not is_safe_top_level_source_path(
            name
        ):
            # ``parent 5`` means that ID 5 is the containing subvolume.
            # It does not mean that the path is one directory component
            # below ID 5. Subvolumes kept under ordinary directories
            # (for example Timeshift storage) must not become Full
            # snapshot members and must not abort discovery.
            continue

        if subvolume_id in seen_ids:
            raise HelperError(
                "Duplicate top-level Btrfs subvolume ID detected."
            )

        if name in seen_names:
            raise HelperError(
                "Duplicate top-level Btrfs subvolume name detected."
            )

        source = read_subvolume_record(
            top_level_mount,
            root_id=subvolume_id,
        )

        if source.parent_id != TOP_LEVEL_SUBVOLUME_ID:
            raise HelperError(
                "The Full snapshot source changed while it was being "
                "validated."
            )

        if source.name != name:
            raise HelperError(
                "The Full snapshot source name changed while it was "
                "being validated."
            )

        if source.name.startswith(RESTORE_INTERNAL_PREFIX):
            raise HelperError(
                "A Synex Snapshots restore staging subvolume cannot be "
                "included as a Full snapshot source."
            )

        if subvolume_is_read_only(
            top_level_mount / source.name
        ):
            continue

        seen_ids.add(
            subvolume_id
        )
        seen_names.add(
            name
        )
        sources.append(
            source
        )

    if not sources:
        raise HelperError(
            "No operational top-level Btrfs subvolumes were found."
        )

    sources.sort(
        key=lambda source: source.name
    )

    return sources


def get_subvolume_mountpoint(
    root: RootFilesystemInfo,
    subvolume_name: str,
) -> str | None:
    """
    Return the current mountpoint of a top-level Btrfs subvolume.
    """

    result = run_command(
        [
            str(FINDMNT_COMMAND),
            "--json",
            "--list",
            "--types",
            "btrfs",
            "--output",
            "TARGET,UUID,FSROOT",
        ],
        check=False,
    )

    if result.returncode != 0 or not result.stdout.strip():
        return None

    try:
        filesystems = json.loads(
            result.stdout
        ).get(
            "filesystems",
            [],
        )
    except json.JSONDecodeError:
        return None

    if not isinstance(filesystems, list):
        return None

    for filesystem in filesystems:
        if not isinstance(filesystem, dict):
            continue

        if filesystem.get("uuid") != root.uuid:
            continue

        fsroot = filesystem.get("fsroot")

        if not isinstance(fsroot, str):
            continue

        if normalize_fsroot(fsroot) != subvolume_name:
            continue

        target = filesystem.get("target")

        if isinstance(target, str) and target:
            return target

    return None


def subvolume_is_read_only(
    path: Path,
) -> bool:
    """
    Return True when a Btrfs subvolume has ro=true.
    """

    result = run_command(
        [
            str(BTRFS_COMMAND),
            "property",
            "get",
            "-ts",
            str(path),
            "ro",
        ]
    )

    return result.stdout.strip() == "ro=true"


def fsync_directory(
    path: Path,
) -> None:
    """
    Flush directory metadata to disk.
    """

    directory_fd = os.open(
        str(path),
        os.O_RDONLY | os.O_DIRECTORY,
    )

    try:
        os.fsync(directory_fd)
    finally:
        os.close(directory_fd)


def choose_snapshot_set_name(
    created_at: datetime,
) -> str:
    """
    Return a readable timestamp-based set name.

    A numeric suffix is used only when a set already exists for the
    same second.
    """

    base_name = created_at.strftime(
        "%Y-%m-%d_%H-%M-%S"
    )
    candidate = base_name
    counter = 2

    while (SNAPSHOT_MOUNTPOINT / candidate).exists():
        candidate = f"{base_name}_{counter}"
        counter += 1

    return candidate


def write_pending_manifest(
    pending_directory: Path,
    manifest: dict[str, object],
) -> None:
    """
    Write and flush the manifest inside a private pending set.

    The whole pending directory is renamed only after validation, so an
    additional temporary manifest name is not required here.
    """

    manifest_path = (
        pending_directory
        / "manifest.json"
    )

    with manifest_path.open(
        "x",
        encoding="utf-8",
    ) as handle:
        json.dump(
            manifest,
            handle,
            ensure_ascii=False,
            indent=2,
        )
        handle.write("\n")
        handle.flush()
        os.fsync(handle.fileno())

    manifest_path.chmod(0o644)


def rollback_pending_snapshot_set(
    pending_directory: Path,
    snapshot_paths: list[Path],
    archive_paths: list[Path],
) -> list[str]:
    """
    Remove only objects created by an incomplete snapshot transaction.

    Snapshot members are deleted in reverse creation order. Regular
    archive files created by the same transaction are removed explicitly.
    """

    errors: list[str] = []

    for snapshot_path in reversed(
        snapshot_paths
    ):
        if not snapshot_path.exists():
            continue

        try:
            run_command(
                [
                    str(BTRFS_COMMAND),
                    "subvolume",
                    "delete",
                    str(snapshot_path),
                ]
            )
        except Exception as exc:
            errors.append(
                f"snapshot {snapshot_path.name}: {exc}"
            )

    known_archive_paths = list(
        archive_paths
    )

    partial_boot_archive = (
        pending_directory
        / "archives"
        / "boot.tar.zst"
    )

    if partial_boot_archive not in known_archive_paths:
        known_archive_paths.append(
            partial_boot_archive
        )

    for archive_path in known_archive_paths:
        if not archive_path.exists():
            continue

        try:
            if archive_path.is_symlink() or not archive_path.is_file():
                raise HelperError(
                    "Rollback found an unexpected archive object."
                )

            archive_path.unlink()
        except Exception as exc:
            errors.append(
                f"archive {archive_path.name}: {exc}"
            )

    manifest_path = (
        pending_directory
        / "manifest.json"
    )

    if manifest_path.exists():
        try:
            manifest_path.unlink()
        except Exception as exc:
            errors.append(
                f"manifest: {exc}"
            )

    archives_directory = (
        pending_directory
        / "archives"
    )

    if archives_directory.exists():
        try:
            archives_directory.rmdir()
        except Exception as exc:
            errors.append(
                f"archives directory: {exc}"
            )

    subvolumes_directory = (
        pending_directory
        / "subvolumes"
    )

    if subvolumes_directory.exists():
        try:
            subvolumes_directory.rmdir()
        except Exception as exc:
            errors.append(
                f"subvolumes directory: {exc}"
            )

    if pending_directory.exists():
        try:
            pending_directory.rmdir()
        except Exception as exc:
            errors.append(
                f"pending directory: {exc}"
            )

    return errors


def create_snapshot_set(
    root: RootFilesystemInfo,
    top_level_mount: Path,
    sources: list[SubvolumeRecord],
    *,
    scope: str,
    purpose: str = "manual",
    source_mountpoints: dict[str, str | None] | None = None,
    root_source_name: str | None = None,
) -> dict[str, object]:
    """
    Create and atomically publish one Synex Snapshots snapshot set.

    All requested members are created read-only below one private
    .pending-* directory. The directory is published only after every
    member and the complete manifest have been validated and flushed.
    """

    if scope not in {
        "single",
        "full",
    }:
        raise HelperError(
            f"Unsupported snapshot creation scope: {scope}"
        )

    if purpose not in SNAPSHOT_PURPOSES:
        raise HelperError(
            f"Unsupported snapshot creation purpose: {purpose}"
        )

    if not sources:
        raise HelperError(
            "Snapshot set requires at least one source subvolume."
        )

    if scope == "single" and len(sources) != 1:
        raise HelperError(
            "Single snapshot set must contain exactly one source."
        )

    source_ids: set[int] = set()
    source_names: set[str] = set()

    for source in sources:
        if source.subvolume_id in source_ids:
            raise HelperError(
                "Snapshot set contains a duplicate source subvolume ID."
            )

        if source.name in source_names:
            raise HelperError(
                "Snapshot set contains a duplicate source subvolume name."
            )

        if source.parent_id != TOP_LEVEL_SUBVOLUME_ID:
            raise HelperError(
                "Nested Btrfs subvolumes cannot be snapshot set sources."
            )

        if source.name == SNAPSHOT_SUBVOLUME:
            raise HelperError(
                "The @snapshots storage subvolume cannot be a snapshot "
                "set source."
            )

        if source.name.startswith(RESTORE_INTERNAL_PREFIX):
            raise HelperError(
                "A Synex Snapshots restore staging subvolume cannot be a "
                "snapshot set source."
            )

        source_ids.add(
            source.subvolume_id
        )
        source_names.add(
            source.name
        )

    if source_mountpoints is not None:
        unknown_mountpoint_sources = (
            set(source_mountpoints)
            - source_names
        )

        if unknown_mountpoint_sources:
            raise HelperError(
                "Snapshot mountpoint metadata contains unknown source "
                "subvolumes."
            )

        for source_name, mountpoint in source_mountpoints.items():
            if mountpoint is None:
                continue

            if (
                not isinstance(mountpoint, str)
                or not mountpoint.startswith("/")
                or "\x00" in mountpoint
            ):
                raise HelperError(
                    f"Invalid persisted mountpoint metadata for "
                    f"{source_name}."
                )

    boot_plan = plan_boot_coverage(
        sources,
        root_source_name=root_source_name,
    )

    created_at = datetime.now().astimezone()
    set_id = str(uuid.uuid4())
    set_name = choose_snapshot_set_name(
        created_at
    )

    pending_directory = (
        SNAPSHOT_MOUNTPOINT
        / f".pending-{set_id}"
    )
    final_directory = (
        SNAPSHOT_MOUNTPOINT
        / set_name
    )

    created_snapshot_paths: list[Path] = []
    created_archive_paths: list[Path] = []
    committed = False

    try:
        pending_directory.mkdir(
            mode=0o755,
            parents=False,
            exist_ok=False,
        )
        pending_directory.chmod(0o755)

        subvolumes_directory = (
            pending_directory
            / "subvolumes"
        )
        subvolumes_directory.mkdir(
            mode=0o755,
            parents=False,
            exist_ok=False,
        )
        subvolumes_directory.chmod(0o755)

        members: list[dict[str, object]] = []

        for source in sources:
            current_source = validate_snapshot_source(
                top_level_mount,
                source.subvolume_id,
            )

            if (
                current_source.name != source.name
                or current_source.uuid != source.uuid
            ):
                raise HelperError(
                    f"Snapshot source changed during transaction: "
                    f"{source.name}"
                )

            source_path = (
                top_level_mount
                / source.name
            )
            snapshot_path = (
                subvolumes_directory
                / source.name
            )

            run_command(
                [
                    str(BTRFS_COMMAND),
                    "subvolume",
                    "snapshot",
                    "-r",
                    str(source_path),
                    str(snapshot_path),
                ]
            )

            created_snapshot_paths.append(
                snapshot_path
            )

            snapshot = read_subvolume_record(
                snapshot_path
            )

            if snapshot.parent_uuid != source.uuid:
                raise HelperError(
                    "The created snapshot does not reference the "
                    f"expected source UUID for {source.name}."
                )

            if not subvolume_is_read_only(
                snapshot_path
            ):
                raise HelperError(
                    f"The created Btrfs snapshot is not read-only: "
                    f"{source.name}"
                )

            members.append(
                {
                    "source": {
                        "id": source.subvolume_id,
                        "path": source.name,
                        "uuid": source.uuid,
                        "mountpoint": (
                            source_mountpoints.get(
                                source.name
                            )
                            if source_mountpoints is not None
                            else get_subvolume_mountpoint(
                                root,
                                source.name,
                            )
                        ),
                    },
                    "snapshot": {
                        "id": snapshot.subvolume_id,
                        "path": f"subvolumes/{source.name}",
                        "uuid": snapshot.uuid,
                        "parent_uuid": snapshot.parent_uuid,
                        "read_only": True,
                    },
                }
            )

        boot_metadata, created_archive_paths = (
            materialize_boot_coverage(
                pending_directory,
                boot_plan,
            )
        )

        manifest: dict[str, object] = {
            "format_version": 1,
            "set_id": set_id,
            "set_name": set_name,
            "created_at": created_at.isoformat(
                timespec="seconds"
            ),
            "scope": scope,
            "purpose": purpose,
            "filesystem_uuid": root.uuid,
            "members": members,
        }

        if boot_metadata is not None:
            manifest["boot"] = boot_metadata

        write_pending_manifest(
            pending_directory,
            manifest,
        )
        fsync_directory(
            subvolumes_directory
        )

        archives_directory = (
            pending_directory
            / "archives"
        )

        if archives_directory.exists():
            fsync_directory(
                archives_directory
            )

        fsync_directory(
            pending_directory
        )

        if final_directory.exists():
            raise HelperError(
                "The final snapshot set directory already exists."
            )

        os.rename(
            pending_directory,
            final_directory,
        )
        fsync_directory(
            SNAPSHOT_MOUNTPOINT
        )
        committed = True

        return manifest

    except Exception as operation_error:
        rollback_errors: list[str] = []

        if (
            not committed
            and final_directory.exists()
            and not pending_directory.exists()
        ):
            try:
                os.rename(
                    final_directory,
                    pending_directory,
                )
            except Exception as exc:
                rollback_errors.append(
                    f"restore pending name: {exc}"
                )

        if (
            not committed
            and pending_directory.exists()
        ):
            rollback_errors.extend(
                rollback_pending_snapshot_set(
                    pending_directory,
                    created_snapshot_paths,
                    created_archive_paths,
                )
            )

        if rollback_errors:
            raise HelperError(
                f"{operation_error} "
                "Rollback was incomplete: "
                + "; ".join(rollback_errors)
            ) from operation_error

        raise



def _parse_automation_enabled(value: str) -> bool:
    """Parse the ENABLED value used by automation.conf."""

    normalized = value.strip().lower()

    if normalized == "true":
        return True

    if normalized == "false":
        return False

    raise HelperError(
        "Automation ENABLED must be true or false."
    )


def _parse_automation_frequency(value: str) -> str:
    """Validate one supported automatic snapshot frequency."""

    normalized = value.strip().lower()

    if normalized not in AUTOMATION_FREQUENCIES:
        raise HelperError(
            "Automation FREQUENCY must be hourly, daily, weekly or monthly."
        )

    return normalized


def _parse_automation_retention(value: str) -> int:
    """Validate automatic snapshot retention."""

    try:
        retention = int(
            value.strip(),
            10,
        )
    except ValueError as exc:
        raise HelperError(
            "Automation RETENTION must be an integer."
        ) from exc

    if (
        retention < 1
        or retention > AUTOMATION_MAX_RETENTION
    ):
        raise HelperError(
            "Automation RETENTION must be between 1 and "
            f"{AUTOMATION_MAX_RETENTION}."
        )

    return retention


def read_automation_config() -> AutomationConfig:
    """
    Read and validate /etc/synex-snapshots/automation.conf.

    A missing file is interpreted as automation disabled using the
    built-in defaults. Invalid existing configuration is never silently
    accepted.
    """

    if not AUTOMATION_CONFIG_PATH.exists():
        return AutomationConfig(
            enabled=AUTOMATION_DEFAULT_ENABLED,
            frequency=AUTOMATION_DEFAULT_FREQUENCY,
            retention=AUTOMATION_DEFAULT_RETENTION,
        )

    if (
        AUTOMATION_CONFIG_PATH.is_symlink()
        or not AUTOMATION_CONFIG_PATH.is_file()
    ):
        raise HelperError(
            "Automation configuration is not a regular file."
        )

    values: dict[str, str] = {}
    allowed_keys = {
        "ENABLED",
        "FREQUENCY",
        "RETENTION",
    }

    try:
        with AUTOMATION_CONFIG_PATH.open(
            "r",
            encoding="utf-8",
        ) as handle:
            lines = handle.readlines()
    except OSError as exc:
        raise HelperError(
            f"Unable to read automation configuration: {exc}"
        ) from exc

    for line_number, raw_line in enumerate(lines, start=1):
        line = raw_line.strip()

        if not line or line.startswith("#"):
            continue

        if "=" not in line:
            raise HelperError(
                "Invalid automation configuration line "
                f"{line_number}."
            )

        key, value = line.split("=", 1)
        key = key.strip().upper()
        value = value.strip()

        if key not in allowed_keys:
            raise HelperError(
                f"Unsupported automation configuration key: {key}"
            )

        if key in values:
            raise HelperError(
                f"Duplicate automation configuration key: {key}"
            )

        values[key] = value

    enabled = _parse_automation_enabled(
        values.get(
            "ENABLED",
            "true" if AUTOMATION_DEFAULT_ENABLED else "false",
        )
    )
    frequency = _parse_automation_frequency(
        values.get(
            "FREQUENCY",
            AUTOMATION_DEFAULT_FREQUENCY,
        )
    )
    retention = _parse_automation_retention(
        values.get(
            "RETENTION",
            str(AUTOMATION_DEFAULT_RETENTION),
        )
    )

    return AutomationConfig(
        enabled=enabled,
        frequency=frequency,
        retention=retention,
    )


def write_automation_config(
    config: AutomationConfig,
) -> None:
    """Atomically persist one validated automation configuration."""

    AUTOMATION_CONFIG_DIR.mkdir(
        mode=0o755,
        parents=True,
        exist_ok=True,
    )

    content = (
        "# Synex Snapshots automatic snapshot configuration.\n"
        f"ENABLED={'true' if config.enabled else 'false'}\n"
        f"FREQUENCY={config.frequency}\n"
        f"RETENTION={config.retention}\n"
    )

    temporary_path = (
        AUTOMATION_CONFIG_DIR
        / f".automation.conf.tmp-{os.getpid()}"
    )

    try:
        with temporary_path.open(
            "w",
            encoding="utf-8",
        ) as handle:
            handle.write(content)
            handle.flush()
            os.fsync(handle.fileno())

        temporary_path.chmod(0o644)
        os.replace(
            temporary_path,
            AUTOMATION_CONFIG_PATH,
        )
        fsync_directory(
            AUTOMATION_CONFIG_DIR
        )

    finally:
        if temporary_path.exists():
            temporary_path.unlink()


def set_automation_config(
    enabled_text: str,
    frequency_text: str,
    retention_text: str,
) -> int:
    """Validate and save automatic snapshot configuration."""

    config = AutomationConfig(
        enabled=_parse_automation_enabled(
            enabled_text
        ),
        frequency=_parse_automation_frequency(
            frequency_text
        ),
        retention=_parse_automation_retention(
            retention_text
        ),
    )

    # Enabling automation is only valid when the running root filesystem
    # is Btrfs and canonical snapshot storage is available.
    #
    # Disabling automation is intentionally always permitted. This allows
    # an old or manually created ENABLED=true configuration to be safely
    # neutralized even when the current system no longer meets automation
    # requirements.
    if config.enabled:
        root = get_root_filesystem_info()

        mount_state = get_snapshot_mount_state(
            root
        )

        if (
            not mount_state.mounted
            or not mount_state.correct
        ):
            raise HelperError(
                "Automatic snapshots require snapshot storage "
                "configured at /.snapshots."
            )

    write_automation_config(
        config
    )

    sys.stdout.write(
        json.dumps(
            {
                "saved": True,
                "enabled": config.enabled,
                "frequency": config.frequency,
                "retention": config.retention,
            },
            ensure_ascii=False,
            indent=2,
        )
        + "\n"
    )

    return 0


def _parse_automatic_created_at(
    value: object,
) -> datetime:
    """Parse a timezone-aware snapshot creation time."""

    if not isinstance(value, str) or not value.strip():
        raise HelperError(
            "Automatic snapshot manifest has an invalid created_at value."
        )

    try:
        created_at = datetime.fromisoformat(
            value.strip()
        )
    except ValueError as exc:
        raise HelperError(
            "Automatic snapshot manifest has an invalid created_at value."
        ) from exc

    if created_at.tzinfo is None:
        raise HelperError(
            "Automatic snapshot manifest creation time has no timezone."
        )

    return created_at


def list_automatic_snapshot_records(
    expected_filesystem_uuid: str,
) -> list[AutomaticSnapshotRecord]:
    """
    Return valid committed purpose=auto snapshot sets.

    Invalid or unrelated sets are ignored. Retention only acts on sets
    that can be identified safely and unambiguously as automatic sets
    belonging to the running Btrfs filesystem.
    """

    records: list[AutomaticSnapshotRecord] = []

    try:
        entries = list(
            SNAPSHOT_MOUNTPOINT.iterdir()
        )
    except OSError as exc:
        raise HelperError(
            f"Unable to read snapshot storage: {exc}"
        ) from exc

    for entry in entries:
        if entry.name.startswith("."):
            continue

        if entry.is_symlink() or not entry.is_dir():
            continue

        manifest_path = entry / "manifest.json"

        if manifest_path.is_symlink() or not manifest_path.is_file():
            continue

        try:
            payload = load_json_object(
                manifest_path
            )

            if payload.get("format_version") != 1:
                continue

            if get_manifest_purpose(payload) != "auto":
                continue

            filesystem_uuid = normalize_uuid_text(
                payload.get("filesystem_uuid"),
                "filesystem_uuid",
            )

            if filesystem_uuid != expected_filesystem_uuid:
                continue

            set_id = normalize_uuid_text(
                payload.get("set_id"),
                "set_id",
            )

            set_name = payload.get("set_name")
            if (
                not isinstance(set_name, str)
                or not set_name
                or set_name != entry.name
            ):
                continue

            created_at = _parse_automatic_created_at(
                payload.get("created_at")
            )

        except HelperError:
            # Uncertain metadata is deliberately excluded from automatic
            # retention. Never delete a set that cannot be validated as an
            # automatic snapshot belonging to this filesystem.
            continue

        records.append(
            AutomaticSnapshotRecord(
                set_id=set_id,
                set_name=set_name,
                created_at=created_at,
            )
        )

    records.sort(
        key=lambda record: record.created_at,
        reverse=True,
    )

    return records


def _add_one_calendar_month(
    value: datetime,
) -> datetime:
    """Return the same local date/time in the following calendar month."""

    if value.month == 12:
        year = value.year + 1
        month = 1
    else:
        year = value.year
        month = value.month + 1

    day = min(
        value.day,
        calendar.monthrange(
            year,
            month,
        )[1],
    )

    return value.replace(
        year=year,
        month=month,
        day=day,
    )


def automatic_snapshot_is_due(
    frequency: str,
    last_created_at: datetime | None,
    now: datetime,
) -> bool:
    """Return whether a new automatic root snapshot is due."""

    if last_created_at is None:
        return True

    if now < last_created_at:
        return False

    if frequency == "hourly":
        due_at = last_created_at + timedelta(hours=1)
    elif frequency == "daily":
        due_at = last_created_at + timedelta(days=1)
    elif frequency == "weekly":
        due_at = last_created_at + timedelta(weeks=1)
    elif frequency == "monthly":
        due_at = _add_one_calendar_month(
            last_created_at
        )
    else:
        raise HelperError(
            f"Unsupported automatic snapshot frequency: {frequency}"
        )

    return now >= due_at


def create_full_snapshot_manifest(
    *,
    purpose: str,
) -> dict[str, object]:
    """Create a Full snapshot set and return its committed manifest."""

    if purpose not in SNAPSHOT_PURPOSES:
        raise HelperError(
            f"Unsupported snapshot creation purpose: {purpose}"
        )

    root = get_root_filesystem_info()
    mount_state = get_snapshot_mount_state(
        root
    )

    if (
        not mount_state.mounted
        or not mount_state.correct
    ):
        raise HelperError(
            "Snapshot storage is not mounted correctly at /.snapshots."
        )

    top_level_mount = mount_top_level(
        root
    )

    try:
        if validate_snapshot_subvolume(
            top_level_mount
        ):
            raise HelperError(
                "Snapshot storage is not configured. "
                "Create @snapshots before creating snapshots."
            )

        sources = list_operational_snapshot_sources(
            top_level_mount
        )

        return create_snapshot_set(
            root,
            top_level_mount,
            sources,
            scope="full",
            purpose=purpose,
        )

    finally:
        unmount_top_level(
            top_level_mount
        )



def create_automatic_root_snapshot_manifest() -> dict[str, object]:
    """Create one automatic Single snapshot of the canonical root."""

    root = get_root_filesystem_info()
    root_name = normalize_fsroot(
        root.fsroot
    )

    if not is_safe_top_level_source_path(
        root_name
    ):
        raise HelperError(
            "Automatic snapshots require the running root to be a "
            "canonical direct top-level Btrfs subvolume."
        )

    mount_state = get_snapshot_mount_state(
        root
    )

    if (
        not mount_state.mounted
        or not mount_state.correct
    ):
        raise HelperError(
            "Snapshot storage is not mounted correctly at /.snapshots."
        )

    top_level_mount = mount_top_level(
        root
    )

    try:
        if validate_snapshot_subvolume(
            top_level_mount
        ):
            raise HelperError(
                "Snapshot storage is not configured. "
                "Create @snapshots before creating snapshots."
            )

        root_path = (
            top_level_mount
            / root_name
        )
        source = read_subvolume_record(
            root_path
        )

        if (
            source.parent_id != TOP_LEVEL_SUBVOLUME_ID
            or source.name != root_name
            or subvolume_is_read_only(root_path)
        ):
            raise HelperError(
                "The running root is not a valid writable top-level "
                "Btrfs snapshot source."
            )

        return create_snapshot_set(
            root,
            top_level_mount,
            [source],
            scope="single",
            purpose="auto",
            source_mountpoints={
                root_name: "/",
            },
            root_source_name=root_name,
        )

    finally:
        unmount_top_level(
            top_level_mount
        )

def apply_automatic_retention(
    expected_filesystem_uuid: str,
    retention: int,
) -> list[str]:
    """
    Delete only excess purpose=auto snapshot sets, oldest first.

    Manual and pre-restore sets are never candidates for automatic
    retention.
    """

    records = list_automatic_snapshot_records(
        expected_filesystem_uuid
    )
    excess_records = records[retention:]
    deleted_set_ids: list[str] = []

    for record in reversed(excess_records):
        captured_stdout = io.StringIO()

        with contextlib.redirect_stdout(
            captured_stdout
        ):
            result = delete_snapshot_set(
                record.set_id
            )

        if result != 0:
            raise HelperError(
                "Unable to delete an expired automatic snapshot set."
            )

        deleted_set_ids.append(
            record.set_id
        )

    return deleted_set_ids


def automation_run() -> int:
    """
    Evaluate automatic snapshot policy and converge retention.

    The systemd timer may invoke this action every hour. The helper is
    responsible for deciding whether the configured Hourly, Daily,
    Weekly or Monthly interval has actually elapsed.

    Unsupported runtime layouts are treated as a clean no-op. This keeps
    the systemd oneshot healthy even if an old configuration still has
    automation enabled on a system that no longer uses Btrfs or no longer
    has canonical snapshot storage available.
    """

    config = read_automation_config()

    if not config.enabled:
        sys.stdout.write(
            json.dumps(
                {
                    "enabled": False,
                    "created": False,
                    "reason": "disabled",
                },
                ensure_ascii=False,
                indent=2,
            )
            + "\n"
        )
        return 0

    if RESTORE_STATE_PATH.exists():
        sys.stdout.write(
            json.dumps(
                {
                    "enabled": True,
                    "created": False,
                    "reason": "restore-pending",
                },
                ensure_ascii=False,
                indent=2,
            )
            + "\n"
        )
        return 0

    try:
        root = get_root_filesystem_info()

    except HelperError as exc:
        if str(exc) != "The root filesystem is not Btrfs.":
            raise

        sys.stdout.write(
            json.dumps(
                {
                    "enabled": True,
                    "frequency": config.frequency,
                    "retention": config.retention,
                    "due": False,
                    "created": False,
                    "reason": "unsupported-root",
                    "deleted_set_ids": [],
                },
                ensure_ascii=False,
                indent=2,
            )
            + "\n"
        )

        return 0

    running_root_name = normalize_fsroot(
        root.fsroot
    )

    if not is_safe_top_level_source_path(
        running_root_name
    ):
        reason = (
            "snapshot-boot"
            if running_root_name.startswith(
                f"{SNAPSHOT_SUBVOLUME}/"
            )
            else "unsupported-root-layout"
        )

        sys.stdout.write(
            json.dumps(
                {
                    "enabled": True,
                    "frequency": config.frequency,
                    "retention": config.retention,
                    "due": False,
                    "created": False,
                    "reason": reason,
                    "deleted_set_ids": [],
                },
                ensure_ascii=False,
                indent=2,
            )
            + "\n"
        )

        return 0

    mount_state = get_snapshot_mount_state(
        root
    )

    if (
        not mount_state.mounted
        or not mount_state.correct
    ):
        sys.stdout.write(
            json.dumps(
                {
                    "enabled": True,
                    "frequency": config.frequency,
                    "retention": config.retention,
                    "due": False,
                    "created": False,
                    "reason": "storage-unavailable",
                    "deleted_set_ids": [],
                },
                ensure_ascii=False,
                indent=2,
            )
            + "\n"
        )

        return 0

    records_before = list_automatic_snapshot_records(
        root.uuid
    )
    last_created_at = (
        records_before[0].created_at
        if records_before
        else None
    )
    now = datetime.now().astimezone()
    due = automatic_snapshot_is_due(
        config.frequency,
        last_created_at,
        now,
    )

    created_manifest: dict[str, object] | None = None

    if due:
        created_manifest = create_automatic_root_snapshot_manifest()

    deleted_set_ids = apply_automatic_retention(
        root.uuid,
        config.retention,
    )

    sys.stdout.write(
        json.dumps(
            {
                "enabled": True,
                "frequency": config.frequency,
                "retention": config.retention,
                "due": due,
                "created": created_manifest is not None,
                "created_set_id": (
                    created_manifest.get("set_id")
                    if created_manifest is not None
                    else None
                ),
                "created_set_name": (
                    created_manifest.get("set_name")
                    if created_manifest is not None
                    else None
                ),
                "deleted_set_ids": deleted_set_ids,
            },
            ensure_ascii=False,
            indent=2,
        )
        + "\n"
    )

    return 0


def create_single_snapshot(
    subvolume_id: int,
) -> int:
    """
    Create one read-only Single snapshot set.
    """

    root = get_root_filesystem_info()
    mount_state = get_snapshot_mount_state(
        root
    )

    if (
        not mount_state.mounted
        or not mount_state.correct
    ):
        raise HelperError(
            "Snapshot storage is not mounted correctly at /.snapshots."
        )

    top_level_mount = mount_top_level(
        root
    )

    try:
        if validate_snapshot_subvolume(
            top_level_mount
        ):
            raise HelperError(
                "Snapshot storage is not configured. "
                "Create @snapshots before creating snapshots."
            )

        source = validate_snapshot_source(
            top_level_mount,
            subvolume_id,
        )

        manifest = create_snapshot_set(
            root,
            top_level_mount,
            [source],
            scope="single",
            purpose="manual",
        )

        sys.stdout.write(
            json.dumps(
                manifest,
                ensure_ascii=False,
                indent=2,
            )
            + "\n"
        )

        return 0

    finally:
        unmount_top_level(
            top_level_mount
        )


def create_full_snapshot() -> int:
    """Create one manual read-only Full snapshot set."""

    manifest = create_full_snapshot_manifest(
        purpose="manual"
    )

    sys.stdout.write(
        json.dumps(
            manifest,
            ensure_ascii=False,
            indent=2,
        )
        + "\n"
    )

    return 0



def normalize_uuid_text(
    value: object,
    field_name: str,
) -> str:
    """
    Validate and normalize a UUID string.
    """

    if not isinstance(value, str) or not value.strip():
        raise HelperError(
            f"Invalid UUID field: {field_name}"
        )

    try:
        normalized = str(
            uuid.UUID(value.strip())
        )
    except (ValueError, AttributeError) as exc:
        raise HelperError(
            f"Invalid UUID field: {field_name}"
        ) from exc

    return normalized


def load_json_object(
    path: Path,
) -> dict[str, object]:
    """
    Read one JSON object from disk.
    """

    if path.is_symlink() or not path.is_file():
        raise HelperError(
            f"Required file is missing or invalid: {path}"
        )

    try:
        with path.open(
            "r",
            encoding="utf-8",
        ) as handle:
            payload = json.load(handle)
    except json.JSONDecodeError as exc:
        raise HelperError(
            f"Invalid JSON file: {path}"
        ) from exc
    except OSError as exc:
        raise HelperError(
            f"Unable to read {path}: {exc}"
        ) from exc

    if not isinstance(payload, dict):
        raise HelperError(
            f"JSON root must be an object: {path}"
        )

    return payload


def find_snapshot_set_directory_by_id(
    set_id: str,
) -> Path | None:
    """
    Find one committed snapshot set by its manifest set_id.

    Hidden transaction directories are not considered committed sets.
    """

    matches: list[Path] = []

    try:
        entries = list(
            SNAPSHOT_MOUNTPOINT.iterdir()
        )
    except OSError as exc:
        raise HelperError(
            f"Unable to read snapshot storage: {exc}"
        ) from exc

    for entry in entries:
        if entry.name.startswith("."):
            continue

        if entry.is_symlink() or not entry.is_dir():
            continue

        manifest_path = (
            entry
            / "manifest.json"
        )

        if manifest_path.is_symlink() or not manifest_path.is_file():
            continue

        try:
            payload = load_json_object(
                manifest_path
            )
        except HelperError:
            continue

        candidate_id = payload.get(
            "set_id"
        )

        if not isinstance(candidate_id, str):
            continue

        try:
            candidate_id = normalize_uuid_text(
                candidate_id,
                "set_id",
            )
        except HelperError:
            continue

        if candidate_id == set_id:
            matches.append(entry)

    if len(matches) > 1:
        raise HelperError(
            "Multiple snapshot sets use the requested set ID. "
            "The operation was aborted because the state is ambiguous."
        )

    if not matches:
        return None

    return matches[0]


def validate_boot_metadata_for_deletion(
    payload: dict[str, object],
    set_directory: Path,
    *,
    allow_missing_files: bool,
) -> tuple[set[str], list[Path]]:
    """
    Validate optional /boot metadata and return allowed root entries and
    archive files eligible for deletion.

    Manifests created before boot coverage was introduced may omit the
    boot field and remain deletable.
    """

    boot_payload = payload.get("boot")

    if boot_payload is None:
        return set(), []

    if not isinstance(boot_payload, dict):
        raise HelperError(
            "Snapshot manifest contains invalid boot metadata."
        )

    mode = boot_payload.get("mode")

    if boot_payload.get("mountpoint") != "/boot":
        raise HelperError(
            "Snapshot manifest contains an invalid boot mountpoint."
        )

    if mode == "included-in-root":
        if set(boot_payload) != {
            "mode",
            "mountpoint",
        }:
            raise HelperError(
                "Snapshot manifest contains unexpected included-root "
                "boot metadata."
            )

        return set(), []

    if mode != "archive":
        raise HelperError(
            "Snapshot manifest contains an unsupported boot mode."
        )

    if set(boot_payload) != {
        "mode",
        "mountpoint",
        "source",
        "fstype",
        "uuid",
        "archive",
    }:
        raise HelperError(
            "Snapshot manifest contains unexpected boot archive metadata."
        )

    source = boot_payload.get("source")
    fstype = boot_payload.get("fstype")
    boot_uuid = normalize_uuid_text(
        boot_payload.get("uuid"),
        "boot.uuid",
    )

    if not isinstance(source, str) or not source.strip():
        raise HelperError(
            "Snapshot manifest contains an invalid boot source."
        )

    if fstype != "ext4":
        raise HelperError(
            "Snapshot manifest contains an unsupported boot filesystem."
        )

    archive = boot_payload.get("archive")

    if not isinstance(archive, dict):
        raise HelperError(
            "Snapshot manifest contains invalid boot archive metadata."
        )

    if set(archive) != {
        "path",
        "compression",
        "size_bytes",
        "sha256",
    }:
        raise HelperError(
            "Snapshot manifest contains unexpected boot archive fields."
        )

    if archive.get("path") != "archives/boot.tar.zst":
        raise HelperError(
            "Snapshot manifest contains an invalid boot archive path."
        )

    if archive.get("compression") != "zstd":
        raise HelperError(
            "Snapshot manifest contains an unsupported boot archive "
            "compression."
        )

    size_value = archive.get("size_bytes")

    if (
        isinstance(size_value, bool)
        or not isinstance(size_value, int)
        or size_value <= 0
    ):
        raise HelperError(
            "Snapshot manifest contains an invalid boot archive size."
        )

    sha256_value = archive.get("sha256")

    if (
        not isinstance(sha256_value, str)
        or re.fullmatch(r"[0-9a-f]{64}", sha256_value) is None
    ):
        raise HelperError(
            "Snapshot manifest contains an invalid boot archive checksum."
        )

    # Keep the normalized UUID validation above even though deletion does
    # not need the value directly. This rejects malformed metadata before
    # any destructive operation is allowed.
    _ = boot_uuid

    archives_directory = (
        set_directory
        / "archives"
    )
    archive_path = (
        archives_directory
        / "boot.tar.zst"
    )

    if archive_path.exists():
        if archive_path.is_symlink() or not archive_path.is_file():
            raise HelperError(
                "Boot archive is not a regular file."
            )

        try:
            actual_size = archive_path.stat().st_size
        except OSError as exc:
            raise HelperError(
                f"Unable to stat the boot archive: {exc}"
            ) from exc

        if actual_size != size_value:
            raise HelperError(
                "Boot archive size does not match the snapshot manifest."
            )

        if sha256_file(archive_path) != sha256_value:
            raise HelperError(
                "Boot archive checksum does not match the snapshot manifest."
            )

    elif not allow_missing_files:
        raise HelperError(
            "Boot archive is missing."
        )

    if archives_directory.exists():
        if archives_directory.is_symlink() or not archives_directory.is_dir():
            raise HelperError(
                "Snapshot archives path is not a directory."
            )

        try:
            actual_archive_names = {
                entry.name
                for entry in archives_directory.iterdir()
            }
        except OSError as exc:
            raise HelperError(
                f"Unable to inspect snapshot archives: {exc}"
            ) from exc

        unexpected_archive_names = (
            actual_archive_names
            - {"boot.tar.zst"}
        )

        if unexpected_archive_names:
            raise HelperError(
                "Snapshot set contains unexpected archive data. "
                "Deletion was aborted: "
                + ", ".join(
                    sorted(unexpected_archive_names)
                )
            )

    elif not allow_missing_files:
        raise HelperError(
            "Snapshot archives directory is missing."
        )

    return {"archives"}, [archive_path]


def get_manifest_purpose(
    payload: dict[str, object],
) -> str:
    """
    Return and validate the snapshot-set purpose.

    Format-version 1 manifests created before purpose was introduced are
    intentionally interpreted as manual snapshots for backward
    compatibility. This is the safest retention behavior.
    """

    value = payload.get(
        "purpose",
        "manual",
    )

    if not isinstance(value, str) or value not in SNAPSHOT_PURPOSES:
        raise HelperError(
            "Snapshot manifest has an unsupported purpose."
        )

    return value


def validate_snapshot_set_for_deletion(
    set_directory: Path,
    *,
    expected_set_id: str,
    expected_filesystem_uuid: str,
    storage_subvolume_id: int,
    allow_missing_snapshots: bool,
    deleting_directory: bool,
) -> tuple[str, list[Path], list[Path]]:
    """
    Validate one snapshot set before or during deletion.

    Only paths explicitly described by a valid Synex Snapshots manifest
    are eligible for deletion. Unexpected files are never removed.
    """

    manifest_path = (
        set_directory
        / "manifest.json"
    )

    payload = load_json_object(
        manifest_path
    )

    format_version = payload.get(
        "format_version"
    )

    if (
        isinstance(format_version, bool)
        or format_version != 1
    ):
        raise HelperError(
            "Unsupported snapshot manifest format."
        )

    set_id = normalize_uuid_text(
        payload.get("set_id"),
        "set_id",
    )

    if set_id != expected_set_id:
        raise HelperError(
            "Snapshot manifest set ID does not match the requested set."
        )

    set_name_value = payload.get(
        "set_name"
    )

    if (
        not isinstance(set_name_value, str)
        or not set_name_value.strip()
    ):
        raise HelperError(
            "Snapshot manifest has an invalid set_name."
        )

    set_name = set_name_value.strip()

    if (
        not deleting_directory
        and set_name != set_directory.name
    ):
        raise HelperError(
            "Snapshot manifest set_name does not match its directory."
        )

    filesystem_uuid = normalize_uuid_text(
        payload.get("filesystem_uuid"),
        "filesystem_uuid",
    )

    if filesystem_uuid != expected_filesystem_uuid:
        raise HelperError(
            "Snapshot set belongs to a different Btrfs filesystem."
        )

    scope = payload.get("scope")

    if scope not in {
        "single",
        "full",
    }:
        raise HelperError(
            "Snapshot manifest has an unsupported scope."
        )

    get_manifest_purpose(
        payload
    )

    members = payload.get(
        "members"
    )

    if not isinstance(members, list) or not members:
        raise HelperError(
            "Snapshot set has no valid members."
        )

    if scope == "single" and len(members) != 1:
        raise HelperError(
            "Single snapshot set must contain exactly one member."
        )

    boot_root_entries, archive_paths = (
        validate_boot_metadata_for_deletion(
            payload,
            set_directory,
            allow_missing_files=allow_missing_snapshots,
        )
    )

    expected_names: set[str] = set()
    snapshot_paths: list[Path] = []
    seen_snapshot_ids: set[int] = set()
    seen_snapshot_uuids: set[str] = set()

    for index, member in enumerate(members):
        if not isinstance(member, dict):
            raise HelperError(
                f"Invalid snapshot member at index {index}."
            )

        source = member.get("source")
        snapshot = member.get("snapshot")

        if not isinstance(source, dict) or not isinstance(snapshot, dict):
            raise HelperError(
                f"Invalid snapshot member at index {index}."
            )

        source_name = source.get("path")

        if (
            not isinstance(source_name, str)
            or not source_name
            or source_name in {".", ".."}
            or "/" in source_name
            or "\x00" in source_name
            or Path(source_name).name != source_name
        ):
            raise HelperError(
                "Snapshot manifest contains an invalid source path."
            )

        if source_name == SNAPSHOT_SUBVOLUME:
            raise HelperError(
                "Snapshot storage cannot be a snapshot member."
            )

        source_uuid = normalize_uuid_text(
            source.get("uuid"),
            "source.uuid",
        )

        snapshot_id_value = snapshot.get("id")

        if (
            isinstance(snapshot_id_value, bool)
            or not isinstance(snapshot_id_value, int)
            or snapshot_id_value <= 0
        ):
            raise HelperError(
                "Snapshot manifest contains an invalid snapshot ID."
            )

        snapshot_id = snapshot_id_value

        if snapshot_id in seen_snapshot_ids:
            raise HelperError(
                "Snapshot manifest contains a duplicate snapshot ID."
            )

        snapshot_relative_path = snapshot.get(
            "path"
        )
        expected_relative_path = (
            f"subvolumes/{source_name}"
        )

        if snapshot_relative_path != expected_relative_path:
            raise HelperError(
                "Snapshot path does not match its source subvolume."
            )

        snapshot_uuid = normalize_uuid_text(
            snapshot.get("uuid"),
            "snapshot.uuid",
        )
        parent_uuid = normalize_uuid_text(
            snapshot.get("parent_uuid"),
            "snapshot.parent_uuid",
        )

        if parent_uuid != source_uuid:
            raise HelperError(
                "Snapshot parent UUID does not match its source UUID."
            )

        if snapshot_uuid == source_uuid:
            raise HelperError(
                "Snapshot UUID matches its source UUID."
            )

        if snapshot_uuid in seen_snapshot_uuids:
            raise HelperError(
                "Snapshot manifest contains a duplicate snapshot UUID."
            )

        if snapshot.get("read_only") is not True:
            raise HelperError(
                "Snapshot manifest does not describe a read-only snapshot."
            )

        snapshot_path = (
            set_directory
            / "subvolumes"
            / source_name
        )

        expected_names.add(
            source_name
        )
        seen_snapshot_ids.add(
            snapshot_id
        )
        seen_snapshot_uuids.add(
            snapshot_uuid
        )
        snapshot_paths.append(
            snapshot_path
        )

        if not snapshot_path.exists():
            if allow_missing_snapshots:
                continue

            raise HelperError(
                f"Snapshot subvolume is missing: {source_name}"
            )

        actual = read_subvolume_record(
            snapshot_path
        )

        if actual.subvolume_id != snapshot_id:
            raise HelperError(
                f"Snapshot ID mismatch for {source_name}."
            )

        if actual.uuid != snapshot_uuid:
            raise HelperError(
                f"Snapshot UUID mismatch for {source_name}."
            )

        if actual.parent_uuid != parent_uuid:
            raise HelperError(
                f"Snapshot parent UUID mismatch for {source_name}."
            )

        if actual.parent_id != storage_subvolume_id:
            raise HelperError(
                f"Snapshot {source_name} is not stored directly "
                "inside @snapshots."
            )

        if not subvolume_is_read_only(
            snapshot_path
        ):
            raise HelperError(
                f"Snapshot {source_name} is no longer read-only. "
                "Deletion was aborted to avoid removing modified data."
            )

    allowed_root_entries = {
        "manifest.json",
        "subvolumes",
        *boot_root_entries,
    }

    try:
        actual_root_entries = {
            entry.name
            for entry in set_directory.iterdir()
        }
    except OSError as exc:
        raise HelperError(
            f"Unable to inspect snapshot set directory: {exc}"
        ) from exc

    unexpected_root_entries = (
        actual_root_entries
        - allowed_root_entries
    )

    if unexpected_root_entries:
        raise HelperError(
            "Snapshot set contains unexpected data. "
            "Deletion was aborted: "
            + ", ".join(
                sorted(unexpected_root_entries)
            )
        )

    subvolumes_directory = (
        set_directory
        / "subvolumes"
    )

    if not subvolumes_directory.is_dir():
        if allow_missing_snapshots:
            actual_names: set[str] = set()
        else:
            raise HelperError(
                "Snapshot set subvolumes directory is missing."
            )
    else:
        try:
            actual_names = {
                entry.name
                for entry in subvolumes_directory.iterdir()
            }
        except OSError as exc:
            raise HelperError(
                f"Unable to inspect snapshot members: {exc}"
            ) from exc

    unexpected_names = (
        actual_names
        - expected_names
    )

    if unexpected_names:
        raise HelperError(
            "Snapshot set contains unexpected subvolume data. "
            "Deletion was aborted: "
            + ", ".join(
                sorted(unexpected_names)
            )
        )

    return (
        set_name,
        snapshot_paths,
        archive_paths,
    )


def finish_empty_deleting_directory(
    deleting_directory: Path,
    set_id: str,
) -> bool:
    """
    Finish final cleanup after an interrupted deletion removed manifest.

    Only empty known transaction directories are removed without a
    manifest. Any other content requires manual inspection.
    """

    manifest_path = (
        deleting_directory
        / "manifest.json"
    )

    if manifest_path.exists():
        return False

    allowed_entries: set[str] = set()

    for directory_name in (
        "subvolumes",
        "archives",
    ):
        directory = (
            deleting_directory
            / directory_name
        )

        if not directory.exists():
            continue

        if directory.is_symlink() or not directory.is_dir():
            raise HelperError(
                "Interrupted deletion contains unexpected data."
            )

        try:
            if any(directory.iterdir()):
                raise HelperError(
                    "Interrupted deletion contains data but its manifest "
                    "is missing. Manual inspection is required."
                )
        except OSError as exc:
            raise HelperError(
                f"Unable to inspect interrupted deletion state: {exc}"
            ) from exc

        allowed_entries.add(
            directory_name
        )

    try:
        entries = list(
            deleting_directory.iterdir()
        )
    except OSError as exc:
        raise HelperError(
            f"Unable to inspect interrupted deletion state: {exc}"
        ) from exc

    unexpected = {
        entry.name
        for entry in entries
    } - allowed_entries

    if unexpected:
        raise HelperError(
            "Interrupted deletion contains unexpected data. "
            "Manual inspection is required."
        )

    for directory_name in (
        "archives",
        "subvolumes",
    ):
        directory = (
            deleting_directory
            / directory_name
        )

        if directory.exists():
            directory.rmdir()

    deleting_directory.rmdir()
    fsync_directory(
        SNAPSHOT_MOUNTPOINT
    )

    sys.stdout.write(
        json.dumps(
            {
                "set_id": set_id,
                "deleted": True,
                "resumed": True,
                "already_absent": False,
            },
            ensure_ascii=False,
            indent=2,
        )
        + "\n"
    )

    return True


def delete_snapshot_set(
    set_id: str,
) -> int:
    """
    Delete one committed Synex Snapshots snapshot set safely.

    The public argument is a UUID set_id, never a filesystem path.
    The set is renamed to .deleting-<set_id> before destructive work.
    If deletion is interrupted, calling the same action again resumes
    the operation from that private transaction directory.
    """

    root = get_root_filesystem_info()
    mount_state = get_snapshot_mount_state(
        root
    )

    if (
        not mount_state.mounted
        or not mount_state.correct
    ):
        raise HelperError(
            "Snapshot storage is not mounted correctly at /.snapshots."
        )

    storage = read_subvolume_record(
        SNAPSHOT_MOUNTPOINT
    )

    if storage.name != SNAPSHOT_SUBVOLUME:
        raise HelperError(
            "Mounted snapshot storage is not @snapshots."
        )

    deleting_directory = (
        SNAPSHOT_MOUNTPOINT
        / f".deleting-{set_id}"
    )

    if deleting_directory.is_symlink():
        raise HelperError(
            "Snapshot deletion marker is a symbolic link. "
            "Deletion was aborted."
        )

    committed_directory = (
        find_snapshot_set_directory_by_id(
            set_id
        )
    )

    if (
        deleting_directory.exists()
        and committed_directory is not None
    ):
        raise HelperError(
            "Both committed and in-progress deletion states exist for "
            "the same snapshot set. Manual inspection is required."
        )

    resumed = False

    if deleting_directory.exists():
        if not deleting_directory.is_dir():
            raise HelperError(
                "Snapshot deletion marker is not a directory."
            )

        if finish_empty_deleting_directory(
            deleting_directory,
            set_id,
        ):
            return 0

        working_directory = (
            deleting_directory
        )
        resumed = True

    elif committed_directory is not None:
        working_directory = (
            committed_directory
        )

    else:
        sys.stdout.write(
            json.dumps(
                {
                    "set_id": set_id,
                    "deleted": False,
                    "resumed": False,
                    "already_absent": True,
                },
                ensure_ascii=False,
                indent=2,
            )
            + "\n"
        )
        return 0

    try:
        set_name, snapshot_paths, archive_paths = (
            validate_snapshot_set_for_deletion(
                working_directory,
                expected_set_id=set_id,
                expected_filesystem_uuid=root.uuid,
                storage_subvolume_id=(
                    storage.subvolume_id
                ),
                allow_missing_snapshots=resumed,
                deleting_directory=resumed,
            )
        )

        if not resumed:
            if deleting_directory.exists():
                raise HelperError(
                    "Snapshot deletion marker already exists."
                )

            os.rename(
                working_directory,
                deleting_directory,
            )
            fsync_directory(
                SNAPSHOT_MOUNTPOINT
            )
            working_directory = (
                deleting_directory
            )

            snapshot_paths = [
                working_directory
                / "subvolumes"
                / path.name
                for path in snapshot_paths
            ]

            archive_paths = [
                working_directory
                / "archives"
                / path.name
                for path in archive_paths
            ]

        for archive_path in archive_paths:
            if not archive_path.exists():
                continue

            if archive_path.is_symlink() or not archive_path.is_file():
                raise HelperError(
                    "Snapshot archive changed during deletion."
                )

            archive_path.unlink()

        archives_directory = (
            working_directory
            / "archives"
        )

        if archives_directory.exists():
            archives_directory.rmdir()

        for snapshot_path in snapshot_paths:
            if not snapshot_path.exists():
                continue

            run_command(
                [
                    str(BTRFS_COMMAND),
                    "subvolume",
                    "delete",
                    str(snapshot_path),
                ]
            )

        subvolumes_directory = (
            working_directory
            / "subvolumes"
        )

        if subvolumes_directory.exists():
            subvolumes_directory.rmdir()

        manifest_path = (
            working_directory
            / "manifest.json"
        )

        if manifest_path.exists():
            manifest_path.unlink()

        fsync_directory(
            working_directory
        )

        working_directory.rmdir()
        fsync_directory(
            SNAPSHOT_MOUNTPOINT
        )

        sys.stdout.write(
            json.dumps(
                {
                    "set_id": set_id,
                    "set_name": set_name,
                    "deleted": True,
                    "resumed": resumed,
                    "already_absent": False,
                },
                ensure_ascii=False,
                indent=2,
            )
            + "\n"
        )

        return 0

    except Exception as exc:
        marker_exists = (
            deleting_directory.exists()
        )

        if marker_exists:
            raise HelperError(
                f"{exc} Snapshot deletion is incomplete and remains "
                f"marked as .deleting-{set_id}. Re-running the same "
                "delete action will resume it safely."
            ) from exc

        raise

def _write_restore_preflight_result(
    result: dict[str, object],
) -> int:
    """
    Write one structured restore preflight result to stdout.

    A completed preflight returns exit code 0 even when preflight_ok is
    false. Command failures and states that prevent evaluation entirely
    continue to raise HelperError and return a non-zero exit status.
    """

    sys.stdout.write(
        json.dumps(
            result,
            ensure_ascii=False,
            indent=2,
        )
        + "\n"
    )

    return 0


def _restore_preflight_base_result(
    *,
    set_id: str,
    set_name: str,
    scope: str,
    purpose: str,
    filesystem_uuid: str,
    current_filesystem_uuid: str,
) -> dict[str, object]:
    """
    Build the common restore preflight result structure.
    """

    return {
        "format_version": 1,
        "operation": "restore-preflight",
        "set_id": set_id,
        "set_name": set_name,
        "scope": scope,
        "purpose": purpose,
        "filesystem_uuid": filesystem_uuid,
        "current_filesystem_uuid": current_filesystem_uuid,
        "filesystem_matches": (
            filesystem_uuid == current_filesystem_uuid
        ),
        "members": [],
        "boot": {
            "required": False,
            "mode": None,
            "compatible": True,
        },
        "contains_root": False,
        "preflight_ok": False,
        "errors": [],
    }


def _parse_restore_source_mountpoint(
    value: object,
    *,
    member_index: int,
) -> str | None:
    """
    Validate the historical source mountpoint stored in one member.
    """

    if value is None:
        return None

    if not isinstance(value, str) or not value.strip():
        raise HelperError(
            f"Invalid source mountpoint at member {member_index}."
        )

    mountpoint = value.strip()

    if not mountpoint.startswith("/"):
        raise HelperError(
            f"Source mountpoint is not absolute at member {member_index}."
        )

    return mountpoint


def _build_restore_target_state(
    root: RootFilesystemInfo,
    top_level_mount: Path,
    source_name: str,
    *,
    original_uuid: str,
) -> tuple[dict[str, object], list[str]]:
    """
    Inspect the current logical restore target for one snapshot member.

    The historical source UUID is intentionally not a restore
    requirement. It is exposed only as provenance information and for
    an informational comparison with the current target UUID.
    """

    errors: list[str] = []
    target_path = (
        top_level_mount
        / source_name
    )
    current_mountpoint = get_subvolume_mountpoint(
        root,
        source_name,
    )

    if target_path.is_symlink():
        errors.append(
            f"Restore target path is a symbolic link: {source_name}"
        )

        return (
            {
                "present": True,
                "id": None,
                "uuid": None,
                "mountpoint": current_mountpoint,
                "mounted": current_mountpoint is not None,
                "read_only": None,
                "matches_original_uuid": None,
            },
            errors,
        )

    if not target_path.exists():
        return (
            {
                "present": False,
                "id": None,
                "uuid": None,
                "mountpoint": current_mountpoint,
                "mounted": current_mountpoint is not None,
                "read_only": None,
                "matches_original_uuid": None,
            },
            errors,
        )

    if not target_path.is_dir():
        errors.append(
            f"Restore target path is occupied by a non-directory "
            f"object: {source_name}"
        )

        return (
            {
                "present": True,
                "id": None,
                "uuid": None,
                "mountpoint": current_mountpoint,
                "mounted": current_mountpoint is not None,
                "read_only": None,
                "matches_original_uuid": None,
            },
            errors,
        )

    try:
        target = read_subvolume_record(
            target_path
        )
    except HelperError as exc:
        errors.append(
            f"Restore target is not a valid Btrfs subvolume "
            f"for {source_name}: {exc}"
        )

        return (
            {
                "present": True,
                "id": None,
                "uuid": None,
                "mountpoint": current_mountpoint,
                "mounted": current_mountpoint is not None,
                "read_only": None,
                "matches_original_uuid": None,
            },
            errors,
        )

    if target.name != source_name:
        errors.append(
            f"Restore target name mismatch for {source_name}."
        )

    if target.parent_id != TOP_LEVEL_SUBVOLUME_ID:
        errors.append(
            f"Restore target is not a top-level Btrfs subvolume: "
            f"{source_name}"
        )

    # Parent UUID is lineage metadata, not an operational-role test.
    # A target created by a previous restore is expected to have one.

    if target.name == SNAPSHOT_SUBVOLUME:
        errors.append(
            "Snapshot storage cannot be used as a restore target."
        )

    target_read_only = subvolume_is_read_only(
        target_path
    )

    if target_read_only:
        errors.append(
            f"Restore target is read-only and cannot be treated as an "
            f"operational subvolume: {source_name}"
        )

    return (
        {
            "present": True,
            "id": target.subvolume_id,
            "uuid": target.uuid,
            "mountpoint": current_mountpoint,
            "mounted": current_mountpoint is not None,
            "read_only": target_read_only,
            "matches_original_uuid": (
                target.uuid == original_uuid
            ),
        },
        errors,
    )


def _build_restore_boot_state(
    payload: dict[str, object],
    set_directory: Path,
    members_result: list[dict[str, object]],
    *,
    contains_root: bool,
) -> tuple[dict[str, object], list[str]]:
    """
    Validate and describe /boot for a local restore preflight.

    Historical /boot UUIDs are provenance information. A current ext4
    /boot is not rejected merely because it has a different UUID.
    Topology and filesystem support are the compatibility requirements.
    """

    errors: list[str] = []
    boot_payload = payload.get("boot")

    if not contains_root:
        if boot_payload is not None:
            errors.append(
                "Snapshot set does not contain root but includes /boot "
                "coverage metadata."
            )

        return (
            {
                "required": False,
                "mode": None,
                "compatible": not errors,
            },
            errors,
        )

    if boot_payload is None:
        errors.append(
            "Root-containing snapshot set has no /boot coverage metadata."
        )

        return (
            {
                "required": True,
                "mode": None,
                "compatible": False,
            },
            errors,
        )

    if not isinstance(boot_payload, dict):
        errors.append(
            "Snapshot manifest contains invalid /boot coverage metadata."
        )

        return (
            {
                "required": True,
                "mode": None,
                "compatible": False,
            },
            errors,
        )

    mode = boot_payload.get("mode")

    try:
        current_boot = get_boot_filesystem_info()
    except HelperError as exc:
        errors.append(
            f"Unable to validate the current /boot layout: {exc}"
        )
        current_boot = None

    current_state: dict[str, object] | None

    if current_boot is None:
        current_state = None
    else:
        current_state = {
            "separate": current_boot.separate,
            "source": current_boot.source,
            "fstype": current_boot.fstype,
            "uuid": current_boot.uuid,
            "mountpoint": current_boot.mountpoint,
        }

    if mode == "included-in-root":
        root_members = [
            member
            for member in members_result
            if (
                isinstance(member.get("source"), dict)
                and member["source"].get("original_mountpoint") == "/"
            )
        ]

        if len(root_members) != 1:
            errors.append(
                "Unable to identify exactly one root snapshot member."
            )
            snapshot_boot_present = False
        else:
            snapshot_info = root_members[0].get("snapshot")

            if not isinstance(snapshot_info, dict):
                snapshot_boot_present = False
            else:
                snapshot_relative_path = snapshot_info.get("path")

                if not isinstance(snapshot_relative_path, str):
                    snapshot_boot_present = False
                else:
                    root_snapshot_path = (
                        set_directory
                        / snapshot_relative_path
                    )
                    snapshot_boot_path = (
                        root_snapshot_path
                        / "boot"
                    )
                    snapshot_boot_present = (
                        snapshot_boot_path.exists()
                        and snapshot_boot_path.is_dir()
                        and not snapshot_boot_path.is_symlink()
                    )

        if not snapshot_boot_present:
            errors.append(
                "The root snapshot does not contain a valid /boot "
                "directory."
            )

        compatible = (
            current_boot is not None
            and not current_boot.separate
            and snapshot_boot_present
        )

        if current_boot is not None and current_boot.separate:
            errors.append(
                "Snapshot expects /boot inside the root Btrfs subvolume, "
                "but the current system uses a separate /boot filesystem."
            )

        return (
            {
                "required": True,
                "mode": "included-in-root",
                "snapshot_boot_present": snapshot_boot_present,
                "current": current_state,
                "compatible": compatible and not errors,
            },
            errors,
        )

    if mode == "archive":
        archive_payload = boot_payload.get("archive")

        if not isinstance(archive_payload, dict):
            errors.append(
                "Snapshot manifest contains invalid /boot archive metadata."
            )

            return (
                {
                    "required": True,
                    "mode": "archive",
                    "current": current_state,
                    "compatible": False,
                },
                errors,
            )

        archive_path = (
            set_directory
            / "archives"
            / "boot.tar.zst"
        )
        archive_present = (
            archive_path.exists()
            and archive_path.is_file()
            and not archive_path.is_symlink()
        )
        expected_size = archive_payload.get("size_bytes")
        expected_sha256 = archive_payload.get("sha256")

        actual_size: int | None = None
        actual_sha256: str | None = None
        size_valid = False
        sha256_valid = False

        if archive_present:
            try:
                actual_size = archive_path.stat().st_size
            except OSError as exc:
                errors.append(
                    f"Unable to stat the /boot archive: {exc}"
                )
            else:
                size_valid = (
                    isinstance(expected_size, int)
                    and not isinstance(expected_size, bool)
                    and actual_size == expected_size
                )

                if not size_valid:
                    errors.append(
                        "/boot archive size does not match the manifest."
                    )
                else:
                    # validate_snapshot_set_for_deletion() has already
                    # calculated and verified the complete SHA-256 before
                    # this plan is built. Do not read a large archive twice
                    # during one preflight.
                    actual_sha256 = (
                        expected_sha256
                        if isinstance(expected_sha256, str)
                        else None
                    )
                    sha256_valid = actual_sha256 is not None
        else:
            errors.append(
                "/boot archive is missing or is not a regular file."
            )

        snapshot_boot_uuid = boot_payload.get("uuid")
        current_uuid = (
            current_boot.uuid
            if current_boot is not None
            else None
        )
        uuid_matches_snapshot = (
            current_uuid == snapshot_boot_uuid
            if current_uuid is not None
            and isinstance(snapshot_boot_uuid, str)
            else None
        )

        topology_compatible = (
            current_boot is not None
            and current_boot.separate
            and current_boot.fstype == "ext4"
        )

        if current_boot is not None:
            if not current_boot.separate:
                errors.append(
                    "Snapshot expects a separate ext4 /boot filesystem, "
                    "but /boot is currently included in the root filesystem."
                )
            elif current_boot.fstype != "ext4":
                errors.append(
                    "Snapshot expects a supported separate ext4 /boot "
                    "filesystem, but the current /boot filesystem differs."
                )

        compatible = (
            topology_compatible
            and archive_present
            and size_valid
            and sha256_valid
            and not errors
        )

        return (
            {
                "required": True,
                "mode": "archive",
                "snapshot": {
                    "source": boot_payload.get("source"),
                    "fstype": boot_payload.get("fstype"),
                    "uuid": snapshot_boot_uuid,
                    "archive": {
                        "path": archive_payload.get("path"),
                        "compression": archive_payload.get("compression"),
                        "size_bytes": expected_size,
                        "sha256": expected_sha256,
                        "present": archive_present,
                        "actual_size_bytes": actual_size,
                        "size_valid": size_valid,
                        "actual_sha256": actual_sha256,
                        "sha256_valid": sha256_valid,
                    },
                },
                "current": current_state,
                "uuid_matches_snapshot": uuid_matches_snapshot,
                "compatible": compatible,
            },
            errors,
        )

    errors.append(
        "Snapshot manifest contains an unsupported /boot coverage mode."
    )

    return (
        {
            "required": True,
            "mode": mode,
            "current": current_state,
            "compatible": False,
        },
        errors,
    )


def build_restore_preflight(
    set_id: str,
) -> dict[str, object]:
    """
    Inspect whether one committed snapshot set can be restored locally.

    This action is strictly read-only. It validates snapshot integrity,
    resolves the current logical targets, validates /boot coverage when
    root is included, and returns a structured restore plan. It never
    renames, creates, deletes or modifies subvolumes or files.
    """

    root = get_root_filesystem_info()
    restore_context = get_restore_context(
        root
    )
    mount_state = get_snapshot_mount_state(
        root
    )

    if (
        not mount_state.mounted
        or not mount_state.correct
    ):
        raise HelperError(
            "Snapshot storage is not mounted correctly at /.snapshots."
        )

    storage = read_subvolume_record(
        SNAPSHOT_MOUNTPOINT
    )

    if storage.name != SNAPSHOT_SUBVOLUME:
        raise HelperError(
            "Mounted snapshot storage is not @snapshots."
        )

    set_directory = find_snapshot_set_directory_by_id(
        set_id
    )

    if set_directory is None:
        raise HelperError(
            "The requested snapshot set was not found."
        )

    payload = load_json_object(
        set_directory
        / "manifest.json"
    )

    format_version = payload.get("format_version")

    if (
        isinstance(format_version, bool)
        or format_version != 1
    ):
        raise HelperError(
            "Unsupported snapshot manifest format."
        )

    manifest_set_id = normalize_uuid_text(
        payload.get("set_id"),
        "set_id",
    )

    if manifest_set_id != set_id:
        raise HelperError(
            "Snapshot manifest set ID does not match the requested set."
        )

    set_name_value = payload.get("set_name")

    if (
        not isinstance(set_name_value, str)
        or not set_name_value.strip()
    ):
        raise HelperError(
            "Snapshot manifest has an invalid set_name."
        )

    set_name = set_name_value.strip()

    if set_name != set_directory.name:
        raise HelperError(
            "Snapshot manifest set_name does not match its directory."
        )

    scope = payload.get("scope")

    if scope not in {
        "single",
        "full",
    }:
        raise HelperError(
            "Snapshot manifest has an unsupported scope."
        )

    purpose = get_manifest_purpose(
        payload
    )

    filesystem_uuid = normalize_uuid_text(
        payload.get("filesystem_uuid"),
        "filesystem_uuid",
    )

    result = _restore_preflight_base_result(
        set_id=set_id,
        set_name=set_name,
        scope=scope,
        purpose=purpose,
        filesystem_uuid=filesystem_uuid,
        current_filesystem_uuid=root.uuid,
    )
    errors = result["errors"]

    if not isinstance(errors, list):
        raise HelperError(
            "Internal restore preflight result error."
        )

    if restore_context.mode == "snapshot-boot":
        if set_name != restore_context.booted_set_name:
            errors.append(
                "Snapshot-boot restore is limited to the snapshot set "
                "that is currently booted. Reboot normally to restore a "
                "different snapshot set."
            )
            return result

    blocking_error = _restore_blocking_error()

    if blocking_error is not None:
        errors.append(
            blocking_error
        )
        return result

    if filesystem_uuid != root.uuid:
        errors.append(
            "Snapshot set belongs to a different Btrfs filesystem. "
            "Local restore requires the filesystem where the snapshot "
            "set was created."
        )

        return result

    try:
        validate_snapshot_set_for_deletion(
            set_directory,
            expected_set_id=set_id,
            expected_filesystem_uuid=root.uuid,
            storage_subvolume_id=(
                storage.subvolume_id
            ),
            allow_missing_snapshots=False,
            deleting_directory=False,
        )
    except HelperError as exc:
        errors.append(
            f"Snapshot integrity validation failed: {exc}"
        )

        return result

    members_payload = payload.get("members")

    if not isinstance(members_payload, list) or not members_payload:
        errors.append(
            "Snapshot set has no valid members."
        )

        return result

    if scope == "single" and len(members_payload) != 1:
        errors.append(
            "Single snapshot set must contain exactly one member."
        )

        return result

    members_result: list[dict[str, object]] = []
    seen_source_ids: set[int] = set()
    seen_source_paths: set[str] = set()
    root_member_count = 0
    root_source_name: str | None = None

    top_level_mount = mount_top_level(
        root
    )

    try:
        for index, member_payload in enumerate(
            members_payload
        ):
            if not isinstance(member_payload, dict):
                errors.append(
                    f"Invalid snapshot member at index {index}."
                )
                continue

            source = member_payload.get("source")
            snapshot = member_payload.get("snapshot")

            if not isinstance(source, dict) or not isinstance(snapshot, dict):
                errors.append(
                    f"Invalid snapshot member at index {index}."
                )
                continue

            source_id_value = source.get("id")

            if (
                isinstance(source_id_value, bool)
                or not isinstance(source_id_value, int)
                or source_id_value <= 0
            ):
                errors.append(
                    f"Invalid historical source ID at member {index}."
                )
                continue

            source_id = source_id_value
            source_name = source.get("path")

            if (
                not isinstance(source_name, str)
                or not source_name
                or source_name in {".", ".."}
                or "/" in source_name
                or "\x00" in source_name
                or Path(source_name).name != source_name
                or source_name == SNAPSHOT_SUBVOLUME
            ):
                errors.append(
                    f"Invalid historical source path at member {index}."
                )
                continue

            if source_id in seen_source_ids:
                errors.append(
                    f"Duplicate historical source ID: {source_id}"
                )
                continue

            if source_name in seen_source_paths:
                errors.append(
                    f"Duplicate historical source path: {source_name}"
                )
                continue

            seen_source_ids.add(
                source_id
            )
            seen_source_paths.add(
                source_name
            )

            try:
                original_uuid = normalize_uuid_text(
                    source.get("uuid"),
                    "source.uuid",
                )
                original_mountpoint = (
                    _parse_restore_source_mountpoint(
                        source.get("mountpoint"),
                        member_index=index,
                    )
                )
            except HelperError as exc:
                errors.append(str(exc))
                continue

            if original_mountpoint == "/":
                root_member_count += 1
                root_source_name = source_name

            target_state, target_errors = (
                _build_restore_target_state(
                    root,
                    top_level_mount,
                    source_name,
                    original_uuid=original_uuid,
                )
            )
            errors.extend(
                target_errors
            )

            snapshot_relative_path = snapshot.get("path")
            snapshot_path = (
                set_directory
                / "subvolumes"
                / source_name
            )

            members_result.append(
                {
                    "source": {
                        "path": source_name,
                        "original_id": source_id,
                        "original_uuid": original_uuid,
                        "original_mountpoint": original_mountpoint,
                    },
                    "snapshot": {
                        "id": snapshot.get("id"),
                        "path": snapshot_relative_path,
                        "uuid": snapshot.get("uuid"),
                        "parent_uuid": snapshot.get("parent_uuid"),
                        "read_only": snapshot.get("read_only") is True,
                        "present": snapshot_path.exists(),
                        "valid": True,
                    },
                    "target": target_state,
                    "valid": not target_errors,
                }
            )

        result["members"] = members_result

        contains_root = (
            root_member_count == 1
        )
        result["contains_root"] = contains_root

        if root_member_count > 1:
            errors.append(
                "Snapshot set contains more than one historical root member."
            )

        if scope == "full" and root_member_count != 1:
            errors.append(
                "Full snapshot set does not contain exactly one root member."
            )

        if contains_root:
            canonical_root_name = (
                restore_context.canonical_root_name
            )

            if root_source_name != canonical_root_name:
                errors.append(
                    "Historical root source path does not match the "
                    "canonical logical root subvolume path."
                )
            else:
                root_target = next(
                    (
                        member.get("target")
                        for member in members_result
                        if (
                            isinstance(member.get("source"), dict)
                            and member["source"].get(
                                "original_mountpoint"
                            ) == "/"
                        )
                    ),
                    None,
                )

                if (
                    not isinstance(root_target, dict)
                    or root_target.get("present") is not True
                ):
                    errors.append(
                        "The canonical root restore target could not be "
                        "identified safely."
                    )

                elif restore_context.mode == "normal":
                    if root_target.get("mountpoint") != "/":
                        errors.append(
                            "The current root restore target is not mounted "
                            "at /."
                        )

                else:
                    if (
                        root_target.get("mounted") is True
                        or root_target.get("mountpoint") is not None
                    ):
                        errors.append(
                            "The canonical root restore target is "
                            "unexpectedly mounted during snapshot boot."
                        )

        boot_state, boot_errors = (
            _build_restore_boot_state(
                payload,
                set_directory,
                members_result,
                contains_root=contains_root,
            )
        )
        result["boot"] = boot_state
        errors.extend(
            boot_errors
        )

    finally:
        unmount_top_level(
            top_level_mount
        )

    result["preflight_ok"] = (
        len(errors) == 0
    )

    return result



def preflight_restore(
    set_id: str,
) -> int:
    """
    Write the structured local restore preflight result.
    """

    return _write_restore_preflight_result(
        build_restore_preflight(
            set_id
        )
    )

def read_boot_id() -> str:
    """
    Return the UUID that identifies the current Linux boot session.
    """

    try:
        value = BOOT_ID_PATH.read_text(
            encoding="utf-8"
        ).strip()
    except OSError as exc:
        raise HelperError(
            f"Unable to read the current boot ID: {exc}"
        ) from exc

    return normalize_uuid_text(
        value,
        "boot_id",
    )


def atomic_write_json(
    path: Path,
    payload: dict[str, object],
    *,
    mode: int = 0o600,
) -> None:
    """
    Atomically replace one JSON state file and flush it to disk.
    """

    temporary_fd: int | None = None
    temporary_name: str | None = None

    try:
        temporary_fd, temporary_name = tempfile.mkstemp(
            prefix=f".{path.name}.",
            dir=str(path.parent),
        )

        os.fchmod(
            temporary_fd,
            mode,
        )

        with os.fdopen(
            temporary_fd,
            "w",
            encoding="utf-8",
            closefd=True,
        ) as handle:
            temporary_fd = None
            json.dump(
                payload,
                handle,
                ensure_ascii=False,
                indent=2,
            )
            handle.write("\n")
            handle.flush()
            os.fsync(handle.fileno())

        os.replace(
            temporary_name,
            path,
        )
        temporary_name = None
        fsync_directory(
            path.parent
        )

    finally:
        if temporary_fd is not None:
            os.close(
                temporary_fd
            )

        if temporary_name is not None:
            try:
                os.unlink(
                    temporary_name
                )
            except FileNotFoundError:
                pass


def remove_restore_state() -> None:
    """
    Remove the persistent restore transaction marker.
    """

    if not RESTORE_STATE_PATH.exists():
        return

    if RESTORE_STATE_PATH.is_symlink() or not RESTORE_STATE_PATH.is_file():
        raise HelperError(
            "Restore state path is not a regular file."
        )

    RESTORE_STATE_PATH.unlink()
    fsync_directory(
        SNAPSHOT_MOUNTPOINT
    )


def load_restore_state() -> dict[str, object] | None:
    """
    Load and minimally validate the persistent restore transaction state.
    """

    if not RESTORE_STATE_PATH.exists():
        return None

    payload = load_json_object(
        RESTORE_STATE_PATH
    )

    if payload.get("format_version") != 1:
        raise HelperError(
            "Unsupported restore state format."
        )

    if payload.get("operation") != "restore":
        raise HelperError(
            "Restore state contains an unexpected operation."
        )

    normalize_uuid_text(
        payload.get("transaction_id"),
        "restore.transaction_id",
    )
    normalize_uuid_text(
        payload.get("restored_set_id"),
        "restore.restored_set_id",
    )
    normalize_uuid_text(
        payload.get("pre_restore_set_id"),
        "restore.pre_restore_set_id",
    )
    normalize_uuid_text(
        payload.get("applied_boot_id"),
        "restore.applied_boot_id",
    )

    phase = payload.get("phase")

    if phase not in {
        "prepared",
        "btrfs-committing",
        "btrfs-committed",
        "boot-restoring",
        "applied",
        "rollback-incomplete",
    }:
        raise HelperError(
            "Restore state contains an unsupported phase."
        )

    members = payload.get("members")
    old_targets = payload.get("old_targets")

    if not isinstance(members, list) or not members:
        raise HelperError(
            "Restore state contains no restored members."
        )

    if not isinstance(old_targets, list):
        raise HelperError(
            "Restore state contains invalid old target metadata."
        )

    return payload


def _restore_blocking_error() -> str | None:
    """
    Return a user-facing reason that blocks another restore, if any.
    """

    state = load_restore_state()

    if state is None:
        return None

    phase = state.get("phase")
    current_boot_id = read_boot_id()
    applied_boot_id = normalize_uuid_text(
        state.get("applied_boot_id"),
        "restore.applied_boot_id",
    )

    if phase == "applied":
        if current_boot_id == applied_boot_id:
            return (
                "A restore has already been applied during this boot. "
                "Restart the system before performing another restore."
            )

        return None

    return (
        "An incomplete restore transaction is present. "
        "Another restore cannot start until that state is resolved."
    )


def _validate_restore_source_name(
    value: object,
    *,
    field_name: str,
) -> str:
    """
    Validate one canonical top-level source name used by restore state.
    """

    if not isinstance(value, str) or not is_safe_top_level_source_path(value):
        raise HelperError(
            f"Invalid restore source path: {field_name}"
        )

    if value == SNAPSHOT_SUBVOLUME or value.startswith(RESTORE_INTERNAL_PREFIX):
        raise HelperError(
            f"Invalid restore source path: {field_name}"
        )

    return value


def _validate_restore_staging_name(
    value: object,
    *,
    field_name: str,
) -> str:
    """
    Validate one Synex-internal restore staging name.
    """

    if (
        not isinstance(value, str)
        or not is_safe_top_level_source_path(value)
        or not value.startswith(RESTORE_INTERNAL_PREFIX)
    ):
        raise HelperError(
            f"Invalid restore staging path: {field_name}"
        )

    return value


def _read_current_source_by_name(
    top_level_mount: Path,
    source_name: str,
) -> SubvolumeRecord:
    """
    Return one currently operational top-level source by logical name.
    """

    target_path = (
        top_level_mount
        / source_name
    )

    if target_path.is_symlink() or not target_path.exists():
        raise HelperError(
            f"Restore target does not exist safely: {source_name}"
        )

    target = read_subvolume_record(
        target_path
    )

    return validate_snapshot_source(
        top_level_mount,
        target.subvolume_id,
    )


def _validate_created_restore_target(
    path: Path,
    *,
    expected_parent_uuid: str,
) -> SubvolumeRecord:
    """
    Validate one writable top-level staging target created from a snapshot.
    """

    if path.is_symlink() or not path.exists():
        raise HelperError(
            f"Restore staging target is missing: {path.name}"
        )

    record = read_subvolume_record(
        path
    )

    if record.parent_id != TOP_LEVEL_SUBVOLUME_ID:
        raise HelperError(
            f"Restore staging target is not top-level: {path.name}"
        )

    if record.name != path.name:
        raise HelperError(
            f"Restore staging target name mismatch: {path.name}"
        )

    if record.parent_uuid != expected_parent_uuid:
        raise HelperError(
            f"Restore staging target lineage mismatch: {path.name}"
        )

    if subvolume_is_read_only(
        path
    ):
        raise HelperError(
            f"Restore staging target is unexpectedly read-only: {path.name}"
        )

    return record


def _delete_restore_staging_subvolume(
    path: Path,
) -> None:
    """
    Delete one known Synex restore staging subvolume.
    """

    if not path.exists():
        return

    if path.is_symlink():
        raise HelperError(
            f"Restore staging path became a symbolic link: {path.name}"
        )

    record = read_subvolume_record(
        path
    )

    if record.parent_id != TOP_LEVEL_SUBVOLUME_ID:
        raise HelperError(
            f"Restore staging path is not top-level: {path.name}"
        )

    if record.name != path.name or not path.name.startswith(RESTORE_INTERNAL_PREFIX):
        raise HelperError(
            f"Refusing to delete an unexpected restore staging path: {path.name}"
        )

    run_command(
        [
            str(BTRFS_COMMAND),
            "subvolume",
            "delete",
            str(path),
        ]
    )


def _decode_mountinfo_path(value: str) -> str:
    """
    Decode the octal escapes used by /proc/self/mountinfo paths.
    """

    replacements = {
        "\\040": " ",
        "\\011": "\t",
        "\\012": "\n",
        "\\134": "\\",
    }

    result = value

    for encoded, decoded in replacements.items():
        result = result.replace(
            encoded,
            decoded,
        )

    return result


def _nested_boot_mountpoints() -> list[str]:
    """
    Return mounted filesystems strictly below /boot, deepest first.
    """

    try:
        lines = Path("/proc/self/mountinfo").read_text(
            encoding="utf-8"
        ).splitlines()
    except OSError as exc:
        raise HelperError(
            f"Unable to inspect nested /boot mounts: {exc}"
        ) from exc

    result: list[str] = []

    for line in lines:
        fields = line.split()

        if len(fields) < 5:
            continue

        mountpoint = _decode_mountinfo_path(
            fields[4]
        )

        if mountpoint.startswith("/boot/"):
            result.append(
                mountpoint
            )

    return sorted(
        set(result),
        key=lambda value: value.count("/"),
        reverse=True,
    )


def _clear_boot_filesystem() -> None:
    """
    Remove every entry from the mounted /boot filesystem.

    Nested mounts must already be unmounted. Symlinks are unlinked rather
    than followed. The /boot filesystem itself is never removed.
    """

    boot_path = Path("/boot")

    if boot_path.is_symlink() or not boot_path.is_dir():
        raise HelperError(
            "/boot is not a safe directory for restore."
        )

    nested = _nested_boot_mountpoints()

    if nested:
        raise HelperError(
            "Nested filesystems remain mounted below /boot: "
            + ", ".join(nested)
        )

    try:
        entries = list(
            os.scandir(
                boot_path
            )
        )
    except OSError as exc:
        raise HelperError(
            f"Unable to inspect /boot before replacement: {exc}"
        ) from exc

    for entry in entries:
        entry_path = Path(
            entry.path
        )

        try:
            if entry.is_symlink():
                entry_path.unlink()
            elif entry.is_dir(
                follow_symlinks=False
            ):
                shutil.rmtree(
                    entry_path
                )
            else:
                entry_path.unlink()
        except OSError as exc:
            raise HelperError(
                f"Unable to remove {entry_path} during /boot restore: {exc}"
            ) from exc

    fsync_directory(
        boot_path
    )


def _extract_boot_archive(
    archive_path: Path,
) -> None:
    """
    Extract one validated Synex /boot archive into the empty filesystem.
    """

    if archive_path.is_symlink() or not archive_path.is_file():
        raise HelperError(
            "Restore /boot archive is missing or invalid."
        )

    ensure_boot_archive_commands()

    run_command(
        [
            str(TAR_COMMAND),
            "--extract",
            f"--use-compress-program={ZSTD_COMMAND}",
            "--acls",
            "--xattrs",
            "--xattrs-include=*",
            "--numeric-owner",
            "--same-owner",
            "--same-permissions",
            "--file",
            str(archive_path),
            "--directory",
            "/boot",
        ]
    )

    fsync_directory(
        Path("/boot")
    )


def _mount_efi_after_boot_restore() -> None:
    """
    Restore the normal /boot/efi mount after /boot replacement.
    """

    efi_path = Path("/boot/efi")

    if efi_path.is_symlink():
        raise HelperError(
            "/boot/efi became a symbolic link after /boot restore."
        )

    if not efi_path.exists() or not efi_path.is_dir():
        raise HelperError(
            "Restored /boot archive does not contain the /boot/efi "
            "mountpoint directory."
        )

    run_command(
        [
            str(MOUNT_COMMAND),
            "/boot/efi",
        ]
    )

    result = run_command(
        [
            str(FINDMNT_COMMAND),
            "--raw",
            "--noheadings",
            "--mountpoint",
            "/boot/efi",
            "--output",
            "TARGET",
        ],
        check=False,
    )

    if result.stdout.strip() != "/boot/efi":
        raise HelperError(
            "/boot/efi could not be remounted after /boot restore."
        )


def _replace_boot_archive_transactionally(
    restore_archive: Path,
    rollback_archive: Path,
) -> dict[str, object]:
    """
    Replace a separate ext4 /boot exactly, with rollback on failure.

    Only /boot/efi is accepted as a nested filesystem. It is unmounted
    before clearing /boot and remounted after extraction, so the ESP is
    never deleted or overwritten by the archive restore.
    """

    current_boot = get_boot_filesystem_info()

    if (
        not current_boot.separate
        or current_boot.fstype != "ext4"
    ):
        raise HelperError(
            "Restore requires the current separate /boot filesystem to "
            "be ext4."
        )

    for archive_path in (
        restore_archive,
        rollback_archive,
    ):
        if archive_path.is_symlink() or not archive_path.is_file():
            raise HelperError(
                f"Required /boot restore archive is missing: {archive_path}"
            )

    nested = _nested_boot_mountpoints()
    unsupported_nested = [
        mountpoint
        for mountpoint in nested
        if mountpoint != "/boot/efi"
    ]

    if unsupported_nested:
        raise HelperError(
            "Unsupported nested filesystems were found below /boot: "
            + ", ".join(unsupported_nested)
        )

    efi_was_mounted = (
        "/boot/efi" in nested
    )
    efi_unmounted = False
    boot_mutated = False

    try:
        if efi_was_mounted:
            run_command(
                [
                    str(UMOUNT_COMMAND),
                    "/boot/efi",
                ]
            )
            efi_unmounted = True

        if _nested_boot_mountpoints():
            raise HelperError(
                "A nested filesystem remained mounted below /boot."
            )

        boot_mutated = True
        _clear_boot_filesystem()
        _extract_boot_archive(
            restore_archive
        )

        if efi_was_mounted:
            _mount_efi_after_boot_restore()
            efi_unmounted = False

        os.sync()

        return {
            "mode": "archive",
            "replaced": True,
            "efi_preserved": True,
            "efi_remounted": efi_was_mounted,
        }

    except Exception as operation_error:
        rollback_errors: list[str] = []

        if not boot_mutated:
            if efi_was_mounted and efi_unmounted:
                try:
                    _mount_efi_after_boot_restore()
                    efi_unmounted = False
                except Exception as exc:
                    rollback_errors.append(
                        f"/boot/efi remount: {exc}"
                    )

            if rollback_errors:
                raise HelperError(
                    f"{operation_error} /boot recovery was incomplete: "
                    + "; ".join(rollback_errors)
                ) from operation_error

            raise

        try:
            current_nested = _nested_boot_mountpoints()

            if "/boot/efi" in current_nested:
                run_command(
                    [
                        str(UMOUNT_COMMAND),
                        "/boot/efi",
                    ]
                )
                efi_unmounted = True

            unexpected_nested = [
                mountpoint
                for mountpoint in _nested_boot_mountpoints()
                if mountpoint != "/boot/efi"
            ]

            if unexpected_nested:
                raise HelperError(
                    "Unexpected nested /boot mounts prevent rollback."
                )

            _clear_boot_filesystem()
            _extract_boot_archive(
                rollback_archive
            )

            if efi_was_mounted:
                _mount_efi_after_boot_restore()
                efi_unmounted = False

            os.sync()

        except Exception as exc:
            rollback_errors.append(
                f"/boot rollback: {exc}"
            )

        if efi_was_mounted and efi_unmounted:
            try:
                _mount_efi_after_boot_restore()
            except Exception as exc:
                rollback_errors.append(
                    f"/boot/efi remount: {exc}"
                )

        if rollback_errors:
            raise HelperError(
                f"{operation_error} /boot rollback was incomplete: "
                + "; ".join(rollback_errors)
            ) from operation_error

        raise


def _prepare_restore_state_for_new_operation() -> None:
    """
    Enforce the one-restore-per-boot contract.

    An applied restore from a previous boot is finalized automatically
    before a new restore. An applied restore from this boot blocks the
    operation. Any incomplete transaction blocks the operation.
    """

    state = load_restore_state()

    if state is None:
        return

    phase = state.get("phase")
    current_boot_id = read_boot_id()
    applied_boot_id = normalize_uuid_text(
        state.get("applied_boot_id"),
        "restore.applied_boot_id",
    )

    if phase != "applied":
        raise HelperError(
            "An incomplete restore transaction is present. "
            "Another restore cannot start until it is resolved."
        )

    if current_boot_id == applied_boot_id:
        raise HelperError(
            "A restore has already been applied during this boot. "
            "Restart the system before performing another restore."
        )

    _finalize_restore_state(
        state
    )


def _finalize_restore_state(
    state: dict[str, object],
) -> dict[str, object]:
    """
    Delete old transaction staging after a successful reboot.

    The permanent pre-restore snapshot set is deliberately retained.
    """

    if state.get("phase") != "applied":
        raise HelperError(
            "Only an applied restore can be finalized automatically."
        )

    current_boot_id = read_boot_id()
    applied_boot_id = normalize_uuid_text(
        state.get("applied_boot_id"),
        "restore.applied_boot_id",
    )

    if current_boot_id == applied_boot_id:
        return {
            "finalized": False,
            "restart_required": True,
            "reason": "The required reboot has not occurred yet.",
            "transaction_id": state.get("transaction_id"),
            "restored_set_id": state.get("restored_set_id"),
            "pre_restore_set_id": state.get("pre_restore_set_id"),
        }

    root = get_root_filesystem_info()
    top_level_mount = mount_top_level(
        root
    )

    try:
        members = state.get("members")
        old_targets = state.get("old_targets")

        if not isinstance(members, list) or not isinstance(old_targets, list):
            raise HelperError(
                "Restore state contains invalid cleanup metadata."
            )

        restored_names: set[str] = set()

        for index, member in enumerate(members):
            if not isinstance(member, dict):
                raise HelperError(
                    "Restore state contains an invalid restored member."
                )

            source_name = _validate_restore_source_name(
                member.get("source_path"),
                field_name=f"members[{index}].source_path",
            )
            restored_uuid = normalize_uuid_text(
                member.get("restored_uuid"),
                f"members[{index}].restored_uuid",
            )
            canonical_path = (
                top_level_mount
                / source_name
            )

            if canonical_path.is_symlink() or not canonical_path.exists():
                raise HelperError(
                    f"Restored target is missing after reboot: {source_name}"
                )

            current = read_subvolume_record(
                canonical_path
            )

            if (
                current.parent_id != TOP_LEVEL_SUBVOLUME_ID
                or current.name != source_name
                or current.uuid != restored_uuid
                or subvolume_is_read_only(canonical_path)
            ):
                raise HelperError(
                    f"Restored target changed before cleanup: {source_name}"
                )

            restored_names.add(
                source_name
            )

        for index, old_target in enumerate(old_targets):
            if not isinstance(old_target, dict):
                raise HelperError(
                    "Restore state contains invalid old target metadata."
                )

            source_name = _validate_restore_source_name(
                old_target.get("source_path"),
                field_name=f"old_targets[{index}].source_path",
            )
            staging_name = _validate_restore_staging_name(
                old_target.get("staging_name"),
                field_name=f"old_targets[{index}].staging_name",
            )
            old_uuid = normalize_uuid_text(
                old_target.get("uuid"),
                f"old_targets[{index}].uuid",
            )
            old_id_value = old_target.get("id")

            if (
                isinstance(old_id_value, bool)
                or not isinstance(old_id_value, int)
                or old_id_value <= 0
            ):
                raise HelperError(
                    "Restore state contains an invalid old target ID."
                )

            if source_name not in restored_names:
                canonical_extra = (
                    top_level_mount
                    / source_name
                )

                if canonical_extra.exists() or canonical_extra.is_symlink():
                    raise HelperError(
                        "A target that should remain absent after Full "
                        f"restore exists again: {source_name}"
                    )

            staging_path = (
                top_level_mount
                / staging_name
            )

            if not staging_path.exists():
                continue

            if staging_path.is_symlink():
                raise HelperError(
                    f"Old restore staging path became a symlink: {staging_name}"
                )

            mounted_at = get_subvolume_mountpoint(
                root,
                staging_name,
            )

            if mounted_at is not None:
                raise HelperError(
                    f"Old restore staging target is still mounted: {staging_name}"
                )

            old_record = read_subvolume_record(
                staging_path
            )

            if (
                old_record.subvolume_id != old_id_value
                or old_record.uuid != old_uuid
                or old_record.parent_id != TOP_LEVEL_SUBVOLUME_ID
                or old_record.name != staging_name
            ):
                raise HelperError(
                    f"Old restore staging metadata changed: {staging_name}"
                )

            _delete_restore_staging_subvolume(
                staging_path
            )
            fsync_directory(
                top_level_mount
            )

    finally:
        unmount_top_level(
            top_level_mount
        )

    remove_restore_state()

    return {
        "finalized": True,
        "restart_required": False,
        "transaction_id": state.get("transaction_id"),
        "restored_set_id": state.get("restored_set_id"),
        "pre_restore_set_id": state.get("pre_restore_set_id"),
    }


def restore_status() -> int:
    """
    Return the persistent local restore state without modifying it.
    """

    state = load_restore_state()

    if state is None:
        payload: dict[str, object] = {
            "pending": False,
            "restart_required": False,
            "cleanup_required": False,
        }
    else:
        current_boot_id = read_boot_id()
        applied_boot_id = normalize_uuid_text(
            state.get("applied_boot_id"),
            "restore.applied_boot_id",
        )
        phase = state.get("phase")
        same_boot = (
            current_boot_id == applied_boot_id
        )
        payload = {
            "pending": True,
            "phase": phase,
            "transaction_id": state.get("transaction_id"),
            "restored_set_id": state.get("restored_set_id"),
            "pre_restore_set_id": state.get("pre_restore_set_id"),
            "applied_boot_id": applied_boot_id,
            "current_boot_id": current_boot_id,
            "restart_required": phase == "applied" and same_boot,
            "cleanup_required": phase == "applied" and not same_boot,
        }

    sys.stdout.write(
        json.dumps(
            payload,
            ensure_ascii=False,
            indent=2,
        )
        + "\n"
    )

    return 0


def finalize_restore() -> int:
    """
    Finalize one successfully applied restore after the required reboot.
    """

    state = load_restore_state()

    if state is None:
        result: dict[str, object] = {
            "finalized": False,
            "already_absent": True,
            "restart_required": False,
        }
    else:
        result = _finalize_restore_state(
            state
        )
        result["already_absent"] = False

    sys.stdout.write(
        json.dumps(
            result,
            ensure_ascii=False,
            indent=2,
        )
        + "\n"
    )

    return 0


def _rollback_btrfs_restore_commit(
    top_level_mount: Path,
    commit_log: list[tuple[str, str, str]],
) -> list[str]:
    """
    Reverse completed Btrfs rename steps in strict reverse order.
    """

    errors: list[str] = []

    for action, source_name, staging_name in reversed(commit_log):
        canonical_path = (
            top_level_mount
            / source_name
        )
        staging_path = (
            top_level_mount
            / staging_name
        )

        try:
            if action == "new-promoted":
                if canonical_path.exists() or canonical_path.is_symlink():
                    if staging_path.exists() or staging_path.is_symlink():
                        raise HelperError(
                            "Restore rollback staging collision."
                        )

                    os.rename(
                        canonical_path,
                        staging_path,
                    )
                    fsync_directory(
                        top_level_mount
                    )

            elif action == "old-moved":
                if staging_path.exists() or staging_path.is_symlink():
                    if canonical_path.exists() or canonical_path.is_symlink():
                        raise HelperError(
                            "Restore rollback target collision."
                        )

                    os.rename(
                        staging_path,
                        canonical_path,
                    )
                    fsync_directory(
                        top_level_mount
                    )

            else:
                raise HelperError(
                    "Unknown restore rollback action."
                )

        except Exception as exc:
            errors.append(
                f"{action} {source_name}: {exc}"
            )

    return errors


def _cleanup_new_restore_staging(
    top_level_mount: Path,
    new_members: list[dict[str, object]],
) -> list[str]:
    """
    Delete restored RW staging targets that are not canonical anymore.
    """

    errors: list[str] = []

    for member in reversed(new_members):
        staging_name = member.get("new_staging_name")

        if not isinstance(staging_name, str):
            continue

        staging_path = (
            top_level_mount
            / staging_name
        )

        if not staging_path.exists():
            continue

        try:
            _delete_restore_staging_subvolume(
                staging_path
            )
            fsync_directory(
                top_level_mount
            )
        except Exception as exc:
            errors.append(
                f"new staging {staging_name}: {exc}"
            )

    return errors


def restore_snapshot_set(
    set_id: str,
) -> int:
    """
    Apply one committed snapshot set to the local running system.

    Every successful restore requires a reboot. A pre-restore snapshot set
    is created first and retained permanently until the user deletes it.
    Single restore currently requires its target to exist. Full restore
    converges the operational top-level layout exactly to the historical
    Full member set.
    """

    _prepare_restore_state_for_new_operation()

    plan = build_restore_preflight(
        set_id
    )

    if plan.get("preflight_ok") is not True:
        plan_errors = plan.get("errors")

        if isinstance(plan_errors, list) and plan_errors:
            reason = "; ".join(
                str(error)
                for error in plan_errors
            )
        else:
            reason = "Restore preflight did not approve this snapshot set."

        raise HelperError(
            f"Restore preflight failed: {reason}"
        )

    root = get_root_filesystem_info()
    restore_context = get_restore_context(
        root
    )
    storage = read_subvolume_record(
        SNAPSHOT_MOUNTPOINT
    )
    set_directory = find_snapshot_set_directory_by_id(
        set_id
    )

    if set_directory is None:
        raise HelperError(
            "The requested snapshot set disappeared after preflight."
        )

    payload = load_json_object(
        set_directory
        / "manifest.json"
    )
    scope = payload.get("scope")
    purpose = get_manifest_purpose(
        payload
    )
    members_payload = payload.get("members")

    if scope not in {"single", "full"}:
        raise HelperError(
            "Snapshot manifest has an unsupported restore scope."
        )

    if not isinstance(members_payload, list) or not members_payload:
        raise HelperError(
            "Snapshot set has no restore members."
        )

    if scope == "single":
        plan_members = plan.get("members")

        if not isinstance(plan_members, list) or len(plan_members) != 1:
            raise HelperError(
                "Single restore preflight has an invalid member plan."
            )

        target_state = plan_members[0].get("target")

        if (
            not isinstance(target_state, dict)
            or target_state.get("present") is not True
        ):
            raise HelperError(
                "Single restore currently requires the operational target "
                "to exist."
            )

    transaction_id = str(
        uuid.uuid4()
    )
    boot_id = read_boot_id()
    top_level_mount = mount_top_level(
        root
    )
    new_members: list[dict[str, object]] = []
    commit_log: list[tuple[str, str, str]] = []
    pre_restore_manifest: dict[str, object] | None = None
    state_written = False
    boot_replaced = False
    rollback_boot_archive_path: Path | None = None
    restore_applied = False

    try:
        if validate_snapshot_subvolume(
            top_level_mount
        ):
            raise HelperError(
                "Snapshot storage is not configured."
            )

        historical_names: list[str] = []
        historical_snapshot_uuids: dict[str, str] = {}

        for index, member in enumerate(members_payload):
            if not isinstance(member, dict):
                raise HelperError(
                    f"Invalid restore member at index {index}."
                )

            source = member.get("source")
            snapshot = member.get("snapshot")

            if not isinstance(source, dict) or not isinstance(snapshot, dict):
                raise HelperError(
                    f"Invalid restore member at index {index}."
                )

            source_name = _validate_restore_source_name(
                source.get("path"),
                field_name=f"members[{index}].source.path",
            )
            snapshot_uuid = normalize_uuid_text(
                snapshot.get("uuid"),
                f"members[{index}].snapshot.uuid",
            )

            historical_names.append(
                source_name
            )
            historical_snapshot_uuids[source_name] = snapshot_uuid

        if len(set(historical_names)) != len(historical_names):
            raise HelperError(
                "Restore manifest contains duplicate source paths."
            )

        if scope == "single":
            current_source = _read_current_source_by_name(
                top_level_mount,
                historical_names[0],
            )
            current_sources = [
                current_source
            ]
        else:
            current_sources = list_operational_snapshot_sources(
                top_level_mount
            )

        current_by_name = {
            source.name: source
            for source in current_sources
        }

        if scope == "single" and historical_names[0] not in current_by_name:
            raise HelperError(
                "Single restore target disappeared after preflight."
            )

        if restore_context.mode == "snapshot-boot":
            pre_restore_mountpoints = (
                get_snapshot_boot_pre_restore_mountpoints(
                    root,
                    top_level_mount,
                    current_sources,
                    required_root_name=(
                        restore_context.canonical_root_name
                    ),
                )
            )

            pre_restore_manifest = create_snapshot_set(
                root,
                top_level_mount,
                current_sources,
                scope=scope,
                purpose="pre-restore",
                source_mountpoints=pre_restore_mountpoints,
                root_source_name=(
                    restore_context.canonical_root_name
                ),
            )

        else:
            pre_restore_manifest = create_snapshot_set(
                root,
                top_level_mount,
                current_sources,
                scope=scope,
                purpose="pre-restore",
            )

        pre_restore_set_id = normalize_uuid_text(
            pre_restore_manifest.get("set_id"),
            "pre_restore.set_id",
        )
        pre_restore_set_name_value = pre_restore_manifest.get("set_name")

        if not isinstance(pre_restore_set_name_value, str):
            raise HelperError(
                "Pre-restore snapshot set has an invalid name."
            )

        pre_restore_set_name = pre_restore_set_name_value
        pre_restore_directory = find_snapshot_set_directory_by_id(
            pre_restore_set_id
        )

        if pre_restore_directory is None:
            raise HelperError(
                "Pre-restore snapshot set could not be found after creation."
            )

        validate_snapshot_set_for_deletion(
            pre_restore_directory,
            expected_set_id=pre_restore_set_id,
            expected_filesystem_uuid=root.uuid,
            storage_subvolume_id=storage.subvolume_id,
            allow_missing_snapshots=False,
            deleting_directory=False,
        )

        old_targets: list[dict[str, object]] = []
        old_staging_by_source: dict[str, str] = {}

        for index, current_source in enumerate(current_sources):
            staging_name = (
                f"{RESTORE_INTERNAL_PREFIX}{transaction_id}-old-{index}"
            )

            if not is_safe_top_level_source_path(staging_name):
                raise HelperError(
                    "Generated restore old-staging name is invalid."
                )

            staging_path = (
                top_level_mount
                / staging_name
            )

            if staging_path.exists() or staging_path.is_symlink():
                raise HelperError(
                    "Restore old-staging path already exists."
                )

            old_staging_by_source[current_source.name] = staging_name
            old_targets.append(
                {
                    "source_path": current_source.name,
                    "staging_name": staging_name,
                    "id": current_source.subvolume_id,
                    "uuid": current_source.uuid,
                    "restored_member": current_source.name in historical_names,
                }
            )

        for index, source_name in enumerate(historical_names):
            new_staging_name = (
                f"{RESTORE_INTERNAL_PREFIX}{transaction_id}-new-{index}"
            )

            if not is_safe_top_level_source_path(new_staging_name):
                raise HelperError(
                    "Generated restore new-staging name is invalid."
                )

            new_staging_path = (
                top_level_mount
                / new_staging_name
            )
            canonical_path = (
                top_level_mount
                / source_name
            )
            snapshot_path = (
                set_directory
                / "subvolumes"
                / source_name
            )

            if new_staging_path.exists() or new_staging_path.is_symlink():
                raise HelperError(
                    "Restore new-staging path already exists."
                )

            run_command(
                [
                    str(BTRFS_COMMAND),
                    "subvolume",
                    "snapshot",
                    str(snapshot_path),
                    str(new_staging_path),
                ]
            )

            restored = _validate_created_restore_target(
                new_staging_path,
                expected_parent_uuid=(
                    historical_snapshot_uuids[source_name]
                ),
            )

            new_members.append(
                {
                    "source_path": source_name,
                    "snapshot_uuid": historical_snapshot_uuids[source_name],
                    "restored_id": restored.subvolume_id,
                    "restored_uuid": restored.uuid,
                    "new_staging_name": new_staging_name,
                    "had_target": source_name in current_by_name,
                    "old_staging_name": old_staging_by_source.get(source_name),
                    "canonical_path": str(canonical_path),
                }
            )

        state: dict[str, object] = {
            "format_version": 1,
            "operation": "restore",
            "transaction_id": transaction_id,
            "restored_set_id": set_id,
            "restored_set_name": payload.get("set_name"),
            "restored_scope": scope,
            "restored_purpose": purpose,
            "pre_restore_set_id": pre_restore_set_id,
            "pre_restore_set_name": pre_restore_set_name,
            "applied_boot_id": boot_id,
            "phase": "prepared",
            "restart_required": False,
            "members": new_members,
            "old_targets": old_targets,
            "boot": {
                "mode": (
                    payload.get("boot", {}).get("mode")
                    if isinstance(payload.get("boot"), dict)
                    else None
                ),
                "replaced": False,
            },
        }

        atomic_write_json(
            RESTORE_STATE_PATH,
            state
        )
        state_written = True

        state["phase"] = "btrfs-committing"
        atomic_write_json(
            RESTORE_STATE_PATH,
            state
        )

        def member_commit_key(member: dict[str, object]) -> tuple[int, str]:
            source_name = str(
                member.get("source_path")
            )
            return (
                (
                    1
                    if source_name
                    == restore_context.canonical_root_name
                    else 0
                ),
                source_name,
            )

        for member in sorted(
            new_members,
            key=member_commit_key,
        ):
            source_name = _validate_restore_source_name(
                member.get("source_path"),
                field_name="restore.member.source_path",
            )
            new_staging_name = _validate_restore_staging_name(
                member.get("new_staging_name"),
                field_name="restore.member.new_staging_name",
            )
            old_staging_name_value = member.get("old_staging_name")
            canonical_path = (
                top_level_mount
                / source_name
            )
            new_staging_path = (
                top_level_mount
                / new_staging_name
            )

            if member.get("had_target") is True:
                old_staging_name = _validate_restore_staging_name(
                    old_staging_name_value,
                    field_name="restore.member.old_staging_name",
                )
                old_staging_path = (
                    top_level_mount
                    / old_staging_name
                )

                if not canonical_path.exists() or canonical_path.is_symlink():
                    raise HelperError(
                        f"Restore target disappeared during commit: {source_name}"
                    )

                os.rename(
                    canonical_path,
                    old_staging_path,
                )
                commit_log.append(
                    (
                        "old-moved",
                        source_name,
                        old_staging_name,
                    )
                )
                fsync_directory(
                    top_level_mount
                )

            elif canonical_path.exists() or canonical_path.is_symlink():
                raise HelperError(
                    f"Restore target appeared unexpectedly: {source_name}"
                )

            os.rename(
                new_staging_path,
                canonical_path,
            )
            commit_log.append(
                (
                    "new-promoted",
                    source_name,
                    new_staging_name,
                )
            )
            fsync_directory(
                top_level_mount
            )

        if scope == "full":
            historical_name_set = set(
                historical_names
            )

            for old_target in old_targets:
                source_name = _validate_restore_source_name(
                    old_target.get("source_path"),
                    field_name="restore.old_target.source_path",
                )

                if source_name in historical_name_set:
                    continue

                old_staging_name = _validate_restore_staging_name(
                    old_target.get("staging_name"),
                    field_name="restore.old_target.staging_name",
                )
                canonical_path = (
                    top_level_mount
                    / source_name
                )
                old_staging_path = (
                    top_level_mount
                    / old_staging_name
                )

                if not canonical_path.exists() or canonical_path.is_symlink():
                    raise HelperError(
                        f"Full restore extra target disappeared: {source_name}"
                    )

                os.rename(
                    canonical_path,
                    old_staging_path,
                )
                commit_log.append(
                    (
                        "old-moved",
                        source_name,
                        old_staging_name,
                    )
                )
                fsync_directory(
                    top_level_mount
                )

        state["phase"] = "btrfs-committed"
        atomic_write_json(
            RESTORE_STATE_PATH,
            state
        )

        boot_payload = payload.get("boot")
        boot_result: dict[str, object]

        if isinstance(boot_payload, dict) and boot_payload.get("mode") == "archive":
            pre_boot_payload = pre_restore_manifest.get("boot")

            if not isinstance(pre_boot_payload, dict) or pre_boot_payload.get("mode") != "archive":
                raise HelperError(
                    "Pre-restore snapshot set does not contain the required "
                    "/boot rollback archive."
                )

            state["phase"] = "boot-restoring"
            atomic_write_json(
                RESTORE_STATE_PATH,
                state
            )

            restore_archive = (
                set_directory
                / "archives"
                / "boot.tar.zst"
            )
            rollback_archive = (
                pre_restore_directory
                / "archives"
                / "boot.tar.zst"
            )
            rollback_boot_archive_path = rollback_archive
            boot_result = _replace_boot_archive_transactionally(
                restore_archive,
                rollback_archive,
            )
            boot_replaced = True

        elif isinstance(boot_payload, dict) and boot_payload.get("mode") == "included-in-root":
            boot_result = {
                "mode": "included-in-root",
                "replaced": True,
                "efi_preserved": True,
            }

        else:
            boot_result = {
                "mode": None,
                "replaced": False,
                "efi_preserved": True,
            }

        for member in new_members:
            source_name = _validate_restore_source_name(
                member.get("source_path"),
                field_name="restore.member.source_path",
            )
            restored_uuid = normalize_uuid_text(
                member.get("restored_uuid"),
                "restore.member.restored_uuid",
            )
            canonical_path = (
                top_level_mount
                / source_name
            )
            restored = read_subvolume_record(
                canonical_path
            )

            if (
                restored.uuid != restored_uuid
                or restored.parent_id != TOP_LEVEL_SUBVOLUME_ID
                or restored.name != source_name
                or subvolume_is_read_only(canonical_path)
            ):
                raise HelperError(
                    f"Restored target validation failed: {source_name}"
                )

        os.sync()

        state["phase"] = "applied"
        state["restart_required"] = True
        state["boot"] = boot_result
        atomic_write_json(
            RESTORE_STATE_PATH,
            state
        )
        restore_applied = True

        result = {
            "format_version": 1,
            "operation": "restore",
            "restored": True,
            "transaction_id": transaction_id,
            "set_id": set_id,
            "set_name": payload.get("set_name"),
            "scope": scope,
            "purpose": purpose,
            "pre_restore": {
                "set_id": pre_restore_set_id,
                "set_name": pre_restore_set_name,
                "scope": scope,
                "purpose": "pre-restore",
            },
            "members": [
                {
                    "source_path": member.get("source_path"),
                    "restored_id": member.get("restored_id"),
                    "restored_uuid": member.get("restored_uuid"),
                }
                for member in new_members
            ],
            "boot": boot_result,
            "restart_required": True,
            "second_restore_blocked_until_reboot": True,
        }

        sys.stdout.write(
            json.dumps(
                result,
                ensure_ascii=False,
                indent=2,
            )
            + "\n"
        )

        return 0

    except Exception as operation_error:
        if restore_applied:
            raise HelperError(
                f"{operation_error} The restore is already applied; "
                "restart the system before doing anything else."
            ) from operation_error

        rollback_errors: list[str] = []

        if boot_replaced:
            if rollback_boot_archive_path is None:
                rollback_errors.append(
                    "/boot was replaced but its rollback archive is unknown"
                )
            else:
                try:
                    _replace_boot_archive_transactionally(
                        rollback_boot_archive_path,
                        rollback_boot_archive_path,
                    )
                    boot_replaced = False
                except Exception as exc:
                    rollback_errors.append(
                        f"/boot restore rollback: {exc}"
                    )

        if commit_log:
            rollback_errors.extend(
                _rollback_btrfs_restore_commit(
                    top_level_mount,
                    commit_log,
                )
            )

        rollback_errors.extend(
            _cleanup_new_restore_staging(
                top_level_mount,
                new_members,
            )
        )

        if state_written and not rollback_errors:
            try:
                remove_restore_state()
                state_written = False
            except Exception as exc:
                rollback_errors.append(
                    f"restore state cleanup: {exc}"
                )

        if rollback_errors:
            if state_written:
                try:
                    state_payload = load_restore_state()

                    if state_payload is not None:
                        state_payload["phase"] = "rollback-incomplete"
                        state_payload["restart_required"] = False
                        atomic_write_json(
                            RESTORE_STATE_PATH,
                            state_payload,
                        )
                except Exception:
                    pass

            raise HelperError(
                f"{operation_error} Restore rollback was incomplete: "
                + "; ".join(rollback_errors)
            ) from operation_error

        raise

    finally:
        unmount_top_level(
            top_level_mount
        )


def scan_subvolumes() -> int:
    """
    List Btrfs subvolumes for the running root filesystem.
    """

    result = run_command(
        [
            str(BTRFS_COMMAND),
            "subvolume",
            "list",
            "-p",
            "-u",
            "-q",
            "/",
        ],
        check=False,
    )

    if result.stdout:
        sys.stdout.write(result.stdout)

    if result.stderr:
        sys.stderr.write(result.stderr)

    return result.returncode


def configure_snapshot_storage() -> int:
    """
    Reconcile the system toward the canonical snapshot storage layout.

    Final state:
        @snapshots exists as a direct top-level Btrfs subvolume.
        /.snapshots exists as its mountpoint.
        /etc/fstab contains one valid persistent mount entry.
        /.snapshots is mounted.

    The operation is idempotent and safe to run repeatedly.
    """

    root = get_root_filesystem_info()

    original_fstab = read_fstab()

    fstab_entries = parse_fstab(
        original_fstab
    )

    snapshot_source = (
        get_snapshot_fstab_source(
            fstab_entries,
            root,
        )
    )

    mount_options = (
        derive_snapshot_mount_options(
            fstab_entries
        )
    )

    fstab_already_valid = (
        validate_fstab_state(
            fstab_entries,
            root,
        )
    )

    mount_state = (
        get_snapshot_mount_state(
            root
        )
    )

    create_mountpoint = (
        validate_mountpoint(
            mount_state
        )
    )

    top_level_mount = mount_top_level(
        root
    )

    created_subvolume = False
    created_mountpoint = False
    changed_fstab = False
    mounted_by_us = False

    try:
        create_subvolume = (
            validate_snapshot_subvolume(
                top_level_mount
            )
        )

        if create_subvolume:
            create_snapshot_subvolume(
                top_level_mount
            )

            created_subvolume = True

        if create_mountpoint:
            SNAPSHOT_MOUNTPOINT.mkdir(
                mode=0o755,
                parents=False,
                exist_ok=False,
            )

            created_mountpoint = True

        if not fstab_already_valid:
            new_fstab = (
                build_fstab_with_snapshot_entry(
                    original_fstab,
                    snapshot_source,
                    mount_options,
                )
            )

            atomic_write_fstab(
                new_fstab
            )

            changed_fstab = True

            daemon_reload()

        current_mount_state = (
            get_snapshot_mount_state(
                root
            )
        )

        if not current_mount_state.mounted:
            run_command(
                [
                    str(MOUNT_COMMAND),
                    str(SNAPSHOT_MOUNTPOINT),
                ]
            )

            mounted_by_us = True

        final_mount_state = (
            get_snapshot_mount_state(
                root
            )
        )

        if (
            not final_mount_state.mounted
            or not final_mount_state.correct
        ):
            raise HelperError(
                "Snapshot storage could not be mounted "
                "with the expected Btrfs layout."
            )

    except Exception as operation_error:
        rollback_errors: list[str] = []

        if mounted_by_us:
            try:
                run_command(
                    [
                        str(UMOUNT_COMMAND),
                        str(SNAPSHOT_MOUNTPOINT),
                    ]
                )
            except Exception as exc:
                rollback_errors.append(
                    f"unmount: {exc}"
                )

        if changed_fstab:
            try:
                atomic_write_fstab(
                    original_fstab
                )

                daemon_reload()

            except Exception as exc:
                rollback_errors.append(
                    f"fstab: {exc}"
                )

        if created_mountpoint:
            try:
                if (
                    SNAPSHOT_MOUNTPOINT.exists()
                    and SNAPSHOT_MOUNTPOINT.is_dir()
                    and not any(
                        SNAPSHOT_MOUNTPOINT.iterdir()
                    )
                ):
                    SNAPSHOT_MOUNTPOINT.rmdir()

            except Exception as exc:
                rollback_errors.append(
                    f"mountpoint: {exc}"
                )

        if created_subvolume:
            try:
                delete_created_snapshot_subvolume(
                    top_level_mount
                )

            except Exception as exc:
                rollback_errors.append(
                    f"subvolume: {exc}"
                )

        if rollback_errors:
            raise HelperError(
                f"{operation_error} "
                "Rollback was incomplete: "
                + "; ".join(rollback_errors)
            ) from operation_error

        raise

    finally:
        unmount_top_level(
            top_level_mount
        )

    return scan_subvolumes()



def acquire_helper_lock() -> int:
    """
    Serialize privileged helper operations across processes.
    """

    try:
        lock_fd = os.open(
            str(HELPER_LOCK_PATH),
            os.O_CREAT | os.O_RDWR,
            0o600,
        )
    except OSError as exc:
        raise HelperError(
            f"Unable to open the Synex Snapshots helper lock: {exc}"
        ) from exc

    try:
        fcntl.flock(
            lock_fd,
            fcntl.LOCK_EX,
        )
    except OSError as exc:
        os.close(
            lock_fd
        )
        raise HelperError(
            f"Unable to acquire the Synex Snapshots helper lock: {exc}"
        ) from exc

    return lock_fd

def main() -> int:
    """
    Execute an explicitly supported privileged action.
    """

    if len(sys.argv) < 2:
        print_error(
            "Usage: synex-snapshots-root-helper ACTION [ARGUMENTS]"
        )
        return 2

    action = sys.argv[1]
    arguments = sys.argv[2:]

    if action not in ALLOWED_ACTIONS:
        print_error(
            f"Unsupported action: {action}"
        )
        return 2

    lock_fd: int | None = None

    try:
        require_root()
        lock_fd = acquire_helper_lock()
        ensure_required_commands()

        if action == "scan-subvolumes":
            if arguments:
                raise HelperError(
                    "scan-subvolumes does not accept arguments."
                )

            return scan_subvolumes()

        if action == "configure-snapshot-storage":
            if arguments:
                raise HelperError(
                    "configure-snapshot-storage does not accept arguments."
                )

            return configure_snapshot_storage()

        if action == "create-single-snapshot":
            if len(arguments) != 1:
                raise HelperError(
                    "create-single-snapshot requires exactly one "
                    "Btrfs subvolume ID."
                )

            try:
                subvolume_id = int(
                    arguments[0],
                    10,
                )
            except ValueError as exc:
                raise HelperError(
                    "The Btrfs subvolume ID must be an integer."
                ) from exc

            if subvolume_id <= 0:
                raise HelperError(
                    "The Btrfs subvolume ID must be greater than zero."
                )

            return create_single_snapshot(
                subvolume_id
            )

        if action == "create-full-snapshot":
            if arguments:
                raise HelperError(
                    "create-full-snapshot does not accept arguments."
                )

            return create_full_snapshot()

        if action == "delete-snapshot-set":
            if len(arguments) != 1:
                raise HelperError(
                    "delete-snapshot-set requires exactly one "
                    "snapshot set UUID."
                )

            set_id = normalize_uuid_text(
                arguments[0],
                "set_id",
            )

            return delete_snapshot_set(
                set_id
            )

        if action == "preflight-restore":
            if len(arguments) != 1:
                raise HelperError(
                    "preflight-restore requires exactly one "
                    "snapshot set UUID."
                )

            set_id = normalize_uuid_text(
                arguments[0],
                "set_id",
            )

            return preflight_restore(
                set_id
            )

        if action == "restore-snapshot-set":
            if len(arguments) != 1:
                raise HelperError(
                    "restore-snapshot-set requires exactly one "
                    "snapshot set UUID."
                )

            set_id = normalize_uuid_text(
                arguments[0],
                "set_id",
            )

            return restore_snapshot_set(
                set_id
            )

        if action == "restore-status":
            if arguments:
                raise HelperError(
                    "restore-status does not accept arguments."
                )

            return restore_status()

        if action == "finalize-restore":
            if arguments:
                raise HelperError(
                    "finalize-restore does not accept arguments."
                )

            return finalize_restore()

        if action == "set-automation-config":
            if len(arguments) != 3:
                raise HelperError(
                    "set-automation-config requires ENABLED, FREQUENCY "
                    "and RETENTION arguments."
                )

            return set_automation_config(
                arguments[0],
                arguments[1],
                arguments[2],
            )

        if action == "automation-run":
            if arguments:
                raise HelperError(
                    "automation-run does not accept arguments."
                )

            return automation_run()

        raise HelperError(
            f"Unsupported action: {action}"
        )

    except HelperError as exc:
        print_error(str(exc))
        return 1

    except Exception as exc:
        print_error(
            f"Unexpected helper error: {exc}"
        )
        return 1

    finally:
        if lock_fd is not None:
            os.close(
                lock_fd
            )

if __name__ == "__main__":
    raise SystemExit(main())
