#!/bin/bash
# CloudNVR host operator command.
#
# The active release is always releases/current. Compose never resolves a
# mutable channel tag. Updates stage exact image digests, prove health, and
# move the current pointer only after activation succeeds.
set -Eeuo pipefail

ROOT="${CLOUDNVR_ROOT:-/opt/dividia}"
ENV_FILE="$ROOT/.env"
MARKER_FILE="$ROOT/.product"
RELEASES_DIR="$ROOT/releases"
STATE_DIR="$ROOT/state"
BIN_DIR="$ROOT/bin"
CURRENT_LINK="$RELEASES_DIR/current"
DEFAULT_LOCK_FILE=/run/lock/cloudnvr/mutation.lock
LOCK_FILE="${CLOUDNVR_LOCK_FILE:-$DEFAULT_LOCK_FILE}"
CRON_FILE="${CLOUDNVR_CRON_FILE:-/etc/cron.d/dividia-cloudnvr-update}"
SERVICE_FILE="${CLOUDNVR_SERVICE_FILE:-/etc/systemd/system/cloudnvr.service}"
CMD_LINK="${CLOUDNVR_CMD_LINK:-/usr/local/bin/nvr}"
DEFAULT_BUNDLE_BASE_URL="http://files.dividia.net/software/cloudnvr-docker"
# ADR-021: the Nano flavor publishes under its own root with the same channels.
NANO_BUNDLE_BASE_URL="http://files.dividia.net/software/cloudnvr-nano"
TEGRA_RELEASE_FILE="${CLOUDNVR_L4T_RELEASE_FILE:-/etc/nv_tegra_release}"
DEVICE_MODEL_FILE="${CLOUDNVR_DEVICE_MODEL_FILE:-/proc/device-tree/model}"
FLAVOR=orin
# Host Python 3.6 (the Nano) has no UTF-8 default without a locale, and cron
# and systemd give it none.
export PYTHONIOENCODING=utf-8
DRAGON_DEADLINE="${CLOUDNVR_DRAGON_ACTIVATION_DEADLINE_SEC:-300}"
CONNECTOR_DEADLINE="${CLOUDNVR_CONNECTOR_ACTIVATION_DEADLINE_SEC:-120}"
HEALTH_POLL="${CLOUDNVR_HEALTH_POLL_SEC:-2}"
SERVICE_STOP_LOCK_TIMEOUT="${CLOUDNVR_SERVICE_STOP_LOCK_TIMEOUT_SEC:-480}"
REMOTED_HEALTH_TIMEOUT="${CLOUDNVR_REMOTED_HEALTH_TIMEOUT_SEC:-120}"
REMOTED_QUARANTINE="$STATE_DIR/remoted-quarantine"
HOST_ACCESS_USER="${CLOUDNVR_HOST_ACCESS_USER:-dividia}"
SUDOERS_FILE="${CLOUDNVR_SUDOERS_FILE:-/etc/sudoers.d/dividia}"
HOSTS_FILE="${CLOUDNVR_HOSTS_FILE:-/etc/hosts}"
ROOT_SSH_DIR="${CLOUDNVR_HOST_SSH_DIR:-/root/.ssh}"
# The shared develop support identity, the key nvr-docker authorizes on hosts
# with no other support key source (nvr docker/nvr). cloudnvr-convert-isite
# carries the same line, so a conversion rollback can take it back.
DEVELOP_SUPPORT_KEY='ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIPGn6gD47+s4ayRHNqb4ZIMRjcvlLmsz7JRw3qCLNAbc'
# The connector's cloudapi profile: the active fleet secret + environment marker,
# and one saved profile per environment for `nvr environment`.
CLOUDAPI_DIR="${CLOUDNVR_CLOUDAPI_DIR:-/etc/dividia/cloudapi}"
SERIAL_FILE="$ROOT/data/cache/dragon/serial"
DRAGON_DEVID_FILE="$ROOT/data/cache/dragon/device-id"
# ADR-023: the zone Dragon applies from the cloud's [global] timezone, a symlink
# to /usr/share/zoneinfo/<zone> that resolves on the host too. The daily update
# runs at 02:MM in it, or in the host zone until the cloud has pushed one.
UPDATE_ZONE_FILE="$ROOT/data/cache/dragon/localtime"
HOST_LOCALTIME="${CLOUDNVR_HOST_LOCALTIME:-/etc/localtime}"
# ADR-021: an iSite conversion (cloudnvr-convert-isite) holds the box while the
# native engine and CloudNVR could both claim the device identity. Only the
# converter's own install may activate dragon and the connector then.
CONVERSION_HOLD="${CLOUDNVR_CONVERSION_DIR:-/var/lib/cloudnvr-conversion}/hold.json"

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

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
}

file_mode() {
    python3 -c 'import os,stat,sys; print(format(stat.S_IMODE(os.lstat(sys.argv[1]).st_mode), "o"))' "$1"
}

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

# `production` is the STABLE FLEET ALIAS. Dated YYYY.M.N channels are immutable
# by design (ADR-009: a production release name is never reused for a different
# source commit), which is right for auditing and rollback but means a box
# pinned to one never moves: its nightly `nvr update` refetches the same release
# forever and reports "already current". The alias is re-pointed at each
# promotion, so a box tracking it picks up the new release on the normal cron
# with no per-box action, while the dated channels stay as audit anchors.
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 it must carry the same
# production posture: the real backend, and the installer's strict gates.
channel_is_production() {
    [[ "$1" == production ]] && return 0
    [[ "$1" =~ ^[0-9]{4}\.([1-9]|1[0-2])\.[1-9][0-9]*$ ]]
}

channel_backend_default() {
    if channel_is_production "$1"; then
        printf '%s\n' engine.cloud.dividia.net
    else
        printf '%s\n' engine-staging.cloud.dividia.net
    fi
}

# ADR-021: a flavor is a hardware target of the one release system. The host
# names its flavor in .env (absent = orin) and it is re-detected on every run.
host_flavor() {
    local release model
    release="$(head -c 512 "$TEGRA_RELEASE_FILE" 2>/dev/null || true)"
    if grep -Eq 'R36\.4|R36.*REVISION:[[:space:]]*4([.,[:space:]]|$)' <<< "$release"; then
        printf 'orin\n'
    elif grep -Eq 'R32.*REVISION:[[:space:]]*[4-7]([.,[:space:]]|$)' <<< "$release"; then
        # JetPack 4.4 through 4.6.x (R32.4-R32.7): every JetPack 4 whose driver
        # runs the image's own CUDA 10.2 runtime (ADR-021 amendment 2).
        # "Jetson Orin Nano" is an Orin; only the original Nano matches here.
        model="$(tr -d '\0' < "$DEVICE_MODEL_FILE" 2>/dev/null || true)"
        [[ "$model" == *"Jetson Nano"* ]] && printf 'nano\n'
    fi
    return 0
}

flavor_product() { case "$FLAVOR" in nano) printf 'cloudnvr-nano\n' ;; *) printf 'cloudnvr\n' ;; esac; }
flavor_dragon_repo() { case "$FLAVOR" in nano) printf 'docker.io/dividia/cloudnvr-dragon-nano\n' ;; *) printf 'docker.io/dividia/cloudnvr-dragon\n' ;; esac; }
flavor_dragon_title() { case "$FLAVOR" in nano) printf 'cloudnvr-dragon-nano\n' ;; *) printf 'cloudnvr-dragon\n' ;; esac; }
flavor_dragon_platform() { case "$FLAVOR" in nano) printf 'jetson-nano-arm64\n' ;; *) printf 'jetson-arm64\n' ;; esac; }
flavor_bundle_base_url() { case "$FLAVOR" in nano) printf '%s\n' "$NANO_BUNDLE_BASE_URL" ;; *) printf '%s\n' "$DEFAULT_BUNDLE_BASE_URL" ;; esac; }

# The cloud a box talks to (engine backend + connector), independent of the
# channel (which software it runs), so a dev box can run a dev build against
# production or a production build against staging.
environment_normalize() {
    case "$1" in
        prod|production) printf '%s\n' prod ;;
        staging) printf '%s\n' staging ;;
        *) return 1 ;;
    esac
}

environment_backend() {
    case "$1" in
        prod) printf '%s\n' engine.cloud.dividia.net ;;
        staging) printf '%s\n' engine-staging.cloud.dividia.net ;;
        *) return 1 ;;
    esac
}

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
}

digest_is_valid() {
    [[ "$1" =~ ^sha256:[0-9a-f]{64}$ ]]
}

release_id_is_valid() {
    [[ "$1" =~ ^[0-9a-f]{64}$ ]]
}

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 ]]
}

require_install() {
    if ! marker_is_ours && [[ "${CLOUDNVR_INSTALLING:-0}" != 1 ]]; then
        die "$ROOT is not a CloudNVR schema 1 installation"
    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
        if [[ "$key" == "$wanted" ]]; then
            found="$value"
        fi
    done < "$file"
    [[ -n "$found" ]] || return 1
    printf '%s\n' "$found"
}

load_host_env() {
    [[ -f "$ENV_FILE" ]] || { die "missing $ENV_FILE"; return 1; }
    CHANNEL="$(kv_get "$ENV_FILE" CHANNEL)" || { die "missing CHANNEL in $ENV_FILE"; return 1; }
    PLATFORM="$(kv_get "$ENV_FILE" PLATFORM)" || { die "missing PLATFORM in $ENV_FILE"; return 1; }
    REGISTRY="$(kv_get "$ENV_FILE" REGISTRY)" || { die "missing REGISTRY in $ENV_FILE"; return 1; }
    COMPOSE_PROJECT_NAME="$(kv_get "$ENV_FILE" COMPOSE_PROJECT_NAME)" \
        || { die "missing COMPOSE_PROJECT_NAME in $ENV_FILE"; return 1; }
    FLAVOR="$(kv_get "$ENV_FILE" FLAVOR 2>/dev/null || true)"
    FLAVOR="${FLAVOR:-orin}"
    [[ "$FLAVOR" == orin || "$FLAVOR" == nano ]] || { die "invalid configured flavor: $FLAVOR"; return 1; }
    local detected_flavor
    detected_flavor="$(host_flavor)"
    [[ -z "$detected_flavor" || "$detected_flavor" == "$FLAVOR" ]] \
        || { die "this host is a Jetson $detected_flavor, but $ENV_FILE is for $FLAVOR"; return 1; }
    BUNDLE_BASE_URL="$(kv_get "$ENV_FILE" BUNDLE_BASE_URL 2>/dev/null || true)"
    BUNDLE_BASE_URL="${BUNDLE_BASE_URL:-$(flavor_bundle_base_url)}"
    channel_is_valid "$CHANNEL" || { die "invalid configured channel: $CHANNEL"; return 1; }
    CLOUD_ENVIRONMENT="$(kv_get "$ENV_FILE" CLOUD_ENVIRONMENT 2>/dev/null || true)"
    if [[ -n "$CLOUD_ENVIRONMENT" ]]; then
        CLOUD_ENVIRONMENT="$(environment_normalize "$CLOUD_ENVIRONMENT")" \
            || { die "invalid configured cloud environment"; return 1; }
    fi
    BACKEND_ADDR="$(kv_get "$ENV_FILE" BACKEND_ADDR 2>/dev/null || true)"
    if [[ -z "$BACKEND_ADDR" && -n "$CLOUD_ENVIRONMENT" ]]; then
        BACKEND_ADDR="$(environment_backend "$CLOUD_ENVIRONMENT")"
    fi
    BACKEND_ADDR="${BACKEND_ADDR:-$(channel_backend_default "$CHANNEL")}"
    backend_address_is_valid "$BACKEND_ADDR" \
        || { die "invalid configured backend address: $BACKEND_ADDR"; return 1; }
    [[ "$PLATFORM" == linux/arm64 ]] || { die "unsupported configured platform: $PLATFORM"; return 1; }
    [[ "$REGISTRY" == docker.io ]] || { die "unsupported configured registry: $REGISTRY"; return 1; }
    [[ "$COMPOSE_PROJECT_NAME" =~ ^[a-z0-9][a-z0-9_-]*$ ]] \
        || { die "unsafe Compose project name"; 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:
    data = sys.stdin.buffer.read()
    with os.fdopen(fd, "wb") as out:
        out.write(data)
        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}"
    [[ -f "$source" && ! -L "$source" && -s "$source" ]] \
        || { die "invalid source file: $source"; return 1; }
    atomic_write "$target" "$mode" < "$source"
}

atomic_symlink() {
    local directory="$1" name="$2" target="$3"
    python3 -c '
import os, pathlib, sys
directory = pathlib.Path(sys.argv[1])
name, target = sys.argv[2], sys.argv[3]
directory.mkdir(parents=True, exist_ok=True)
tmp = directory / ("." + name + ".new.%d" % os.getpid())
committed = False
try:
    os.symlink(target, tmp)
    os.replace(tmp, directory / name)
    committed = True
    dfd = os.open(directory, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
    try:
        if os.environ.get("CLOUDNVR_TEST_SYMLINK_DIR_FSYNC_FAIL") == name:
            raise OSError("injected symlink directory sync failure")
        os.fsync(dfd)
    except OSError:
        raise SystemExit(2)
    finally: os.close(dfd)
except BaseException:
    if not committed:
        try: tmp.unlink()
        except FileNotFoundError: pass
    raise
' "$directory" "$name" "$target"
}

fsync_tree() {
    [[ "${CLOUDNVR_TEST_FSYNC_TREE_FAIL:-0}" != 1 ]] || return 1
    python3 -c '
import os, pathlib, sys
root = pathlib.Path(sys.argv[1])
for path in sorted(root.rglob("*")):
    if path.is_file() and not path.is_symlink():
        fd = os.open(path, os.O_RDONLY)
        try: os.fsync(fd)
        finally: os.close(fd)
for path in sorted((p for p in root.rglob("*") if p.is_dir()), reverse=True):
    fd = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
    try: os.fsync(fd)
    finally: os.close(fd)
fd = os.open(root, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
try: os.fsync(fd)
finally: os.close(fd)
' "$1"
}

manifest_load() {
    local file="$1" parsed
    parsed="$(python3 - "$file" "$(flavor_product)" "$(flavor_dragon_repo)" <<'PY'
import hashlib, json, re, sys
path, product, dragon_repository = sys.argv[1:4]
with open(path, encoding="utf-8") as handle:
    value = json.load(handle)
schema = value.get("schema")
if type(schema) is not int or schema not in (1, 2) or value.get("product") != product:
    raise SystemExit("wrong product or schema")
images = value.get("images")
if not isinstance(images, dict):
    raise SystemExit("missing images")
services = ["dragon", "connector"] + (["remoted"] if schema == 2 else [])
if set(images) != set(services): raise SystemExit("unexpected image services")
if any(not isinstance(images[k], dict) for k in services): raise SystemExit("invalid image")
dragon = images.get("dragon", {})
connector = images.get("connector", {})
fields = [
    value.get("release_id", ""), value.get("platform", ""),
    value.get("source_revision", ""),
    value.get("connector_nvr_source_revision", ""),
    dragon.get("repository", ""), dragon.get("digest", ""),
    dragon.get("reference", ""), dragon.get("revision_tag", ""),
    connector.get("repository", ""), connector.get("digest", ""),
    connector.get("reference", ""), connector.get("revision_tag", ""),
    value.get("parent_release_id") or "",
]
if any(not isinstance(item, str) or "\t" in item or "\n" in item for item in fields):
    raise SystemExit("invalid release field")
rid, platform, source, nvr, dr, dd, dref, dtag, cr, cd, cref, ctag, parent = fields
if not re.fullmatch(r"[0-9a-f]{64}", rid): raise SystemExit("bad release id")
if platform != "linux/arm64": raise SystemExit("bad platform")
if not re.fullmatch(r"[0-9a-f]{40}", source): raise SystemExit("bad source revision")
if not re.fullmatch(r"[0-9a-f]{40}", nvr): raise SystemExit("bad NVR revision")
if dr != dragon_repository: raise SystemExit("bad Dragon repository")
if cr != "docker.io/dividia/cloudnvr-connector": raise SystemExit("bad connector repository")
if not re.fullmatch(r"sha256:[0-9a-f]{64}", dd): raise SystemExit("bad Dragon digest")
if not re.fullmatch(r"sha256:[0-9a-f]{64}", cd): raise SystemExit("bad connector digest")
if dref != dr + "@" + dd or cref != cr + "@" + cd: raise SystemExit("bad digest reference")
if not re.fullmatch(r"[A-Za-z0-9._/:@-]+", dtag): raise SystemExit("bad Dragon revision tag")
if not re.fullmatch(r"[A-Za-z0-9._/:@-]+", ctag): raise SystemExit("bad connector revision tag")
if parent and not re.fullmatch(r"[0-9a-f]{64}", parent): raise SystemExit("bad parent")
if parent: raise SystemExit("release manifests cannot contain activation lineage")
canonical = "\n".join([
    "schema=" + str(schema), "product=" + product, "platform=" + platform,
    "source_revision=" + source, "nvr_source_revision=" + nvr,
    "dragon=" + dr + "@" + dd, "connector=" + cr + "@" + cd, "",
])
remoted = images.get("remoted", {})
rr = remoted.get("repository", "-")
rd = remoted.get("digest", "-")
rf = remoted.get("reference", "-")
rt = remoted.get("revision_tag", "-")
rn = value.get("remoted_nvr_source_revision", "-")
if schema == 2:
    if not all(isinstance(v, str) and "\t" not in v and "\n" not in v for v in (rr, rd, rf, rt, rn)):
        raise SystemExit("invalid remoted field")
    if rr != "docker.io/dividia/cloudnvr-remoted": raise SystemExit("bad remoted repository")
    if not re.fullmatch(r"sha256:[0-9a-f]{64}", rd): raise SystemExit("bad remoted digest")
    if rf != rr + "@" + rd: raise SystemExit("bad remoted reference")
    if not re.fullmatch(r"[A-Za-z0-9._/:@-]+", rt): raise SystemExit("bad remoted revision tag")
    if not re.fullmatch(r"[0-9a-f]{40}", rn): raise SystemExit("bad remoted NVR revision")
    canonical += "remoted_nvr_source_revision=" + rn + "\nremoted=" + rf + "\n"
fields[-1] = "-"
fields.extend([str(schema), rr, rd, rf, rt, rn])
if hashlib.sha256(canonical.encode()).hexdigest() != rid:
    raise SystemExit("release identity mismatch")
print("\t".join(fields))
PY
)" || { die "invalid release manifest: $file"; return 1; }
    # shellcheck disable=SC2034 # Repository and revision-tag fields are validated by the parser.
    IFS=$'\t' read -r \
        MANIFEST_RELEASE_ID MANIFEST_PLATFORM MANIFEST_SOURCE_REV MANIFEST_NVR_REV \
        MANIFEST_DRAGON_REPO MANIFEST_DRAGON_DIGEST MANIFEST_DRAGON_REF MANIFEST_DRAGON_TAG \
        MANIFEST_CONNECTOR_REPO MANIFEST_CONNECTOR_DIGEST MANIFEST_CONNECTOR_REF MANIFEST_CONNECTOR_TAG \
        MANIFEST_PARENT MANIFEST_SCHEMA MANIFEST_REMOTED_REPO MANIFEST_REMOTED_DIGEST \
        MANIFEST_REMOTED_REF MANIFEST_REMOTED_TAG MANIFEST_REMOTED_NVR_REV <<< "$parsed"
    [[ "$MANIFEST_PARENT" != - ]] || MANIFEST_PARENT=""
    if [[ "$MANIFEST_SCHEMA" == 1 ]]; then
        MANIFEST_REMOTED_REF=""
        MANIFEST_REMOTED_DIGEST=""
    fi
}

current_release_id() {
    local target
    [[ -L "$CURRENT_LINK" ]] || return 1
    target="$(readlink "$CURRENT_LINK")" || return 1
    release_id_is_valid "$target" || return 1
    [[ -d "$RELEASES_DIR/$target" && ! -L "$RELEASES_DIR/$target" ]] || return 1
    printf '%s\n' "$target"
}

release_path() {
    local release_id="$1"
    release_id_is_valid "$release_id" || return 1
    printf '%s/%s\n' "$RELEASES_DIR" "$release_id"
}

# ADR-021: an Orin runs the host's Compose plugin. A Nano has none, so each Nano
# release carries its own Compose binary (from its dragon image): a candidate is
# validated with its own Compose and a rollback runs the old release's.
NANO_TOOLS="compose compose.sha256"

sha256_file() {
    python3 -c 'import hashlib,sys; print(hashlib.sha256(open(sys.argv[1],"rb").read()).hexdigest())' "$1"
}

# Hashing a 58 MB binary on every Compose call would cost the Nano about half a
# second each, so a directory is checked once per process (until it changes).
COMPOSE_VERIFIED=""

release_compose_valid() {
    local dir="$1" want stamp
    [[ -f "$dir/compose" && ! -L "$dir/compose" && -x "$dir/compose" \
       && -f "$dir/compose.sha256" && ! -L "$dir/compose.sha256" ]] || return 1
    stamp="$dir|$(ls -ln "$dir/compose" 2>/dev/null | awk '{print $5}')|$(cat "$dir/compose.sha256" 2>/dev/null)"
    [[ $'\n'"$COMPOSE_VERIFIED"$'\n' == *$'\n'"$stamp"$'\n'* ]] && return 0
    want="$(awk '{print $1; exit}' "$dir/compose.sha256" 2>/dev/null)"
    [[ "$want" =~ ^[0-9a-f]{64}$ && "$(sha256_file "$dir/compose")" == "$want" ]] || return 1
    COMPOSE_VERIFIED+="$stamp"$'\n'
}

# A missing or damaged binary comes back from the release's own local image, so
# remoted (the lifeline) never depends on a host file surviving (ADR-017).
ensure_release_compose() {
    local dir="$1" ref cid name
    release_compose_valid "$dir" && return 0
    [[ -f "$dir/release.json" ]] || { die "the Compose binary in $dir is missing or damaged"; return 1; }
    ref="$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["images"]["dragon"]["reference"])' "$dir/release.json")" \
        || { die "cannot read the dragon image of ${dir##*/}"; return 1; }
    warn "the Compose binary of release ${dir##*/} is missing or damaged; restoring it from $ref"
    cid="$(docker create "$ref")" || { die "cannot restore Compose: $ref is not present"; return 1; }
    local work
    work="$(mktemp -d "$dir/.compose-restore.XXXXXX")" || { docker rm -f "$cid" >/dev/null 2>&1; return 1; }
    for name in $NANO_TOOLS; do
        docker cp "$cid:/usr/share/cloudnvr/$name" "$work/$name" >/dev/null \
            || { docker rm -f "$cid" >/dev/null 2>&1; rm -rf "$work"; die "cannot restore Compose from $ref"; return 1; }
    done
    docker rm -f "$cid" >/dev/null 2>&1 || true
    chmod 0755 "$work/compose" && chmod 0644 "$work/compose.sha256" \
        && mv -f "$work/compose.sha256" "$dir/compose.sha256" && mv -f "$work/compose" "$dir/compose" \
        || { rm -rf "$work"; return 1; }
    rm -rf "$work"
    release_compose_valid "$dir" || { die "the restored Compose binary of ${dir##*/} does not match its checksum"; return 1; }
}

run_compose() {
    local dir="$1"
    shift
    if [[ "$FLAVOR" == nano ]]; then
        ensure_release_compose "$dir" || return 1
        install_security_profiles "$dir" || return 1
        "$dir/compose" "$@"
    else
        docker compose "$@"
    fi
}

# Compose resolves `seccomp:./security/<name>` against the project directory.
# Profile names are versioned and a file is never changed in place, so any
# release, including a rollback target, still finds the profile it names.
install_security_profiles() {
    local dir="$1" source name target
    [[ -d "$dir/security" ]] || return 0
    [[ ! -L "$ROOT/security" ]] || { die "$ROOT/security must not be a symlink"; return 1; }
    mkdir -p "$ROOT/security" && chmod 0755 "$ROOT/security" || return 1
    for source in "$dir"/security/*.json; do
        [[ -e "$source" ]] || continue
        name="${source##*/}"
        [[ "$name" =~ ^[a-z0-9-]+\.json$ && -f "$source" && ! -L "$source" ]] \
            || { die "invalid security profile in release: $name"; return 1; }
        target="$ROOT/security/$name"
        if [[ -e "$target" || -L "$target" ]]; then
            [[ -f "$target" && ! -L "$target" ]] && cmp -s "$source" "$target" \
                || { die "security profile $name differs from the installed one; a changed profile needs a new name"; return 1; }
        else
            atomic_copy "$source" "$target" 0644 || return 1
        fi
    done
}

