#!/bin/bash
# Install or adopt the CloudNVR two-service Docker product.
set -Eeuo pipefail

ROOT="${CLOUDNVR_ROOT:-/opt/dividia}"
DEFAULT_BASE_URL="http://files.dividia.net/software/cloudnvr-docker"
BASE_URL="${CLOUDNVR_BUNDLE_BASE_URL:-$DEFAULT_BASE_URL}"
CHANNEL="${CLOUDNVR_CHANNEL:-dev}"
CHANNEL_EXPLICIT=false
PLATFORM=linux/arm64
REGISTRY=docker.io
COMPOSE_PROJECT_NAME=cloudnvr
BACKEND_ADDR="${CLOUDNVR_BACKEND_ADDR:-}"
BACKEND_ADDR_EXPLICIT=false
[[ -n "$BACKEND_ADDR" ]] && BACKEND_ADDR_EXPLICIT=true
BUNDLE_DIR=""
CONFIG_DIR=""
VIDEO_SOURCE=""
ALLOW_ROOT_VIDEO=false
GENERATE_SELF_SIGNED=false
ADOPT_EXISTING=false
LEGACY_DRAGON_DIR=""
MARKER_FILE="$ROOT/.product"
ENV_FILE="$ROOT/.env"
SERVICE_FILE="${CLOUDNVR_SERVICE_FILE:-/etc/systemd/system/cloudnvr.service}"
MANUAL_RECOVERY_UNIT_NAME=cloudnvr-legacy-dragon-recovery.service
MANUAL_RECOVERY_UNIT="${CLOUDNVR_MANUAL_RECOVERY_UNIT:-/etc/systemd/system/$MANUAL_RECOVERY_UNIT_NAME}"
MANUAL_RECOVERY_SCRIPT="$ROOT/state/manual-dragon-recovery.py"
MANUAL_RECOVERY_STATE="$ROOT/state/manual-dragon-recovery.json"
MANUAL_RECOVERY_LOG="$ROOT/state/manual-dragon-recovery.log"
FSTAB_FILE="${CLOUDNVR_FSTAB_FILE:-/etc/fstab}"
TLS_CA_FILE="${CLOUDNVR_TLS_CA_FILE:-}"
TLS_EXPECTED_NAME="${CLOUDNVR_TLS_EXPECTED_NAME:-}"
PROC_ROOT="${CLOUDNVR_PROC_ROOT:-/proc}"
DEFAULT_LOCK_FILE=/run/lock/cloudnvr/mutation.lock
LOCK_FILE="${CLOUDNVR_LOCK_FILE:-$DEFAULT_LOCK_FILE}"
SKIP_HOST_VALIDATION="${CLOUDNVR_SKIP_HOST_VALIDATION:-0}"
SKIP_PORT_CHECK="${CLOUDNVR_SKIP_PORT_CHECK:-0}"
BACKUP_DIR=""
LEGACY_DRAGON_PID=""
LEGACY_DRAGON_EXE=""
LEGACY_DRAGON_STARTTIME=""
LEGACY_DRAGON_CONF=""
LEGACY_DRAGON_SERVICE_ACTIVE=false
LEGACY_CONNECTOR_SERVICE_ACTIVE=false
LEGACY_DRAGON_SERVICE_ENABLED=false
LEGACY_CONNECTOR_SERVICE_ENABLED=false
LEGACY_DRAGON_UNIT_HASH=""
LEGACY_CONNECTOR_UNIT_HASH=""
LEGACY_CONNECTOR_PID=""
LEGACY_CONNECTOR_CONF=""
LEGACY_CONNECTOR_DB=""
LEGACY_MANUAL_STATE=""
LEGACY_MANUAL_STOPPED=false
MANUAL_RECOVERY_INSTALLED=false
VIDEO_BIND_ADDED=false
FSTAB_BACKUP_FILE=""
FSTAB_WRITE_FILE=""
FSTAB_EXISTED=false
FSTAB_CHANGED=false
SUPPORT_REGISTRATION=true

log() { printf '%s\n' "$*"; }
step() { printf '\n==> %s\n' "$*"; }
warn() { printf 'WARNING: %s\n' "$*" >&2; }
die() { printf 'ERROR: %s\n' "$*" >&2; return 1; }

usage() {
    printf '%s\n' \
        'Usage: sudo install-cloudnvr.sh [options]' \
        '' \
        '  --channel CHANNEL          dev, dev-<suffix>, or YYYY.MM.INCREMENT' \
        '  --backend-address HOST     engine backend DNS name or IPv4 address' \
        '  --bundle-dir DIR           use an already downloaded verified bundle' \
        '  --config-dir DIR           approved dragon.conf, connector.conf, TLS, users.htpasswd, and' \
        '                             connector/cloudapi.secret (required when [cloud] enabled=1' \
        '                             without a static api_token)' \
        '  --video-source DIR         bind an existing video filesystem at /opt/dividia/data/video' \
        '  --allow-root-video         allow video below the root filesystem for a bench install' \
        '  --generate-self-signed     create a bench-only self-signed TLS certificate' \
        '  --tls-name NAME            expected production certificate DNS name or IP address' \
        '  --adopt-existing           adopt a recognized manual Dragon installation' \
        '  --legacy-dragon-dir DIR    expected directory for a manual Dragon binary' \
        '  --no-support-registration  leave remote support enrollment to staff approval' \
        '  --help'
}

# `production` is the stable fleet alias, re-pointed at every promotion. Install
# a customer appliance onto it rather than a dated YYYY.M.N channel, or that box
# never moves forward without someone tunnelling in to run `nvr channel`. The
# dated channels remain valid install targets for pinning a box deliberately.
channel_is_valid() {
    [[ "$1" =~ ^(production|dev(-[a-z0-9-]+)?|[0-9]{4}\.(1[0-2]|[1-9])\.[1-9][0-9]*)$ ]]
}

# The alias only ever points at a production bundle, so every strict production
# gate below (auth required, https_enabled, TLS name) must apply to it too.
channel_is_production() {
    [[ "$1" == production ]] && return 0
    [[ "$1" =~ ^[0-9]{4}\.([1-9]|1[0-2])\.[1-9][0-9]*$ ]]
}

backend_address_is_valid() {
    local address="$1" label
    local -a labels
    [[ -n "$address" && ${#address} -le 253 ]] || return 1
    [[ "$address" =~ ^[A-Za-z0-9.-]+$ ]] || return 1
    [[ "$address" != .* && "$address" != *. && "$address" != *..* ]] || return 1
    IFS='.' read -r -a labels <<< "$address"
    for label in "${labels[@]}"; do
        [[ ${#label} -le 63 ]] || return 1
        [[ "$label" =~ ^[A-Za-z0-9]([A-Za-z0-9-]*[A-Za-z0-9])?$ ]] || return 1
    done
}

finalize_backend_address() {
    if [[ -z "$BACKEND_ADDR" ]]; then
        if channel_is_production "$CHANNEL"; then
            BACKEND_ADDR="engine.cloud.dividia.net"
        else
            BACKEND_ADDR="engine-staging.cloud.dividia.net"
        fi
    fi
    backend_address_is_valid "$BACKEND_ADDR" \
        || { die "installed backend address is invalid: $BACKEND_ADDR"; return 1; }
}

parse_args() {
    while [[ "$#" -gt 0 ]]; do
        case "$1" in
            --channel|--backend-address|--bundle-dir|--config-dir|--video-source|--legacy-dragon-dir|--tls-name)
                [[ "$#" -ge 2 ]] || { die "$1 requires a value"; return 2; }
                case "$1" in
                    --channel) CHANNEL="$2" ;;
                    --backend-address) BACKEND_ADDR="$2"; BACKEND_ADDR_EXPLICIT=true ;;
                    --bundle-dir) BUNDLE_DIR="$2" ;;
                    --config-dir) CONFIG_DIR="$2" ;;
                    --video-source) VIDEO_SOURCE="$2" ;;
                    --legacy-dragon-dir) LEGACY_DRAGON_DIR="$2" ;;
                    --tls-name) TLS_EXPECTED_NAME="$2" ;;
                esac
                [[ "$1" == --channel ]] && CHANNEL_EXPLICIT=true
                shift 2
                ;;
            --allow-root-video) ALLOW_ROOT_VIDEO=true; shift ;;
            --generate-self-signed) GENERATE_SELF_SIGNED=true; shift ;;
            --adopt-existing) ADOPT_EXISTING=true; shift ;;
            --no-support-registration) SUPPORT_REGISTRATION=false; shift ;;
            -h|--help) usage; exit 0 ;;
            *) die "unknown option: $1"; usage >&2; return 2 ;;
        esac
    done
    channel_is_valid "$CHANNEL" || { die "invalid channel: $CHANNEL"; return 2; }
    if [[ "$BACKEND_ADDR_EXPLICIT" == true ]] && ! backend_address_is_valid "$BACKEND_ADDR"; then
        die "invalid backend address: $BACKEND_ADDR"
        return 2
    fi
    # Self-signed is permitted on a production channel as the BOOTSTRAP state
    # (ADR-015). These appliances ship with no DNS name and take a DHCP address on
    # the customer LAN, so at install time there is no stable identity for a SAN and
    # no CA a client would already trust. The original gate assumed a managed
    # deployment with real names; on a fielded Orin it only blocked the install.
    # require_auth and HTTPS still hold, so admin credentials stay encrypted in
    # transit. A trusted certificate arrives later over the profile channel.
    if channel_is_production "$CHANNEL" && [[ -z "$TLS_EXPECTED_NAME" && "$GENERATE_SELF_SIGNED" != true ]]; then
        die "production channels require --tls-name for a supplied certificate"
        return 2
    fi
}

require_root() {
    if [[ "${CLOUDNVR_ALLOW_NON_ROOT:-0}" != 1 && "$EUID" -ne 0 ]]; then
        die "the installer requires root"
    fi
}

kv_get() {
    local file="$1" wanted="$2" key value found=""
    [[ -f "$file" ]] || return 1
    while IFS='=' read -r key value || [[ -n "$key$value" ]]; do
        [[ "$key" == "$wanted" ]] && found="$value"
    done < "$file"
    [[ -n "$found" ]] || return 1
    printf '%s\n' "$found"
}

marker_is_ours() {
    [[ -f "$MARKER_FILE" ]] || return 1
    [[ "$(kv_get "$MARKER_FILE" product 2>/dev/null || true)" == cloudnvr ]] || return 1
    [[ "$(kv_get "$MARKER_FILE" schema 2>/dev/null || true)" == 1 ]]
}

validate_existing_root() {
    python3 - "$ROOT" "${CLOUDNVR_ALLOW_NON_ROOT:-0}" <<'PY' || {
import os, pathlib, stat, sys
root = pathlib.Path(sys.argv[1])
allow_non_root = sys.argv[2] == "1"
if os.path.lexists(root):
    root_stat = os.lstat(root)
    if not stat.S_ISDIR(root_stat.st_mode) or stat.S_ISLNK(root_stat.st_mode):
        raise SystemExit("CloudNVR product root must be a real directory")
    paths = [root]
    paths += [root / name for name in (
        "bin", "config", "data", "releases", "state", "adoption-backups")]
    for path in paths:
        if not os.path.lexists(path):
            continue
        item = os.lstat(path)
        if not stat.S_ISDIR(item.st_mode) or stat.S_ISLNK(item.st_mode):
            raise SystemExit("CloudNVR product directories cannot be symlinks")
        if not allow_non_root and (item.st_uid != 0 or stat.S_IMODE(item.st_mode) & 0o022):
            raise SystemExit("CloudNVR product directories must be root-owned and not writable by group or other")
    for name in (".product", ".env"):
        path = root / name
        if not os.path.lexists(path):
            continue
        item = os.lstat(path)
        if not stat.S_ISREG(item.st_mode) or stat.S_ISLNK(item.st_mode):
            raise SystemExit("CloudNVR product metadata must be regular files")
        if not allow_non_root and (item.st_uid != 0 or stat.S_IMODE(item.st_mode) & 0o022):
            raise SystemExit("CloudNVR product metadata must be root-owned and not writable by group or other")
PY
        die "unsafe CloudNVR product root: $ROOT"
        return 1
    }
    if [[ -f "$MARKER_FILE" ]]; then
        marker_is_ours || { die "$ROOT belongs to another product or schema"; return 1; }
        return 0
    fi
    if [[ -d "$ROOT" && -n "$(find "$ROOT" -mindepth 1 -maxdepth 1 -print -quit 2>/dev/null)" ]]; then
        die "$ROOT is non-empty and has no valid CloudNVR marker"
        return 1
    fi
}