compose_cmd() {
    local release_id="$1"
    shift
    local dir
    dir="$(release_path "$release_id")" || return 1
    [[ -f "$dir/release.env" && -f "$dir/docker-compose.yml" ]] \
        || { die "release $release_id is incomplete"; return 1; }
    if [[ "${1:-}" == up && -d "$ROOT/data/remoted" && " $(release_services "$release_id") " != *" remoted "* ]]; then
        local remoted_cid
        remoted_cid="$(docker ps -aq --filter "label=com.docker.compose.project=$COMPOSE_PROJECT_NAME" --filter label=com.docker.compose.service=remoted)" || return 1
        if [[ -n "$remoted_cid" ]]; then
            [[ "$remoted_cid" != *$'\n'* ]] || { die "ambiguous remoted container"; return 1; }
            docker rm -f "$remoted_cid" >/dev/null || return 1
        fi
    fi
    if [[ "${1:-}" == up ]] && ! compose_args_name_services "${@:2}" && remoted_held "$release_id"; then
        set -- "$@" dragon connector
    fi
    BACKEND_ADDR="$BACKEND_ADDR" run_compose "$dir" --project-directory "$ROOT" --env-file "$ROOT/.env" \
        --env-file "$dir/release.env" \
        -f "$dir/docker-compose.yml" "$@"
}

candidate_compose_cmd() {
    local release_id="$1"
    shift
    local dir
    dir="$(release_path "$release_id")" || return 1
    [[ -f "$dir/release.env" && -f "$dir/docker-compose.yml" \
       && -f "$dir/activation-compose.yml" ]] \
        || { die "release $release_id is incomplete"; return 1; }
    if [[ "${1:-}" == up && -d "$ROOT/data/remoted" && " $(release_services "$release_id") " != *" remoted "* ]]; then
        local remoted_cid
        remoted_cid="$(docker ps -aq --filter "label=com.docker.compose.project=$COMPOSE_PROJECT_NAME" --filter label=com.docker.compose.service=remoted)" || return 1
        if [[ -n "$remoted_cid" ]]; then
            [[ "$remoted_cid" != *$'\n'* ]] || { die "ambiguous remoted container"; return 1; }
            docker rm -f "$remoted_cid" >/dev/null || return 1
        fi
    fi
    if [[ "${1:-}" == up ]] && ! compose_args_name_services "${@:2}" && remoted_held "$release_id"; then
        set -- "$@" dragon connector
    fi
    BACKEND_ADDR="$BACKEND_ADDR" run_compose "$dir" --project-directory "$ROOT" --env-file "$ROOT/.env" \
        --env-file "$dir/release.env" \
        -f "$dir/docker-compose.yml" -f "$dir/activation-compose.yml" "$@"
}

container_health() {
    local release_id="$1" service="$2" cid state expected actual digest_key repository
    cid="$(compose_cmd "$release_id" ps -q "$service" 2>/dev/null || true)"
    [[ -n "$cid" ]] || { printf 'missing\n'; return 1; }
    if [[ "$service" == dragon ]]; then
        digest_key=DRAGON_DIGEST
        repository="$(flavor_dragon_repo)"
    elif [[ "$service" == remoted ]]; then
        digest_key=REMOTED_DIGEST
        repository=docker.io/dividia/cloudnvr-remoted
    else
        digest_key=CONNECTOR_DIGEST
        repository=docker.io/dividia/cloudnvr-connector
    fi
    expected="$repository@$(kv_get "$RELEASES_DIR/$release_id/release.env" "$digest_key" 2>/dev/null || true)"
    actual="$(docker inspect --format '{{.Config.Image}}' "$cid" 2>/dev/null || true)"
    if [[ "$actual" != "$expected" ]] \
       && ! { [[ "$service" == remoted ]] && remoted_same_content "$actual" "$expected"; }; then
        printf 'wrong-image\n'
        return 1
    fi
    state="$(docker inspect --format '{{if .State.Health}}{{.State.Health.Status}}{{else}}{{.State.Status}}{{end}}' "$cid" 2>/dev/null || true)"
    printf '%s\n' "${state:-unknown}"
    [[ "$state" == healthy ]]
}

release_services() {
    python3 -c 'import json,sys; v=json.load(open(sys.argv[1])); print(" ".join(v["images"]))' "$RELEASES_DIR/$1/release.json"
}

# ADR-017: remoted is the support lifeline, so the core lifecycle never gates on
# it, never stops it, and never recreates it onto an image this box has already
# watched fail. The quarantine holds that image's identity and expires on its
# own once a release carries a different one.
release_has_remoted() {
    [[ " $(release_services "$1") " == *" remoted "* ]]
}

release_remoted_digest() {
    kv_get "$RELEASES_DIR/$1/release.env" REMOTED_DIGEST 2>/dev/null
}

release_remoted_ref() {
    local digest
    digest="$(release_remoted_digest "$1")" && [[ -n "$digest" ]] || return 1
    printf 'docker.io/dividia/cloudnvr-remoted@%s\n' "$digest"
}

# ADR-017: every release rebuilds remoted, so its digest always moves. The
# content key label covers only what the image is made of, so a box can tell a
# rebuild of the same tunnel from a real change without anyone publishing
# anything in order. Images from before the label have no key and never match.
remoted_content_key() {
    local key
    key="$(docker image inspect --format '{{index .Config.Labels "net.dividia.cloudnvr.remoted-content"}}' "$1" 2>/dev/null || true)"
    [[ "$key" =~ ^[0-9a-f]{64}$ ]] || return 1
    printf '%s\n' "$key"
}

remoted_same_content() {
    local running expected
    running="$(remoted_content_key "$1")" || return 1
    expected="$(remoted_content_key "$2")" || return 1
    [[ "$running" == "$expected" ]]
}

# What a quarantine names: the content key, so a rebuild of the same bad
# content stays quarantined; the digest for images from before the label.
remoted_identity() {
    local ref
    ref="$(release_remoted_ref "$1")" || return 1
    remoted_content_key "$ref" 2>/dev/null && return 0
    printf '%s\n' "${ref##*@}"
}

quarantined_remoted_identity() {
    [[ -f "$REMOTED_QUARANTINE" && ! -L "$REMOTED_QUARANTINE" ]] || return 1
    head -n 1 "$REMOTED_QUARANTINE" 2>/dev/null
}

remoted_quarantined() {
    local identity quarantined
    release_has_remoted "$1" 2>/dev/null || return 1
    identity="$(remoted_identity "$1")" || return 1
    quarantined="$(quarantined_remoted_identity)" || return 1
    [[ -n "$identity" && "$identity" == "$quarantined" ]]
}

# The image of the one remoted container, only while it is actually running:
# a stopped or exited tunnel must never be "kept".
remoted_running_image() {
    local cid running
    cid="$(docker ps -aq --filter "label=com.docker.compose.project=$COMPOSE_PROJECT_NAME" \
        --filter label=com.docker.compose.service=remoted 2>/dev/null || true)"
    [[ -n "$cid" && "$cid" != *$'\n'* ]] || return 1
    running="$(docker inspect --format '{{.State.Running}}' "$cid" 2>/dev/null || true)"
    [[ "$running" == true ]] || return 1
    docker inspect --format '{{.Config.Image}}' "$cid" 2>/dev/null
}

# The running tunnel is a different image of the same content as the release's.
remoted_content_unchanged() {
    local running expected
    release_has_remoted "$1" 2>/dev/null || return 1
    running="$(remoted_running_image)" || return 1
    expected="$(release_remoted_ref "$1")" || return 1
    [[ "$running" != "$expected" ]] && remoted_same_content "$running" "$expected"
}

# Leave the running remoted alone: its content is unchanged, or the release's
# image is one this box already watched fail.
remoted_held() {
    remoted_quarantined "$1" || remoted_content_unchanged "$1"
}

# Whether a `compose up` argument list names services. Every up flag this CLI
# passes (-d, --force-recreate, --no-deps) is valueless.
compose_args_name_services() {
    local arg
    for arg in "$@"; do
        [[ "$arg" == -* ]] || return 0
    done
    return 1
}

# Converge remoted onto the target release before the core stack moves, so
# whatever fails after this still leaves a way in. Never fails the caller. A
# changed remoted that will not go healthy is rolled back to the prior release's
# image and quarantined, so the next nightly does not retry the same content.
remoted_reconcile() {
    local target="$1" prior="${2:-}" identity prior_identity="" prior_ref old_quarantine=""
    release_has_remoted "$target" 2>/dev/null || return 0
    identity="$(remoted_identity "$target")" || return 0
    if old_quarantine="$(quarantined_remoted_identity)"; then
        if [[ "$old_quarantine" == "$identity" ]]; then
            log "remoted $identity is quarantined; leaving the running support tunnel in place"
            return 0
        fi
        rm -f -- "$REMOTED_QUARANTINE" || warn "could not clear the expired remoted quarantine"
    fi
    if remoted_content_unchanged "$target"; then
        log "remoted content is unchanged; keeping the running support tunnel"
        return 0
    fi
    [[ "$(container_health "$target" remoted 2>/dev/null || true)" == healthy ]] && return 0
    log "Converging remoted onto $identity"
    if compose_cmd "$target" up -d remoted && wait_service "$target" remoted "$REMOTED_HEALTH_TIMEOUT"; then
        return 0
    fi
    if [[ -n "$prior" && "$prior" != "$target" ]] && release_has_remoted "$prior" 2>/dev/null \
       && prior_ref="$(release_remoted_ref "$prior")" && docker image inspect "$prior_ref" >/dev/null 2>&1; then
        prior_identity="$(remoted_identity "$prior" || true)"
    fi
    if [[ -z "$prior_identity" || "$prior_identity" == "$identity" || "$prior_identity" == "$old_quarantine" ]]; then
        warn "remoted is unhealthy and there is no local known-good image to return to; continuing"
        return 0
    fi
    warn "remoted $identity is unhealthy; returning the support tunnel to $prior_identity"
    if compose_cmd "$prior" up -d remoted; then
        printf '%s\n' "$identity" | atomic_write "$REMOTED_QUARANTINE" 0644 \
            || warn "could not record the remoted quarantine"
    else
        warn "remoted rollback start failed; not quarantining so the next run retries"
    fi
    return 0
}

release_services_present() {
    local release_id="$1" service
    for service in $(release_services "$release_id"); do
        [[ "$service" != remoted ]] || continue
        [[ -n "$(compose_cmd "$release_id" ps -q "$service" 2>/dev/null || true)" ]] || return 1
    done
}

prepare_lock_file() {
    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
}

acquire_mutation_lock() {
    prepare_lock_file || { die "unsafe CloudNVR mutation lock: $LOCK_FILE"; return 1; }
    if [[ "${CLOUDNVR_INHERITED_LOCK:-0}" == 1 ]]; then
        : >&9 2>/dev/null \
            || { die "the inherited CloudNVR mutation lock is unavailable"; return 1; }
        flock -n 9 \
            || { die "the inherited CloudNVR mutation lock is not owned"; return 1; }
        return 0
    fi
    exec 9>>"$LOCK_FILE"
    flock -n 9 || { die "another CloudNVR mutation owns the lock"; return 1; }
}

acquire_service_lock() {
    local action="$1"
    [[ "$SERVICE_STOP_LOCK_TIMEOUT" =~ ^[1-9][0-9]*$ ]] \
        || { die "invalid service lock timeout"; return 1; }
    prepare_lock_file || { die "unsafe CloudNVR mutation lock: $LOCK_FILE"; return 1; }
    exec 9>>"$LOCK_FILE"
    flock -w "$SERVICE_STOP_LOCK_TIMEOUT" 9 \
        || { die "CloudNVR service $action timed out waiting for the mutation lock"; return 1; }
}

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_inactive_verified() {
    local rc
    if systemd_active_status "$1"; then
        return 1
    else
        rc=$?
    fi
    [[ "$rc" -eq 1 ]]
}

# Print why a service failed its health gate, BEFORE the caller rolls back and
# destroys the container. Without this the operator gets only "last state:
# missing", and the actual reason -- which the container logged before it died --
# is gone. A crash-looping container is the common case and the worst one: its
# `ps -q` is empty, so even the container id is unavailable unless we ask for
# stopped ones too.
report_unhealthy() {
    local release_id="$1" service="$2" health="$3" cid
    cid="$(compose_cmd "$release_id" ps -q "$service" 2>/dev/null || true)"
    [[ -n "$cid" ]] || cid="$(compose_cmd "$release_id" ps -aq "$service" 2>/dev/null | head -1 || true)"
    if [[ -z "$cid" ]]; then
        log "$service: no container was created; check the Compose file and image digests"
        return 0
    fi
    local status restarts exitcode
    # Bounded: this runs on the failure path just before rollback, so a wedged
    # docker must not hold the rollback open.
    local dk=(docker); command -v timeout >/dev/null 2>&1 && dk=(timeout 10 docker)
    status="$("${dk[@]}" inspect --format '{{.State.Status}}' "$cid" 2>/dev/null || true)"
    restarts="$("${dk[@]}" inspect --format '{{.RestartCount}}' "$cid" 2>/dev/null || true)"
    exitcode="$("${dk[@]}" inspect --format '{{.State.ExitCode}}' "$cid" 2>/dev/null || true)"
    log "$service diagnostics: state=${status:-?} restarts=${restarts:-?} exit=${exitcode:-?} health=$health"
    # Guard the numeric test: docker inspect can return empty, and a stubbed or
    # unusual docker can return a non-numeric string. `[[ str -gt 3 ]]` evaluates
    # its operands ARITHMETICALLY, so a bare word there is read as a variable name
    # and trips `set -u`, aborting the caller mid-rollback.
    if [[ "$restarts" =~ ^[0-9]+$ ]] && (( restarts > 3 )); then
        log "$service is crash-looping; the lines below are why"
    fi
    log "--- last 40 log lines from $service ---"
    # Diagnostic output, not secret-safe: connector lines can carry POS refs and
    # plate reads. Fine for an installer console, do not ship it anywhere else.
    "${dk[@]}" logs --tail 40 "$cid" 2>&1 | sed "s/^/    /" || true
    log "--- end $service log ---"
}

wait_service() {
    local release_id="$1" service="$2" deadline="$3" started="$SECONDS" health
    while :; do
        health="$(container_health "$release_id" "$service" || true)"
        if [[ "$health" == healthy ]]; then
            log "$service is healthy"
            return 0
        fi
        if (( SECONDS - started >= deadline )); then
            report_unhealthy "$release_id" "$service" "$health"
            die "$service did not become healthy within ${deadline}s (last state: $health)"
            return 1
        fi
        sleep "$HEALTH_POLL"
    done
}

wait_release() {
    local release_id="$1"
    wait_service "$release_id" dragon "$DRAGON_DEADLINE" || return 1
    wait_service "$release_id" connector "$CONNECTOR_DEADLINE" || return 1
}

dragon_activation_ready() {
    local release_id="$1" cid health
    health="$(container_health "$release_id" dragon || true)"
    [[ "$health" == healthy ]] || return 1
    cid="$(compose_cmd "$release_id" ps -q dragon 2>/dev/null || true)"
    [[ -n "$cid" ]] || return 1
    docker exec "$cid" /usr/local/bin/dragon-healthcheck --activation \
        >/dev/null 2>&1
}

wait_dragon_activation() {
    local release_id="$1" started="$SECONDS"
    while :; do
        if dragon_activation_ready "$release_id"; then
            log "dragon activation is ready"
            return 0
        fi
        if (( SECONDS - started >= DRAGON_DEADLINE )); then
            die "dragon did not reach activation readiness within ${DRAGON_DEADLINE}s"
            return 1
        fi
        sleep "$HEALTH_POLL"
    done
}

image_label() {
    docker image inspect --format "{{index .Config.Labels \"$2\"}}" "$1" 2>/dev/null
}

verify_image() {
    local ref="$1" product="$2" expected_source="$3" expected_nvr="${4:-}"
    local architecture title revision profile nvr_revision version
    architecture="$(docker image inspect --format '{{.Architecture}}' "$ref" 2>/dev/null || true)"
    [[ "$architecture" == arm64 ]] || { die "$product image architecture is '$architecture', expected arm64"; return 1; }
    title="$(image_label "$ref" org.opencontainers.image.title || true)"
    revision="$(image_label "$ref" org.opencontainers.image.revision || true)"
    version="$(image_label "$ref" org.opencontainers.image.version || true)"
    [[ "$title" == "$product" ]] || { die "$ref has product label '$title'"; return 1; }
    [[ "$revision" == "$expected_source" ]] || { die "$ref has source revision '$revision'"; return 1; }
    [[ -n "$version" ]] || { die "$ref has no version label"; return 1; }
    if [[ "$product" == "$(flavor_dragon_title)" ]]; then
        [[ "$(image_label "$ref" net.dividia.cloudnvr.platform || true)" == "$(flavor_dragon_platform)" ]] \
            || { die "$ref is not a Jetson $FLAVOR image"; return 1; }
        profile="$(image_label "$ref" net.dividia.cloudnvr.profile || true)"
        [[ "$profile" == production ]] || { die "$ref is not a strict production image"; return 1; }
    else
        nvr_revision="$(image_label "$ref" net.dividia.cloudnvr.nvr-source-rev || true)"
        [[ "$nvr_revision" == "$expected_nvr" ]] \
            || { die "$ref has connector NVR revision '$nvr_revision'"; return 1; }
    fi
}

extract_host_tools() {
    local image="$1" destination="$2" cid="" name rc=0
    mkdir -p "$destination"
    cid="$(docker create "$image")" || { die "cannot create a tool-extraction container"; return 1; }
    for name in docker-compose.yml install-cloudnvr.sh nvr cloudnvr.service; do
        if ! docker cp "$cid:/usr/share/cloudnvr/$name" "$destination/$name" >/dev/null; then
            die "candidate image is missing /usr/share/cloudnvr/$name"
            rc=1
            break
        fi
    done
    if [[ "$rc" -eq 0 && "$FLAVOR" == nano ]]; then
        for name in $NANO_TOOLS security; do
            if ! docker cp "$cid:/usr/share/cloudnvr/$name" "$destination/$name" >/dev/null; then
                die "candidate Nano image is missing /usr/share/cloudnvr/$name"
                rc=1
                break
            fi
        done
    fi
    docker rm -f "$cid" >/dev/null 2>&1 || true
    [[ "$rc" -eq 0 ]] || return "$rc"
    for name in docker-compose.yml install-cloudnvr.sh nvr cloudnvr.service; do
        [[ -f "$destination/$name" && ! -L "$destination/$name" && -s "$destination/$name" ]] \
            || { die "invalid extracted host tool: $name"; return 1; }
    done
    if [[ "$FLAVOR" == nano ]]; then
        chmod 0755 "$destination/compose" || return 1
        release_compose_valid "$destination" || { die "candidate Compose binary does not match its checksum"; return 1; }
        [[ -d "$destination/security" && ! -L "$destination/security" ]] \
            || { die "candidate Nano image has no security profiles"; return 1; }
        for name in "$destination"/security/* "$destination"/security/.[!.]*; do
            [[ -e "$name" || -L "$name" ]] || continue
            [[ "${name##*/}" =~ ^[a-z0-9-]+\.json$ && -f "$name" && ! -L "$name" ]] \
                || { die "candidate Nano image carries an unexpected security entry: ${name##*/}"; return 1; }
        done
    fi
    bash -n "$destination/nvr" || { die "candidate nvr script failed bash -n"; return 1; }
    bash -n "$destination/install-cloudnvr.sh" \
        || { die "candidate installer failed bash -n"; return 1; }
}

write_release_env() {
    local path="$1" dragon_digest="$2" connector_digest="$3"
    {
        printf 'DRAGON_DIGEST=%s\n' "$dragon_digest"
        printf 'CONNECTOR_DIGEST=%s\n' "$connector_digest"
        if [[ -n "${MANIFEST_REMOTED_DIGEST:-}" ]]; then printf 'REMOTED_DIGEST=%s\n' "$MANIFEST_REMOTED_DIGEST"; fi
        printf 'COMPOSE_PROJECT_NAME=%s\n' "$COMPOSE_PROJECT_NAME"
    } | atomic_write "$path" 0600
}

activation_compose_body() {
    printf '%s\n' 'services:' '  dragon:' '    restart: "no"' '  connector:' '    restart: "no"'
    if [[ "${MANIFEST_SCHEMA:-1}" == 2 ]]; then printf '%s\n' '  remoted:' '    restart: "no"'; fi
}

write_activation_compose() {
    activation_compose_body | atomic_write "$1" 0644
}

validate_compose() {
    local tools="$1" env_file="$2" images services rendered expected_services expected_images
    install_security_profiles "$tools" || return 1
    expected_services=$'connector\ndragon'
    expected_images="$MANIFEST_CONNECTOR_REF"$'\n'"$MANIFEST_DRAGON_REF"
    if [[ "${MANIFEST_SCHEMA:-1}" == 2 ]]; then
        expected_services+=$'\nremoted'
        expected_images+=$'\n'"$MANIFEST_REMOTED_REF"
    fi
    services="$(BACKEND_ADDR="$BACKEND_ADDR" run_compose "$tools" --project-directory "$ROOT" --env-file "$ROOT/.env" \
        --env-file "$env_file" \
        -f "$tools/docker-compose.yml" config --services 2>/dev/null | sort)" \
        || { die "candidate Compose file does not render"; return 1; }
    [[ "$services" == "$expected_services" ]] \
        || { die "candidate Compose services must match the manifest"; return 1; }
    images="$(BACKEND_ADDR="$BACKEND_ADDR" run_compose "$tools" --project-directory "$ROOT" --env-file "$ROOT/.env" \
        --env-file "$env_file" \
        -f "$tools/docker-compose.yml" config --images 2>/dev/null | sort -u)" \
        || { die "candidate Compose image list failed"; return 1; }
    [[ "$images" == "$expected_images" ]] \
        || { die "candidate Compose does not use the manifest digests"; return 1; }
    rendered="$(BACKEND_ADDR="$BACKEND_ADDR" run_compose "$tools" --project-directory "$ROOT" --env-file "$ROOT/.env" \
        --env-file "$env_file" \
        -f "$tools/docker-compose.yml" config 2>/dev/null)" \
        || { die "candidate Compose file does not render"; return 1; }
    if grep -Eq '^[[:space:]]+build:' <<< "$rendered"; then
        die "candidate Compose contains a build directive"
        return 1
    fi
}

prepare_release() {
    local manifest="$1" tools="$2" target staging name expected_activation
    target="$RELEASES_DIR/$MANIFEST_RELEASE_ID"
    if [[ -e "$target" ]]; then
        [[ -d "$target" && ! -L "$target" && "$(file_mode "$target")" == 700 ]] \
            || { die "release target is not a private directory: $target"; return 1; }
        cmp -s "$manifest" "$target/release.json" || {
            die "installed release metadata does not match candidate $MANIFEST_RELEASE_ID"
            return 1
        }
        for name in release.json docker-compose.yml activation-compose.yml cloudnvr.service; do
            [[ -f "$target/$name" && ! -L "$target/$name" \
               && "$(file_mode "$target/$name")" == 644 ]] \
                || { die "installed $name has an invalid type or mode"; return 1; }
        done
        [[ -f "$target/release.env" && ! -L "$target/release.env" \
           && "$(file_mode "$target/release.env")" == 600 ]] \
            || { die "installed release.env has an invalid type or mode"; return 1; }
        for name in install-cloudnvr.sh nvr; do
            [[ -f "$target/$name" && ! -L "$target/$name" \
               && "$(file_mode "$target/$name")" == 755 ]] \
                || { die "installed $name has an invalid type or mode"; return 1; }
        done
        for name in docker-compose.yml install-cloudnvr.sh nvr cloudnvr.service; do
            cmp -s "$tools/$name" "$target/$name" \
                || { die "installed $name differs from candidate $MANIFEST_RELEASE_ID"; return 1; }
        done
        if [[ "$FLAVOR" == nano ]]; then
            ensure_release_compose "$target" || return 1
            cmp -s "$tools/compose.sha256" "$target/compose.sha256" \
                || { die "installed Compose binary differs from candidate $MANIFEST_RELEASE_ID"; return 1; }
            diff -r "$tools/security" "$target/security" >/dev/null 2>&1 \
                || { die "installed security profiles differ from candidate $MANIFEST_RELEASE_ID"; return 1; }
        fi
        expected_activation="$(activation_compose_body)"
        [[ -f "$target/activation-compose.yml" \
           && "$(cat "$target/activation-compose.yml")" == "$expected_activation" ]] \
            || { die "installed activation Compose file differs from the release contract"; return 1; }
        python3 - "$target/release.env" "$MANIFEST_DRAGON_DIGEST" \
            "$MANIFEST_CONNECTOR_DIGEST" "$COMPOSE_PROJECT_NAME" "${MANIFEST_REMOTED_DIGEST:-}" <<'PY' || {
import pathlib, sys
path, dragon, connector, project, remoted = sys.argv[1:]
expected = (
    f"DRAGON_DIGEST={dragon}\n"
    f"CONNECTOR_DIGEST={connector}\n"
    + (f"REMOTED_DIGEST={remoted}\n" if remoted else "")
    + f"COMPOSE_PROJECT_NAME={project}\n"
).encode()
if pathlib.Path(path).read_bytes() != expected:
    raise SystemExit("installed release.env differs from the release contract")
PY
            die "installed release.env differs from candidate $MANIFEST_RELEASE_ID"
            return 1
        }
        manifest_load "$target/release.json" || return 1
        [[ "$MANIFEST_RELEASE_ID" == "${target##*/}" ]] || return 1
        validate_compose "$target" "$target/release.env"
        return
    fi

    staging="$RELEASES_DIR/.staging-$MANIFEST_RELEASE_ID-$$"
    [[ ! -e "$staging" ]] || { die "staging path already exists: $staging"; return 1; }
    mkdir -m 0700 "$staging" || return 1
    atomic_copy "$manifest" "$staging/release.json" 0644 || return 1
    atomic_copy "$tools/docker-compose.yml" "$staging/docker-compose.yml" 0644 || return 1
    atomic_copy "$tools/install-cloudnvr.sh" "$staging/install-cloudnvr.sh" 0755 || return 1
    atomic_copy "$tools/nvr" "$staging/nvr" 0755 || return 1
    atomic_copy "$tools/cloudnvr.service" "$staging/cloudnvr.service" 0644 || return 1
    if [[ "$FLAVOR" == nano ]]; then
        atomic_copy "$tools/compose" "$staging/compose" 0755 || return 1
        atomic_copy "$tools/compose.sha256" "$staging/compose.sha256" 0644 || return 1
        mkdir -m 0755 "$staging/security" || return 1
        for name in "$tools"/security/*.json; do
            [[ -e "$name" ]] || continue
            atomic_copy "$name" "$staging/security/${name##*/}" 0644 || return 1
        done
    fi
    write_activation_compose "$staging/activation-compose.yml" || return 1
    write_release_env "$staging/release.env" "$MANIFEST_DRAGON_DIGEST" "$MANIFEST_CONNECTOR_DIGEST" \
        || return 1
    validate_compose "$staging" "$staging/release.env" || return 1
    fsync_tree "$staging" || return 1
    python3 -c '
import os, sys
source, target, parent = sys.argv[1:]
os.replace(source, target)
fd = os.open(parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
try: os.fsync(fd)
finally: os.close(fd)
' "$staging" "$target" "$RELEASES_DIR" || return 1
}

cron_minute() {
    local seed
    [[ "${CLOUDNVR_TEST_CRON_MINUTE_FAIL:-0}" != 1 ]] || return 1
    if [[ -s "${CLOUDNVR_DEVICE_ID_FILE:-/etc/machine-id}" ]]; then
        seed="$(tr -d '[:space:]' < "${CLOUDNVR_DEVICE_ID_FILE:-/etc/machine-id}")"
    else
        seed="$(hostname 2>/dev/null || printf cloudnvr)"
    fi
    printf '%s' "$seed" | cksum | awk '{print $1 % 60}'
}