load_existing_install_settings() {
    marker_is_ours || return 0
    if [[ "$CHANNEL_EXPLICIT" != true ]]; then
        CHANNEL="$(kv_get "$ENV_FILE" CHANNEL 2>/dev/null || printf '%s' "$CHANNEL")"
    fi
    if [[ -z "$VIDEO_SOURCE" ]]; then
        VIDEO_SOURCE="$(kv_get "$ENV_FILE" VIDEO_SOURCE 2>/dev/null || true)"
    fi
    if [[ "$(kv_get "$ENV_FILE" ALLOW_ROOT_VIDEO 2>/dev/null || true)" == 1 ]]; then
        ALLOW_ROOT_VIDEO=true
    fi
    if [[ "$BACKEND_ADDR_EXPLICIT" != true ]]; then
        BACKEND_ADDR="$(kv_get "$ENV_FILE" BACKEND_ADDR 2>/dev/null || true)"
    fi
    BASE_URL="$(kv_get "$ENV_FILE" BUNDLE_BASE_URL 2>/dev/null || printf '%s' "$BASE_URL")"
    channel_is_valid "$CHANNEL" || { die "installed channel is invalid: $CHANNEL"; return 1; }
}

atomic_write() {
    local path="$1" mode="${2:-0644}"
    python3 -c '
import os, pathlib, sys, tempfile
path = pathlib.Path(sys.argv[1]); mode = int(sys.argv[2], 8)
path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp = tempfile.mkstemp(prefix="." + path.name + ".", dir=str(path.parent))
try:
    with os.fdopen(fd, "wb") as out:
        out.write(sys.stdin.buffer.read()); out.flush(); os.fsync(out.fileno())
    os.chmod(tmp, mode); os.replace(tmp, path)
    dfd = os.open(path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
    try: os.fsync(dfd)
    finally: os.close(dfd)
except BaseException:
    try: os.unlink(tmp)
    except FileNotFoundError: pass
    raise
' "$path" "$mode"
}

durable_sync_dirs() {
    python3 - "$@" <<'PY'
import os, pathlib, stat, sys

fail = os.environ.get("CLOUDNVR_TEST_DIR_FSYNC_FAIL", "")
paths = set()
for raw in sys.argv[1:]:
    path = pathlib.Path(raw)
    info = os.lstat(path)
    if not stat.S_ISDIR(info.st_mode) or stat.S_ISLNK(info.st_mode):
        raise SystemExit("durability target is not a real directory: %s" % path)
    paths.add(path)
    paths.add(path.parent)

for path in sorted(paths, key=lambda item: len(item.parts), reverse=True):
    if fail and (str(path) == fail or path.name == fail):
        raise OSError("injected directory sync failure: %s" % path)
    fd = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
    try:
        os.fsync(fd)
    finally:
        os.close(fd)
PY
}

atomic_copy() {
    local source="$1" target="$2" mode="${3:-0644}"
    case "${CLOUDNVR_TEST_INSTALLER_COPY_FAIL:-}:$target" in
        "host-nvr:$ROOT/bin/nvr"|"config-connector:$ROOT/config/connector.conf") return 1 ;;
    esac
    [[ -f "$source" && ! -L "$source" && -s "$source" ]] \
        || { die "invalid source file: $source"; return 1; }
    atomic_write "$target" "$mode" < "$source"
}

atomic_replace_file() {
    local path="$1" data="$2" metadata="$3" preserve_times="${4:-0}"
    python3 - "$path" "$data" "$metadata" "$preserve_times" <<'PY'
import os, pathlib, subprocess, sys, tempfile
path, data, metadata = map(pathlib.Path, sys.argv[1:4])
preserve_times = sys.argv[4] == "1"
source_stat = os.stat(metadata)
path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp = tempfile.mkstemp(prefix="." + path.name + ".", dir=str(path.parent))
try:
    os.close(fd)
    os.unlink(tmp)
    subprocess.run(["cp", "-a", str(metadata), tmp], check=True)
    with open(tmp, "wb") as out, open(data, "rb") as source:
        while True:
            block = source.read(1024 * 1024)
            if not block: break
            out.write(block)
        out.flush()
        os.fsync(out.fileno())
    if preserve_times:
        os.utime(tmp, ns=(source_stat.st_atime_ns, source_stat.st_mtime_ns))
    sync_fd = os.open(tmp, os.O_RDONLY)
    try: os.fsync(sync_fd)
    finally: os.close(sync_fd)
    os.replace(tmp, path)
    dfd = os.open(path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
    try: os.fsync(dfd)
    finally: os.close(dfd)
except BaseException:
    try: os.unlink(tmp)
    except FileNotFoundError: pass
    raise
PY
}

sha256_file() {
    if command -v sha256sum >/dev/null 2>&1; then
        sha256sum "$1" | awk '{print $1}'
    else
        shasum -a 256 "$1" | awk '{print $1}'
    fi
}

resolve_bundle() {
    local destination="$1" name url
    local -a curl_protocols
    if [[ -n "$BUNDLE_DIR" ]]; then
        BUNDLE_DIR="$(cd "$BUNDLE_DIR" && pwd -P)" || return 1
        validate_bundle "$BUNDLE_DIR"
        return
    fi
    BUNDLE_DIR="$destination/bundle"
    mkdir -p "$BUNDLE_DIR"
    # files.dividia.net serves plain HTTP, the same as the rest of the product.
    # .manifest.sha256 comes down with the bundle and catches a corrupt or
    # truncated download. --bundle-dir takes a bundle staged from the publish
    # host when you want to skip the fetch entirely.
    case "$BASE_URL" in
        https://*) curl_protocols=(--proto '=https' --proto-redir '=https') ;;
        http://*)  curl_protocols=(--proto '=http,https' --proto-redir '=http,https') ;;
        *) die "install bundle URL uses an unsupported protocol: $BASE_URL"; return 1 ;;
    esac
    for name in docker-compose.yml install-cloudnvr.sh nvr cloudnvr.service release.json .manifest.sha256; do
        url="${BASE_URL%/}/$CHANNEL/$name"
        curl -fsSL "${curl_protocols[@]}" --connect-timeout 10 --max-time 60 \
            "$url" -o "$BUNDLE_DIR/$name" \
            || { die "cannot download $url"; return 1; }
    done
    validate_bundle "$BUNDLE_DIR"
}

validate_bundle() {
    local directory="$1" expected name actual count=0 seen=" " required
    local manifest="$directory/.manifest.sha256"
    [[ -f "$manifest" && ! -L "$manifest" && -s "$manifest" ]] \
        || { die "bundle is missing .manifest.sha256"; return 1; }
    while read -r expected name || [[ -n "$expected$name" ]]; do
        name="${name# }"
        [[ "$expected" =~ ^[0-9a-f]{64}$ ]] || { die "invalid bundle hash"; return 1; }
        case "$name" in
            docker-compose.yml|install-cloudnvr.sh|nvr|cloudnvr.service|release.json) ;;
            *) die "unexpected bundle manifest entry: $name"; return 1 ;;
        esac
        case "$seen" in
            *" $name "*) die "duplicate bundle manifest entry: $name"; return 1 ;;
        esac
        seen+="$name "
        [[ -f "$directory/$name" && ! -L "$directory/$name" && -s "$directory/$name" ]] \
            || { die "invalid bundle file: $name"; return 1; }
        actual="$(sha256_file "$directory/$name")"
        [[ "$actual" == "$expected" ]] || { die "bundle hash mismatch: $name"; return 1; }
        count=$((count + 1))
    done < "$manifest"
    [[ "$count" -eq 5 ]] || { die "bundle manifest must contain exactly five entries"; return 1; }
    for required in docker-compose.yml install-cloudnvr.sh nvr cloudnvr.service release.json; do
        case "$seen" in *" $required "*) ;; *) die "bundle manifest omitted $required"; return 1 ;; esac
    done
    [[ "$(find "$directory" -mindepth 1 -maxdepth 1 | wc -l | tr -d ' ')" -eq 6 ]] \
        || { die "bundle has missing or extra files"; return 1; }
    bash -n "$directory/nvr" || { die "bundle nvr script failed bash -n"; return 1; }
    bash -n "$directory/install-cloudnvr.sh" || { die "bundle installer failed bash -n"; return 1; }
}

manifest_read() {
    local file="$1" parsed
    parsed="$(python3 - "$file" <<'PY'
import hashlib, json, re, sys
v = json.load(open(sys.argv[1], encoding="utf-8")); i = v.get("images", {})
d, c = i.get("dragon", {}), i.get("connector", {})
f = [v.get("release_id", ""), v.get("platform", ""), v.get("source_revision", ""),
     v.get("connector_nvr_source_revision", ""), d.get("repository", ""), d.get("digest", ""),
     d.get("reference", ""), c.get("repository", ""), c.get("digest", ""), c.get("reference", "")]
schema=v.get("schema")
if type(schema) is not int or schema not in (1,2) or v.get("product") != "cloudnvr": raise SystemExit("bad schema")
if set(i) != ({"dragon","connector","remoted"} if schema == 2 else {"dragon","connector"}): raise SystemExit("bad services")
if any(not isinstance(x, str) or "\t" in x or "\n" in x for x in f): raise SystemExit("bad field")
rid,p,s,n,dr,dd,dref,cr,cd,cref=f
if p != "linux/arm64" or not re.fullmatch(r"[0-9a-f]{64}",rid): raise SystemExit("bad target")
if not re.fullmatch(r"[0-9a-f]{40}",s) or not re.fullmatch(r"[0-9a-f]{40}",n): raise SystemExit("bad source")
if dr != "docker.io/dividia/cloudnvr-dragon" or cr != "docker.io/dividia/cloudnvr-connector": raise SystemExit("bad repo")
if not re.fullmatch(r"sha256:[0-9a-f]{64}",dd) or not re.fullmatch(r"sha256:[0-9a-f]{64}",cd): raise SystemExit("bad digest")
if dref != dr+"@"+dd or cref != cr+"@"+cd: raise SystemExit("bad reference")
canonical="\n".join(["schema="+str(schema),"product=cloudnvr","platform="+p,"source_revision="+s,"nvr_source_revision="+n,"dragon="+dr+"@"+dd,"connector="+cr+"@"+cd,""])
if schema == 2:
    r=i["remoted"]; rn=v.get("remoted_nvr_source_revision", "")
    rr=r.get("repository", ""); rd=r.get("digest", ""); rf=r.get("reference", "")
    if rr != "docker.io/dividia/cloudnvr-remoted" or not re.fullmatch(r"sha256:[0-9a-f]{64}", rd): raise SystemExit("bad remoted image")
    if rf != rr+"@"+rd or not re.fullmatch(r"[0-9a-f]{40}", rn): raise SystemExit("bad remoted source")
    canonical += "remoted_nvr_source_revision="+rn+"\nremoted="+rf+"\n"
if hashlib.sha256(canonical.encode()).hexdigest()!=rid: raise SystemExit("bad identity")
print("\t".join(f))
PY
)" || { die "invalid release.json"; return 1; }
    # shellcheck disable=SC2034 # All fields are validated before the installer delegates to nvr.
    IFS=$'\t' read -r RELEASE_ID RELEASE_PLATFORM RELEASE_SOURCE RELEASE_NVR_SOURCE \
        DRAGON_REPO DRAGON_DIGEST DRAGON_REF CONNECTOR_REPO CONNECTOR_DIGEST CONNECTOR_REF <<< "$parsed"
}