# ADR-023: cron runs in the host zone (UTC on a stock install) and has no
# CRON_TZ, and the cloud pushes the device zone to Dragon, not to the host. So
# the job wakes every 15 minutes (every current zone offset is a multiple of
# 15) and runs only inside its 15-minute window at 02:MM local time, reading
# the zone when it fires: a zone change needs no rewrite, DST needs no rewrite,
# and the host zone applies until the cloud pushes one. The job only ever runs
# `nvr update`, so any release's CLI can execute it, including after a
# rollback to one older than this format.
ensure_cron() {
    [[ "${CLOUDNVR_SKIP_HOST_ARTIFACTS:-0}" != 1 ]] || return 0
    local minute base start end body
    minute="$(cron_minute)" || return 1
    base=$((minute % 15))
    # The leading 1 keeps 02MM a plain decimal for test(1).
    start="$(printf '102%02d' "$minute")"
    if ((minute < 45)); then end="$(printf '102%02d' $((minute + 15)))"
    else end="$(printf '103%02d' $((minute - 45)))"; fi
    body="# Daily CloudNVR update at 02:$(printf '%02d' "$minute") local time (ADR-023). Managed by /opt/dividia/bin/nvr.
# Local time is the zone the cloud pushed to Dragon, else the host zone.
SHELL=/bin/sh
PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
$base,$((base + 15)),$((base + 30)),$((base + 45)) * * * * root z=$UPDATE_ZONE_FILE; [ -e \"\$z\" ] || z=$HOST_LOCALTIME; t=\"\$(TZ=\":\$z\" date +1\\%H\\%M)\"; [ \"\$t\" -ge $start ] && [ \"\$t\" -lt $end ] || exit 0; exec $CMD_LINK update >>$ROOT/data/logs/update.log 2>&1
"
    printf '%s' "$body" | atomic_write "$CRON_FILE" 0644
}

# zone_name LINK: the IANA name a localtime symlink points at, if it is one.
zone_name() {
    local target
    target="$(readlink "$1" 2>/dev/null)" || return 1
    [[ "$target" == */zoneinfo/* ]] || return 1
    printf '%s\n' "${target#*/zoneinfo/}"
}

# update_schedule: the `nvr status` phrase for the installed daily job.
update_schedule() {
    local at zone
    [[ -f "$CRON_FILE" ]] || { printf 'not installed\n'; return 0; }
    at="$(sed -n 's/.* -ge 10\([0-9]\)\([0-5][0-9]\) .*/0\1:\2/p' "$CRON_FILE" | head -n 1)"
    if [[ -z "$at" ]]; then
        # A release before ADR-023 wrote a plain 02:MM line in the host zone. The
        # next update this CLI runs rewrites it.
        at="$(awk '$1 ~ /^[0-9]+$/ && $2 == 2 && / update( |$)/ {printf "02:%02d", $1; exit}' "$CRON_FILE")"
        [[ -n "$at" ]] || { printf 'installed but invalid\n'; return 0; }
        zone="$(zone_name "$HOST_LOCALTIME" || printf 'UTC')"
        printf 'daily at %s %s (host timezone; the next update moves it to the cloud timezone)\n' "$at" "$zone"
    elif [[ -e "$UPDATE_ZONE_FILE" ]]; then
        zone="$(zone_name "$UPDATE_ZONE_FILE" || printf 'unknown')"
        printf 'daily at %s %s (cloud timezone)\n' "$at" "$zone"
    else
        zone="$(zone_name "$HOST_LOCALTIME" || printf 'UTC')"
        printf 'daily at %s %s (host timezone; no cloud timezone yet)\n' "$at" "$zone"
    fi
}

prepare_host_dispatch() {
    local release_id="$1" tools="$2" current=""
    mkdir -p "$BIN_DIR" "$ROOT/data/logs" || return 1
    durable_sync_dirs "$BIN_DIR" "$ROOT/data/logs" || return 1
    [[ -s "$RELEASES_DIR/$release_id/nvr" \
       && -s "$RELEASES_DIR/$release_id/install-cloudnvr.sh" \
       && -s "$RELEASES_DIR/$release_id/cloudnvr.service" ]] \
        || { die "release $release_id is missing its host tools"; return 1; }
    current="$(current_release_id 2>/dev/null || true)"
    if [[ -n "$current" ]]; then
        [[ -s "$RELEASES_DIR/$current/nvr" \
           && -s "$RELEASES_DIR/$current/install-cloudnvr.sh" \
           && -s "$RELEASES_DIR/$current/cloudnvr.service" ]] \
            || { die "current release $current is missing its host tools"; return 1; }
        atomic_symlink "$BIN_DIR" nvr ../releases/current/nvr || return 1
        atomic_symlink "$BIN_DIR" install-cloudnvr.sh ../releases/current/install-cloudnvr.sh || return 1
    else
        # The first install has no current pointer yet. Keep complete matching
        # bootstrap tools available until the pointer commits.
        atomic_copy "$tools/nvr" "$BIN_DIR/nvr" 0755 || return 1
        atomic_copy "$tools/install-cloudnvr.sh" "$BIN_DIR/install-cloudnvr.sh" 0755 || return 1
    fi
    atomic_symlink "$ROOT" nvr bin/nvr || return 1
    atomic_symlink "$ROOT" docker-compose.yml releases/current/docker-compose.yml || return 1
    if [[ "${CLOUDNVR_SKIP_HOST_ARTIFACTS:-0}" != 1 ]]; then
        if [[ -n "$current" ]]; then
            atomic_symlink "${SERVICE_FILE%/*}" "${SERVICE_FILE##*/}" \
                "$RELEASES_DIR/current/cloudnvr.service" || return 1
        else
            atomic_copy "$tools/cloudnvr.service" "$SERVICE_FILE" 0644 || return 1
        fi
        install_operator_command || return 1
        ensure_cron || return 1
        systemctl daemon-reload || return 1
        systemctl enable cloudnvr.service >/dev/null || return 1
        systemctl is-enabled --quiet cloudnvr.service || return 1
    fi
}

commit_host_dispatch() {
    atomic_symlink "$BIN_DIR" nvr ../releases/current/nvr || return 1
    atomic_symlink "$BIN_DIR" install-cloudnvr.sh ../releases/current/install-cloudnvr.sh || return 1
    if [[ "${CLOUDNVR_SKIP_HOST_ARTIFACTS:-0}" != 1 ]]; then
        atomic_symlink "${SERVICE_FILE%/*}" "${SERVICE_FILE##*/}" \
            "$RELEASES_DIR/current/cloudnvr.service" || return 1
        systemctl daemon-reload || return 1
        systemctl enable cloudnvr.service >/dev/null || return 1
        systemctl is-enabled --quiet cloudnvr.service || return 1
    fi
}

record_release_parent() {
    local release_id="$1" parent="$2"
    local path="$STATE_DIR/release-parents/$release_id"
    [[ "${CLOUDNVR_TEST_RELEASE_PARENT_FAIL:-}" != "$release_id" ]] || return 1
    release_id_is_valid "$release_id" || return 1
    [[ -z "$parent" ]] || release_id_is_valid "$parent" || return 1
    mkdir -p "${path%/*}" || return 1
    durable_sync_dirs "${path%/*}" || return 1
    printf '%s\n' "$parent" | atomic_write "$path" 0644
}

release_parent_id() {
    local release_id="$1" parent
    release_id_is_valid "$release_id" || return 1
    parent="$(head -n 1 "$STATE_DIR/release-parents/$release_id" 2>/dev/null || true)"
    release_id_is_valid "$parent" || return 1
    printf '%s\n' "$parent"
}

json_state() {
    local file="$1" result="$2" action="$3" release_id="$4" stage="$5" reason="$6"
    [[ "${CLOUDNVR_TEST_STATE_WRITE_FAIL:-}" != "$result" ]] || return 1
    python3 -c '
import json, sys
from datetime import datetime, timezone
value = {
  "schema": 1, "result": sys.argv[2], "action": sys.argv[3],
  "release_id": sys.argv[4] or None, "stage": sys.argv[5] or None,
  "reason": sys.argv[6] or None,
  "timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
}
json.dump(value, sys.stdout, sort_keys=True, indent=2)
sys.stdout.write("\n")
' "$file" "$result" "$action" "$release_id" "$stage" "$reason" | atomic_write "$file" 0644
}

mark_rejected() {
    local release_id="$1" stage="$2" reason="$3"
    json_state "$STATE_DIR/rejected-release.json" rejected update "$release_id" "$stage" "$reason"
}

rejected_release_id() {
    [[ -f "$STATE_DIR/rejected-release.json" ]] || return 1
    python3 -c 'import json,sys; print(json.load(open(sys.argv[1])).get("release_id") or "")' \
        "$STATE_DIR/rejected-release.json" 2>/dev/null
}

release_was_committed() {
    local wanted="$1"
    [[ -f "$STATE_DIR/accepted-releases/$wanted" ]]
}

mark_release_accepted() {
    local release_id="$1"
    release_id_is_valid "$release_id" || return 1
    mkdir -p "$STATE_DIR/accepted-releases" || return 1
    durable_sync_dirs "$STATE_DIR/accepted-releases" || return 1
    printf 'accepted=%s\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" \
        | atomic_write "$STATE_DIR/accepted-releases/$release_id" 0644
}

clear_rejected() {
    local rejected
    [[ "${CLOUDNVR_TEST_CLEAR_REJECTED_FAIL:-0}" != 1 ]] || return 1
    rejected="$(rejected_release_id 2>/dev/null || true)"
    if [[ -n "$rejected" && "$rejected" == "$1" ]]; then
        rm -f "$STATE_DIR/rejected-release.json" || return 1
        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)' "$STATE_DIR"
    fi
}

set_hold() {
    local reason="${1:-operator}"
    [[ "${CLOUDNVR_TEST_HOLD_FAIL:-0}" != 1 ]] || return 1
    python3 -c '
import json,sys
from datetime import datetime, timezone
json.dump({"schema":1,"enabled":True,"reason":sys.argv[1],"timestamp":datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")}, sys.stdout, sort_keys=True, indent=2)
sys.stdout.write("\n")
' "$reason" | atomic_write "$STATE_DIR/update-hold" 0644
}

disable_update_cron() {
    [[ -e "$CRON_FILE" || -L "$CRON_FILE" ]] || return 0
    rm -f "$CRON_FILE" || return 1
    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)' \
        "${CRON_FILE%/*}"
}

reject_release() {
    local release_id="$1" stage="$2" reason="$3"
    if mark_rejected "$release_id" "$stage" "$reason"; then
        return 0
    fi
    warn "could not record rejected release $release_id; enabling the update hold"
    if ! set_hold rejection-state-failure; then
        warn "could not enable the update hold after rejection-state failure; removing automatic update cron"
        if ! disable_update_cron; then
            warn "could not remove automatic update cron after rejection-state failure"
        fi
    fi
    return 1
}

clear_hold() {
    rm -f "$STATE_DIR/update-hold"
    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)' "$STATE_DIR"
}

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

restore_prior_runtime() {
    local prior="$1"
    RESTORE_RUNTIME_ALIGNED=false
    if [[ -n "$prior" ]]; then
        log "Restoring prior release $prior"
        compose_cmd "$prior" up -d dragon connector || return 1
        RESTORE_RUNTIME_ALIGNED=true
        wait_release "$prior" && wait_dragon_activation "$prior"
    else
        BACKEND_ADDR="$BACKEND_ADDR" run_compose "$2" --project-directory "$ROOT" --env-file "$ROOT/.env" \
            --env-file "$2/release.env" \
            -f "$2/docker-compose.yml" rm -s -f dragon connector >/dev/null 2>&1 || return 1
        RESTORE_RUNTIME_ALIGNED=true
    fi
}

restore_prior_or_stop_candidate() {
    local prior="$1" candidate_dir="$2"
    if restore_prior_runtime "$prior" "$candidate_dir"; then
        return 0
    fi
    if [[ "$RESTORE_RUNTIME_ALIGNED" != true ]]; then
        warn "prior runtime could not start; stopping the candidate to preserve the current pointer"
        BACKEND_ADDR="$BACKEND_ADDR" run_compose "$candidate_dir" --project-directory "$ROOT" --env-file "$ROOT/.env" \
            --env-file "$candidate_dir/release.env" \
            -f "$candidate_dir/docker-compose.yml" stop dragon connector >/dev/null 2>&1 \
            || warn "candidate stop also failed; runtime state needs operator repair"
    fi
    if ! set_hold runtime-restoration-failure; then
        warn "could not enable the update hold after runtime restoration failed"
        disable_update_cron || warn "could not remove automatic update cron"
    fi
    return 1
}

activate_manifest() {
    local manifest="$1" tools="$2" action="$3" prior="" candidate_dir stage reason
    local pointer_rc=0 observed_current=""
    local postcommit_ok=true postcommit_stage="" postcommit_reason=""
    local candidate_id="$MANIFEST_RELEASE_ID"
    prior="$(current_release_id 2>/dev/null || true)"
    if ! prepare_release "$manifest" "$tools" "$prior"; then
        json_state "$STATE_DIR/last-update.json" failed "$action" "$candidate_id" prepare \
            "candidate release staging failed" || warn "could not record update failure state"
        return 1
    fi
    candidate_dir="$RELEASES_DIR/$candidate_id"

    if [[ "$candidate_id" == "$prior" ]]; then
        remoted_reconcile "$candidate_id"
        if ! release_services_present "$candidate_id"; then
            log "Restarting missing current-release containers with the activation policy"
            if ! candidate_compose_cmd "$candidate_id" up -d dragon connector; then
                json_state "$STATE_DIR/last-update.json" failed "$action" "$candidate_id" compose-start \
                    "current release containers could not be restarted" || true
                return 1
            fi
        fi
        if ! wait_release "$candidate_id"; then
            json_state "$STATE_DIR/last-update.json" failed "$action" "$candidate_id" health \
                "current release is unhealthy" || true
            return 1
        fi
        if ! wait_dragon_activation "$candidate_id"; then
            json_state "$STATE_DIR/last-update.json" failed "$action" "$candidate_id" activation \
                "current release did not reach activation readiness" || true
            return 1
        fi
        if ! prepare_host_dispatch "$candidate_id" "$tools"; then
            json_state "$STATE_DIR/last-update.json" failed "$action" "$candidate_id" host-tools \
                "host tool convergence failed" || warn "could not record update failure state"
            return 1
        fi
        commit_host_dispatch || { die "host tool activation failed"; return 1; }
        compose_cmd "$candidate_id" up -d || { die "restart-policy convergence failed"; return 1; }
        mark_release_accepted "$candidate_id" || { die "accepted-release evidence could not be recorded"; return 1; }
        clear_rejected "$candidate_id" \
            || warn "current release is healthy, but rejected-release state could not be cleared"
        json_state "$STATE_DIR/last-update.json" current "$action" "$candidate_id" current "already current" \
            || warn "could not record current release state"
        log "Release $candidate_id is already current"
        return 0
    fi

    if [[ -n "$prior" ]] && ! mark_release_accepted "$prior"; then
        die "cannot preserve accepted-release evidence for current release $prior"
        return 1
    fi

    remoted_reconcile "$candidate_id" "$prior"
    log "Starting candidate release $candidate_id"
    if ! candidate_compose_cmd "$candidate_id" up -d dragon connector; then
        stage=compose-start
        reason="candidate Compose start failed"
    elif ! wait_release "$candidate_id"; then
        stage=health
        reason="candidate health check failed"
    elif ! wait_dragon_activation "$candidate_id"; then
        stage=activation
        reason="candidate activation readiness failed"
    else
        stage=""
        reason=""
    fi

    if [[ -n "$stage" ]]; then
        if ! restore_prior_or_stop_candidate "$prior" "$candidate_dir"; then
            reason="$reason; prior release health restoration failed"
        fi
        reject_release "$candidate_id" "$stage" "$reason" || true
        json_state "$STATE_DIR/last-update.json" failed "$action" "$candidate_id" "$stage" "$reason" \
            || warn "could not record update failure state"
        die "$reason"
        return 1
    fi

    if ! record_release_parent "$candidate_id" "$prior"; then
        reason="rollback lineage commit failed"
        if ! restore_prior_or_stop_candidate "$prior" "$candidate_dir"; then
            reason="$reason; prior runtime restoration failed"
        fi
        reject_release "$candidate_id" lineage "rollback lineage commit failed" || true
        json_state "$STATE_DIR/last-update.json" failed "$action" "$candidate_id" lineage \
            "$reason" || warn "could not record update failure state"
        die "$reason"
        return 1
    fi
    maybe_power_loss before-current
    if ! prepare_host_dispatch "$candidate_id" "$tools"; then
        reason="host tool convergence failed"
        if ! restore_prior_or_stop_candidate "$prior" "$candidate_dir"; then
            reason="$reason; prior runtime restoration failed"
        fi
        reject_release "$candidate_id" host-tools "$reason" || true
        json_state "$STATE_DIR/last-update.json" failed "$action" "$candidate_id" host-tools \
            "$reason" || warn "could not record update failure state"
        die "$reason"
        return 1
    fi
    atomic_symlink "$RELEASES_DIR" current "$candidate_id" || pointer_rc=$?
    if [[ "$pointer_rc" -ne 0 ]]; then
        observed_current="$(current_release_id 2>/dev/null || true)"
    fi
    if [[ "$pointer_rc" -ne 0 && "$observed_current" != "$candidate_id" ]]; then
        reason="current pointer commit failed"
        if ! restore_prior_or_stop_candidate "$prior" "$candidate_dir"; then
            reason="$reason; prior runtime restoration failed"
        fi
        reject_release "$candidate_id" pointer "$reason" || true
        json_state "$STATE_DIR/last-update.json" failed "$action" "$candidate_id" pointer \
            "$reason" || warn "could not record update failure state"
        die "$reason"
        return 1
    fi
    if [[ "$pointer_rc" -ne 0 ]]; then
        postcommit_ok=false
        postcommit_stage=pointer-sync
        postcommit_reason="release pointer changed, but its directory sync failed"
        warn "$postcommit_reason"
    fi
    maybe_power_loss after-current
    if ! commit_host_dispatch; then
        if [[ "$postcommit_ok" == true ]]; then
            postcommit_ok=false
            postcommit_stage=host-tools
            postcommit_reason="release committed, but host service metadata refresh failed"
        fi
        warn "release committed, but host service metadata could not be refreshed"
    fi
    if ! compose_cmd "$candidate_id" up -d; then
        if [[ "$postcommit_ok" == true ]]; then
            postcommit_ok=false
            postcommit_stage=restart-policy
            postcommit_reason="release committed, but normal container restart policy could not be restored"
        fi
        warn "release committed, but normal container restart policy could not be restored"
    fi
    if ! mark_release_accepted "$candidate_id"; then
        if [[ "$postcommit_ok" == true ]]; then
            postcommit_ok=false
            postcommit_stage=accepted
            postcommit_reason="release committed, but accepted-release evidence could not be recorded"
        fi
        warn "release committed, but accepted-release evidence could not be recorded"
    fi
    if [[ "$postcommit_ok" == true ]]; then
        json_state "$STATE_DIR/last-update.json" success "$action" "$candidate_id" activated "" \
            || warn "release committed, but success state could not be recorded"
    else
        json_state "$STATE_DIR/last-update.json" degraded "$action" "$candidate_id" "$postcommit_stage" \
            "$postcommit_reason" \
            || warn "release committed, but degraded state could not be recorded"
    fi
    clear_rejected "$candidate_id" \
        || warn "release committed, but rejected-release state could not be cleared"
    log "Activated release $candidate_id"
    [[ "$postcommit_ok" == true ]] || return 2
}

# fetch_url URL OUTPUT: curl where installed; an iSite Jetson Nano ships without
# it (ADR-021), so otherwise python3's urllib under the same rules: an https
# URL may only redirect to https, an http URL to http or https.
fetch_url() {
    local url="$1" output="$2"
    case "$url" in
        https://*|http://*) ;;
        *) die "unsupported URL protocol: $url"; return 1 ;;
    esac
    if [[ "${CLOUDNVR_FETCH_TOOL:-}" != python ]] && command -v curl >/dev/null 2>&1; then
        if [[ "$url" == https://* ]]; then
            curl -fsSL --proto '=https' --proto-redir '=https' --connect-timeout 10 --max-time 60 "$url" -o "$output"
        else
            curl -fsSL --proto '=http,https' --proto-redir '=http,https' --connect-timeout 10 --max-time 60 "$url" -o "$output"
        fi
        return
    fi
    python3 - "$url" "$output" "${CLOUDNVR_FETCH_MAX_TIME:-60}" <<'PY'
import math, os, signal, sys, urllib.error, urllib.request
url, output, max_time = sys.argv[1], sys.argv[2], float(sys.argv[3])
secure = url.startswith("https://")
limit = 64 * 1024 * 1024
def fail(message):
    sys.stderr.write("download failed: %s: %s\n" % (url, message))
    raise SystemExit(22)
def expired(signum, frame):
    fail("exceeded %ds" % max_time)
# curl --max-time is a total deadline; a blocking read never sees a clock check.
signal.signal(signal.SIGALRM, expired)
signal.alarm(max(1, int(math.ceil(max_time))))
class Redirects(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        if not (newurl.startswith("https://") or (not secure and newurl.startswith("http://"))):
            raise urllib.error.HTTPError(newurl, code, "refused redirect", headers, fp)
        return urllib.request.HTTPRedirectHandler.redirect_request(self, req, fp, code, msg, headers, newurl)
tmp = output + ".part"
try:
    chunks, size = [], 0
    with urllib.request.build_opener(Redirects).open(url, timeout=10) as response:
        while True:
            chunk = response.read(65536)
            if not chunk:
                break
            size += len(chunk)
            if size > limit:
                fail("larger than %d bytes" % limit)
            chunks.append(chunk)
    with open(tmp, "wb") as stream:
        stream.write(b"".join(chunks))
    os.replace(tmp, output)
except SystemExit:
    raise
except Exception as error:
    fail(error)
finally:
    signal.alarm(0)
    if os.path.exists(tmp):
        os.unlink(tmp)
PY
}

download_candidate_manifest() {
    local output="$1" url
    if [[ -n "${CLOUDNVR_CANDIDATE_MANIFEST:-}" ]]; then
        atomic_copy "$CLOUDNVR_CANDIDATE_MANIFEST" "$output" 0644
        return
    fi
    url="${BUNDLE_BASE_URL%/}/$CHANNEL/release.json"
    # release.json is only a pointer, so the transport is not what protects it.
    # manifest_read() hardcodes the docker.io/dividia/cloudnvr-* repositories,
    # requires an explicit sha256 digest and reference == repo@digest, Docker
    # pulls those digests from Hub over its own TLS, and verify_image() rechecks
    # the labels baked into the image. Do not reintroduce an HTTPS requirement
    # here: it bought nothing the digests do not, and it is what put a private
    # bench CA on files.dividia.net that no clean appliance trusts.
    case "$url" in
        https://*|http://*) ;;
        *) die "release manifest URL uses an unsupported protocol: $url"; return 1 ;;
    esac
    fetch_url "$url" "$output"
}

pull_candidate_images() {
    log "Pulling candidate images while the current release stays online"
    docker pull "$MANIFEST_DRAGON_REF" || { die "Dragon image pull failed"; return 1; }
    docker pull "$MANIFEST_CONNECTOR_REF" || { die "connector image pull failed"; return 1; }
    if [[ -n "${MANIFEST_REMOTED_REF:-}" ]]; then docker pull "$MANIFEST_REMOTED_REF" || return 1; fi
}

verify_candidate_images() {
    verify_image "$MANIFEST_DRAGON_REF" "$(flavor_dragon_title)" "$MANIFEST_SOURCE_REV" || return 1
    verify_image "$MANIFEST_CONNECTOR_REF" cloudnvr-connector "$MANIFEST_SOURCE_REV" "$MANIFEST_NVR_REV" || return 1
    if [[ -n "${MANIFEST_REMOTED_REF:-}" ]]; then
        verify_image "$MANIFEST_REMOTED_REF" cloudnvr-remoted "$MANIFEST_SOURCE_REV" "$MANIFEST_REMOTED_NVR_REV" || return 1
    fi
}

ensure_rollback_images() {
    if ! docker image inspect "$MANIFEST_DRAGON_REF" >/dev/null 2>&1; then
        docker pull "$MANIFEST_DRAGON_REF" || { die "Dragon rollback image is unavailable"; return 1; }
    fi
    if ! docker image inspect "$MANIFEST_CONNECTOR_REF" >/dev/null 2>&1; then
        docker pull "$MANIFEST_CONNECTOR_REF" || { die "connector rollback image is unavailable"; return 1; }
    fi
    if [[ -n "${MANIFEST_REMOTED_REF:-}" ]] && ! docker image inspect "$MANIFEST_REMOTED_REF" >/dev/null 2>&1; then
        docker pull "$MANIFEST_REMOTED_REF" || return 1
    fi
    verify_candidate_images
}

cleanup_staging() {
    local path
    shopt -s nullglob
    for path in "$RELEASES_DIR"/.staging-*; do
        [[ -d "$path" && ! -L "$path" ]] || continue
        rm -rf -- "$path"
    done
    shopt -u nullglob
}

prune_old_releases() {
    [[ "${CLOUDNVR_SKIP_PRUNE:-0}" != 1 ]] || return 0
    [[ "${CLOUDNVR_TEST_RELEASE_PRUNE_FAIL:-0}" != 1 ]] || return 1
    local current parent="" rejected=""
    current="$(current_release_id 2>/dev/null || true)"
    if [[ -n "$current" ]]; then
        parent="$(release_parent_id "$current" 2>/dev/null || true)"
    fi
    rejected="$(rejected_release_id 2>/dev/null || true)"
    python3 - "$RELEASES_DIR" "$current" "$parent" "$rejected" <<'PY' || return 1
import re, shutil, sys, time
from pathlib import Path
root, current, parent, rejected = Path(sys.argv[1]), sys.argv[2], sys.argv[3], sys.argv[4]
dirs = [p for p in root.iterdir()
        if p.is_dir() and not p.is_symlink() and re.fullmatch(r"[0-9a-f]{64}", p.name)]
dirs.sort(key=lambda p: p.stat().st_mtime, reverse=True)
cutoff = time.time() - 7 * 86400
for index, path in enumerate(dirs):
    if index < 2 or path.name in (current, parent, rejected) or path.stat().st_mtime >= cutoff:
        continue
    shutil.rmtree(path)
PY
    prune_product_images
}

prune_product_images() {
    local path ref id keep_ids="" image_ids image_id created old
    shopt -s nullglob
    for path in "$RELEASES_DIR"/[0-9a-f]*; do
        [[ -d "$path" && ! -L "$path" && "${path##*/}" =~ ^[0-9a-f]{64}$ ]] || continue
        while IFS= read -r ref; do
            [[ -n "$ref" ]] || continue
            while IFS= read -r image_id; do
                [[ -n "$image_id" ]] && keep_ids+="$image_id"$'\n'
            done < <(docker image inspect --format '{{.Id}}' "$ref" 2>/dev/null || true)
        done < <(python3 - "$path/release.json" <<'PY' 2>/dev/null || true
import json, sys
value=json.load(open(sys.argv[1], encoding="utf-8"))
print(value["images"]["dragon"]["reference"])
print(value["images"]["connector"]["reference"])
if "remoted" in value["images"]: print(value["images"]["remoted"]["reference"])
PY
)
    done
    shopt -u nullglob
    ref="$(remoted_running_image 2>/dev/null || true)"
    if [[ -n "$ref" ]]; then
        keep_ids+="$(docker image inspect --format '{{.Id}}' "$ref" 2>/dev/null || true)"$'\n'
    fi
    image_ids="$({
        docker image ls --no-trunc --filter "label=org.opencontainers.image.title=$(flavor_dragon_title)" --format '{{.ID}}' 2>/dev/null || true
        docker image ls --no-trunc --filter label=org.opencontainers.image.title=cloudnvr-connector --format '{{.ID}}' 2>/dev/null || true
        docker image ls --no-trunc --filter label=org.opencontainers.image.title=cloudnvr-remoted --format '{{.ID}}' 2>/dev/null || true
    } | sort -u)"
    while IFS= read -r id; do
        [[ -n "$id" ]] || continue
        grep -qxF "$id" <<< "$keep_ids" && continue
        created="$(docker image inspect --format '{{.Created}}' "$id" 2>/dev/null || true)"
        [[ -n "$created" ]] || continue
        old="$(python3 - "$created" <<'PY' 2>/dev/null || printf no
# Python 3.6 (the Nano) has no datetime.fromisoformat, and none before 3.11
# accepts the nine fractional digits Docker writes, so parse RFC 3339 by hand.
import calendar, re, sys, time
m = re.fullmatch(r"(\d{4})-(\d\d)-(\d\d)T(\d\d):(\d\d):(\d\d)(?:\.\d+)?(Z|[+-]\d\d:\d\d)?", sys.argv[1])
if not m: print("no"); raise SystemExit
seconds = calendar.timegm(tuple(int(g) for g in m.groups()[:6]) + (0, 0, 0))
zone = m.group(7) or "Z"
if zone != "Z":
    offset = int(zone[1:3]) * 3600 + int(zone[4:6]) * 60
    seconds -= offset if zone[0] == "+" else -offset
print("yes" if time.time() - seconds > 7 * 86400 else "no")
PY
)"
        [[ "$old" == yes ]] || continue
        docker image rm "$id" >/dev/null 2>&1 || true
    done <<< "$image_ids"
}

ensure_host_tmux() {
    # Host packages follow the existing test/preflight host-artifact boundary.
    [[ "${CLOUDNVR_SKIP_HOST_ARTIFACTS:-0}" != 1 ]] || return 0
    if command -v tmux >/dev/null 2>&1 && tmux -V >/dev/null 2>&1; then
        return 0
    fi
    if [[ "$FLAVOR" == nano ]]; then
        # ADR-021: no host packages by apt on a Nano (JetPack 4 mirrors are
        # fragile); browser terminals simply do not persist without tmux.
        warn "tmux is not installed; persistent browser terminals are unavailable on this Nano"
        return 0
    fi
    command -v apt-get >/dev/null 2>&1 \
        || { die "tmux is missing and this CloudNVR host has no apt-get"; return 1; }
    log "Installing tmux for persistent browser terminals"
    DEBIAN_FRONTEND=noninteractive apt-get update -o Acquire::http::Timeout=30 -o Acquire::https::Timeout=30 -o Acquire::Retries=2 || return 1
    DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends tmux -o Acquire::http::Timeout=30 -o Acquire::https::Timeout=30 -o Acquire::Retries=2 || return 1
    tmux -V >/dev/null 2>&1 \
        || { die "tmux is not usable after package installation"; return 1; }
}

cmd_ensure_host_tmux() {
    [[ "$#" -eq 0 ]] || { die "usage: nvr ensure-host-tmux"; return 2; }
    require_root || return 1
    # Only the flavor: this host step must not start failing on anything else in .env.
    FLAVOR="$(kv_get "$ENV_FILE" FLAVOR 2>/dev/null || true)"
    [[ "$FLAVOR" == nano ]] || FLAVOR=orin
    ensure_host_tmux
}

# The operator command is a small script, not a symlink. $ROOT is root-only
# (0750, releases 0700) because it holds config and secrets, so a symlink into
# it is unreachable for anyone else: bash reports "command not found" and a
# direct call "Permission denied". The script runs the real CLI as root and
# sends every other caller through sudo, which the dividia sudoers grant makes
# silent.
install_operator_command() {
    local body
    printf -v body '#!/bin/sh\n# CloudNVR operator command. Managed by %s.\nif [ "$(id -u)" -eq 0 ]; then exec %q "$@"; fi\nexec sudo %q "$@"\n' \
        "$BIN_DIR/nvr" "$BIN_DIR/nvr" "$BIN_DIR/nvr"
    printf '%s' "$body" | atomic_write "$CMD_LINK" 0755
}

host_access_home() {
    if [[ -n "${CLOUDNVR_HOST_ACCESS_HOME:-}" ]]; then
        printf '%s\n' "$CLOUDNVR_HOST_ACCESS_HOME"
        return 0
    fi
    getent passwd "$HOST_ACCESS_USER" | cut -d: -f6
}

# The support account, created exactly as nvr-docker's installer does
# (nvr docker/install-nvr.sh create_directory_structure). Here rather than only
# at install, so the nightly update gives every existing box the same account.
# useradd leaves its password locked; ensure_host_password then gives it root's.
ensure_host_account() {
    id "$HOST_ACCESS_USER" >/dev/null 2>&1 && return 0
    if ! getent group docker >/dev/null 2>&1; then
        groupadd docker || { warn "could not create the docker group"; return 1; }
    fi
    useradd -r -m -G docker -s /bin/bash "$HOST_ACCESS_USER" \
        || { warn "could not create the $HOST_ACCESS_USER support account"; return 1; }
    log "Created the $HOST_ACCESS_USER support account"
}