validate_host() {
    [[ "$SKIP_HOST_VALIDATION" != 1 ]] || return 0
    local arch l4t runtimes free_kib min_free
    arch="${CLOUDNVR_HOST_ARCH:-$(uname -m)}"
    [[ "$arch" == aarch64 || "$arch" == arm64 ]] \
        || { die "CloudNVR requires ARM64; host reports $arch"; return 1; }
    l4t="${CLOUDNVR_L4T_RELEASE_FILE:-/etc/nv_tegra_release}"
    [[ -r "$l4t" ]] || { die "missing NVIDIA L4T release file: $l4t"; return 1; }
    grep -Eq 'R36\.4|R36.*REVISION:[[:space:]]*4([.,[:space:]]|$)' "$l4t" \
        || { die "CloudNVR requires JetPack 6 with L4T R36.4"; return 1; }
    command -v docker >/dev/null 2>&1 || { die "Docker is not installed"; return 1; }
    docker compose version >/dev/null 2>&1 || { die "Docker Compose v2 is not installed"; return 1; }
    runtimes="$(docker info --format '{{json .Runtimes}}' 2>/dev/null || true)"
    grep -q 'nvidia' <<< "$runtimes" || { die "Docker does not report the NVIDIA runtime"; return 1; }
    command -v flock >/dev/null 2>&1 || { die "flock is required"; return 1; }
    command -v python3 >/dev/null 2>&1 || { die "python3 is required"; return 1; }
    min_free="${CLOUDNVR_MIN_FREE_KIB:-10485760}"
    free_kib="$(df -Pk "${ROOT%/*}" 2>/dev/null | awk 'NR==2 {print $4}')"
    [[ "$free_kib" =~ ^[0-9]+$ && "$free_kib" -ge "$min_free" ]] \
        || { die "installation filesystem has less than $((min_free / 1024 / 1024)) GiB free"; return 1; }
}

validate_video_storage() {
    [[ "$SKIP_HOST_VALIDATION" != 1 ]] || return 0
    local check_path free_kib min_free
    min_free="${CLOUDNVR_MIN_VIDEO_FREE_KIB:-10485760}"
    if [[ -n "$VIDEO_SOURCE" ]]; then
        VIDEO_SOURCE="$(cd "$VIDEO_SOURCE" && pwd -P)" \
            || { die "video source is not a directory: $VIDEO_SOURCE"; return 1; }
        [[ ! "$VIDEO_SOURCE" =~ [[:space:]] ]] \
            || { die "video source paths with whitespace are not supported in fstab"; return 1; }
        if [[ "$ALLOW_ROOT_VIDEO" != true ]]; then
            mountpoint -q "$VIDEO_SOURCE" \
                || { die "video source is not a mounted filesystem: $VIDEO_SOURCE"; return 1; }
        fi
        [[ -w "$VIDEO_SOURCE" ]] || { die "video source is not writable: $VIDEO_SOURCE"; return 1; }
        check_path="$VIDEO_SOURCE"
    elif [[ "$ALLOW_ROOT_VIDEO" == true ]]; then
        check_path="${ROOT%/*}"
    elif marker_is_ours && mountpoint -q "$ROOT/data/video" 2>/dev/null; then
        check_path="$ROOT/data/video"
    else
        die "use --video-source for the appliance video filesystem, or --allow-root-video for a bench"
        return 1
    fi
    free_kib="$(df -Pk "$check_path" 2>/dev/null | awk 'NR==2 {print $4}')"
    [[ "$free_kib" =~ ^[0-9]+$ && "$free_kib" -ge "$min_free" ]] \
        || { die "video filesystem has less than $((min_free / 1024 / 1024)) GiB free"; return 1; }
}

ini_value() {
    local file="$1" section="$2" key="$3"
    python3 - "$file" "$section" "$key" <<'PY'
import re, sys
path, wanted_section, wanted_key = sys.argv[1:]
section = ""
value = None
for raw in open(path, encoding="utf-8"):
    line = raw.strip()
    match = re.match(r"^\[([^]]+)\]$", line)
    if match:
        section = match.group(1).strip().lower(); continue
    if section != wanted_section.lower() or not line or line[0] in ";#" or "=" not in line: continue
    key, item = line.split("=", 1)
    if key.strip().lower() == wanted_key.lower():
        value = item.split(";",1)[0].split("#",1)[0].strip().strip("\"")
if value is None: raise SystemExit(1)
print(value)
PY
}

canonical_port() {
    python3 - "$1" <<'PY'
import re, sys

value = sys.argv[1].strip()
if not re.fullmatch(r"[0-9]+", value):
    raise SystemExit(1)
port = int(value, 10)
if not 1 <= port <= 65535:
    raise SystemExit(1)
print(port)
PY
}

canonical_bool() {
    case "$(printf '%s' "$1" | tr '[:upper:]' '[:lower:]')" in
        1|true|yes|on) printf '1\n' ;;
        0|false|no|off) printf '0\n' ;;
        *) return 1 ;;
    esac
}

validate_accounts() {
    local file="$1"
    python3 - "$file" <<'PY'
import re, sys

# Accounts are validated for SHAPE only. The standard dividia/admin/user logins
# ship on every unit by design, so a password matching the seed is expected here
# rather than a reason to abort an install.
admins = 0; seen = set()
for number, raw in enumerate(open(sys.argv[1], encoding="utf-8"), 1):
    line = raw.rstrip("\n")
    if not line: continue
    parts = line.split(":")
    # Accept either scheme http.c verifies: a 40-char SHA-1 digest (legacy files)
    # or a crypt(3) bcrypt string, which is what 6dd7069 made the product generate.
    sha1 = re.fullmatch(r"[0-9a-f]{40}", parts[1]) is not None
    bcrypt = re.fullmatch(r"\$2[aby]\$[0-9]{2}\$[./A-Za-z0-9]{53}", parts[1]) is not None
    if (len(parts) not in (2, 3)
        or not re.fullmatch(r"[A-Za-z0-9_.@-]{1,31}", parts[0])
        or not (sha1 or bcrypt)):
        raise SystemExit("invalid account record on line %d" % number)
    role = parts[2] if len(parts) == 3 else "admin"
    if role not in ("admin", "user") or parts[0] in seen: raise SystemExit("invalid account role or duplicate")
    seen.add(parts[0]); admins += role == "admin"
if not seen or not admins: raise SystemExit("at least one administrator is required")
PY
}

acquire_install_lock() {
    command -v flock >/dev/null 2>&1 || { die "flock is required"; return 1; }
    python3 - "$LOCK_FILE" "${CLOUDNVR_ALLOW_NON_ROOT:-0}" "$DEFAULT_LOCK_FILE" <<'PY' || {
import os, pathlib, stat, sys

path = pathlib.Path(sys.argv[1])
allow_non_root = sys.argv[2] == "1"
default_path = pathlib.Path(sys.argv[3])
if not path.is_absolute():
    raise SystemExit("CloudNVR lock path must be absolute")
if not allow_non_root and path != default_path:
    raise SystemExit("CloudNVR lock overrides are test-only")
parent = path.parent
parent.mkdir(parents=True, exist_ok=True, mode=0o700)
parent_stat = os.lstat(parent)
if not stat.S_ISDIR(parent_stat.st_mode) or stat.S_ISLNK(parent_stat.st_mode):
    raise SystemExit("CloudNVR lock directory is not a real directory")
if not allow_non_root:
    if parent_stat.st_uid != 0:
        raise SystemExit("CloudNVR lock directory is not root-owned")
    os.chmod(parent, 0o700)
if os.path.lexists(path):
    lock_stat = os.lstat(path)
    if not stat.S_ISREG(lock_stat.st_mode) or stat.S_ISLNK(lock_stat.st_mode):
        raise SystemExit("CloudNVR lock file is not a regular file")
    if not allow_non_root and lock_stat.st_uid != 0:
        raise SystemExit("CloudNVR lock file is not root-owned")
else:
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
    flags |= getattr(os, "O_NOFOLLOW", 0)
    fd = os.open(path, flags, 0o600)
    os.close(fd)
if not allow_non_root:
    os.chmod(path, 0o600)
PY
        die "unsafe CloudNVR mutation lock: $LOCK_FILE"
        return 1
    }
    exec 9>>"$LOCK_FILE"
    flock -n 9 || { die "another CloudNVR mutation owns the lock"; return 1; }
}

validate_config_tree() {
    local directory="$1"
    local dragon="$directory/dragon.conf" connector="$directory/connector.conf"
    local accounts="$directory/users.htpasswd" auth cert key cert_ref key_ref https_enabled leaf
    local vs1 vs2 dragon_http dragon_https connector_http connector_host connector_db connector_tickets connector_logs connector_conf
    local dragon_http_raw dragon_https_raw connector_http_raw rpc_port_raw
    local cloud_enabled connector_serial connector_secret_file connector_api_token dragon_use_backend identity_kind rpc_bind rpc_port san_text
    [[ -s "$dragon" && -s "$connector" && -s "$accounts" ]] \
        || { die "config input requires dragon.conf, connector.conf, and users.htpasswd"; return 1; }
    grep -Eq '^[[:space:]]*\[global\][[:space:]]*$' "$dragon" \
        || { die "dragon.conf is missing [global]"; return 1; }
    grep -Eq '^[[:space:]]*\[server\][[:space:]]*$' "$dragon" \
        || { die "dragon.conf is missing [server]"; return 1; }
    grep -Eq '^[[:space:]]*\[global\][[:space:]]*$' "$connector" \
        || { die "connector.conf is missing [global]"; return 1; }
    grep -Eq '^[[:space:]]*\[dragon\][[:space:]]*$' "$connector" \
        || { die "connector.conf is missing [dragon]"; return 1; }
    auth="$(ini_value "$dragon" server require_auth 2>/dev/null || printf 0)"
    https_enabled="$(ini_value "$dragon" server https_enabled 2>/dev/null || printf 0)"
    cert_ref="$(ini_value "$dragon" server cert_file 2>/dev/null || printf /etc/ssl/certs/dragon.pem)"
    key_ref="$(ini_value "$dragon" server key_file 2>/dev/null || printf /etc/ssl/private/dragon.key)"
    [[ "$cert_ref" == /etc/ssl/certs/dragon.pem && "$key_ref" == /etc/ssl/private/dragon.key ]] \
        || { die "dragon.conf TLS paths do not match the read-only Compose mounts"; return 1; }
    if channel_is_production "$CHANNEL" && [[ "$auth" != 1 ]]; then
        die "production dragon.conf requires require_auth=1"
        return 1
    fi
    if channel_is_production "$CHANNEL" && [[ "$https_enabled" != 1 ]]; then
        die "production dragon.conf requires https_enabled=1"
        return 1
    fi
    cert="$directory/tls/dragon.pem"
    key="$directory/tls/dragon.key"
    [[ -s "$cert" && -s "$key" ]] || { die "TLS certificate and key are required"; return 1; }
    openssl x509 -in "$cert" -noout >/dev/null 2>&1 || { die "invalid TLS certificate"; return 1; }
    openssl pkey -in "$key" -noout >/dev/null 2>&1 || { die "invalid TLS private key"; return 1; }
    [[ "$(openssl x509 -in "$cert" -pubkey -noout | sha256_file /dev/stdin)" == \
       "$(openssl pkey -in "$key" -pubout 2>/dev/null | sha256_file /dev/stdin)" ]] \
        || { die "TLS certificate and key do not match"; return 1; }
    if channel_is_production "$CHANNEL"; then
        identity_kind="$(python3 - "$TLS_EXPECTED_NAME" <<'PY'
import ipaddress, re, sys
value = sys.argv[1]
try:
    ipaddress.ip_address(value)
except ValueError:
    if len(value) > 253 or value.endswith("."):
        raise SystemExit("invalid expected TLS DNS name")
    labels = value.split(".")
    if len(labels) < 2 or any(
        not label or len(label) > 63
        or not re.fullmatch(r"[A-Za-z0-9](?:[A-Za-z0-9-]*[A-Za-z0-9])?", label)
        for label in labels
    ):
        raise SystemExit("invalid expected TLS DNS name")
    print("dns")
else:
    print("ip")
PY
)" || { die "invalid production TLS identity: $TLS_EXPECTED_NAME"; return 1; }
        leaf="$(mktemp "${TMPDIR:-/tmp}/cloudnvr-leaf.XXXXXXXX.pem")" || return 1
        if ! openssl x509 -in "$cert" -out "$leaf" >/dev/null 2>&1; then
            rm -f "$leaf"
            die "invalid production TLS leaf certificate"
            return 1
        fi
        local -a identity_args trust_args
        if [[ "$identity_kind" == ip ]]; then
            identity_args=(-verify_ip "$TLS_EXPECTED_NAME")
        else
            identity_args=(-verify_hostname "$TLS_EXPECTED_NAME")
        fi
        if [[ -n "$TLS_CA_FILE" ]]; then
            trust_args=(-CAfile "$TLS_CA_FILE")
        else
            trust_args=(-CApath /etc/ssl/certs)
        fi
        san_text="$(openssl x509 -in "$leaf" -noout -ext subjectAltName 2>/dev/null)" \
            || { rm -f "$leaf"; die "production TLS certificate has no subject alternative name"; return 1; }
        python3 - "$TLS_EXPECTED_NAME" "$identity_kind" "$san_text" <<'PY' || {
import ipaddress, re, sys
expected, kind, text = sys.argv[1:]
if kind == "ip":
    wanted = ipaddress.ip_address(expected)
    values = re.findall(r"IP Address:([^,\s]+)", text)
    matched = any(ipaddress.ip_address(value) == wanted for value in values)
else:
    wanted = expected.lower()
    values = [value.lower().rstrip(".") for value in re.findall(r"DNS:([^,\s]+)", text)]
    def matches(pattern):
        if pattern == wanted:
            return True
        if not pattern.startswith("*."):
            return False
        return wanted.split(".")[1:] == pattern.split(".")[1:] and len(wanted.split(".")) == len(pattern.split("."))
    matched = any(matches(value) for value in values)
if not matched:
    raise SystemExit("expected identity is absent from the certificate SAN")
PY
            rm -f "$leaf"
            die "production TLS certificate SAN does not match $TLS_EXPECTED_NAME"
            return 1
        }
        if ! openssl verify -purpose sslserver "${identity_args[@]}" \
            "${trust_args[@]}" -untrusted "$cert" "$leaf" >/dev/null 2>&1; then
            rm -f "$leaf"
            die "production TLS certificate is not trusted for $TLS_EXPECTED_NAME"
            return 1
        fi
        rm -f "$leaf" || return 1
    fi
    validate_accounts "$accounts" || { die "invalid users.htpasswd"; return 1; }
    vs1="$(ini_value "$dragon" videostore vs1 2>/dev/null || printf /mnt/vs1)"
    vs2="$(ini_value "$dragon" videostore vs2 2>/dev/null || true)"
    [[ -z "$vs2" ]] || { die "Phase 4 supports one appliance video filesystem"; return 1; }
    [[ "$vs1" == /mnt/vs1 || "$ADOPT_EXISTING" == true ]] \
        || { die "dragon.conf [videostore] vs1 must be /mnt/vs1"; return 1; }
    dragon_http_raw="$(ini_value "$dragon" server http_port 2>/dev/null || printf 80)"
    dragon_https_raw="$(ini_value "$dragon" server https_port 2>/dev/null || printf 443)"
    dragon_http="$(canonical_port "$dragon_http_raw")" \
        || { die "invalid Dragon HTTP port: $dragon_http_raw"; return 1; }
    dragon_https="$(canonical_port "$dragon_https_raw")" \
        || { die "invalid Dragon HTTPS port: $dragon_https_raw"; return 1; }
    if [[ "$https_enabled" == 1 && "$dragon_http" == "$dragon_https" ]]; then
        die "Dragon HTTP and HTTPS ports must differ when HTTPS is enabled"
        return 1
    fi
    connector_http_raw="$(ini_value "$connector" dragon port 2>/dev/null || printf 80)"
    connector_http="$(canonical_port "$connector_http_raw")" \
        || { die "invalid connector Dragon port: $connector_http_raw"; return 1; }
    connector_host="$(ini_value "$connector" dragon host 2>/dev/null || printf 127.0.0.1)"
    connector_db="$(ini_value "$connector" global dbfile 2>/dev/null || printf /var/lib/connector/connector.db)"
    connector_tickets="$(ini_value "$connector" global ticketsdir 2>/dev/null || printf /videostore/vs1/dividia/tickets)"
    connector_logs="$(ini_value "$connector" global logdir 2>/dev/null || printf /var/log)"
    connector_conf="$(ini_value "$connector" dragon conf 2>/dev/null || printf /usr/local/etc/dragon.conf)"
    cloud_enabled="$(ini_value "$connector" cloud enabled 2>/dev/null || printf 0)"
    connector_serial="$(ini_value "$connector" global serial 2>/dev/null || true)"
    dragon_use_backend="$(ini_value "$dragon" global use_backend 2>/dev/null || printf 0)"
    cloud_enabled="$(canonical_bool "$cloud_enabled")" \
        || { die "invalid connector cloud enabled value"; return 1; }
    [[ "$dragon_use_backend" == 0 || "$dragon_use_backend" == 1 ]] \
        || { die "Dragon use_backend must be 0 or 1"; return 1; }
    rpc_bind="$(ini_value "$connector" rpc bind 2>/dev/null || printf 127.0.0.1)"
    rpc_port_raw="$(ini_value "$connector" rpc port 2>/dev/null || printf 43204)"
    rpc_port="$(canonical_port "$rpc_port_raw")" \
        || { die "invalid connector RPC port: $rpc_port_raw"; return 1; }
    [[ "$connector_host" == 127.0.0.1 || "$connector_host" == localhost ]] \
        || { die "connector must use the local Dragon service"; return 1; }
    python3 - "$rpc_bind" <<'PY' \
        || { die "connector [rpc] bind must be a loopback address"; return 1; }
import ipaddress, sys
value = sys.argv[1].strip().lower()
if value == "localhost":
    value = "127.0.0.1"
if not ipaddress.ip_address(value).is_loopback:
    raise SystemExit(1)
PY
    [[ "$rpc_port" == 43204 ]] \
        || { die "connector [rpc] port must be 43204 for legacy clients"; return 1; }
    [[ "$connector_http" == "$dragon_http" ]] \
        || { die "connector Dragon port does not match dragon.conf"; return 1; }
    [[ "$connector_db" == /var/lib/connector/connector.db ]] \
        || { die "connector dbfile does not match the persistent mount"; return 1; }
    [[ "$connector_tickets" == /videostore/vs1/* ]] \
        || { die "connector ticketsdir is outside the video mount"; return 1; }
    [[ "$connector_logs" == /var/log ]] \
        || { die "connector logdir does not match the persistent mount"; return 1; }
    [[ "$connector_conf" == /usr/local/etc/dragon.conf ]] \
        || { die "connector Dragon config path does not match the read-only mount"; return 1; }
    if [[ -n "$connector_serial" ]]; then
        if ! [[ "$connector_serial" =~ ^[0-9]{1,10}$ ]]; then
            die "connector serial must be a nonzero uint32 decimal value"
            return 1
        fi
        if (( 10#$connector_serial < 1 || 10#$connector_serial > 4294967295 )); then
            die "connector serial must be a nonzero uint32 decimal value"
            return 1
        fi
    fi
    if [[ "$cloud_enabled" == 1 && "$dragon_use_backend" != 1 && -z "$connector_serial" ]]; then
        die "cloud-enabled connector needs Dragon use_backend=1 or an explicit connector serial"
        return 1
    fi
    # The connector MINTS its cloud bearer token (7-day HS256 serial JWT) from
    # /etc/dividia/cloudapi/cloudapi.secret unless connector.conf carries a
    # static [cloud] api_token override.
    connector_api_token="$(ini_value "$connector" cloud api_token 2>/dev/null || true)"
    if [[ -e "$directory/connector/cloudapi.secret" ]]; then
        [[ -f "$directory/connector/cloudapi.secret" ]] \
            || { die "connector/cloudapi.secret must be a regular file"; return 1; }
        [[ ! -s "$directory/connector/cloudapi.secret" ]] \
            || [[ "$(wc -l < "$directory/connector/cloudapi.secret")" -le 1 ]] \
            || { die "connector/cloudapi.secret must be a single line"; return 1; }
    fi
    if [[ "$cloud_enabled" == 1 && -z "$connector_api_token" ]]; then
        [[ -s "$directory/connector/cloudapi.secret" ]] \
            || { die "cloud-enabled connector needs connector/cloudapi.secret in the config bundle (or a static [cloud] api_token)"; return 1; }
    fi
    if grep -Eq 'rtsp://\.\.\.|CHANGE_ME|placeholder' "$dragon" "$connector"; then
        die "placeholder configuration is not allowed"
        return 1
    fi
}

prepare_config_stage() {
    local destination="$1" supplied="" name
    mkdir -p "$destination/tls" "$destination/connector" || return 1
    if [[ -n "$CONFIG_DIR" ]]; then
        supplied="$(cd "$CONFIG_DIR" && pwd -P)" || return 1
    fi
    if marker_is_ours; then
        for name in dragon.conf connector.conf; do
            [[ ! -s "$ROOT/config/$name" ]] \
                || cp -p "$ROOT/config/$name" "$destination/$name" || return 1
        done
        [[ ! -s "$ROOT/data/dragon/users.htpasswd" ]] \
            || cp -p "$ROOT/data/dragon/users.htpasswd" "$destination/users.htpasswd" || return 1
        for name in dragon.pem dragon.key; do
            [[ ! -s "$ROOT/config/tls/$name" ]] \
                || cp -p "$ROOT/config/tls/$name" "$destination/tls/$name" || return 1
        done
        [[ ! -s "$ROOT/config/connector/cloudapi.secret" ]] \
            || cp -p "$ROOT/config/connector/cloudapi.secret" "$destination/connector/cloudapi.secret" || return 1
    elif [[ -z "$supplied" ]]; then
        die "use --config-dir with an approved configuration bundle"
        return 1
    fi
    if [[ -n "$supplied" ]]; then
        for name in dragon.conf connector.conf users.htpasswd; do
            [[ -s "$destination/$name" || ! -s "$supplied/$name" ]] \
                || cp -p "$supplied/$name" "$destination/$name" || return 1
        done
        for name in dragon.pem dragon.key; do
            [[ -s "$destination/tls/$name" || ! -s "$supplied/tls/$name" ]] \
                || cp -p "$supplied/tls/$name" "$destination/tls/$name" || return 1
        done
        [[ -s "$destination/connector/cloudapi.secret" || ! -s "$supplied/connector/cloudapi.secret" ]] \
            || cp -p "$supplied/connector/cloudapi.secret" "$destination/connector/cloudapi.secret" || return 1
    fi

    if [[ "$ADOPT_EXISTING" == true && -n "$LEGACY_DRAGON_CONF" ]]; then
        cp -p "$LEGACY_DRAGON_CONF" "$destination/dragon.conf" || return 1
    fi
    if [[ "$ADOPT_EXISTING" == true && -n "$LEGACY_CONNECTOR_CONF" ]]; then
        cp -p "$LEGACY_CONNECTOR_CONF" "$destination/connector.conf" || return 1
    fi
    if [[ "$GENERATE_SELF_SIGNED" == true ]]; then
        local cn ip4 san
        cn="$(hostname 2>/dev/null || printf cloudnvr)"
        ip4="$(ip -4 -o addr show scope global 2>/dev/null | awk '{split($4,a,"/"); print a[1]; exit}')"
        # Derived whenever self-signed is in effect, NOT only when the certificate is
        # generated: a rerun over an existing tls/dragon.pem must still know the
        # identity to verify against, or the production identity check fails empty.
        [[ -n "$TLS_EXPECTED_NAME" ]] || TLS_EXPECTED_NAME="${ip4:-$cn}"
        [[ -n "$TLS_CA_FILE" ]] || TLS_CA_FILE="$destination/tls/dragon.pem"
    fi
    # A DHCP lease can move under a unit between installs. The bootstrap certificate
    # is pinned to the address it was generated for, so a rerun would otherwise
    # validate the NEW identity against the OLD SAN and fail. Discard a stale one.
    if [[ "$GENERATE_SELF_SIGNED" == true && -s "$destination/tls/dragon.pem" ]]; then
        if ! openssl x509 -in "$destination/tls/dragon.pem" -noout -ext subjectAltName 2>/dev/null \
            | grep -qF "$TLS_EXPECTED_NAME"; then
            step "Bootstrap certificate no longer covers $TLS_EXPECTED_NAME; regenerating"
            rm -f "$destination/tls/dragon.pem" "$destination/tls/dragon.key"
        fi
    fi
    if [[ "$GENERATE_SELF_SIGNED" == true && ! -s "$destination/tls/dragon.pem" ]]; then
        step "Generating a self-signed TLS certificate"
        # A CN-only certificate is rejected outright by every modern client, so always
        # emit a SAN. Cover the hostname and the address the unit currently holds. On
        # DHCP that address can move, in which case regenerate rather than expecting
        # this to stay accurate (ADR-015 replaces this with an issued certificate).
        san="DNS:$cn"
        [[ -z "$ip4" ]] || san="$san,IP:$ip4"
        openssl req -x509 -newkey rsa:2048 -nodes -days 3650 \
            -subj "/CN=$cn" -addext "subjectAltName=$san" \
            -addext "basicConstraints=critical,CA:TRUE" \
            -keyout "$destination/tls/dragon.key" -out "$destination/tls/dragon.pem" >/dev/null 2>&1 \
            || return 1
    fi
    if [[ "$ADOPT_EXISTING" == true ]]; then
        normalize_adopted_videostore "$destination/dragon.conf" || return 1
    fi
    validate_config_tree "$destination" || return 1
}

normalize_adopted_videostore() {
    local file="$1"
    [[ "${CLOUDNVR_TEST_NORMALIZE_FAIL:-0}" != 1 ]] || return 1
    python3 - "$file" <<'PY' | atomic_write "$file.normalized" 0600
import re, sys
lines = open(sys.argv[1], encoding="utf-8").read().splitlines()
section = ""; found_section = False; replaced = False; out=[]
for raw in lines:
    m = re.match(r"^\s*\[([^]]+)\]\s*$", raw)
    if m:
        if section == "videostore" and not replaced: out.append("vs1=/mnt/vs1")
        section=m.group(1).strip().lower(); found_section |= section == "videostore"; replaced=False
        out.append(raw); continue
    if section == "videostore" and re.match(r"^\s*vs1\s*=", raw, re.I):
        out.append("vs1=/mnt/vs1"); replaced=True
    else: out.append(raw)
if section == "videostore" and not replaced: out.append("vs1=/mnt/vs1")
if not found_section: out.extend(["", "[videostore]", "vs1=/mnt/vs1"])
print("\n".join(out) + "\n", end="")
PY
    mv "$file.normalized" "$file" || return 1
}

validate_ports() {
    local config="$1" port raw_port listeners https_enabled ports
    [[ "$SKIP_PORT_CHECK" != 1 ]] || return 0
    command -v ss >/dev/null 2>&1 || { die "ss is required for port validation"; return 1; }
    listeners="$(ss -H -ltn 2>/dev/null)" || { die "cannot inspect listening TCP ports"; return 1; }
    ports="$(ini_value "$config" server http_port 2>/dev/null || printf 80)"
    https_enabled="$(ini_value "$config" server https_enabled 2>/dev/null || printf 0)"
    [[ "$https_enabled" == 0 || "$https_enabled" == 1 ]] \
        || { die "invalid configured https_enabled value: $https_enabled"; return 1; }
    if [[ "$https_enabled" == 1 ]]; then
        ports="$ports $(ini_value "$config" server https_port 2>/dev/null || printf 443)"
    fi
    for raw_port in $ports; do
        port="$(canonical_port "$raw_port")" \
            || { die "invalid configured server port: $raw_port"; return 1; }
        if grep -Fq ":$port " <<< "$listeners"; then
            if [[ "$ADOPT_EXISTING" != true ]] && ! marker_is_ours; then
                die "TCP port $port is already in use"
                return 1
            fi
        fi
    done
}

systemd_active_status() {
    local state
    state="$(systemctl show -p ActiveState --value "$1" 2>/dev/null)" || return 2
    case "$state" in
        active|activating|reloading|deactivating) return 0 ;;
        inactive|failed) return 1 ;;
        *) return 2 ;;
    esac
}

systemd_enabled_status() {
    local state
    state="$(systemctl show -p UnitFileState --value "$1" 2>/dev/null)" || return 2
    case "$state" in
        enabled|enabled-runtime|linked|linked-runtime|alias) return 0 ;;
        disabled|masked|masked-runtime|static|indirect|generated|transient) return 1 ;;
        *) return 2 ;;
    esac
}

systemd_inactive_verified() {
    local rc
    if systemd_active_status "$1"; then
        return 1
    else
        rc=$?
    fi
    [[ "$rc" -eq 1 ]]
}

systemd_disabled_verified() {
    local rc
    if systemd_enabled_status "$1"; then
        return 1
    else
        rc=$?
    fi
    [[ "$rc" -eq 1 ]]
}

read_legacy_unit_state() {
    local unit="$1" active_var="$2" enabled_var="$3" load_state rc
    local active=false enabled=false
    load_state="$(systemctl show -p LoadState --value "$unit" 2>/dev/null)" \
        || { die "cannot inspect legacy $unit load state"; return 1; }
    case "$load_state" in
        not-found)
            printf -v "$active_var" '%s' false
            printf -v "$enabled_var" '%s' false
            return 0
            ;;
        loaded|masked) ;;
        *) die "legacy $unit has unsafe load state: $load_state"; return 1 ;;
    esac
    if systemd_active_status "$unit"; then
        active=true
    else
        rc=$?
        [[ "$rc" -eq 1 ]] \
            || { die "cannot inspect legacy $unit active state"; return 1; }
    fi
    if systemd_enabled_status "$unit"; then
        enabled=true
    else
        rc=$?
        [[ "$rc" -eq 1 ]] \
            || { die "cannot inspect legacy $unit enablement"; return 1; }
    fi
    printf -v "$active_var" '%s' "$active"
    printf -v "$enabled_var" '%s' "$enabled"
}

inventory_legacy() {
    local pids pid exe expected conf env_conf manual_connector host_pids="" connector_args connector_env dragon_cwd unit
    local connector_cwd="" connector_db_base="" legacy_uid
    command -v pgrep >/dev/null 2>&1 \
        || { die "pgrep is required for legacy process inventory"; return 1; }
    [[ -d "$PROC_ROOT" ]] \
        || { die "process inventory root is unavailable: $PROC_ROOT"; return 1; }
    pids="$(pgrep -x dragon 2>/dev/null || true)"
    for pid in $pids; do
        if grep -Eq '(^|/)(docker([/-]|$)|docker-[^.]+\.scope|containerd([/-]|$)|kubepods([/-]|$))' \
            "$PROC_ROOT/$pid/cgroup" 2>/dev/null; then
            continue
        fi
        host_pids+="${host_pids:+ }$pid"
    done
    pids="$host_pids"
    if [[ "$ADOPT_EXISTING" != true ]]; then
        read_legacy_unit_state dragon.service LEGACY_DRAGON_SERVICE_ACTIVE \
            LEGACY_DRAGON_SERVICE_ENABLED || return 1
        read_legacy_unit_state connector.service LEGACY_CONNECTOR_SERVICE_ACTIVE \
            LEGACY_CONNECTOR_SERVICE_ENABLED || return 1
        if [[ "$LEGACY_DRAGON_SERVICE_ACTIVE" == true \
           || "$LEGACY_DRAGON_SERVICE_ENABLED" == true ]]; then
            die "dragon.service is active or enabled; use --adopt-existing to replace a legacy installation"
            return 1
        fi
        if [[ "$LEGACY_CONNECTOR_SERVICE_ACTIVE" == true \
           || "$LEGACY_CONNECTOR_SERVICE_ENABLED" == true ]]; then
            die "connector.service is active or enabled; use --adopt-existing to replace a legacy installation"
            return 1
        fi
        [[ -z "$pids" ]] \
            || { die "a manual Dragon process is running; use --adopt-existing"; return 1; }
        manual_connector="$(python3 - "$PROC_ROOT" <<'PY' 2>/dev/null || true
import os, sys
from pathlib import Path
root = Path(sys.argv[1])
for proc in root.glob("[0-9]*"):
    try:
        cgroup = (proc / "cgroup").read_text(errors="replace")
    except (FileNotFoundError, PermissionError, ProcessLookupError):
        cgroup = ""
    if any(token in cgroup for token in ("/docker/", "docker-", "/containerd/", "/kubepods/")):
        continue
    try:
        args = [os.fsdecode(v) for v in (proc / "cmdline").read_bytes().split(b"\0") if v]
        cwd = os.readlink(proc / "cwd")
    except (FileNotFoundError, PermissionError, ProcessLookupError):
        continue
    joined = " ".join(args)
    if any(Path(arg).name == "main.py" for arg in args) and ("connector" in cwd.lower() or "connector" in joined.lower()):
        print(proc.name + " " + joined)
        break
PY
)"
        [[ -z "$manual_connector" ]] \
            || { die "a manual connector process is running; use --adopt-existing"; return 1; }
        return 0
    fi
    step "Inventorying the manual installation"
    if [[ -n "$LEGACY_DRAGON_DIR" ]]; then
        LEGACY_DRAGON_DIR="$(cd "$LEGACY_DRAGON_DIR" && pwd -P)" || return 1
        expected="$LEGACY_DRAGON_DIR/dragon"
        [[ -x "$expected" ]] || { die "missing legacy Dragon binary: $expected"; return 1; }
    fi
    [[ "$(wc -w <<< "$pids")" -le 1 ]] || { die "more than one Dragon process is running"; return 1; }
    pid="${pids%%[[:space:]]*}"
    if [[ -n "$pid" ]]; then
        exe="$(readlink "$PROC_ROOT/$pid/exe" 2>/dev/null || true)"
        [[ -n "$exe" && "${exe##*/}" == dragon ]] \
            || { die "PID $pid is not a recognized Dragon executable"; return 1; }
        if [[ "${CLOUDNVR_ALLOW_NON_ROOT:-0}" != 1 ]]; then
            legacy_uid="$(awk '/^Uid:/ { print $2 ":" $3 ":" $4 ":" $5 }' \
                "$PROC_ROOT/$pid/status" 2>/dev/null || true)"
            [[ "$legacy_uid" == 0:0:0:0 ]] \
                || { die "manual Dragon PID $pid does not run with root credentials"; return 1; }
            python3 - "$exe" <<'PY' || {
import os, pathlib, stat, sys
path = pathlib.Path(sys.argv[1]).resolve(strict=True)
for item_path in (path, *path.parents):
    item = os.lstat(item_path)
    expected = stat.S_ISREG(item.st_mode) if item_path == path else stat.S_ISDIR(item.st_mode)
    if not expected or stat.S_ISLNK(item.st_mode) or item.st_uid != 0 or stat.S_IMODE(item.st_mode) & 0o022:
        raise SystemExit("manual Dragon executable path must be root-owned and not writable by group or other")
PY
                die "unsafe manual Dragon executable: $exe"
                return 1
            }
        fi
        if [[ -n "${expected:-}" && "$exe" != "$expected" ]]; then
            die "PID $pid runs $exe, outside --legacy-dragon-dir"
            return 1
        fi
        LEGACY_DRAGON_PID="$pid"
        LEGACY_DRAGON_EXE="$exe"
        LEGACY_DRAGON_STARTTIME="$(python3 - "$PROC_ROOT/$pid/stat" <<'PY' 2>/dev/null || true
import sys
text=open(sys.argv[1], encoding="utf-8").read()
tail=text[text.rfind(")") + 2:].split()
print(tail[19] if len(tail) > 19 else "")
PY
)"
        [[ "$LEGACY_DRAGON_STARTTIME" =~ ^[0-9]+$ ]] \
            || { die "cannot record the identity of Dragon PID $pid"; return 1; }
        LEGACY_DRAGON_DIR="${exe%/*}"
        env_conf="$(tr '\0' '\n' < "$PROC_ROOT/$pid/environ" 2>/dev/null | sed -n 's/^DRAGON_CONF=//p' | tail -1)"
        if [[ -n "$env_conf" && "$env_conf" != /* ]]; then
            dragon_cwd="$(readlink "$PROC_ROOT/$pid/cwd" 2>/dev/null || true)"
            [[ -n "$dragon_cwd" ]] \
                || { die "cannot resolve relative DRAGON_CONF for PID $pid"; return 1; }
            env_conf="$dragon_cwd/$env_conf"
        fi
        if [[ -n "$env_conf" && -s "$env_conf" ]]; then conf="$env_conf"; else conf="$LEGACY_DRAGON_DIR/dragon.conf"; fi
        [[ -s "$conf" ]] || { die "cannot find the recognized Dragon configuration for PID $pid"; return 1; }
        LEGACY_DRAGON_CONF="$conf"
    elif [[ -n "$LEGACY_DRAGON_DIR" ]]; then
        LEGACY_DRAGON_EXE="$LEGACY_DRAGON_DIR/dragon"
        LEGACY_DRAGON_CONF="$LEGACY_DRAGON_DIR/dragon.conf"
        [[ -s "$LEGACY_DRAGON_CONF" ]] || { die "missing $LEGACY_DRAGON_CONF"; return 1; }
    else
        die "--adopt-existing needs a running recognized Dragon or --legacy-dragon-dir"
        return 1
    fi
    VIDEO_SOURCE="${VIDEO_SOURCE:-$(ini_value "$LEGACY_DRAGON_CONF" videostore vs1 2>/dev/null || printf /mnt/vs1)}"
    [[ -d "$VIDEO_SOURCE" ]] || { die "legacy video path is not a directory: $VIDEO_SOURCE"; return 1; }
    read_legacy_unit_state dragon.service LEGACY_DRAGON_SERVICE_ACTIVE \
        LEGACY_DRAGON_SERVICE_ENABLED || return 1
    if [[ "$LEGACY_DRAGON_SERVICE_ACTIVE" == true \
       && "$LEGACY_DRAGON_SERVICE_ENABLED" != true ]]; then
        die "active dragon.service is disabled and has no reboot owner"
        return 1
    fi
    if [[ "$LEGACY_DRAGON_SERVICE_ACTIVE" == true || "$LEGACY_DRAGON_SERVICE_ENABLED" == true ]]; then
        local exec_start
        exec_start="$(systemctl show -p ExecStart --value dragon.service 2>/dev/null)" \
            || { die "cannot inspect dragon.service command"; return 1; }
        [[ "$exec_start" == *"$LEGACY_DRAGON_EXE"* ]] \
            || { die "active dragon.service does not run $LEGACY_DRAGON_EXE"; return 1; }
        LEGACY_DRAGON_UNIT_HASH="$(systemctl cat dragon.service 2>/dev/null | sha256_file /dev/stdin)" \
            || { die "cannot record the identity of dragon.service"; return 1; }
    fi
    read_legacy_unit_state connector.service LEGACY_CONNECTOR_SERVICE_ACTIVE \
        LEGACY_CONNECTOR_SERVICE_ENABLED || return 1
    if [[ "$LEGACY_CONNECTOR_SERVICE_ACTIVE" == true \
       && "$LEGACY_CONNECTOR_SERVICE_ENABLED" != true ]]; then
        die "active connector.service is disabled and has no reboot owner"
        return 1
    fi
    if [[ "$LEGACY_CONNECTOR_SERVICE_ACTIVE" == true || "$LEGACY_CONNECTOR_SERVICE_ENABLED" == true ]]; then
        local connector_exec
        connector_exec="$(systemctl show -p ExecStart --value connector.service 2>/dev/null)" \
            || { die "cannot inspect connector.service command"; return 1; }
        [[ "$connector_exec" == *python* && "$connector_exec" == *main.py* ]] \
            || { die "active connector.service is not the recognized connector"; return 1; }
        LEGACY_CONNECTOR_UNIT_HASH="$(systemctl cat connector.service 2>/dev/null | sha256_file /dev/stdin)" \
            || { die "cannot record the identity of connector.service"; return 1; }
        if [[ "$LEGACY_CONNECTOR_SERVICE_ACTIVE" == true ]]; then
            LEGACY_CONNECTOR_PID="$(systemctl show -p MainPID --value connector.service 2>/dev/null)" \
                || { die "cannot inspect connector.service main PID"; return 1; }
            [[ "$LEGACY_CONNECTOR_PID" =~ ^[1-9][0-9]*$ ]] \
                || { die "active connector.service has no recognized main PID"; return 1; }
        fi
    fi
    LEGACY_CONNECTOR_CONF="${CLOUDNVR_LEGACY_CONNECTOR_CONF:-}"
    if [[ -z "$LEGACY_CONNECTOR_CONF" && -n "$LEGACY_CONNECTOR_PID" ]]; then
        connector_cwd="$(readlink "$PROC_ROOT/$LEGACY_CONNECTOR_PID/cwd" 2>/dev/null || true)"
        connector_env="$(tr '\0' '\n' < "$PROC_ROOT/$LEGACY_CONNECTOR_PID/environ" 2>/dev/null \
            | sed -n 's/^CONNECTOR_CONF=//p' | tail -1)"
        connector_args="$(tr '\0' '\n' < "$PROC_ROOT/$LEGACY_CONNECTOR_PID/cmdline" 2>/dev/null || true)"
        if [[ -n "$connector_env" ]]; then
            LEGACY_CONNECTOR_CONF="$connector_env"
        else
            LEGACY_CONNECTOR_CONF="$(python3 - "$connector_args" <<'PY' 2>/dev/null || true
import sys
args=sys.argv[1].splitlines()
for index, arg in enumerate(args):
    if arg.startswith("--conf="):
        print(arg.split("=", 1)[1]); break
    if arg == "--conf" and index + 1 < len(args):
        print(args[index + 1]); break
PY
)"
        fi
    fi
    if [[ -n "$LEGACY_CONNECTOR_CONF" && "$LEGACY_CONNECTOR_CONF" != /* ]]; then
        if [[ -n "$LEGACY_CONNECTOR_PID" && -z "$connector_cwd" \
           && -z "${CLOUDNVR_LEGACY_CONNECTOR_CONF:-}" ]]; then
            die "cannot resolve relative connector config for PID $LEGACY_CONNECTOR_PID"
            return 1
        fi
        LEGACY_CONNECTOR_CONF="${connector_cwd:-$PWD}/$LEGACY_CONNECTOR_CONF"
    fi
    if [[ -z "$LEGACY_CONNECTOR_CONF" && -s /usr/local/etc/connector.conf ]]; then
        LEGACY_CONNECTOR_CONF=/usr/local/etc/connector.conf
    fi
    [[ -z "$LEGACY_CONNECTOR_CONF" || -s "$LEGACY_CONNECTOR_CONF" ]] \
        || { die "legacy connector configuration is unreadable: $LEGACY_CONNECTOR_CONF"; return 1; }
    LEGACY_CONNECTOR_DB="${CLOUDNVR_LEGACY_CONNECTOR_DB:-}"
    if [[ -z "$LEGACY_CONNECTOR_DB" && -n "$LEGACY_CONNECTOR_CONF" ]]; then
        LEGACY_CONNECTOR_DB="$(ini_value "$LEGACY_CONNECTOR_CONF" global dbfile 2>/dev/null \
            || printf /var/lib/connector/connector.db)"
    fi
    if [[ -n "$LEGACY_CONNECTOR_DB" && "$LEGACY_CONNECTOR_DB" != /* ]]; then
        connector_db_base="${connector_cwd:-${LEGACY_CONNECTOR_CONF%/*}}"
        LEGACY_CONNECTOR_DB="$connector_db_base/$LEGACY_CONNECTOR_DB"
    fi
    [[ -z "$LEGACY_CONNECTOR_DB" || -s "$LEGACY_CONNECTOR_DB" ]] \
        || { die "legacy connector database is unreadable: $LEGACY_CONNECTOR_DB"; return 1; }
    if [[ "$LEGACY_CONNECTOR_SERVICE_ACTIVE" == true || "$LEGACY_CONNECTOR_SERVICE_ENABLED" == true ]]; then
        [[ -n "$LEGACY_CONNECTOR_CONF" && -n "$LEGACY_CONNECTOR_DB" ]] \
            || { die "recognized connector state paths could not be resolved"; return 1; }
    fi
    manual_connector="$(python3 - "$PROC_ROOT" "$LEGACY_CONNECTOR_PID" <<'PY' 2>/dev/null || true
import os, sys
from pathlib import Path
root, known_pid = Path(sys.argv[1]), sys.argv[2]
for proc in root.glob("[0-9]*"):
    if proc.name == known_pid:
        continue
    try:
        cgroup=(proc/"cgroup").read_text(errors="replace")
    except (FileNotFoundError, PermissionError, ProcessLookupError):
        cgroup=""
    if any(token in cgroup for token in ("/docker/", "docker-", "/containerd/", "/kubepods/")):
        continue
    try:
        args=[os.fsdecode(v) for v in (proc/"cmdline").read_bytes().split(b"\0") if v]
        cwd=os.readlink(proc/"cwd")
    except (FileNotFoundError, PermissionError, ProcessLookupError):
        continue
    joined=" ".join(args)
    if any(Path(arg).name == "main.py" for arg in args) and ("connector" in cwd.lower() or "connector" in joined.lower()):
        print(proc.name + " " + joined)
        break
PY
)"
    if [[ -n "$manual_connector" ]]; then
        die "refusing to stop an unknown connector process: $manual_connector"
        return 1
    fi
    log "Recognized Dragon: ${LEGACY_DRAGON_EXE} (PID ${LEGACY_DRAGON_PID:-stopped})"
    log "Legacy config: $LEGACY_DRAGON_CONF"
    log "Legacy video path: $VIDEO_SOURCE"
}

create_layout() {
    local -a layout=(
        "$ROOT" "$ROOT/bin" "$ROOT/config" "$ROOT/data"
        "$ROOT/data/video" "$ROOT/data/cache/dragon" "$ROOT/data/dragon"
        "$ROOT/data/connector" "$ROOT/data/logs/connector" "$ROOT/data/logs"
        "$ROOT/releases" "$ROOT/state" "$ROOT/adoption-backups"
    )
    if [[ "${CLOUDNVR_ALLOW_NON_ROOT:-0}" == 1 ]]; then
        install -d -m 0750 "${layout[@]}" || return 1
        install -d -m 0700 "$ROOT/config/tls" "$ROOT/config/connector" || return 1
    else
        install -o root -g root -d -m 0750 "${layout[@]}" || return 1
        install -o root -g root -d -m 0700 "$ROOT/config/tls" "$ROOT/config/connector" || return 1
    fi
    chmod 0700 "$ROOT/releases" "$ROOT/state" || return 1
    durable_sync_dirs "${layout[@]}" "$ROOT/config/tls" "$ROOT/config/connector" || return 1
    if [[ ! -e "$ROOT/data/logs/update.log" ]]; then
        printf '' | atomic_write "$ROOT/data/logs/update.log" 0640 || return 1
    fi
}

write_marker() {
    [[ -f "$MARKER_FILE" ]] && return 0
    {
        printf 'product=cloudnvr\n'
        printf 'schema=1\n'
        printf 'flow=%s\n' "$([[ "$ADOPT_EXISTING" == true ]] && printf adoption || printf clean)"
        printf 'installed=%s\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
    } | atomic_write "$MARKER_FILE" 0644
}

write_host_env() {
    if [[ -f "$ENV_FILE" ]]; then
        local existing_channel
        existing_channel="$(kv_get "$ENV_FILE" CHANNEL 2>/dev/null || true)"
        [[ -z "$existing_channel" || "$existing_channel" == "$CHANNEL" ]] \
            || { die "rerun channel $CHANNEL differs from installed channel $existing_channel; use 'nvr channel'"; return 1; }
    fi
    {
        printf 'PRODUCT=cloudnvr\n'
        printf 'CHANNEL=%s\n' "$CHANNEL"
        printf 'PLATFORM=%s\n' "$PLATFORM"
        printf 'REGISTRY=%s\n' "$REGISTRY"
        printf 'COMPOSE_PROJECT_NAME=%s\n' "$COMPOSE_PROJECT_NAME"
        printf 'BACKEND_ADDR=%s\n' "$BACKEND_ADDR"
        printf 'BUNDLE_BASE_URL=%s\n' "$BASE_URL"
        printf 'VIDEO_SOURCE=%s\n' "$VIDEO_SOURCE"
        printf 'ALLOW_ROOT_VIDEO=%s\n' "$([[ "$ALLOW_ROOT_VIDEO" == true ]] && printf 1 || printf 0)"
    } | atomic_write "$ENV_FILE" 0600
}

install_config() {
    local stage="$1" name
    for name in dragon.conf connector.conf; do
        if [[ ! -e "$ROOT/config/$name" ]]; then
            atomic_copy "$stage/$name" "$ROOT/config/$name" 0600 || return 1
        fi
    done
    for name in dragon.pem dragon.key; do
        if [[ ! -e "$ROOT/config/tls/$name" ]]; then
            atomic_copy "$stage/tls/$name" "$ROOT/config/tls/$name" \
                "$([[ "$name" == dragon.key ]] && printf 0600 || printf 0644)" || return 1
        fi
    done
    if [[ ! -e "$ROOT/data/dragon/users.htpasswd" ]]; then
        atomic_copy "$stage/users.htpasswd" "$ROOT/data/dragon/users.htpasswd" 0600 || return 1
    fi
    if [[ -s "$stage/connector/cloudapi.secret" ]]; then
        if [[ ! -e "$ROOT/config/connector/cloudapi.secret" ]]; then
            atomic_copy "$stage/connector/cloudapi.secret" "$ROOT/config/connector/cloudapi.secret" 0600 || return 1
        fi
        # Real installs run as root; bench/contract tests may lack /etc write access.
        if mkdir -p /etc/dividia/cloudapi 2>/dev/null; then
            chmod 0700 /etc/dividia/cloudapi || return 1
            if [[ ! -e /etc/dividia/cloudapi/cloudapi.secret ]]; then
                atomic_copy "$stage/connector/cloudapi.secret" /etc/dividia/cloudapi/cloudapi.secret 0600 || return 1
            fi
            if [[ ! -e /etc/dividia/cloudapi/environment ]]; then
                case "${BACKEND_ADDR:-}" in
                    *staging*) cloud_env=staging ;;
                    *) cloud_env=prod ;;
                esac
                printf '%s\n' "$cloud_env" | atomic_write /etc/dividia/cloudapi/environment 0644 || return 1
            fi
        fi
    fi
    if [[ ! -e "$ROOT/data/connector/connector.db" ]]; then
        if [[ "$ADOPT_EXISTING" != true ]]; then
            printf '' | atomic_write "$ROOT/data/connector/connector.db" 0600 || return 1
        fi
    fi
    return 0
}

backup_legacy() {
    [[ "$ADOPT_EXISTING" == true ]] || return 0
    local stamp
    stamp="$(date -u +%Y%m%dT%H%M%SZ)"
    BACKUP_DIR="$(mktemp -d "$ROOT/adoption-backups/$stamp.XXXXXXXX")" || return 1
    chmod 0700 "$BACKUP_DIR" || return 1
    durable_sync_dirs "$BACKUP_DIR" || return 1
    [[ "${CLOUDNVR_TEST_INSTALLER_COPY_FAIL:-}" != legacy-dragon ]] || return 1
    cp -p "$LEGACY_DRAGON_EXE" "$BACKUP_DIR/dragon" || return 1
    cp -p "$LEGACY_DRAGON_CONF" "$BACKUP_DIR/dragon.conf" || return 1
    if [[ -n "$LEGACY_CONNECTOR_CONF" ]]; then
        cp -p "$LEGACY_CONNECTOR_CONF" "$BACKUP_DIR/connector.conf" || return 1
    fi
    {
        printf 'dragon_executable=%s\n' "$LEGACY_DRAGON_EXE"
        printf 'dragon_pid=%s\n' "$LEGACY_DRAGON_PID"
        printf 'dragon_config=%s\n' "$LEGACY_DRAGON_CONF"
        printf 'video_source=%s\n' "$VIDEO_SOURCE"
        printf 'dragon_service_active=%s\n' "$LEGACY_DRAGON_SERVICE_ACTIVE"
        printf 'dragon_service_enabled=%s\n' "$LEGACY_DRAGON_SERVICE_ENABLED"
        printf 'connector_service_active=%s\n' "$LEGACY_CONNECTOR_SERVICE_ACTIVE"
        printf 'connector_service_enabled=%s\n' "$LEGACY_CONNECTOR_SERVICE_ENABLED"
    } | atomic_write "$BACKUP_DIR/inventory" 0600 || return 1
    if [[ -n "$LEGACY_DRAGON_PID" && "$LEGACY_DRAGON_SERVICE_ACTIVE" != true ]]; then
        LEGACY_MANUAL_STATE="$BACKUP_DIR/manual-dragon.json"
        if ! python3 - "$PROC_ROOT" "$LEGACY_DRAGON_PID" "$LEGACY_DRAGON_EXE" \
            "$LEGACY_MANUAL_STATE" <<'PY'
import json, os, pathlib, sys
proc_root, pid, executable, output = sys.argv[1:]
base = pathlib.Path(proc_root) / pid
args = (base / "cmdline").read_bytes().split(b"\0")
args = [os.fsdecode(item) for item in args if item]
env = {}
for item in (base / "environ").read_bytes().split(b"\0"):
    if b"=" in item:
        key, value = item.split(b"=", 1); env[os.fsdecode(key)] = os.fsdecode(value)
value = {"args": args, "cwd": os.readlink(base / "cwd"), "env": env,
         "executable": executable}
with open(output, "w", encoding="utf-8") as handle: json.dump(value, handle)
os.chmod(output, 0o600)
PY
        then
            return 1
        fi
    fi
    log "Legacy backup: $BACKUP_DIR"
}

record_legacy_unit_identities() {
    local directory="$ROOT/state/legacy-units"
    [[ "$ADOPT_EXISTING" == true ]] || return 0
    mkdir -p "$directory" || return 1
    chmod 0700 "$directory" || return 1
    durable_sync_dirs "$directory" || return 1
    if [[ -n "$LEGACY_DRAGON_UNIT_HASH" ]]; then
        printf '%s\n' "$LEGACY_DRAGON_UNIT_HASH" \
            | atomic_write "$directory/dragon.service.sha256" 0600 || return 1
    fi
    if [[ -n "$LEGACY_CONNECTOR_UNIT_HASH" ]]; then
        printf '%s\n' "$LEGACY_CONNECTOR_UNIT_HASH" \
            | atomic_write "$directory/connector.service.sha256" 0600 || return 1
    fi
}

configure_video_storage() {
    local target="$ROOT/data/video" line source_identity target_identity filtered
    if [[ -z "$VIDEO_SOURCE" ]]; then
        if [[ "$ALLOW_ROOT_VIDEO" == true || "${CLOUDNVR_ALLOW_ROOT_VIDEO:-0}" == 1 ]]; then
            warn "Using the installation filesystem for video. This is for bench use only."
            return 0
        fi
        if mountpoint -q "$target" 2>/dev/null; then return 0; fi
        die "use --video-source for the appliance video filesystem, or --allow-root-video for a bench"
        return 1
    fi
    VIDEO_SOURCE="$(cd "$VIDEO_SOURCE" && pwd -P)" || return 1
    [[ "$VIDEO_SOURCE" != "$target" ]] || return 0
    if mountpoint -q "$target" 2>/dev/null; then
        source_identity="$(stat -Lc '%d:%i' "$VIDEO_SOURCE" 2>/dev/null)" \
            || { die "cannot identify video source directory: $VIDEO_SOURCE"; return 1; }
        target_identity="$(stat -Lc '%d:%i' "$target" 2>/dev/null)" \
            || { die "cannot identify video target directory: $target"; return 1; }
        [[ "$source_identity" == "$target_identity" ]] \
            || { die "$target is not bound from $VIDEO_SOURCE"; return 1; }
    else
        [[ -z "$(find "$target" -mindepth 1 -maxdepth 1 -print -quit 2>/dev/null)" ]] \
            || { die "$target is non-empty; refusing to hide it with a mount"; return 1; }
        if [[ "${CLOUDNVR_SKIP_MOUNTS:-0}" == 1 ]]; then return 0; fi
        mount --bind "$VIDEO_SOURCE" "$target" \
            || { die "cannot bind $VIDEO_SOURCE at $target"; return 1; }
        VIDEO_BIND_ADDED=true
    fi
    line="$VIDEO_SOURCE $target none bind 0 0"
    if [[ -e "$FSTAB_FILE" || -L "$FSTAB_FILE" ]]; then
        FSTAB_WRITE_FILE="$(python3 -c 'import os,sys; print(os.path.realpath(sys.argv[1]))' "$FSTAB_FILE")" \
            || return 1
        [[ -f "$FSTAB_WRITE_FILE" && ! -L "$FSTAB_WRITE_FILE" ]] \
            || { die "fstab target is not a regular file: $FSTAB_FILE"; return 1; }
    else
        FSTAB_WRITE_FILE="$FSTAB_FILE"
    fi
    filtered="$FSTAB_BACKUP_FILE.desired"
    if [[ -f "$FSTAB_WRITE_FILE" ]]; then
        [[ "${CLOUDNVR_TEST_FSTAB_FILTER_FAIL:-0}" != 1 ]] || return 1
        awk -v target="$target" '$2 != target {print}' "$FSTAB_WRITE_FILE" > "$filtered" \
            || return 1
        FSTAB_EXISTED=true
    else
        : > "$filtered" || return 1
        FSTAB_EXISTED=false
    fi
    printf '%s\n' "$line" >> "$filtered" || return 1
    if [[ "$FSTAB_EXISTED" == true ]] && cmp -s "$filtered" "$FSTAB_WRITE_FILE"; then
        return 0
    fi
    if [[ "$FSTAB_EXISTED" == true ]]; then
        cp -a "$FSTAB_WRITE_FILE" "$FSTAB_BACKUP_FILE" || return 1
    fi
    FSTAB_CHANGED=true
    if [[ "$FSTAB_EXISTED" == true ]]; then
        atomic_replace_file "$FSTAB_WRITE_FILE" "$filtered" "$FSTAB_BACKUP_FILE" 0 || return 1
    else
        atomic_write "$FSTAB_WRITE_FILE" 0644 < "$filtered" || return 1
    fi
    [[ "${CLOUDNVR_TEST_FSTAB_POST_WRITE_FAIL:-0}" != 1 ]] || return 1
}

cleanup_failed_candidate() {
    local directory="$ROOT/releases/$RELEASE_ID"
    [[ -f "$directory/release.env" && -f "$directory/docker-compose.yml" ]] || return 0
    docker compose --project-directory "$ROOT" --env-file "$ROOT/.env" \
        --env-file "$directory/release.env" \
        -f "$directory/docker-compose.yml" down >/dev/null 2>&1
}

preflight_candidate() {
    step "Pulling and validating the complete release before activation"
    CLOUDNVR_ROOT="$ROOT" CLOUDNVR_INSTALLING=1 CLOUDNVR_INHERITED_LOCK=1 \
        CLOUDNVR_CANDIDATE_MANIFEST="$BUNDLE_DIR/release.json" \
        CLOUDNVR_PREFLIGHT_ONLY=1 CLOUDNVR_SKIP_HOST_ARTIFACTS=1 \
        bash "$BUNDLE_DIR/nvr" update --force
}

install_boot_owner() {
    local committed_before="$1" current_target="$2" source="$BUNDLE_DIR/cloudnvr.service"
    [[ "${CLOUDNVR_SKIP_HOST_ARTIFACTS:-0}" != 1 ]] || return 0
    step "Installing the reboot-safe CloudNVR boot owner"
    if [[ "$committed_before" == true ]]; then
        source="$ROOT/releases/$current_target/cloudnvr.service"
    fi
    atomic_copy "$source" "$SERVICE_FILE" 0644 \
        || return 1
    systemctl daemon-reload || return 1
    systemctl enable cloudnvr.service >/dev/null || return 1
    systemctl is-enabled --quiet cloudnvr.service \
        || { die "cloudnvr.service did not become enabled"; return 1; }
}

remove_manual_recovery_owner() {
    local rc=0
    [[ "$MANUAL_RECOVERY_INSTALLED" == true || -e "$MANUAL_RECOVERY_UNIT" \
       || -e "$MANUAL_RECOVERY_SCRIPT" || -e "$MANUAL_RECOVERY_STATE" ]] || return 0
    if [[ "${CLOUDNVR_SKIP_HOST_ARTIFACTS:-0}" != 1 ]]; then
        systemctl disable "$MANUAL_RECOVERY_UNIT_NAME" >/dev/null 2>&1 || rc=1
    fi
    rm -f -- "$MANUAL_RECOVERY_UNIT" "$MANUAL_RECOVERY_SCRIPT" "$MANUAL_RECOVERY_STATE" || rc=1
    if [[ "${CLOUDNVR_SKIP_HOST_ARTIFACTS:-0}" != 1 ]]; then
        systemctl daemon-reload || rc=1
    fi
    MANUAL_RECOVERY_INSTALLED=false
    return "$rc"
}

install_manual_recovery_owner() {
    [[ "$ADOPT_EXISTING" == true && -n "$LEGACY_MANUAL_STATE" ]] || return 0
    [[ "$LEGACY_DRAGON_SERVICE_ENABLED" != true ]] || return 0
    [[ "${CLOUDNVR_SKIP_HOST_ARTIFACTS:-0}" != 1 ]] || return 0
    step "Installing temporary reboot recovery for the manual Dragon process"
    atomic_copy "$LEGACY_MANUAL_STATE" "$MANUAL_RECOVERY_STATE" 0600 || return 1
    cat <<'PY' | atomic_write "$MANUAL_RECOVERY_SCRIPT" 0755 || return 1
#!/usr/bin/python3
import json
import os
import sys

state_path, log_path = sys.argv[1:]
with open(state_path, encoding="utf-8") as handle:
    state = json.load(handle)
args = state.get("args")
cwd = state.get("cwd")
env = state.get("env")
executable = state.get("executable")
if not isinstance(args, list) or not args or not all(isinstance(value, str) for value in args):
    raise SystemExit("invalid manual Dragon arguments")
if not isinstance(cwd, str) or not isinstance(env, dict) or not isinstance(executable, str):
    raise SystemExit("invalid manual Dragon process state")
if not os.path.isabs(executable):
    raise SystemExit("manual Dragon executable is not absolute")
os.chdir(cwd)
log_fd = os.open(log_path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)
os.dup2(log_fd, 1)
os.dup2(log_fd, 2)
if log_fd > 2:
    os.close(log_fd)
os.execve(executable, args, {str(key): str(value) for key, value in env.items()})
PY
    {
        printf '%s\n' \
            '[Unit]' \
            'Description=Temporary recovery for an adopted manual Dragon process' \
            'After=network-online.target' \
            'Wants=network-online.target' \
            "ConditionPathExists=!$ROOT/releases/current" \
            '' \
            '[Service]' \
            'Type=simple'
        printf 'ExecStart=/usr/bin/python3 "%s" "%s" "%s"\n' \
            "$MANUAL_RECOVERY_SCRIPT" "$MANUAL_RECOVERY_STATE" "$MANUAL_RECOVERY_LOG"
        printf '%s\n' \
            'Restart=on-failure' \
            'RestartSec=5' \
            '' \
            '[Install]' \
            'WantedBy=multi-user.target'
    } | atomic_write "$MANUAL_RECOVERY_UNIT" 0644 || return 1
    MANUAL_RECOVERY_INSTALLED=true
    if ! systemctl daemon-reload \
        || ! systemctl enable "$MANUAL_RECOVERY_UNIT_NAME" >/dev/null \
        || ! systemctl is-enabled --quiet "$MANUAL_RECOVERY_UNIT_NAME"; then
        remove_manual_recovery_owner || true
        die "temporary manual Dragon recovery did not become enabled"
        return 1
    fi
}

process_starttime() {
    python3 - "$1" <<'PY'
import sys
text=open(sys.argv[1], encoding="utf-8").read()
tail=text[text.rfind(")") + 2:].split()
print(tail[19] if len(tail) > 19 else "")
PY
}

manual_dragon_identity_matches() {
    local exe starttime
    [[ "${CLOUDNVR_TEST_MANUAL_PID_REUSE:-0}" != 1 ]] || return 1
    [[ -n "$LEGACY_DRAGON_PID" && -n "$LEGACY_DRAGON_EXE" \
       && -n "$LEGACY_DRAGON_STARTTIME" ]] || return 1
    exe="$(readlink "$PROC_ROOT/$LEGACY_DRAGON_PID/exe" 2>/dev/null || true)"
    starttime="$(process_starttime "$PROC_ROOT/$LEGACY_DRAGON_PID/stat" 2>/dev/null || true)"
    [[ "$exe" == "$LEGACY_DRAGON_EXE" && "$starttime" == "$LEGACY_DRAGON_STARTTIME" ]]
}

stop_legacy() {
    [[ "$ADOPT_EXISTING" == true ]] || return 0
    step "Stopping only the recognized legacy services"
    if [[ "$LEGACY_CONNECTOR_SERVICE_ACTIVE" == true ]]; then
        systemctl stop connector.service || { die "cannot stop connector.service"; return 1; }
        if ! systemd_inactive_verified connector.service; then
            die "connector.service stop state could not be verified"
            return 1
        fi
    fi
    if [[ "$LEGACY_DRAGON_SERVICE_ACTIVE" == true ]]; then
        systemctl stop dragon.service || { die "cannot stop dragon.service"; return 1; }
        if ! systemd_inactive_verified dragon.service; then
            die "dragon.service stop state could not be verified"
            return 1
        fi
    elif [[ -n "$LEGACY_DRAGON_PID" ]]; then
        manual_dragon_identity_matches \
            || { die "recognized Dragon PID $LEGACY_DRAGON_PID changed identity before stop"; return 1; }
        install_manual_recovery_owner || return 1
        if [[ "${CLOUDNVR_TEST_MANUAL_STOP:-0}" == 1 ]]; then
            LEGACY_MANUAL_STOPPED=true
        else
            kill -TERM "$LEGACY_DRAGON_PID" \
                || { die "cannot stop recognized Dragon PID $LEGACY_DRAGON_PID"; return 1; }
            local waited=0
            while manual_dragon_identity_matches && (( waited < 30 )); do
                sleep 1
                waited=$((waited + 1))
            done
            if manual_dragon_identity_matches; then
                die "recognized Dragon PID $LEGACY_DRAGON_PID did not stop"
                return 1
            fi
            LEGACY_MANUAL_STOPPED=true
        fi
    fi
}

capture_legacy_connector_state() {
    local committed_before="$1" destination="$ROOT/data/connector/connector.db"
    [[ "$ADOPT_EXISTING" == true ]] || return 0
    if [[ -z "$LEGACY_CONNECTOR_DB" ]]; then
        [[ -e "$destination" ]] \
            || printf '' | atomic_write "$destination" 0600 || return 1
        return 0
    fi
    step "Capturing a consistent legacy connector database"
    if [[ "${CLOUDNVR_TEST_CONNECTOR_CAPTURE_FAIL:-0}" == 1 ]]; then
        die "cannot capture the legacy connector database"
        return 1
    fi
    if ! python3 - "$LEGACY_CONNECTOR_DB" "$BACKUP_DIR/connector.db" <<'PY'
import os, pathlib, sqlite3, sys, tempfile
source, target = map(pathlib.Path, sys.argv[1:])
target.parent.mkdir(parents=True, exist_ok=True)
fd, temporary = tempfile.mkstemp(prefix=".connector.db.", dir=str(target.parent))
os.close(fd)
try:
    source_db = sqlite3.connect(source.resolve(strict=True).as_uri() + "?mode=ro", uri=True)
    target_db = sqlite3.connect(temporary)
    try:
        source_db.backup(target_db)
        result = target_db.execute("PRAGMA integrity_check").fetchone()
        if not result or result[0] != "ok":
            raise RuntimeError("connector database integrity check failed")
    finally:
        target_db.close()
        source_db.close()
    os.chmod(temporary, 0o600)
    sync_fd = os.open(temporary, os.O_RDONLY)
    try:
        os.fsync(sync_fd)
    finally:
        os.close(sync_fd)
    os.replace(temporary, target)
    directory_fd = os.open(target.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
    try:
        os.fsync(directory_fd)
    finally:
        os.close(directory_fd)
except BaseException:
    try:
        os.unlink(temporary)
    except FileNotFoundError:
        pass
    raise
PY
    then
        die "cannot capture the legacy connector database"
        return 1
    fi
    if [[ "$committed_before" != true ]]; then
        atomic_copy "$BACKUP_DIR/connector.db" "$destination" 0600 || return 1
    elif [[ ! -e "$destination" ]]; then
        die "committed CloudNVR connector state is missing"
        return 1
    fi
}

retire_legacy() {
    local rc=0
    remove_manual_recovery_owner \
        || { warn "cannot remove temporary manual Dragon recovery"; rc=1; }
    [[ "$ADOPT_EXISTING" == true ]] || return "$rc"
    if [[ "$LEGACY_CONNECTOR_SERVICE_ENABLED" == true ]]; then
        systemctl disable connector.service >/dev/null \
            || { warn "cannot disable connector.service"; rc=1; }
        if ! systemd_disabled_verified connector.service; then
            warn "connector.service disable state could not be verified"
            rc=1
        fi
    fi
    if [[ "$LEGACY_DRAGON_SERVICE_ENABLED" == true ]]; then
        systemctl disable dragon.service >/dev/null \
            || { warn "cannot disable dragon.service"; rc=1; }
        if ! systemd_disabled_verified dragon.service; then
            warn "dragon.service disable state could not be verified"
            rc=1
        fi
    fi
    return "$rc"
}

restore_legacy() {
    local rc=0
    [[ "$ADOPT_EXISTING" == true ]] || return 0
    warn "Restoring the recognized legacy installation"
    if [[ "$LEGACY_DRAGON_SERVICE_ENABLED" == true ]]; then
        if ! systemctl enable dragon.service >/dev/null \
            || ! systemd_enabled_status dragon.service; then
            warn "could not restore dragon.service enablement"
            rc=1
        fi
    fi
    if [[ "$LEGACY_DRAGON_SERVICE_ACTIVE" == true ]]; then
        if ! systemctl start dragon.service \
            || ! systemd_active_status dragon.service; then
            warn "could not restart dragon.service"
            rc=1
        fi
    elif [[ "$LEGACY_MANUAL_STOPPED" == true \
       && -n "$LEGACY_MANUAL_STATE" && -f "$LEGACY_MANUAL_STATE" ]]; then
        if ! python3 - "$LEGACY_MANUAL_STATE" "$BACKUP_DIR/manual-dragon.log" <<'PY'
import json, os, subprocess, sys, time
state=json.load(open(sys.argv[1], encoding="utf-8"))
args=state["args"]
env={str(key): str(value) for key, value in state["env"].items()}
program=state["executable"]
if not os.path.isabs(program):
    raise SystemExit("manual Dragon executable is not absolute")
with open(sys.argv[2], "ab", buffering=0) as log:
    process=subprocess.Popen(args, executable=program, cwd=state["cwd"], env=env, stdin=subprocess.DEVNULL, stdout=log, stderr=log, start_new_session=True)
    time.sleep(0.25)
    if process.poll() is not None:
        raise SystemExit("manual Dragon process exited during restoration")
PY
        then
            warn "could not restart the manual Dragon process"
            rc=1
        fi
    fi
    if [[ "$LEGACY_CONNECTOR_SERVICE_ENABLED" == true ]]; then
        if ! systemctl enable connector.service >/dev/null \
            || ! systemd_enabled_status connector.service; then
            warn "could not restore connector.service enablement"
            rc=1
        fi
    fi
    if [[ "$LEGACY_CONNECTOR_SERVICE_ACTIVE" == true ]]; then
        if ! systemctl start connector.service \
            || ! systemd_active_status connector.service; then
            warn "could not restart connector.service"
            rc=1
        fi
    fi
    if [[ "$rc" -eq 0 ]]; then
        remove_manual_recovery_owner \
            || { warn "could not remove temporary manual Dragon recovery"; rc=1; }
    fi
    return "$rc"
}

maybe_installer_power_loss() {
    if [[ "${CLOUDNVR_TEST_INSTALLER_POWER_LOSS_AT:-}" == "$1" ]]; then
        exit 99
    fi
}

rollback_video_bind() {
    if [[ "$VIDEO_BIND_ADDED" == true && "${CLOUDNVR_SKIP_MOUNTS:-0}" != 1 ]]; then
        umount "$ROOT/data/video" || warn "could not remove the failed installation's video bind"
    fi
    if [[ "$FSTAB_CHANGED" == true ]]; then
        if [[ "$FSTAB_EXISTED" == true ]]; then
            atomic_replace_file "$FSTAB_WRITE_FILE" "$FSTAB_BACKUP_FILE" "$FSTAB_BACKUP_FILE" 1 \
                || warn "could not restore the prior fstab"
        else
            if rm -f "$FSTAB_WRITE_FILE"; then
                python3 -c 'import os,sys; fd=os.open(sys.argv[1], os.O_RDONLY|getattr(os,"O_DIRECTORY",0)); os.fsync(fd); os.close(fd)' \
                    "${FSTAB_WRITE_FILE%/*}" \
                    || warn "could not sync the failed installation's fstab removal"
            else
                warn "could not remove the failed installation's fstab file"
            fi
        fi
    fi
}

main() {
    local temp config_stage installed_before=false committed_before=false current_target="" activation_rc=0
    parse_args "$@" || return $?
    require_root || return 1
    acquire_install_lock || return 1
    validate_existing_root || return 1
    load_existing_install_settings || return 1
    finalize_backend_address || return 1
    marker_is_ours && installed_before=true
    if [[ -L "$ROOT/releases/current" ]]; then
        current_target="$(readlink "$ROOT/releases/current" 2>/dev/null || true)"
        if [[ "$current_target" =~ ^[0-9a-f]{64}$ \
           && -s "$ROOT/releases/$current_target/release.json" \
           && -s "$ROOT/releases/$current_target/release.env" \
           && -s "$ROOT/releases/$current_target/docker-compose.yml" \
           && -s "$ROOT/releases/$current_target/activation-compose.yml" \
           && -s "$ROOT/releases/$current_target/nvr" \
           && -s "$ROOT/releases/$current_target/install-cloudnvr.sh" \
           && -s "$ROOT/releases/$current_target/cloudnvr.service" ]]; then
            committed_before=true
        else
            die "the current release pointer is incomplete or invalid"
            return 1
        fi
    fi
    inventory_legacy || return 1
    temp="$(mktemp -d "${TMPDIR:-/tmp}/cloudnvr-install.XXXXXXXX")"
    trap 'rm -rf -- "${temp:-}"' EXIT
    FSTAB_BACKUP_FILE="$temp/fstab.before"
    resolve_bundle "$temp" || return 1
    manifest_read "$BUNDLE_DIR/release.json" || return 1
    validate_host || return 1
    config_stage="$temp/config"
    prepare_config_stage "$config_stage" || return 1
    validate_video_storage || return 1
    validate_ports "$config_stage/dragon.conf" || return 1

    step "Creating the CloudNVR host layout"
    create_layout || return 1
    write_marker || return 1
    write_host_env || return 1
    backup_legacy || return 1
    record_legacy_unit_identities || return 1
    install_config "$config_stage" || return 1
    configure_video_storage || { rollback_video_bind; return 1; }
    if [[ "$committed_before" != true ]]; then
        atomic_copy "$BUNDLE_DIR/nvr" "$ROOT/bin/nvr" 0755 \
            || { rollback_video_bind; return 1; }
        atomic_copy "$BUNDLE_DIR/install-cloudnvr.sh" "$ROOT/bin/install-cloudnvr.sh" 0755 \
            || { rollback_video_bind; return 1; }
    fi

    if ! preflight_candidate; then
        rollback_video_bind
        return 1
    fi
    # Run verified candidate host code before stopping a legacy recorder. This
    # is separate from preflight, which must not install host packages.
    if ! bash "$BUNDLE_DIR/nvr" ensure-host-tmux; then
        warn "Host tmux is unavailable; persistent terminals will retry on the next update"
    fi
    if ! install_boot_owner "$committed_before" "$current_target"; then
        rollback_video_bind
        return 1
    fi
    if ! stop_legacy; then
        restore_legacy || warn "legacy restoration failed after a stop error"
        rollback_video_bind
        return 1
    fi
    if ! capture_legacy_connector_state "$committed_before"; then
        restore_legacy || warn "legacy restoration failed after connector database capture"
        rollback_video_bind
        return 1
    fi
    maybe_installer_power_loss after-legacy-stop
    step "Activating the exact release digests"
    if [[ "${CLOUDNVR_TEST_INSTALLER_POWER_LOSS_AT:-}" == during-candidate-before-current ]]; then
        CLOUDNVR_ROOT="$ROOT" CLOUDNVR_INSTALLING=1 CLOUDNVR_INHERITED_LOCK=1 \
            CLOUDNVR_CANDIDATE_MANIFEST="$BUNDLE_DIR/release.json" \
            CLOUDNVR_TEST_POWER_LOSS_AT=before-current \
            bash "$BUNDLE_DIR/nvr" update --force || activation_rc=$?
        [[ "$activation_rc" -ne 99 ]] || exit 99
    else
        CLOUDNVR_ROOT="$ROOT" CLOUDNVR_INSTALLING=1 CLOUDNVR_INHERITED_LOCK=1 \
            CLOUDNVR_CANDIDATE_MANIFEST="$BUNDLE_DIR/release.json" \
            bash "$BUNDLE_DIR/nvr" update --force || activation_rc=$?
    fi
    if [[ "$activation_rc" -ne 0 ]]; then
        current_target="$(readlink "$ROOT/releases/current" 2>/dev/null || true)"
        if [[ "$current_target" == "$RELEASE_ID" ]]; then
            die "CloudNVR release $RELEASE_ID committed, but host service metadata needs repair"
            return 1
        fi
        if [[ "$committed_before" != true ]]; then
            if ! cleanup_failed_candidate; then
                rollback_video_bind
                die "candidate cleanup failed; legacy services remain stopped"
                return 1
            fi
            if ! restore_legacy; then
                rollback_video_bind
                die "candidate cleanup succeeded, but legacy restoration failed"
                return 1
            fi
        else
            remove_manual_recovery_owner \
                || warn "could not remove temporary manual Dragon recovery after failed update"
        fi
        rollback_video_bind
        return 1
    fi
    maybe_installer_power_loss after-current
    if ! retire_legacy; then
        die "CloudNVR is committed, but one or more legacy services remain enabled"
        return 1
    fi
    log "CloudNVR release $RELEASE_ID is installed and healthy."
    if [[ "$installed_before" == true ]]; then
        log "The rerun preserved the installed configuration and persistent data."
    fi
    register_remote_support
}

# ADR-020 amendment 1: vouch for the new box's remote-support key while a
# person is here, so it enrolls by itself. Never fails the install; without a
# credential the box waits for staff approval instead.
register_remote_support() {
    [[ "$SUPPORT_REGISTRATION" == true ]] || return 0
    # Contract-test installs (non-root) have no live stack to register.
    [[ "${CLOUDNVR_ALLOW_NON_ROOT:-0}" != 1 ]] || return 0
    # On the office network this needs nothing from anyone. Elsewhere, with a
    # terminal, it offers a login; without one the box waits for approval.
    # Dragon registers within seconds on the office network; do not stall an
    # unattended install for long when the backend is unreachable.
    local -a args=(remoted register --wait 300)
    [[ -t 0 ]] || args=(remoted register --wait 90 --no-login)
    step "Registering the remote support key"
    CLOUDNVR_ROOT="$ROOT" bash "$ROOT/bin/nvr" "${args[@]}" \
        || warn "remote support key not preauthorized; the box will wait for staff approval (sudo nvr remoted register retries)"
}

main "$@"