# Passwordless sudo for the support account, matching nvr-docker. Validated
# with visudo before it goes live: a broken drop-in disables sudo for everyone.
ensure_host_sudoers() {
    local want tmp
    want="Defaults:$HOST_ACCESS_USER !requiretty
$HOST_ACCESS_USER ALL=(ALL) NOPASSWD: ALL"
    if [[ -f "$SUDOERS_FILE" && ! -L "$SUDOERS_FILE" && "$(cat "$SUDOERS_FILE")" == "$want" ]]; then
        [[ "$(file_mode "$SUDOERS_FILE")" == 440 ]] || chmod 0440 "$SUDOERS_FILE"
        return
    fi
    command -v visudo >/dev/null 2>&1 || { warn "visudo is missing; not installing $SUDOERS_FILE"; return 1; }
    mkdir -p "${SUDOERS_FILE%/*}" || return 1
    tmp="$(mktemp "${SUDOERS_FILE%/*}/.dividia.XXXXXX")" || return 1
    if printf '%s\n' "$want" > "$tmp" && chmod 0440 "$tmp" && visudo -cf "$tmp" >/dev/null \
        && mv -f "$tmp" "$SUDOERS_FILE"; then
        log "Installed $SUDOERS_FILE"
        return 0
    fi
    rm -f "$tmp"
    warn "could not install $SUDOERS_FILE"
    return 1
}

# The support account's password is root's, as nvr-docker's installer makes it
# (nvr docker/install-nvr.sh create_directory_structure): techs know the shared
# root password and expect dividia to take the same one. The hash is copied, so
# no password is ever handled here, and it goes to chpasswd on stdin: an argv
# (usermod -p) is readable by every process on the host while it runs. Unlike nvr-docker this runs on every
# update, so a root password change reaches dividia and a distinct dividia
# password does not survive. A root with no usable password leaves dividia as
# it is.
ensure_host_password() {
    local want have
    want="$(getent shadow root 2>/dev/null | cut -d: -f2)" || want=""
    [[ -n "$want" && "$want" != '!'* && "$want" != '*'* ]] || return 0
    have="$(getent shadow "$HOST_ACCESS_USER" 2>/dev/null | cut -d: -f2)" || have=""
    [[ "$have" != "$want" ]] || return 0
    if printf '%s:%s\n' "$HOST_ACCESS_USER" "$want" | chpasswd -e; then
        log "Set the $HOST_ACCESS_USER password to root's"
        return 0
    fi
    warn "could not set the $HOST_ACCESS_USER password to root's"
    return 1
}

# docker for container inspection, systemd-journal for host logs, as nvr-docker.
ensure_host_groups() {
    local group rc=0
    for group in docker systemd-journal; do
        getent group "$group" >/dev/null 2>&1 || continue
        [[ " $(id -nG "$HOST_ACCESS_USER") " != *" $group "* ]] || continue
        if usermod -aG "$group" "$HOST_ACCESS_USER"; then
            log "Added $HOST_ACCESS_USER to the $group group"
        else
            warn "could not add $HOST_ACCESS_USER to $group"
            rc=1
        fi
    done
    return "$rc"
}

# Copy root's support keys into the support account, append-only and without
# duplicates, as nvr-docker does. Keys the account already has are kept.
ensure_host_support_keys() {
    local home dir auth tmp
    [[ -s "$ROOT_SSH_DIR/authorized_keys" ]] || return 0
    home="$(host_access_home)" || return 1
    dir="$home/.ssh"
    auth="$dir/authorized_keys"
    [[ -n "$home" && -d "$home" && ! -L "$home" && ! -L "$dir" && ! -L "$auth" ]] \
        || { warn "refusing to update $auth (missing home or symlinked path)"; return 1; }
    [[ ! -e "$auth" || -f "$auth" ]] || { warn "refusing to update $auth (not a regular file)"; return 1; }
    if [[ -f "$auth" ]] && awk 'FILENAME == ARGV[1] { have[$0] = 1; next } NF && !($0 in have) { exit 1 }' \
        "$auth" "$ROOT_SSH_DIR/authorized_keys"; then
        return 0
    fi
    [[ -d "$dir" ]] || mkdir -m 0700 "$dir" || return 1
    chown "$HOST_ACCESS_USER:" "$dir" && chmod 0700 "$dir" || return 1
    tmp="$(mktemp "$dir/.authorized_keys.nvr.XXXXXX")" || return 1
    # awk treats each file's end as a record boundary, so a missing final
    # newline cannot splice two keys into one line.
    if { [[ ! -f "$auth" ]] || awk '{ print }' "$auth"; } > "$tmp" \
        && awk 'FILENAME == ARGV[1] { have[$0] = 1; next } NF && !seen[$0]++ && !($0 in have)' \
            "$tmp" "$ROOT_SSH_DIR/authorized_keys" >> "$tmp" \
        && chmod 0600 "$tmp" && chown "$HOST_ACCESS_USER:" "$tmp" && mv -f "$tmp" "$auth"; then
        log "Synced root's support keys into $auth"
        return 0
    fi
    rm -f "$tmp"
    warn "could not update $auth"
    return 1
}

# The develop support key in root's authorized_keys, so a tech's develop
# identity logs in as root, and as dividia once ensure_host_support_keys copies
# root's keys across. nvr-docker boxes carry it from their installer; nothing
# else puts it on a CloudNVR, where the relay only pipes the connection through
# and the remoted support key is the relay's own. Append-only: a line already
# holding the key, even with options, counts, and nothing is ever removed.
# Like the rest of host access it converges nightly: a deleted key comes back.
ensure_develop_support_key() {
    local dir="$ROOT_SSH_DIR" auth="$ROOT_SSH_DIR/authorized_keys" blob tmp
    blob="${DEVELOP_SUPPORT_KEY#* }"
    [[ ! -L "$dir" && ! -L "$auth" ]] || { warn "refusing to update $auth (symlinked path)"; return 1; }
    [[ ! -e "$auth" || -f "$auth" ]] || { warn "refusing to update $auth (not a regular file)"; return 1; }
    if [[ -f "$auth" ]] && awk -v blob="$blob" '{ for (i = 1; i <= NF; i++) if ($i == blob) found = 1 }
        END { exit !found }' "$auth"; then
        return 0
    fi
    [[ -d "$dir" ]] || mkdir -m 0700 "$dir" || return 1
    tmp="$(mktemp "$dir/.authorized_keys.nvr.XXXXXX")" || return 1
    # awk ends every existing line, so the key never splices onto the last one.
    if { [[ ! -f "$auth" ]] || awk '{ print }' "$auth"; } > "$tmp" \
        && printf '%s\n' "$DEVELOP_SUPPORT_KEY" >> "$tmp" && chmod 0600 "$tmp" && mv -f "$tmp" "$auth"; then
        log "Authorized the develop support key in $auth"
        return 0
    fi
    rm -f "$tmp"
    warn "could not authorize the develop support key in $auth"
    return 1
}

# cs<serial>.dividia.net, the nvr-docker convention. The serial is assigned by
# the backend at first registration, so this converges once dragon has cached
# it. Until then the box is cloudnvr-<devid> (dragon caches the DEVID at start,
# reachable backend or not; 000000000000, dragon's no-NIC sentinel, is not one), so the console's "<host> login:" prompt never
# shows whatever name the OS image was flashed with. A cs<serial> name is never
# taken back to the DEVID form. /etc/hosts changes first so sudo can always
# resolve its own host name.
ensure_host_name() {
    local serial devid want short hosts managed='^cs[1-9][0-9]{4,9}(\.dividia\.net)?$'
    serial="$(head -n 1 "$SERIAL_FILE" 2>/dev/null | tr -d '[:space:]')" || true
    devid="$(head -n 1 "$DRAGON_DEVID_FILE" 2>/dev/null | tr -d '[:space:]')" || true
    if [[ "$serial" =~ ^[1-9][0-9]{4,9}$ ]]; then
        want="cs$serial.dividia.net"
        short="cs$serial"
    elif [[ "$devid" =~ ^[0-9A-F]{12}$ && "$devid" != 000000000000 ]] && ! [[ "$(hostname)" =~ $managed ]]; then
        want="cloudnvr-$(printf '%s' "$devid" | tr 'A-F' 'a-f')"
        short="$want"
    else
        return 0
    fi
    hosts="$(python3 - "$HOSTS_FILE" "$want" "$short" <<'PY'
import sys
path, fqdn, short = sys.argv[1:]
want = "127.0.1.1\t%s" % fqdn if fqdn == short else "127.0.1.1\t%s %s" % (fqdn, short)
try:
    lines = open(path).read().splitlines()
except FileNotFoundError:
    lines = ["127.0.0.1\tlocalhost"]
out, done = [], False
for line in lines:
    if line.split()[:1] == ["127.0.1.1"]:
        if not done:
            out.append(want)
            done = True
        continue
    out.append(line)
if not done:
    at = next((i + 1 for i, l in enumerate(out) if l.split()[:1] == ["127.0.0.1"]), 0)
    out.insert(at, want)
print("\n".join(out))
PY
)" || return 1
    if [[ ! -f "$HOSTS_FILE" || "$(cat "$HOSTS_FILE")" != "$hosts" ]]; then
        printf '%s\n' "$hosts" | atomic_write "$HOSTS_FILE" 0644 || return 1
        log "Updated $HOSTS_FILE for $want"
    fi
    [[ "$(hostname)" != "$want" ]] || return 0
    hostnamectl set-hostname "$want" || { warn "could not set the host name to $want"; return 1; }
    log "Set the host name to $want"
}

# Host access matching nvr-docker: passwordless sudo, groups, root's password,
# the develop support key, support keys, and the cs<serial> host name. Each step is independent and non-fatal: a
# failure is reported and retried on the next update, never blocks one.
ensure_host_access() {
    [[ "${CLOUDNVR_SKIP_HOST_ARTIFACTS:-0}" != 1 ]] || return 0
    local rc=0
    ensure_host_name || rc=1
    ensure_host_account || rc=1
    ensure_develop_support_key || rc=1
    if id "$HOST_ACCESS_USER" >/dev/null 2>&1; then
        local account_rc=0
        ensure_host_sudoers || account_rc=1
        ensure_host_groups || account_rc=1
        ensure_host_password || account_rc=1
        ensure_host_support_keys || account_rc=1
        # Only once dividia is fully usable does the login it replaces go.
        if [[ "$account_rc" -eq 0 ]]; then
            retire_provision_login || rc=1
        else
            rc=1
        fi
    fi
    ensure_remoted_host_access || rc=1
    return "$rc"
}

# provision-sd.sh before dragon 65dd838 created the image's first (uid 1000)
# user as `dragon`: an admin login (sudo, docker, the shop password) that
# nothing managed. The first user is dividia now; on boxes flashed before, the
# dragon login is removed. Matched exactly (Orin, name, uid 1000, /home/dragon),
# so an account anyone else made is never touched. A login with anything
# running (someone at the console) is left for a later update.
LEGACY_PROVISION_USER=dragon
retire_provision_login() {
    local user="$LEGACY_PROVISION_USER" entry name uid home
    [[ "$(kv_get "$ENV_FILE" FLAVOR 2>/dev/null || true)" != nano ]] || return 0
    entry="$(getent passwd "$user" 2>/dev/null)" || return 0
    IFS=: read -r name _ uid _ _ home _ <<< "$entry"
    [[ "$name" == "$user" && "$uid" == 1000 && "$home" == "/home/$user" ]] || return 0
    # A card flashed with no root password leaves dividia without one too, and
    # dragon is then the only password login (the monitor, a tech without a
    # key): keep it until dividia has a usable password.
    local hash
    hash="$(getent shadow "$HOST_ACCESS_USER" 2>/dev/null | cut -d: -f2)" || hash=""
    if [[ -z "$hash" || "$hash" == '!'* || "$hash" == '*'* ]]; then
        log "Keeping the old $user login: $HOST_ACCESS_USER has no password yet"
        return 0
    fi
    if pgrep -u "$user" >/dev/null 2>&1; then
        log "The old $user login has processes running; it will be removed on a later update"
        return 0
    fi
    crontab -r -u "$user" >/dev/null 2>&1 || true
    local path="${CLOUDNVR_HOST_FS_ROOT:-}$home"
    # userdel's status is not the verdict (it can fail on a missing mail spool
    # after the account is gone): the passwd entry is. Nothing in the home is
    # touched unless the account really went.
    if [[ -L "$path" ]]; then
        userdel "$user" >/dev/null 2>&1 || true
    else
        userdel -r "$user" >/dev/null 2>&1 || true
    fi
    if getent passwd "$user" >/dev/null 2>&1; then
        warn "could not remove the old $user login"
        return 1
    fi
    if [[ -L "$path" ]]; then
        rm -f "$path"  # never follow a symlinked home out of /home
    elif [[ -d "$path" ]]; then
        rm -rf -- "$path"
    fi
    log "Removed the old $user login (provision-sd.sh's first user before dividia)"
}

# ADR-020: a box that enrolled by itself gets the host SSH path the manual
# enroll links, on the next update. Only once remoted holds a relay config, so a
# box that never enrolled never lets the container's key file log in as root.
# A refusal is reported and retried, never fatal.
ensure_remoted_host_access() {
    local conf="$ROOT/data/remoted/remoted-relay.conf" message serial
    [[ -f "$conf" && ! -L "$conf" ]] || return 0
    # The container writes this file, so the gate is only as strong as the
    # enrolled container (root-equivalent by ADR-014 design). Still require the
    # shape an enrollment produces, for this box's own serial, before linking.
    serial="$(tr -d '[:space:]' < "$SERIAL_FILE" 2>/dev/null || true)"
    [[ "$serial" =~ ^[1-9][0-9]{4,9}$ ]] || return 0
    [[ "$(wc -c < "$conf")" -le 400 ]] &&
        grep -Eqx 'host=relay(-staging)?\.cloud\.dividia\.net' "$conf" &&
        grep -Fqx "serial=$(printf '%012d' "$serial")" "$conf" &&
        grep -Eqx 'secret=[0-9a-f]{64}' "$conf" || {
        warn "remote support host SSH not linked: remoted-relay.conf is not a completed enrollment for serial $serial"
        return 1
    }
    if ! message="$(ensure_remoted_host_authorization 2>&1)"; then
        warn "remote support host SSH not linked: ${message#ERROR: }"
        return 1
    fi
}

# The host name step alone, for the console's reconcile: run when dragon caches
# its serial, so the login prompt does not wait for the next nightly update.
cmd_ensure_host_name() {
    [[ "$#" -eq 0 ]] || { die "usage: nvr ensure-host-name"; return 2; }
    require_root || return 1
    require_install || return 1
    [[ "${CLOUDNVR_SKIP_HOST_ARTIFACTS:-0}" != 1 ]] || return 0
    ensure_host_name
}

cmd_ensure_host_access() {
    [[ "$#" -eq 0 ]] || { die "usage: nvr ensure-host-access"; return 2; }
    require_root || return 1
    require_install || return 1
    if [[ "${CLOUDNVR_SKIP_HOST_ARTIFACTS:-0}" != 1 ]]; then
        install_operator_command || return 1
    fi
    ensure_host_access
}

# The local console (FINAL CloudNVR-Local-Console-TUI plan). The tree ships in
# the active release's dragon image at CONSOLE_DIST; its own console-install
# does the host work, so that logic changes with each release and this
# function stays a small, stable loader (an old CLI would otherwise lag a
# release behind, as host access does). A release whose image predates the
# console, a Nano (out of scope), or a test harness leaves the host alone.
CONSOLE_DIST=/opt/cloudnvr-console-dist
CONSOLE_DIR="${CLOUDNVR_CONSOLE_DIR:-/opt/cloudnvr-console}"

active_dragon_image() {
    local digest
    digest="$(kv_get "$CURRENT_LINK/release.env" DRAGON_DIGEST 2>/dev/null)" || return 1
    digest_is_valid "$digest" || return 1
    printf '%s@%s\n' "$(flavor_dragon_repo)" "$digest"
}

ensure_host_console() {
    [[ "${CLOUDNVR_SKIP_HOST_ARTIFACTS:-0}" != 1 ]] || return 0
    [[ "$FLAVOR" != nano ]] || return 0
    local image work cid rc=0
    image="$(active_dragon_image)" || return 0
    work="$(mktemp -d "${TMPDIR:-/tmp}/cloudnvr-console.XXXXXXXX")" || return 1
    cid="$(docker create "$image" 2>/dev/null)" || { rm -rf -- "$work"; warn "cannot create a console-extraction container"; return 1; }
    if ! docker cp "$cid:$CONSOLE_DIST" "$work/dist" >/dev/null 2> "$work/cp.err"; then
        docker rm -f "$cid" >/dev/null 2>&1 || true
        # An image that predates the console has no tree: nothing to do. Any
        # other failure (disk full, daemon trouble) is reported and retried.
        if grep -qi 'could not find the file\|no such file or directory' "$work/cp.err"; then
            rm -rf -- "$work"
            return 0
        fi
        warn "could not copy the console tree out of the release image: $(head -c 200 "$work/cp.err")"
        rm -rf -- "$work"
        return 1
    fi
    docker rm -f "$cid" >/dev/null 2>&1 || true
    if [[ ! -d "$work/dist" || -L "$work/dist" || -n "$(find "$work/dist" -type l -print -quit)" ]]; then
        rm -rf -- "$work"
        warn "the release's console tree is malformed; the console was not updated"
        return 1
    fi
    python3 -IB "$work/dist/libexec/console-install" --install "$work/dist" || rc=1
    rm -rf -- "$work"
    return "$rc"
}

cmd_ensure_host_console() {
    local remove=false
    local resend=false
    case "${1:-}" in
        "") ;;
        --remove) remove=true; shift ;;
        --resend) resend=true; shift ;;
        *) die "usage: nvr ensure-host-console [--remove | --resend]"; return 2 ;;
    esac
    [[ "$#" -eq 0 ]] || { die "usage: nvr ensure-host-console [--remove | --resend]"; return 2; }
    require_root || return 1
    require_install || return 1
    FLAVOR="$(kv_get "$ENV_FILE" FLAVOR 2>/dev/null || true)"
    [[ "$FLAVOR" == nano ]] || FLAVOR=orin
    # Install and remove share staging directories with the nightly update's
    # own ensure_host_console: never run both at once.
    if [[ "$resend" != true ]]; then
        acquire_mutation_lock || return 1
    fi
    if [[ "$remove" == true ]]; then
        [[ -f "$CONSOLE_DIR/libexec/console-install" ]] || { log "The local console is not installed"; return 0; }
        python3 -IB "$CONSOLE_DIR/libexec/console-install" --remove
        return
    fi
    if [[ "$resend" == true ]]; then
        # ADR-024: re-send the pending escrow as is; dragon picks it up on its
        # next 60 s pass.
        [[ -f "$CONSOLE_DIR/libexec/console-password" ]] || { log "The local console is not installed"; return 0; }
        python3 -IB "$CONSOLE_DIR/libexec/console-password" --resend
        return
    fi
    [[ "$FLAVOR" != nano ]] || { log "The local console is not available on the Nano flavor"; return 0; }
    ensure_host_console
}

cmd_update() (
    local force=false candidate_tmp tools_tmp rejected current activate_rc=0
    [[ "${1:-}" == --force ]] && { force=true; shift; }
    [[ "$#" -eq 0 ]] || { die "usage: nvr update [--force]"; return 2; }
    require_root || return 1
    acquire_mutation_lock || {
        log "Another CloudNVR update owns the lock. Exiting."
        [[ "${CLOUDNVR_INSTALLING:-0}" != 1 ]] || return 1
        return 0
    }
    require_install || return 1
    load_host_env || return 1
    mkdir -p "$RELEASES_DIR" "$STATE_DIR" || return 1
    durable_sync_dirs "$RELEASES_DIR" "$STATE_DIR" || return 1
    if [[ -e "$CONVERSION_HOLD" && "${CLOUDNVR_INSTALLING:-0}" != 1 ]]; then
        log "An iSite conversion holds this box; finish it with install-cloudnvr.sh --convert-isite recover."
        return 0
    fi
    if [[ -f "$STATE_DIR/update-hold" ]]; then
        log "Updates are held. Run 'nvr hold off' before updating."
        # A held box still converges host access: the serial, and so the host
        # name, often arrives after the install that set the hold.
        ensure_host_access || warn "Host access is incomplete; it will retry on the next update"
        ensure_host_console || warn "The local console is incomplete; it will retry on the next update"
        [[ "${CLOUDNVR_INSTALLING:-0}" != 1 ]] || return 1
        return 0
    fi
    if [[ -L "$ROOT/data/remoted" ]]; then die "remote support directory cannot be a symlink"; return 1; fi
    install -d -m 0700 "$ROOT/data/remoted" || return 1
    cleanup_staging
    candidate_tmp="$(mktemp "${TMPDIR:-/tmp}/cloudnvr-release.json.XXXXXXXX")"
    tools_tmp="$(mktemp -d "${TMPDIR:-/tmp}/cloudnvr-tools.XXXXXXXX")"
    trap 'rm -f "${candidate_tmp:-}"; [[ -z "${tools_tmp:-}" ]] || rm -rf -- "$tools_tmp"' EXIT

    if ! download_candidate_manifest "$candidate_tmp"; then
        json_state "$STATE_DIR/last-update.json" failed update "" resolve "cannot resolve channel manifest"
        return 1
    fi
    if ! manifest_load "$candidate_tmp"; then
        json_state "$STATE_DIR/last-update.json" failed update "" manifest "invalid channel release manifest"
        return 1
    fi
    [[ "$MANIFEST_PLATFORM" == "$PLATFORM" ]] || { die "candidate platform mismatch"; return 1; }
    current="$(current_release_id 2>/dev/null || true)"
    if [[ "$MANIFEST_RELEASE_ID" == "$current" ]]; then
        log "Release $current is already current"
    fi
    rejected="$(rejected_release_id 2>/dev/null || true)"
    if [[ "$MANIFEST_RELEASE_ID" == "$rejected" \
       && "$MANIFEST_RELEASE_ID" != "$current" \
       && "$force" != true ]] \
       && ! release_was_committed "$MANIFEST_RELEASE_ID"; then
        log "Skipping rejected release $rejected. Use 'nvr update --force' to retry it."
        ensure_host_access || warn "Host access is incomplete; it will retry on the next update"
        ensure_host_console || warn "The local console is incomplete; it will retry on the next update"
        json_state "$STATE_DIR/last-update.json" skipped update "$rejected" rejected "candidate is marked rejected"
        return 0
    fi
    if ! pull_candidate_images; then
        json_state "$STATE_DIR/last-update.json" failed update "$MANIFEST_RELEASE_ID" pull "image pull failed"
        return 1
    fi
    if ! verify_candidate_images; then
        reject_release "$MANIFEST_RELEASE_ID" image "image labels or architecture failed validation" || true
        json_state "$STATE_DIR/last-update.json" failed update "$MANIFEST_RELEASE_ID" image \
            "image validation failed" || warn "could not record update failure state"
        return 1
    fi
    if ! extract_host_tools "$MANIFEST_DRAGON_REF" "$tools_tmp"; then
        reject_release "$MANIFEST_RELEASE_ID" host-tools "candidate host tools failed validation" || true
        json_state "$STATE_DIR/last-update.json" failed update "$MANIFEST_RELEASE_ID" host-tools \
            "candidate host tools failed validation" || warn "could not record update failure state"
        return 1
    fi
    write_release_env "$tools_tmp/release.env" "$MANIFEST_DRAGON_DIGEST" "$MANIFEST_CONNECTOR_DIGEST"
    if ! validate_compose "$tools_tmp" "$tools_tmp/release.env"; then
        reject_release "$MANIFEST_RELEASE_ID" compose "candidate Compose contract failed validation" || true
        json_state "$STATE_DIR/last-update.json" failed update "$MANIFEST_RELEASE_ID" compose \
            "candidate Compose contract failed validation" || warn "could not record update failure state"
        return 1
    fi
    if [[ "${CLOUDNVR_PREFLIGHT_ONLY:-0}" == 1 ]]; then
        log "Candidate $MANIFEST_RELEASE_ID passed host preflight"
        return 0
    fi
    ensure_host_tmux || warn "Host tmux is unavailable; persistent terminals will retry on the next update"
    activate_manifest "$candidate_tmp" "$tools_tmp" update || activate_rc=$?
    # After activation, so a fresh install already has its backend serial.
    ensure_host_access || warn "Host access is incomplete; it will retry on the next update"
    ensure_host_console || warn "The local console is incomplete; it will retry on the next update"
    [[ "$activate_rc" -eq 0 ]] || return "$activate_rc"
    prune_old_releases || warn "old release pruning failed"
)

cmd_rollback() (
    local current current_dir parent parent_dir tools_tmp hold_was_enabled=false restore_ok postcommit_ok=true
    local postcommit_stage="" postcommit_reason="" pointer_rc=0 observed_current=""
    require_root || return 1
    acquire_mutation_lock || return 1
    require_install || return 1
    load_host_env || return 1
    refuse_conversion_hold || return 1
    current="$(current_release_id)" || { die "there is no current release"; return 1; }
    current_dir="$RELEASES_DIR/$current"
    manifest_load "$current_dir/release.json" || return 1
    parent="$(release_parent_id "$current" 2>/dev/null || true)"
    release_id_is_valid "$parent" || { die "release $current has no rollback parent"; return 1; }
    parent_dir="$RELEASES_DIR/$parent"
    manifest_load "$parent_dir/release.json" || return 1
    ensure_rollback_images || return 1
    tools_tmp="$(mktemp -d "${TMPDIR:-/tmp}/cloudnvr-rollback-tools.XXXXXXXX")"
    trap 'rm -rf -- "${tools_tmp:-}"' EXIT
    extract_host_tools "$MANIFEST_DRAGON_REF" "$tools_tmp" || return 1
    prepare_release "$parent_dir/release.json" "$tools_tmp" "" \
        || { die "stored rollback release failed validation"; return 1; }
    [[ -f "$STATE_DIR/update-hold" ]] && hold_was_enabled=true
    if [[ "$hold_was_enabled" != true ]] && ! set_hold manual-rollback; then
        die "cannot enable the update hold for rollback"
        return 1
    fi
    remoted_reconcile "$parent" "$current"
    log "Starting rollback release $parent"
    if ! candidate_compose_cmd "$parent" up -d dragon connector || ! wait_release "$parent" \
        || ! wait_dragon_activation "$parent"; then
        warn "rollback candidate failed; restoring current release $current"
        restore_ok=true
        if ! restore_prior_or_stop_candidate "$current" "$parent_dir"; then
            restore_ok=false
            warn "current release restoration failed during rollback"
        fi
        if [[ "$hold_was_enabled" != true && "$restore_ok" == true ]]; then clear_hold || true; fi
        json_state "$STATE_DIR/last-update.json" failed rollback "$parent" health "rollback release was unhealthy"
        return 1
    fi
    prepare_host_dispatch "$parent" "$tools_tmp" || {
        restore_ok=true
        if ! restore_prior_or_stop_candidate "$current" "$parent_dir"; then
            restore_ok=false
            warn "current release restoration failed after host tool failure"
        fi
        if [[ "$hold_was_enabled" != true && "$restore_ok" == true ]]; then clear_hold || true; fi
        return 1
    }
    if ! record_release_parent "$parent" "$current"; then
        restore_ok=true
        if ! restore_prior_or_stop_candidate "$current" "$parent_dir"; then
            restore_ok=false
            warn "current release restoration failed after rollback lineage failure"
        fi
        if [[ "$hold_was_enabled" != true && "$restore_ok" == true ]]; then clear_hold || true; fi
        json_state "$STATE_DIR/last-update.json" failed rollback "$parent" lineage \
            "rollback lineage commit failed" || true
        return 1
    fi
    atomic_symlink "$RELEASES_DIR" current "$parent" || pointer_rc=$?
    if [[ "$pointer_rc" -ne 0 ]]; then
        observed_current="$(current_release_id 2>/dev/null || true)"
    fi
    if [[ "$pointer_rc" -ne 0 && "$observed_current" != "$parent" ]]; then
        restore_ok=true
        if ! restore_prior_or_stop_candidate "$current" "$parent_dir"; then
            restore_ok=false
            warn "current release restoration failed after pointer failure"
        fi
        if [[ "$hold_was_enabled" != true && "$restore_ok" == true ]]; then clear_hold || true; fi
        json_state "$STATE_DIR/last-update.json" failed rollback "$parent" pointer "rollback pointer commit failed"
        return 1
    fi
    if [[ "$pointer_rc" -ne 0 ]]; then
        postcommit_ok=false
        postcommit_stage=pointer-sync
        postcommit_reason="rollback pointer changed, but its directory sync failed"
        warn "$postcommit_reason"
    fi
    if ! commit_host_dispatch; then
        if [[ "$postcommit_ok" == true ]]; then
            postcommit_ok=false
            postcommit_stage=host-tools
            postcommit_reason="rollback committed, but host service metadata refresh failed"
        fi
        warn "rollback committed, but host service metadata could not be refreshed"
    fi
    if ! compose_cmd "$parent" up -d; then
        if [[ "$postcommit_ok" == true ]]; then
            postcommit_ok=false
            postcommit_stage=restart-policy
            postcommit_reason="rollback committed, but normal container restart policy could not be restored"
        fi
        warn "rollback committed, but normal container restart policy could not be restored"
    fi
    if ! mark_release_accepted "$parent"; then
        if [[ "$postcommit_ok" == true ]]; then
            postcommit_ok=false
            postcommit_stage=accepted
            postcommit_reason="rollback committed, but accepted-release evidence could not be recorded"
        fi
        warn "rollback committed, but accepted-release evidence could not be recorded"
    fi
    if [[ "$postcommit_ok" == true ]]; then
        json_state "$STATE_DIR/last-update.json" success rollback "$parent" activated "update hold enabled" \
            || warn "rollback committed, but success state could not be recorded"
    else
        json_state "$STATE_DIR/last-update.json" degraded rollback "$parent" "$postcommit_stage" \
            "$postcommit_reason" \
            || warn "rollback committed, but degraded state could not be recorded"
    fi
    log "Rolled back to $parent. Automatic updates are held."
    [[ "$postcommit_ok" == true ]] || return 2
)

# ADR-021: while an iSite conversion holds the box, native may still own the
# device identity, and the converter set dragon and the connector to restart=no.
# Only the converter's own install (CLOUDNVR_INSTALLING=1) may bring them up;
# remoted is identity-free and always may.
refuse_conversion_hold() {
    [[ -e "$CONVERSION_HOLD" && "${CLOUDNVR_INSTALLING:-0}" != 1 ]] || return 0
    die "an iSite conversion holds this box; run install-cloudnvr.sh --convert-isite recover (or --rollback-isite)"
    return 1
}

cmd_start() {
    local service="${1:-}" current
    [[ -z "$service" || "$service" == dragon || "$service" == connector || "$service" == remoted ]] \
        || { die "service must be dragon, connector, or remoted"; return 2; }
    acquire_mutation_lock || return 1
    require_install || return 1
    load_host_env || return 1
    [[ "$service" == remoted ]] || refuse_conversion_hold || return 1
    current="$(current_release_id)" || { die "there is no current release"; return 1; }
    if [[ -n "$service" ]]; then
        compose_cmd "$current" up -d "$service"
        wait_service "$current" "$service" "$([[ "$service" == dragon ]] && printf '%s' "$DRAGON_DEADLINE" || printf '%s' "$CONNECTOR_DEADLINE")"
    else
        compose_cmd "$current" up -d
        wait_release "$current"
    fi
}

cmd_stop() {
    local service="${1:-}" current
    [[ -z "$service" || "$service" == dragon || "$service" == connector || "$service" == remoted ]] \
        || { die "service must be dragon, connector, or remoted"; return 2; }
    acquire_mutation_lock || return 1
    require_install || return 1
    load_host_env || return 1
    current="$(current_release_id)" || { die "there is no current release"; return 1; }
    if [[ -n "$service" ]]; then
        compose_cmd "$current" stop "$service"
    else
        compose_cmd "$current" down
    fi
}

cmd_service_stop() {
    local current
    [[ "$#" -eq 0 ]] || { die "service-stop does not accept arguments"; return 2; }
    require_root || return 1
    acquire_service_lock stop || return 1
    require_install || return 1
    load_host_env || return 1
    current="$(current_release_id)" || { die "there is no current release"; return 1; }
    compose_cmd "$current" down
}

legacy_stop_locked() {
    local unit load_state expected actual active_rc rc=0 identity_dir="$STATE_DIR/legacy-units"
    for unit in dragon.service connector.service; do
        if ! load_state="$(systemctl show -p LoadState --value "$unit" 2>/dev/null)"; then
            die "cannot inspect legacy $unit"
            rc=1
            continue
        fi
        case "$load_state" in
            not-found) continue ;;
            loaded|masked) ;;
            *)
                die "legacy $unit has unsafe load state: $load_state"
                rc=1
                continue
                ;;
        esac
        if systemd_active_status "$unit"; then
            :
        else
            active_rc=$?
            if [[ "$active_rc" -eq 1 ]]; then
                continue
            fi
            die "cannot inspect legacy $unit active state"
            rc=1
            continue
        fi
        if [[ ! -f "$identity_dir/$unit.sha256" || -L "$identity_dir/$unit.sha256" ]]; then
            die "loaded $unit has no adopted CloudNVR identity"
            rc=1
            continue
        fi
        expected="$(cat "$identity_dir/$unit.sha256")"
        if [[ ! "$expected" =~ ^[0-9a-f]{64}$ ]] \
           || ! actual="$(systemctl cat "$unit" 2>/dev/null | sha256_file /dev/stdin)" \
           || [[ "$actual" != "$expected" ]]; then
            die "loaded $unit does not match its adopted identity"
            rc=1
            continue
        fi
        if ! systemctl stop "$unit"; then
            die "cannot stop legacy $unit"
            rc=1
            continue
        fi
        if ! systemd_inactive_verified "$unit"; then
            die "legacy $unit stop state could not be verified"
            rc=1
        fi
    done
    return "$rc"
}

cmd_legacy_stop() {
    [[ "$#" -eq 0 ]] || { die "legacy-stop does not accept arguments"; return 2; }
    require_root || return 1
    acquire_mutation_lock || return 1
    require_install || return 1
    legacy_stop_locked
}

cmd_service_start() {
    local current
    [[ "$#" -eq 0 ]] || { die "service-start does not accept arguments"; return 2; }
    require_root || return 1
    acquire_service_lock start || return 1
    require_install || return 1
    load_host_env || return 1
    current="$(current_release_id)" || { die "there is no current release"; return 1; }
    legacy_stop_locked || return 1
    if release_has_remoted "$current" && ! remoted_held "$current"; then
        compose_cmd "$current" up -d remoted || warn "remoted did not start; continuing with the core stack"
    fi
    # remoted is the identity-free support lifeline and runs during a hold.
    if [[ -e "$CONVERSION_HOLD" ]]; then
        log "An iSite conversion holds this box; dragon and the connector stay stopped until it is recovered."
        return 0
    fi
    compose_cmd "$current" up -d dragon connector || return 1
    wait_release "$current" && wait_dragon_activation "$current"
}

cmd_restart() {
    local service="${1:-}" current
    [[ -z "$service" || "$service" == dragon || "$service" == connector || "$service" == remoted ]] \
        || { die "service must be dragon, connector, or remoted"; return 2; }
    acquire_mutation_lock || return 1
    require_install || return 1
    load_host_env || return 1
    [[ "$service" == remoted ]] || refuse_conversion_hold || return 1
    current="$(current_release_id)" || { die "there is no current release"; return 1; }
    if [[ -n "$service" ]]; then
        compose_cmd "$current" up -d --force-recreate "$service"
        wait_service "$current" "$service" "$([[ "$service" == dragon ]] && printf '%s' "$DRAGON_DEADLINE" || printf '%s' "$CONNECTOR_DEADLINE")"
    else
        compose_cmd "$current" up -d --force-recreate
        wait_release "$current"
    fi
}

cmd_logs() {
    local service="" follow="" current arg
    for arg in "$@"; do
        case "$arg" in
            dragon|connector|remoted) [[ -z "$service" ]] || { die "choose one service"; return 2; }; service="$arg" ;;
            -f|--follow) follow=-f ;;
            *) die "usage: nvr logs [dragon|connector|remoted] [-f]"; return 2 ;;
        esac
    done
    require_install || return 1
    load_host_env || return 1
    current="$(current_release_id)" || { die "there is no current release"; return 1; }
    if [[ -n "$service" ]]; then
        compose_cmd "$current" logs ${follow:+"$follow"} "$service"
    else
        compose_cmd "$current" logs ${follow:+"$follow"}
    fi
}

cmd_channel() {
    local requested="${1:-}" backend_addr old_backend_default
    if [[ -z "$requested" ]]; then
        require_install || return 1
        load_host_env || return 1
        printf '%s\n' "$CHANNEL"
        return
    fi
    [[ "$#" -eq 1 ]] || { die "usage: nvr channel [CHANNEL]"; return 2; }
    require_root || return 1
    channel_is_valid "$requested" || { die "invalid channel: $requested"; return 1; }
    acquire_mutation_lock || return 1
    require_install || return 1
    load_host_env || return 1
    backend_addr="$(kv_get "$ENV_FILE" BACKEND_ADDR 2>/dev/null || true)"
    old_backend_default="$(channel_backend_default "$CHANNEL")"
    if [[ -n "$CLOUD_ENVIRONMENT" ]]; then
        # An explicit `nvr environment` owns the backend; the channel only
        # chooses the software.
        backend_addr="$(environment_backend "$CLOUD_ENVIRONMENT")"
    elif [[ -z "$backend_addr" || "$backend_addr" == "$old_backend_default" ]]; then
        backend_addr="$(channel_backend_default "$requested")"
    fi
    python3 - "$ENV_FILE" "$requested" "$backend_addr" <<'PY' | atomic_write "$ENV_FILE" 0600
import sys
path, channel, backend_addr = sys.argv[1:]
lines = open(path, encoding="utf-8").read().splitlines()
seen_channel = False
seen_backend = False
for line in lines:
    if line.startswith("CHANNEL="):
        print("CHANNEL=" + channel)
        seen_channel = True
    elif line.startswith("BACKEND_ADDR="):
        print("BACKEND_ADDR=" + backend_addr)
        seen_backend = True
    else:
        print(line)
if not seen_channel: print("CHANNEL=" + channel)
if not seen_backend: print("BACKEND_ADDR=" + backend_addr)
PY
    log "Channel set to $requested with backend $backend_addr. Run 'nvr update' to activate it."
}

env_file_set() {  # env_file_set KEY=VALUE...: rewrite those keys in $ENV_FILE, append new ones
    python3 - "$ENV_FILE" "$@" <<'PY' | atomic_write "$ENV_FILE" 0600
import sys
path, pairs = sys.argv[1], [a.split("=", 1) for a in sys.argv[2:]]
wanted = dict(pairs)
seen = set()
for line in open(path, encoding="utf-8").read().splitlines():
    key = line.split("=", 1)[0]
    if key in wanted:
        print(key + "=" + wanted[key]); seen.add(key)
    else:
        print(line)
for key, value in pairs:
    if key not in seen: print(key + "=" + value)
PY
}

# A fleet cloudapi.secret: a regular, non-symlink file holding one line of
# printable, non-space ASCII. Checked before any secret is read or copied.
cloudapi_secret_is_valid() {
    [[ -f "$1" && ! -L "$1" ]] || return 1
    python3 - "$1" <<'PY'
import os, sys
fd = os.open(sys.argv[1], os.O_RDONLY | os.O_NOFOLLOW)
with os.fdopen(fd, "rb") as stream:
    data = stream.read(1025)
text = data[:-1] if data.endswith(b"\n") else data
sys.exit(0 if 16 <= len(text) <= 512 and all(0x21 <= b <= 0x7e for b in text) else 1)
PY
}

active_connector_environment() {
    local value
    value="$(tr -d '[:space:]' < "$CLOUDAPI_DIR/environment" 2>/dev/null)" || return 1
    environment_normalize "$value"
}

environment_status() {
    local shown="${CLOUD_ENVIRONMENT:-}" connector name
    if [[ -z "$shown" ]]; then
        case "$BACKEND_ADDR" in
            engine.cloud.dividia.net) shown="prod (from the backend)" ;;
            engine-staging.cloud.dividia.net) shown="staging (from the backend)" ;;
            *) shown="custom" ;;
        esac
    fi
    connector="$(active_connector_environment 2>/dev/null || echo unprofiled)"
    printf 'Flavor: %s\nBundle root: %s\n' "$FLAVOR" "$BUNDLE_BASE_URL"
    printf 'Environment: %s\nChannel: %s\nBackend: %s\nConnector: %s\n' "$shown" "$CHANNEL" "$BACKEND_ADDR" "$connector"
    for name in prod staging; do
        if cloudapi_secret_is_valid "$CLOUDAPI_DIR/profiles/$name/cloudapi.secret"; then
            printf 'Profile %s: saved\n' "$name"
        elif [[ -e "$CLOUDAPI_DIR/profiles/$name/cloudapi.secret" || -L "$CLOUDAPI_DIR/profiles/$name/cloudapi.secret" ]]; then
            printf 'Profile %s: invalid\n' "$name"
        else
            printf 'Profile %s: missing\n' "$name"
        fi
    done
}

# nvr environment [production|staging [--secret-file FILE]]: move the box to
# another cloud. The engine backend and the connector's cloudapi profile (fleet
# secret + environment marker) move together; the channel is untouched. Each
# backend keeps one JSDevice row per MAC, so the box gets back the same serial
# every time it returns to an environment. The first move saves the current
# profile so the way back needs nothing; a new environment's fleet secret is
# supplied once with --secret-file (0600, as in the install bundle).
cmd_environment() {
    local requested="" secret_file="" target backend active current
    while [[ "$#" -gt 0 ]]; do
        case "$1" in
            --secret-file) [[ "$#" -ge 2 ]] || { die "--secret-file requires a value"; return 2; }; secret_file="$2"; shift 2 ;;
            -*) die "usage: nvr environment [production|staging [--secret-file FILE]]"; return 2 ;;
            *) [[ -z "$requested" ]] || { die "usage: nvr environment [production|staging [--secret-file FILE]]"; return 2; }
               requested="$1"; shift ;;
        esac
    done
    if [[ -z "$requested" ]]; then
        [[ -z "$secret_file" ]] || { die "--secret-file needs an environment"; return 2; }
        require_install || return 1
        load_host_env || return 1
        environment_status
        return
    fi
    target="$(environment_normalize "$requested")" || { die "environment must be production or staging"; return 2; }
    require_root || return 1
    if [[ -n "$secret_file" ]]; then
        [[ -f "$secret_file" && ! -L "$secret_file" && "$(stat -c %a "$secret_file" 2>/dev/null || stat -f %Lp "$secret_file")" == 600 ]] \
            || { die "--secret-file must be a regular mode-0600 file"; return 1; }
        cloudapi_secret_is_valid "$secret_file" || { die "--secret-file is not a fleet cloudapi.secret (one line)"; return 1; }
    fi
    acquire_mutation_lock || return 1
    require_install || return 1
    load_host_env || return 1
    refuse_conversion_hold || return 1
    backend="$(environment_backend "$target")"
    local dir
    for dir in "$CLOUDAPI_DIR" "$CLOUDAPI_DIR/profiles" "$CLOUDAPI_DIR/profiles/prod" "$CLOUDAPI_DIR/profiles/staging"; do
        [[ ! -L "$dir" ]] || { die "$dir must not be a symlink"; return 1; }
        mkdir -p "$dir" && chmod 0700 "$dir" || return 1
    done
    # Save the active profile first, so the way back exists before anything moves.
    if active="$(active_connector_environment)" && cloudapi_secret_is_valid "$CLOUDAPI_DIR/cloudapi.secret" \
        && [[ ! -e "$CLOUDAPI_DIR/profiles/$active/cloudapi.secret" && ! -L "$CLOUDAPI_DIR/profiles/$active/cloudapi.secret" ]]; then
        atomic_write "$CLOUDAPI_DIR/profiles/$active/cloudapi.secret" 0600 < "$CLOUDAPI_DIR/cloudapi.secret" || return 1
    fi
    if [[ -n "$secret_file" ]]; then
        atomic_write "$CLOUDAPI_DIR/profiles/$target/cloudapi.secret" 0600 < "$secret_file" || return 1
    fi
    cloudapi_secret_is_valid "$CLOUDAPI_DIR/profiles/$target/cloudapi.secret" || {
        die "this box has no valid $target connector profile. Supply the $target fleet secret once:
       nvr environment $requested --secret-file FILE"; return 1; }
    # The connector's secret and marker move together, then the engine backend.
    atomic_write "$CLOUDAPI_DIR/cloudapi.secret" 0600 < "$CLOUDAPI_DIR/profiles/$target/cloudapi.secret" || return 1
    printf '%s\n' "$target" | atomic_write "$CLOUDAPI_DIR/environment" 0644 || return 1
    env_file_set "CLOUD_ENVIRONMENT=$target" "BACKEND_ADDR=$backend" || return 1
    BACKEND_ADDR="$backend"; CLOUD_ENVIRONMENT="$target"
    if ! current="$(current_release_id 2>/dev/null)"; then
        log "Environment set to $target (backend $backend). No release is running; the next start uses it."
        return 0
    fi
    # dragon re-registers with the new backend and receives that environment's
    # serial; the connector reads its profile only at start.
    # Every step above is idempotent, so a failed restart is converged by
    # running the same command again (or the old environment's, to go back).
    compose_cmd "$current" up -d --force-recreate dragon connector \
        || { die "the files now name $target, but dragon and connector did not restart. Re-run: nvr environment $requested"; return 1; }
    wait_release "$current" \
        || { die "dragon or connector is unhealthy after the move to $target. Re-run: nvr environment $requested (or move back)"; return 1; }
    log "Environment set to $target: backend $backend, connector $target, channel $CHANNEL."
}

cmd_hold() {
    local action="${1:-status}"
    case "$action" in
        on)
            require_root || return 1
            acquire_mutation_lock || return 1
            require_install || return 1
            mkdir -p "$STATE_DIR" || return 1
            durable_sync_dirs "$STATE_DIR" || return 1
            set_hold operator
            log "Automatic updates are held."
            ;;
        off)
            require_root || return 1
            acquire_mutation_lock || return 1
            require_install || return 1
            mkdir -p "$STATE_DIR" || return 1
            durable_sync_dirs "$STATE_DIR" || return 1
            clear_hold
            log "Automatic updates are enabled."
            ;;
        status)
            require_install || return 1
            if [[ -f "$STATE_DIR/update-hold" ]]; then
                printf 'on\n'
            else
                printf 'off\n'
            fi
            ;;
        *) die "usage: nvr hold [on|off|status]"; return 2 ;;
    esac
}

cmd_status() {
    local current="" health release_file hold=off rejected=none last=none usage runtime gpu schedule
    local runtimes gpu_output
    require_install || return 1
    load_host_env || return 1
    current="$(current_release_id 2>/dev/null || true)"
    [[ -f "$STATE_DIR/update-hold" ]] && hold=on
    rejected="$(rejected_release_id 2>/dev/null || printf none)"
    [[ -n "$rejected" ]] || rejected=none
    if [[ -f "$STATE_DIR/last-update.json" ]]; then
        last="$(python3 -c 'import json,sys; v=json.load(open(sys.argv[1])); print("%s at %s" % (v.get("result","?"),v.get("timestamp","?")))' "$STATE_DIR/last-update.json" 2>/dev/null || printf invalid)"
    fi
    usage="$(df -P "$ROOT/data/video" 2>/dev/null | awk 'NR==2 {print $5 " used (" $4 " KiB free)"}' || true)"
    runtimes="$(docker info --format '{{json .Runtimes}}' 2>/dev/null || true)"
    if grep -q 'nvidia' <<< "$runtimes"; then runtime=available; else runtime=unavailable; fi
    gpu_output="$(command -v nvidia-smi >/dev/null 2>&1 && nvidia-smi -L 2>/dev/null || true)"
    gpu="${gpu_output%%$'\n'*}"
    schedule="$(update_schedule)"
    printf 'Product: cloudnvr\n'
    printf 'Flavor: %s\n' "$FLAVOR"
    printf 'Bundle root: %s\n' "$BUNDLE_BASE_URL"
    printf 'Host architecture: %s\n' "$(uname -m)"
    printf 'Channel: %s\n' "$CHANNEL"
    if [[ -n "$current" ]]; then
        release_file="$RELEASES_DIR/$current/release.json"
        printf 'Release: %s\n' "$(release_display "$release_file")"
    fi
    printf 'Active release: %s\n' "${current:-none}"
    if [[ -n "$current" ]]; then
        manifest_load "$release_file" || return 1
        printf 'Dragon image: %s\n' "$MANIFEST_DRAGON_REF"
        printf 'Connector image: %s\n' "$MANIFEST_CONNECTOR_REF"
        if [[ -n "${MANIFEST_REMOTED_REF:-}" ]]; then printf 'Remoted image: %s\n' "$MANIFEST_REMOTED_REF"; fi
        local service_label
        for service in $(release_services "$current"); do
            health="$(container_health "$current" "$service" || true)"
            service_label="$service"
            printf '%s health: %s\n' "$service_label" "$health"
        done
        if remoted_quarantined "$current"; then
            printf 'Remoted quarantine: %s (running the prior support tunnel)\n' "$(quarantined_remoted_identity)"
        elif remoted_content_unchanged "$current"; then
            printf 'Remoted: same content as the release, kept running as %s\n' "$(remoted_running_image)"
        fi
    fi
    printf 'Last update: %s\n' "$last"
    printf 'Rejected release: %s\n' "$rejected"
    printf 'Automatic update: %s, hold %s\n' "$schedule" "$hold"
    printf 'Video filesystem: %s\n' "${usage:-unavailable}"
    printf 'NVIDIA runtime: %s\n' "$runtime"
    printf 'GPU: %s\n' "${gpu:-not visible from host probe}"
}

# The release a box runs, as users see it: V from the release's own stored
# release.json ("release_version", written by the publisher inside the hashed
# bundle), or dev-<commit12> for a build that is not a release candidate. V is
# display only and gates nothing. "(candidate)" comes from the box's explicit
# channel setting, never from the bytes, which are identical to production.
release_display() {
    local file="$1" shown
    shown="$(python3 - "$file" <<'PY'
import json, re, sys
try:
    value = json.load(open(sys.argv[1], encoding="utf-8"))
except Exception:
    print("unknown"); raise SystemExit(0)
version = value.get("release_version")
source = value.get("source_revision")
if isinstance(version, str) and re.fullmatch(r"[0-9]{4}\.([1-9]|1[0-2])\.[1-9][0-9]*", version):
    print(version)
elif isinstance(source, str) and re.fullmatch(r"[0-9a-f]{40}", source):
    print("dev-" + source[:12])
else:
    print("unknown")
PY
)" || shown=unknown
    case "$shown" in
        dev-*|unknown) ;;
        *) channel_is_production "${CHANNEL:-}" || shown="$shown (candidate)" ;;
    esac
    printf '%s\n' "$shown"
}

cmd_version() {
    local current version channel
    require_install || return 1
    load_host_env || return 1
    current="$(current_release_id)" || { die "there is no current release"; return 1; }
    manifest_load "$RELEASES_DIR/$current/release.json" || return 1
    printf 'Release: %s\n' "$(release_display "$RELEASES_DIR/$current/release.json")"
    printf 'Release ID: %s\n' "$MANIFEST_RELEASE_ID"
    printf 'Dragon source: %s\n' "$MANIFEST_SOURCE_REV"
    printf 'Connector source: %s\n' "$MANIFEST_SOURCE_REV"
    printf 'Connector NVR source: %s\n' "$MANIFEST_NVR_REV"
    version="$(image_label "$MANIFEST_DRAGON_REF" org.opencontainers.image.version 2>/dev/null || true)"
    channel="$(image_label "$MANIFEST_DRAGON_REF" channel 2>/dev/null || true)"
    printf 'Dragon image: %s (version %s, build channel %s)\n' "$MANIFEST_DRAGON_REF" "${version:-unknown}" "${channel:-unknown}"
    version="$(image_label "$MANIFEST_CONNECTOR_REF" org.opencontainers.image.version 2>/dev/null || true)"
    channel="$(image_label "$MANIFEST_CONNECTOR_REF" channel 2>/dev/null || true)"
    printf 'Connector image: %s (version %s, build channel %s)\n' "$MANIFEST_CONNECTOR_REF" "${version:-unknown}" "${channel:-unknown}"
}

ensure_remoted_host_authorization() {
    local key_dir="${CLOUDNVR_HOST_SSH_DIR:-/root/.ssh}" sshd_bin="${CLOUDNVR_SSHD:-/usr/sbin/sshd}"
    local target="$ROOT/data/remoted/authorized_keys" secondary keys
    secondary="$key_dir/authorized_keys2"
    keys="$("$sshd_bin" -T -C user=root,host=localhost,addr=127.0.0.1 | awk '$1 == "authorizedkeysfile" { $1=""; print }')" || return 1
    case " $keys " in
        *" .ssh/authorized_keys2 "*|*" $secondary "*) ;;
        *) die "host sshd must enable the secondary root authorized-keys path for managed support"; return 1 ;;
    esac
    [[ ! -L "$key_dir" && ! -L "$ROOT/data/remoted" ]] || {
        die "remote support key directories cannot be symlinks"; return 1;
    }
    if [[ -L "$secondary" ]]; then
        [[ "$(readlink "$secondary")" == "$target" ]] || {
            die "host secondary authorized-keys path belongs to another configuration"; return 1;
        }
        return 0
    fi
    [[ ! -e "$secondary" ]] || { die "host secondary authorized-keys file already exists"; return 1; }
    install -d -m 0700 "$key_dir" "$ROOT/data/remoted" || return 1
    # sshd follows the managed file after the client atomically rotates it.
    # The container never receives the host's existing keys or private keys.
    ln -s "$target" "$secondary"
}

# ADR-020: what the bootstrap is doing, the key fingerprint staff confirm before
# approving, and whether host SSH is linked. Informational only: it never
# changes the status result that callers (the office controller) check.
remoted_enrollment_status() {
    local state link
    if state="$(compose_cmd "$1" exec -T remoted python3 /usr/local/libexec/remoted-bootstrap.py status 2>/dev/null)" && [[ -n "$state" ]]; then
        printf 'Enrollment: %s\n' "$state"
    else
        echo 'Enrollment: unknown (this remoted has no automatic enrollment)'
    fi
    link="$(readlink "${CLOUDNVR_HOST_SSH_DIR:-/root/.ssh}/authorized_keys2" 2>/dev/null || true)"
    if [[ "$link" == "$ROOT/data/remoted/authorized_keys" ]]; then
        echo 'Host SSH access: linked'
    else
        echo 'Host SSH access: not linked (linked by the next update after enrollment)'
    fi
}

# ADR-020 amendment 1: vouch for this box's remoted key so it enrolls with no
# staff click. A Dividia master logs in for this device's MAC and names the key
# (the token then vouches for that key only, once); cloudapi records a
# preauthorization that the device consumes on its own proof of possession.
# On the office network it needs no credential at all (the WAF-gated office
# path); elsewhere it falls back to the login unless --no-login.
# Idempotent and never destructive. Exit 3 = skipped (no credential available).
# --token-file is for automation only: the token must have been minted for this
# device *naming this key's fingerprint* (shown by `nvr remoted status`), within
# its 30-minute life. An ordinary migration token is refused.
remoted_register() {
    local current="$1" token_file="" wait=0 rc=0 login=1
    shift
    while [[ "$#" -gt 0 ]]; do
        case "$1" in
            --token-file) [[ "$#" -ge 2 ]] || { die "--token-file requires a value"; return 2; }; token_file="$2"; shift 2 ;;
            --wait) [[ "$#" -ge 2 && "$2" =~ ^[0-9]{1,4}$ ]] || { die "--wait requires seconds"; return 2; }; wait="$2"; shift 2 ;;
            --no-login) login=0; shift ;;
            *) die "usage: nvr remoted register [--token-file FILE] [--wait SECONDS] [--no-login]"; return 2 ;;
        esac
    done
    if [[ -n "$token_file" ]]; then
        [[ -f "$token_file" && ! -L "$token_file" && "$(stat -c %a "$token_file" 2>/dev/null || stat -f %Lp "$token_file")" == 600 ]] \
            || { die "--token-file must be a regular mode-0600 file"; return 1; }
    fi
    # dragon publishes its MAC and serial after its first registration, and
    # remoted generates its key when it first starts: wait for all three.
    local key="$ROOT/data/remoted/id_rsa.pub" device="$ROOT/data/cache/dragon/device-id" deadline=$((SECONDS + wait))
    until [[ -s "$key" && -s "$device" && -s "$SERIAL_FILE" ]]; do
        (( SECONDS < deadline )) || { die "remote support key, device ID or serial is not ready yet"; return 1; }
        sleep 5
    done
    python3 - "$key" "$device" "$SERIAL_FILE" "$token_file" "$login" <<'PY' || rc=$?
import base64, binascii, getpass, hashlib, json, os, re, stat, struct, sys, urllib.error, urllib.request
key_path, device_path, serial_path, token_file, allow_login = sys.argv[1:6]

def read_small(path, cap):
    # The remoted container writes id_rsa.pub: no symlinks, regular files only, bounded.
    fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
    try:
        info = os.fstat(fd)
        if not stat.S_ISREG(info.st_mode) or info.st_size > cap:
            sys.exit('ERROR: %s is not a small regular file' % path)
        return os.read(fd, cap).decode('ascii', 'replace')
    finally:
        os.close(fd)

fields = read_small(key_path, 16384).split()
if len(fields) < 2 or fields[0] not in ('ssh-rsa', 'ssh-ed25519'):
    sys.exit('ERROR: the remoted public key is not an ssh-rsa/ssh-ed25519 key')
try:
    blob = base64.b64decode(fields[1], validate=True)
except (binascii.Error, ValueError):
    sys.exit('ERROR: the remoted public key is not valid base64')
if len(blob) < 4 or blob[4:4 + struct.unpack('>I', blob[:4])[0]] != fields[0].encode():
    sys.exit('ERROR: the remoted public key blob does not match its type')
public_key = fields[0] + ' ' + fields[1]
fingerprint = 'SHA256:' + base64.b64encode(hashlib.sha256(blob).digest()).decode().rstrip('=')
device_id = read_small(device_path, 64).strip()
if not re.fullmatch('[0-9A-F]{12}', device_id) or device_id == '000000000000':
    sys.exit('ERROR: dragon has not published a valid device ID')
serial = int(read_small(serial_path, 64).strip() or 0)
# cloudapi ADR-024: the serial band names the environment.
origin = ('https://api.cloud.dividia.net' if 10000 <= serial <= 14999 else
          'https://api-staging.cloud.dividia.net' if 15000 <= serial <= 19999 else None)
if os.environ.get('CLOUDNVR_ALLOW_NON_ROOT') == '1' and os.environ.get('CLOUDNVR_TEST_API_BASE'):
    origin = os.environ['CLOUDNVR_TEST_API_BASE']
if not origin:
    sys.exit('ERROR: serial %d is not a CloudNVR serial' % serial)

class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, *args):
        return None
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect())

def post(path, body, token=None):
    headers = {'Content-Type': 'application/json'}
    if token:
        headers['Authorization'] = 'Bearer ' + token
    request = urllib.request.Request(origin + path, data=json.dumps(body).encode(), headers=headers, method='POST')
    try:
        with opener.open(request, timeout=20) as response:
            return response.status, json.loads(response.read(20001) or b'{}')
    except urllib.error.HTTPError as error:
        try:
            return error.code, json.loads(error.read(20001) or b'{}')
        except ValueError:
            return error.code, {}          # e.g. the WAF's own non-JSON 403
    except (urllib.error.URLError, OSError, ValueError):
        return 0, {}                       # offline, DNS, TLS, or a garbled reply

print('Remote support key %s for device %s (serial %d)' % (fingerprint, device_id, serial))

def refused(reply, status):
    error = reply.get('error')
    # Routes answer {"error":{"code":..}}; the auth layer answers {"error":"<code>"}.
    code = error.get('code') if isinstance(error, dict) else error if isinstance(error, str) else None
    sys.exit('ERROR: preauthorization refused (%s); remote support will need staff approval' % (code or status))

# On Dividia's office network no credential is needed: cloudapi's WAF admits
# this route only from office egress, and pins the record to it.
if not token_file:
    status, reply = post('/ve/nvr/cloudnvr-support-keys', {'device_id': device_id, 'public_key': public_key})
    if status == 200 and reply.get('already_bound'):
        print('Remote support: this key is already bound; nothing to do.')
        sys.exit(0)
    if status == 201:
        print('Remote support: key preauthorized from the office network until %s; the box enrolls by itself.' % reply.get('expires_at'))
        sys.exit(0)
    error = reply.get('error')
    code = error.get('code') if isinstance(error, dict) else None
    # Only a real application answer is final. The WAF's block (a 403 with no
    # JSON code, or office_path_required) means "not the office"; anything else
    # (a cloudapi without this route, disabled, rate limited, offline) means the
    # office path is unavailable. Either way, fall back rather than give up.
    # Final: the key or serial itself is refused. Anything else (e.g. this box's
    # registration went out another office link) falls back like "not office".
    if status == 409 and code in ('bound_to_another_key', 'not_cloudnvr'):
        refused(reply, status)
    if status == 403 and code in (None, 'office_path_required'):
        print('Not on the Dividia office network.')
    else:
        print('The office registration path is unavailable (%s).' % (code or status or 'no connection'))
    if allow_login != '1':
        print('Remote support will need staff approval in Monarch (or run: sudo nvr remoted register).')
        sys.exit(3)
if token_file:
    token = open(token_file).read().strip()
else:
    try:
        tty = open('/dev/tty', 'r+')
    except OSError:
        print('No terminal and no --token-file: remote support will need staff approval in Monarch.')
        sys.exit(3)
    tty.write('Dividia cloud login (a master account) to vouch for this key\nUsername: ')
    tty.flush()
    username = tty.readline().strip()
    password = getpass.getpass('Password: ', stream=tty)
    status, reply = post('/ve/login', {'username': username, 'password': password,
                                        'migrationDeviceId': device_id, 'supportFingerprint': fingerprint})
    password = None
    token = reply.get('nvrMigrationPersonToken') if status == 200 else None
    if not token:
        code = reply.get('error') if isinstance(reply.get('error'), str) else 'login_failed'
        sys.exit('ERROR: cloud login did not grant this device (%s); remote support will need staff approval' % code)
status, reply = post('/ve/cloudnvr/support-preauthorizations?device=' + device_id,
                     {'public_key': public_key, 'source': 'installer'}, token)
if status == 0:
    sys.exit('ERROR: cloudapi is unreachable; remote support will need staff approval (sudo nvr remoted register retries)')
token = None
if status == 200 and reply.get('already_bound'):
    print('Remote support: this key is already bound; nothing to do.')
elif status == 201:
    print('Remote support: key preauthorized until %s; the box enrolls by itself.' % reply.get('expires_at'))
else:
    refused(reply, status)
PY
    [[ "$rc" -eq 0 ]] || return "$rc"
    # Release the bootstrap's backoff so it presents now.
    compose_cmd "$current" exec -T remoted python3 /usr/local/libexec/remoted-bootstrap.py retry >/dev/null 2>&1 || true
}

# ADR-021: the iSite conversion names the support key in its migration token,
# so the key has to exist before remoted first runs. This generates it on the
# host with the remoted client's own parameters (the client adopts an existing
# key) and never replaces a key that is already there.
ensure_remoted_key() {
    local dir="$ROOT/data/remoted" name part
    for part in "$ROOT" "$ROOT/data" "$dir"; do
        [[ ! -L "$part" ]] || { die "$part must not be a symlink"; return 1; }
        [[ ! -e "$part" || -d "$part" ]] || { die "$part is not a directory"; return 1; }
    done
    mkdir -p "$dir" && chmod 0700 "$dir" || return 1
    for name in id_rsa id_rsa.pub; do
        if [[ -L "$dir/$name" ]] || { [[ -e "$dir/$name" ]] && [[ ! -f "$dir/$name" ]]; }; then
            die "$dir/$name is not a regular file"; return 1
        fi
    done
    if [[ ! -e "$dir/id_rsa" ]]; then
        rm -f "$dir/id_rsa.pub"
        ssh-keygen -q -t rsa -b 3072 -N '' -C remoted -f "$dir/id_rsa" </dev/null >/dev/null \
            || { die "ssh-keygen could not create the remote support key"; return 1; }
    elif [[ ! -s "$dir/id_rsa.pub" ]]; then
        ssh-keygen -y -f "$dir/id_rsa" </dev/null > "$dir/.id_rsa.pub.tmp" \
            && mv -f "$dir/.id_rsa.pub.tmp" "$dir/id_rsa.pub" \
            || { rm -f "$dir/.id_rsa.pub.tmp"; die "cannot derive the remote support public key"; return 1; }
    fi
    chmod 0600 "$dir/id_rsa" && chmod 0644 "$dir/id_rsa.pub"
}

remoted_key_fingerprint() {
    python3 - "$ROOT/data/remoted/id_rsa.pub" <<'PY'
import base64, binascii, hashlib, os, stat, struct, sys
fd = os.open(sys.argv[1], os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
try:
    info = os.fstat(fd)
    if not stat.S_ISREG(info.st_mode) or info.st_size > 16384:
        sys.exit("ERROR: the remote support public key is not a small regular file")
    fields = os.read(fd, 16384).decode("ascii", "replace").split()
finally:
    os.close(fd)
if len(fields) < 2 or fields[0] not in ("ssh-rsa", "ssh-ed25519"):
    sys.exit("ERROR: the remote support public key is not an ssh-rsa/ssh-ed25519 key")
try:
    blob = base64.b64decode(fields[1], validate=True)
except (binascii.Error, ValueError):
    sys.exit("ERROR: the remote support public key is not valid base64")
if len(blob) < 4 or blob[4:4 + struct.unpack(">I", blob[:4])[0]] != fields[0].encode():
    sys.exit("ERROR: the remote support public key blob does not match its type")
print("SHA256:" + base64.b64encode(hashlib.sha256(blob).digest()).decode().rstrip("="))
PY
}

cmd_remoted() {
    local action="${1:-status}" current environment relay token extra result
    [[ "$#" -eq 0 ]] || shift
    require_root || return 1
    if [[ "$action" == key ]]; then
        # Works before any install (the conversion needs it first).
        [[ "$#" -eq 0 ]] || { die "usage: nvr remoted key"; return 2; }
        ensure_remoted_key && remoted_key_fingerprint
        return
    fi
    require_install || return 1
    if [[ "$action" == enroll ]]; then acquire_mutation_lock || return 1; fi
    load_host_env || return 1
    current="$(current_release_id)" || return 1
    manifest_load "$RELEASES_DIR/$current/release.json" || return 1
    [[ "$MANIFEST_SCHEMA" == 2 ]] || { die "this release has no remote support service"; return 1; }
    if [[ "$action" == status && "$#" -eq 0 ]]; then
        local rc=0
        container_health "$current" remoted || rc=$?
        remoted_enrollment_status "$current"
        return "$rc"
    fi
    if [[ "$action" == register ]]; then
        remoted_register "$current" "$@"
        return
    fi
    if [[ "$action" == retry && "$#" -eq 0 ]]; then
        # ADR-020: release the bootstrap's backoff, e.g. right after approval.
        compose_cmd "$current" exec -T remoted python3 /usr/local/libexec/remoted-bootstrap.py retry
        return
    fi
    [[ "$action" == enroll && "$#" -eq 3 && "$1" == --environment && "$3" == --token-stdin ]] || {
        die "usage: nvr remoted status|retry|register [--token-file FILE] [--wait SECONDS]|enroll --environment prod|staging --token-stdin"; return 2;
    }
    environment="$2"
    case "$environment" in
        staging)
            [[ "$CHANNEL" == dev || "$CHANNEL" == dev-* ]] || { die "staging requires a development channel"; return 1; }
            [[ -f "$ROOT/config/remoted-staging-approved" && ! -L "$ROOT/config/remoted-staging-approved" ]] || {
                die "staging requires remoted-staging-approved"; return 1;
            }
            relay=relay-staging.cloud.dividia.net ;;
        prod)
            channel_is_production "$CHANNEL" || { die "production enrollment requires a production channel"; return 1; }
            relay=relay.cloud.dividia.net ;;
        *) die "invalid enrollment environment"; return 2 ;;
    esac
    ensure_remoted_host_authorization || return 1
    set +x
    IFS= read -r -n 65 token || [[ ${#token} -eq 64 ]] || return 1
    [[ ${#token} -eq 64 && "$token" != *[!0-9a-f]* ]] || return 1
    if IFS= read -r -n 1 extra; then token=; return 1; fi
    result=$(printf '%s\n' "$token" | compose_cmd "$current" exec -T remoted /usr/local/bin/remoted enroll \
        --host "$relay" --token-file /dev/stdin) || { token=; return 1; }
    token=
    [[ "$result" == 'Relay enrollment complete' ]] || return 1
    compose_cmd "$current" exec -T remoted test -s /root/.ssh/remoted-relay.conf || return 1
    compose_cmd "$current" restart remoted >/dev/null || return 1
    echo "Remote support enrollment complete"
}

usage() {
    printf '%s\n' \
        'Usage: nvr COMMAND [arguments]' \
        '' \
        '  status' \
        '  remoted status|retry|register [--token-file FILE] [--wait SECONDS]' \
        '  remoted enroll --environment prod|staging --token-stdin' \
        '  remoted key' \
        '  logs [dragon|connector|remoted] [-f]' \
        '  update [--force]' \
        '  ensure-host-tmux' \
        '  ensure-host-access' \
        '  ensure-host-name' \
        '  ensure-host-console [--remove | --resend]' \
        '  version' \
        '  channel [dev|dev-<suffix>|YYYY.MM.INCREMENT]' \
        '  environment [production|staging [--secret-file FILE]]' \
        '  start [dragon|connector|remoted]' \
        '  stop [dragon|connector|remoted]' \
        '  restart [dragon|connector|remoted]' \
        '  rollback' \
        '  hold [on|off|status]'
}

main() {
    local command="${1:-help}"
    [[ "$#" -eq 0 ]] || shift
    case "$command" in
        remoted) cmd_remoted "$@" ;;
        status) cmd_status "$@" ;;
        logs) cmd_logs "$@" ;;
        update) cmd_update "$@" ;;
        ensure-host-tmux) cmd_ensure_host_tmux "$@" ;;
        ensure-host-access) cmd_ensure_host_access "$@" ;;
        ensure-host-name) cmd_ensure_host_name "$@" ;;
        ensure-host-console) cmd_ensure_host_console "$@" ;;
        version) cmd_version "$@" ;;
        channel) cmd_channel "$@" ;;
        environment) cmd_environment "$@" ;;
        start) cmd_start "$@" ;;
        stop) cmd_stop "$@" ;;
        service-start) cmd_service_start "$@" ;;
        service-stop) cmd_service_stop "$@" ;;
        legacy-stop) cmd_legacy_stop "$@" ;;
        restart) cmd_restart "$@" ;;
        rollback) [[ "$#" -eq 0 ]] || { die "usage: nvr rollback"; return 2; }; cmd_rollback ;;
        hold) cmd_hold "$@" ;;
        help|-h|--help) usage ;;
        *) die "unknown command: $command"; usage >&2; return 2 ;;
    esac
}

main "$@"