#!/usr/bin/env bash
# =============================================================================
# P0-OBS-002 - isolated ClickHouse native RESTORE drill
#
# This verifier never joins a production network and never restores into an
# existing ClickHouse instance.  Apply mode stages the read-only source artifact
# through an exact-image, network-none container into a fresh writable volume,
# then creates one Docker internal network, one exact local ZooKeeper container,
# one exact local ClickHouse container and fresh ephemeral volumes.  Every owned
# resource is removed on every terminal.
# =============================================================================

set -euo pipefail

ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
INVENTORY_HELPER="${CLICKHOUSE_RESTORE_INVENTORY_HELPER:-${ROOT_DIR}/scripts/backup/clickhouse-restore-inventory.py}"
BACKUP_DISK_CONFIG="${CLICKHOUSE_RESTORE_BACKUP_DISK_CONFIG:-${ROOT_DIR}/ops/signoz/clickhouse/config.d/backup_disk.xml}"
ISOLATED_CLUSTER_CONFIG="${CLICKHOUSE_RESTORE_CLUSTER_CONFIG:-${ROOT_DIR}/ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml}"

BACKUP_DISK="backups"
CONTAINER_ARTIFACT_NAME="native-backup.zip"
CONTAINER_ARTIFACT_PATH="/backups/${CONTAINER_ARTIFACT_NAME}"
STAGING_SOURCE_ARTIFACT_PATH="/tmp/awoooi-source-${CONTAINER_ARTIFACT_NAME}"
RESTORE_SCOPE_SQL="DATABASE signoz_analytics,DATABASE signoz_logs,DATABASE signoz_metadata,DATABASE signoz_meter,DATABASE signoz_metrics,DATABASE signoz_traces"
DATABASE_IN_SQL="'signoz_analytics','signoz_logs','signoz_metadata','signoz_meter','signoz_metrics','signoz_traces'"
EMPTY_SHA256="E3B0C44298FC1C149AFBF4C8996FB92427AE41E4649B934CA495991B7852B855"

COMMAND_TIMEOUT_SECONDS="${CLICKHOUSE_RESTORE_COMMAND_TIMEOUT_SECONDS:-60}"
STARTUP_TIMEOUT_SECONDS="${CLICKHOUSE_RESTORE_STARTUP_TIMEOUT_SECONDS:-180}"
RESTORE_TIMEOUT_SECONDS="${CLICKHOUSE_RESTORE_TIMEOUT_SECONDS:-1800}"
POLL_INTERVAL_SECONDS="${CLICKHOUSE_RESTORE_POLL_INTERVAL_SECONDS:-2}"
KILL_AFTER_SECONDS="${CLICKHOUSE_RESTORE_KILL_AFTER_SECONDS:-15}"
MAX_LOG_BYTES="${CLICKHOUSE_RESTORE_MAX_LOG_BYTES:-262144}"
LOG_TAIL_LINES="${CLICKHOUSE_RESTORE_LOG_TAIL_LINES:-200}"

MODE=""
TRACE_ID="${TRACE_ID:-}"
RUN_ID="${RUN_ID:-}"
WORK_ITEM_ID="${WORK_ITEM_ID:-}"
ARTIFACT_PATH="${CLICKHOUSE_NATIVE_ARTIFACT_PATH:-}"
MANIFEST_PATH="${CLICKHOUSE_NATIVE_MANIFEST_PATH:-}"
SOURCE_INVENTORY_PATH="${CLICKHOUSE_NATIVE_SOURCE_INVENTORY_PATH:-}"
CLICKHOUSE_IMAGE="${CLICKHOUSE_RESTORE_CLICKHOUSE_IMAGE:-clickhouse/clickhouse-server:25.5.6}"
CLICKHOUSE_IMAGE_ID_EXPECTED="${CLICKHOUSE_RESTORE_CLICKHOUSE_IMAGE_ID:-sha256:8248e5926d7304400e44ecdf0c7181fbde28e6a9b1d647780eca839f34c00cc6}"
ZOOKEEPER_IMAGE="${CLICKHOUSE_RESTORE_ZOOKEEPER_IMAGE:-signoz/zookeeper:3.7.1}"
ZOOKEEPER_IMAGE_ID_EXPECTED="${CLICKHOUSE_RESTORE_ZOOKEEPER_IMAGE_ID:-sha256:3ab0e8f032ab58f14ac1e929ac40305a4a464cf320bdfe4151d62d6922729ba0}"
RECEIPT_ROOT="${CLICKHOUSE_RESTORE_RECEIPT_ROOT:-/backup/logs/clickhouse-native-restore-drill}"
HOOK_ARTIFACT_SHA256="${CLICKHOUSE_NATIVE_ARTIFACT_SHA256:-}"
HOOK_OPERATION_ID="${CLICKHOUSE_NATIVE_OPERATION_ID:-}"
HOOK_EXPECTED_DATABASES="${CLICKHOUSE_NATIVE_EXPECTED_DATABASES:-}"

IDENTITY_HASH=""
SHORT_ID=""
RESTORE_REPLICA=""
RESTORE_OPERATION_ID=""
NETWORK_NAME=""
STAGING_CONTAINER=""
CLICKHOUSE_CONTAINER=""
ZOOKEEPER_CONTAINER=""
BACKUP_VOLUME=""
CLICKHOUSE_VOLUME=""
ZOOKEEPER_DATA_VOLUME=""
ZOOKEEPER_LOG_VOLUME=""

RECEIPT_DIR=""
RECEIPT_LOG=""
STATUS_FILE=""
PREFLIGHT_FILE=""
PREFLIGHT_CANDIDATE=""
LOCK_DIR=""
INVOCATION_ID=""
RECEIPTS_READY=0
COMMAND_SEQUENCE=0
LAST_STDOUT=""
LAST_STDERR=""
LAST_EXIT_CODE=0

ARTIFACT_SHA256=""
STAGED_ARTIFACT_SHA256=""
CLICKHOUSE_ARTIFACT_SHA256=""
SOURCE_INVENTORY_SHA256=""
BACKUP_OPERATION_ID=""
SOURCE_TABLE_COUNT=0
SOURCE_DATABASE_COUNT=0
REPLICATED_TABLE_COUNT=0
DISTRIBUTED_TABLE_COUNT=0
PHYSICAL_TABLE_COUNT=0
CLICKHOUSE_IMAGE_ID=""
ZOOKEEPER_IMAGE_ID=""
BACKUP_CONFIG_SHA256=""
CLUSTER_CONFIG_SHA256=""

RESTORE_TERMINAL="not_started"
RESTORE_NUM_FILES=0
RESTORE_TOTAL_SIZE=0
MANIFEST_PARITY=0
RESTORED_DATABASE_COUNT=0
RESTORED_TABLE_COUNT=0
CRITICAL_NONZERO_COUNT=0
CHECK_TABLE_COUNT=0
CHECK_TABLE_PASS_COUNT=0
SOURCE_TOTAL_ROWS=0
RESTORED_TOTAL_ROWS=0
ROW_DELTA=0
SOURCE_TOTAL_BYTES=0
RESTORED_TOTAL_BYTES=0
BYTE_DELTA=0

NETWORK_CREATED=0
STAGING_CONTAINER_CREATED=0
CLICKHOUSE_CONTAINER_CREATED=0
ZOOKEEPER_CONTAINER_CREATED=0
BACKUP_VOLUME_CREATED=0
CLICKHOUSE_VOLUME_CREATED=0
ZOOKEEPER_DATA_VOLUME_CREATED=0
ZOOKEEPER_LOG_VOLUME_CREATED=0
EPHEMERAL_RUNTIME_CREATED=0
ARTIFACT_STAGING_VERIFIED=0
CLICKHOUSE_ARTIFACT_READBACK_VERIFIED=0
APPLY_RESOURCE_SCOPE=0
CHECK_PASS=0
APPLY_PASS=0
CLEANUP_PASS=1
FINAL_REASON="not_started"

usage() {
    cat <<'USAGE'
Usage:
  clickhouse-native-restore-drill.sh --check|--apply \
    --trace-id ID --run-id ID --work-item-id ID \
    --artifact ABSOLUTE_PATH --manifest ABSOLUTE_PATH \
    --source-inventory ABSOLUTE_PATH \
    [--clickhouse-image REF --clickhouse-image-id sha256:...] \
    [--zookeeper-image signoz/zookeeper:3.7.1] \
    [--zookeeper-image-id sha256:...] [--receipt-root ABSOLUTE_PATH]

--check performs local artifact/config validation plus read-only Docker image
inspection.  It creates no container, volume or network.  --apply requires the
same run's check receipt, stages the source through an exact-image network-none
container into a writable owned backup volume, then restores only into fresh
ephemeral volumes on an internal-only Docker network.  It never attaches to
production and publishes no ports.  All ephemeral resources are removed on
every terminal.

When invoked as CLICKHOUSE_NATIVE_POST_VERIFY_HOOK, TRACE_ID/RUN_ID/
WORK_ITEM_ID and CLICKHOUSE_NATIVE_{ARTIFACT,MANIFEST,SOURCE_INVENTORY}_PATH
are accepted directly from the adapter's bounded environment.  Deployments may
set CLICKHOUSE_RESTORE_{INVENTORY_HELPER,BACKUP_DISK_CONFIG,CLUSTER_CONFIG} to
reviewed absolute host paths; no env file is read.
USAGE
}

log_info() { printf '[INFO] %s\n' "$*"; }
log_error() { printf '[ERROR] %s\n' "$*" >&2; }

valid_identity() {
    [[ "$1" =~ ^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$ ]]
}

valid_positive_integer() {
    [[ "$1" =~ ^[0-9]+$ ]] && [ "$1" -gt 0 ]
}

valid_image_ref() {
    [[ "$1" =~ ^[A-Za-z0-9][A-Za-z0-9._/@:-]{0,255}$ ]]
}

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

require_value() {
    local option="$1"
    local value="${2:-}"
    [ -n "${value}" ] || { log_error "${option} requires a value"; exit 64; }
}

set_mode() {
    local requested="$1"
    [ -z "${MODE}" ] || { log_error "exactly one of --check or --apply is required"; exit 64; }
    MODE="${requested}"
}

parse_args() {
    while [ "$#" -gt 0 ]; do
        case "$1" in
            --check) set_mode check ;;
            --apply) set_mode apply ;;
            --trace-id)
                require_value "$1" "${2:-}"
                TRACE_ID="$2"
                shift
                ;;
            --run-id)
                require_value "$1" "${2:-}"
                RUN_ID="$2"
                shift
                ;;
            --work-item-id)
                require_value "$1" "${2:-}"
                WORK_ITEM_ID="$2"
                shift
                ;;
            --artifact)
                require_value "$1" "${2:-}"
                ARTIFACT_PATH="$2"
                shift
                ;;
            --manifest)
                require_value "$1" "${2:-}"
                MANIFEST_PATH="$2"
                shift
                ;;
            --source-inventory)
                require_value "$1" "${2:-}"
                SOURCE_INVENTORY_PATH="$2"
                shift
                ;;
            --clickhouse-image)
                require_value "$1" "${2:-}"
                CLICKHOUSE_IMAGE="$2"
                shift
                ;;
            --clickhouse-image-id)
                require_value "$1" "${2:-}"
                CLICKHOUSE_IMAGE_ID_EXPECTED="$2"
                shift
                ;;
            --zookeeper-image)
                require_value "$1" "${2:-}"
                ZOOKEEPER_IMAGE="$2"
                shift
                ;;
            --zookeeper-image-id)
                require_value "$1" "${2:-}"
                ZOOKEEPER_IMAGE_ID_EXPECTED="$2"
                shift
                ;;
            --receipt-root)
                require_value "$1" "${2:-}"
                RECEIPT_ROOT="$2"
                shift
                ;;
            -h|--help)
                usage
                exit 0
                ;;
            *)
                log_error "unsupported argument: $1"
                usage >&2
                exit 64
                ;;
        esac
        shift
    done
}

validate_regular_file() {
    local label="$1"
    local path="$2"
    [ -n "${path}" ] && [[ "${path}" == /* ]] && [[ "${path}" != *$'\n'* ]] \
        || { log_error "${label} must be an absolute single-line path"; exit 64; }
    [ -f "${path}" ] && [ ! -L "${path}" ] \
        || { log_error "${label} must be a non-symlink regular file"; exit 64; }
}

validate_contract() {
    [ -n "${MODE}" ] || { log_error "exactly one of --check or --apply is required"; exit 64; }
    valid_identity "${TRACE_ID}" || { log_error "--trace-id is required and path-safe"; exit 64; }
    valid_identity "${RUN_ID}" || { log_error "--run-id is required and path-safe"; exit 64; }
    valid_identity "${WORK_ITEM_ID}" || { log_error "--work-item-id is required and path-safe"; exit 64; }
    valid_image_ref "${CLICKHOUSE_IMAGE}" || { log_error "invalid ClickHouse image reference"; exit 64; }
    valid_image_id "${CLICKHOUSE_IMAGE_ID_EXPECTED}" || { log_error "invalid expected ClickHouse image ID"; exit 64; }
    valid_image_ref "${ZOOKEEPER_IMAGE}" || { log_error "invalid ZooKeeper image reference"; exit 64; }
    [ "${ZOOKEEPER_IMAGE}" = "signoz/zookeeper:3.7.1" ] \
        || { log_error "ZooKeeper image must be the reviewed signoz/zookeeper:3.7.1 reference"; exit 64; }
    valid_image_id "${ZOOKEEPER_IMAGE_ID_EXPECTED}" || { log_error "invalid expected ZooKeeper image ID"; exit 64; }
    [[ "${RECEIPT_ROOT}" == /* ]] && [[ "${RECEIPT_ROOT}" != *$'\n'* ]] \
        || { log_error "receipt root must be an absolute single-line path"; exit 64; }
    [ ! -L "${RECEIPT_ROOT}" ] || { log_error "receipt root symlink is unsafe"; exit 64; }
    validate_regular_file artifact "${ARTIFACT_PATH}"
    validate_regular_file manifest "${MANIFEST_PATH}"
    validate_regular_file source_inventory "${SOURCE_INVENTORY_PATH}"
    validate_regular_file backup_disk_config "${BACKUP_DISK_CONFIG}"
    validate_regular_file isolated_cluster_config "${ISOLATED_CLUSTER_CONFIG}"
    validate_regular_file inventory_helper "${INVENTORY_HELPER}"
    if [ -n "${HOOK_EXPECTED_DATABASES}" ]; then
        [ "${HOOK_EXPECTED_DATABASES}" = "signoz_analytics,signoz_logs,signoz_metadata,signoz_meter,signoz_metrics,signoz_traces" ] \
            || { log_error "hook database allowlist mismatch"; exit 64; }
    fi

    for value in "${COMMAND_TIMEOUT_SECONDS}" "${STARTUP_TIMEOUT_SECONDS}" \
        "${RESTORE_TIMEOUT_SECONDS}" "${POLL_INTERVAL_SECONDS}" \
        "${KILL_AFTER_SECONDS}" "${MAX_LOG_BYTES}" "${LOG_TAIL_LINES}"; do
        valid_positive_integer "${value}" || { log_error "timeout/log limits must be positive integers"; exit 64; }
    done
    for command_name in awk basename cat cmp date docker grep head mkdir mktemp mv python3 rm \
        rmdir sed sha256sum sleep timeout touch tr unzip wc; do
        command -v "${command_name}" >/dev/null 2>&1 \
            || { log_error "required command missing: ${command_name}"; exit 64; }
    done

    grep -F -q '<allowed_disk>backups</allowed_disk>' "${BACKUP_DISK_CONFIG}" \
        || { log_error "repo-owned backup disk config does not allow backups disk"; exit 64; }
    grep -F -q '<allowed_path>/backups/</allowed_path>' "${BACKUP_DISK_CONFIG}" \
        || { log_error "repo-owned backup disk config path mismatch"; exit 64; }
    grep -F -q '<host>restore-zookeeper</host>' "${ISOLATED_CLUSTER_CONFIG}" \
        || { log_error "isolated ZooKeeper host contract missing"; exit 64; }
    grep -F -q '<cluster>' "${ISOLATED_CLUSTER_CONFIG}" \
        || { log_error "isolated cluster name contract missing"; exit 64; }
    grep -F -q '<shard>01</shard>' "${ISOLATED_CLUSTER_CONFIG}" \
        || { log_error "isolated shard macro contract missing"; exit 64; }
    grep -F -q 'from_env="CLICKHOUSE_RESTORE_REPLICA"' "${ISOLATED_CLUSTER_CONFIG}" \
        || { log_error "run-derived replica macro contract missing"; exit 64; }
}

sha256_file() {
    sha256sum "$1" | awk '{print $1}'
}

emit_receipt() {
    local stage="$1"
    local terminal="$2"
    local detail="$3"
    local observed_at
    [ "${RECEIPTS_READY}" -eq 1 ] || return 0
    observed_at="$(date -u '+%Y-%m-%dT%H:%M:%SZ')"
    printf '{"trace_id":"%s","run_id":"%s","work_item_id":"%s","invocation_id":"%s","observed_at":"%s","stage":"%s","terminal":"%s","detail":"%s"}\n' \
        "${TRACE_ID}" "${RUN_ID}" "${WORK_ITEM_ID}" "${INVOCATION_ID}" \
        "${observed_at}" "${stage}" "${terminal}" "${detail}" >> "${RECEIPT_LOG}"
}

fail_closed() {
    FINAL_REASON="$1"
    log_error "${FINAL_REASON}"
    emit_receipt failure failed "${FINAL_REASON}"
    exit 1
}

setup_receipts() {
    local identity_file identity_content identity_tmp

    IDENTITY_HASH="$(printf '%s\000%s\000%s' "${TRACE_ID}" "${RUN_ID}" "${WORK_ITEM_ID}" | sha256sum | awk '{print $1}')"
    SHORT_ID="${IDENTITY_HASH:0:16}"
    RESTORE_REPLICA="restore-${SHORT_ID}"
    RESTORE_OPERATION_ID="restore-${IDENTITY_HASH}"
    NETWORK_NAME="awoooi-ch-restore-${SHORT_ID}-net"
    STAGING_CONTAINER="awoooi-ch-restore-${SHORT_ID}-stage"
    CLICKHOUSE_CONTAINER="awoooi-ch-restore-${SHORT_ID}"
    ZOOKEEPER_CONTAINER="awoooi-zk-restore-${SHORT_ID}"
    BACKUP_VOLUME="awoooi-ch-restore-${SHORT_ID}-backup"
    CLICKHOUSE_VOLUME="awoooi-ch-restore-${SHORT_ID}-data"
    ZOOKEEPER_DATA_VOLUME="awoooi-zk-restore-${SHORT_ID}-data"
    ZOOKEEPER_LOG_VOLUME="awoooi-zk-restore-${SHORT_ID}-log"

    mkdir -p "${RECEIPT_ROOT}"
    RECEIPT_DIR="${RECEIPT_ROOT}/${RUN_ID}"
    [ ! -L "${RECEIPT_DIR}" ] || { log_error "receipt directory symlink is unsafe"; exit 64; }
    mkdir -m 700 -p "${RECEIPT_DIR}"
    LOCK_DIR="${RECEIPT_DIR}/active.lock"
    mkdir "${LOCK_DIR}" 2>/dev/null \
        || { log_error "same restore-drill run is active or has a stale lock"; exit 1; }

    RECEIPT_LOG="${RECEIPT_DIR}/receipts.jsonl"
    STATUS_FILE="${RECEIPT_DIR}/status.json"
    PREFLIGHT_FILE="${RECEIPT_DIR}/preflight.contract"
    INVOCATION_ID="${MODE}-$(date -u '+%Y%m%dT%H%M%SZ')-$$"
    PREFLIGHT_CANDIDATE="${RECEIPT_DIR}/${INVOCATION_ID}.preflight.candidate"
    identity_file="${RECEIPT_DIR}/identity.json"
    identity_content="{\"trace_id\":\"${TRACE_ID}\",\"run_id\":\"${RUN_ID}\",\"work_item_id\":\"${WORK_ITEM_ID}\",\"identity_hash\":\"${IDENTITY_HASH}\"}"
    if [ -e "${identity_file}" ]; then
        [ -f "${identity_file}" ] && [ ! -L "${identity_file}" ] \
            || { rmdir "${LOCK_DIR}"; log_error "existing identity receipt is unsafe"; exit 1; }
        [ "$(sed -n '1p' "${identity_file}")" = "${identity_content}" ] \
            || { rmdir "${LOCK_DIR}"; log_error "run identity conflicts with durable receipt"; exit 1; }
    else
        identity_tmp="$(mktemp "${RECEIPT_DIR}/.identity.XXXXXX")"
        printf '%s\n' "${identity_content}" > "${identity_tmp}"
        mv "${identity_tmp}" "${identity_file}"
    fi
    touch "${RECEIPT_LOG}"
    RECEIPTS_READY=1
    trap cleanup EXIT
    trap 'on_signal 1' HUP
    trap 'on_signal 2' INT
    trap 'on_signal 15' TERM
}

run_captured() {
    local step="$1"
    local stdout_tmp stderr_tmp stdout_size stderr_size stdout_hash stderr_hash truncated=0
    shift
    [[ "${step}" =~ ^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$ ]] || fail_closed invalid_command_step

    COMMAND_SEQUENCE=$((COMMAND_SEQUENCE + 1))
    LAST_STDOUT="${RECEIPT_DIR}/${INVOCATION_ID}-$(printf '%03d' "${COMMAND_SEQUENCE}")-${step}.stdout"
    LAST_STDERR="${RECEIPT_DIR}/${INVOCATION_ID}-$(printf '%03d' "${COMMAND_SEQUENCE}")-${step}.stderr"
    stdout_tmp="$(mktemp "${RECEIPT_DIR}/.stdout.XXXXXX")"
    stderr_tmp="$(mktemp "${RECEIPT_DIR}/.stderr.XXXXXX")"

    set +e
    "$@" >"${stdout_tmp}" 2>"${stderr_tmp}"
    LAST_EXIT_CODE=$?
    set -e
    stdout_size="$(wc -c < "${stdout_tmp}" | tr -d ' ')"
    stderr_size="$(wc -c < "${stderr_tmp}" | tr -d ' ')"
    head -c "${MAX_LOG_BYTES}" "${stdout_tmp}" > "${LAST_STDOUT}"
    head -c "${MAX_LOG_BYTES}" "${stderr_tmp}" > "${LAST_STDERR}"
    rm -f "${stdout_tmp}" "${stderr_tmp}"
    if [ "${stdout_size}" -gt "${MAX_LOG_BYTES}" ] || [ "${stderr_size}" -gt "${MAX_LOG_BYTES}" ]; then
        truncated=1
        [ "${LAST_EXIT_CODE}" -ne 0 ] || LAST_EXIT_CODE=70
    fi
    stdout_hash="$(sha256_file "${LAST_STDOUT}")"
    stderr_hash="$(sha256_file "${LAST_STDERR}")"
    printf '%s\n' "${LAST_EXIT_CODE}" > "${LAST_STDOUT}.status"
    emit_receipt command "$([ "${LAST_EXIT_CODE}" -eq 0 ] && printf pass || printf failed)" \
        "step_${step}_exit_${LAST_EXIT_CODE}_stdout_bytes_${stdout_size}_stderr_bytes_${stderr_size}_truncated_${truncated}_stdout_${stdout_hash}_stderr_${stderr_hash}"
    return "${LAST_EXIT_CODE}"
}

run_sql() {
    local step="$1"
    local query="$2"
    shift 2
    run_captured "${step}" timeout --kill-after="${KILL_AFTER_SECONDS}" \
        "${COMMAND_TIMEOUT_SECONDS}" docker exec "${CLICKHOUSE_CONTAINER}" \
        clickhouse-client --format TSVRaw "$@" --query "${query}"
}

read_key() {
    local key="$1"
    local file="$2"
    awk -F= -v wanted="${key}" '$1 == wanted {sub(/^[^=]*=/, ""); print; exit}' "${file}"
}

validate_preflight() {
    local image_line

    run_captured artifact_zip timeout --kill-after="${KILL_AFTER_SECONDS}" \
        "${COMMAND_TIMEOUT_SECONDS}" unzip -tq "${ARTIFACT_PATH}" \
        || fail_closed native_backup_zip_integrity_failed
    run_captured manifest_inventory python3 "${INVENTORY_HELPER}" preflight \
        --manifest "${MANIFEST_PATH}" --artifact "${ARTIFACT_PATH}" \
        --source-inventory "${SOURCE_INVENTORY_PATH}" \
        || fail_closed native_backup_manifest_or_source_inventory_failed

    ARTIFACT_SHA256="$(read_key artifact_sha256 "${LAST_STDOUT}")"
    SOURCE_INVENTORY_SHA256="$(read_key source_inventory_sha256 "${LAST_STDOUT}")"
    BACKUP_OPERATION_ID="$(read_key backup_operation_id "${LAST_STDOUT}")"
    SOURCE_TABLE_COUNT="$(read_key source_table_count "${LAST_STDOUT}")"
    SOURCE_DATABASE_COUNT="$(read_key source_database_count "${LAST_STDOUT}")"
    REPLICATED_TABLE_COUNT="$(read_key replicated_table_count "${LAST_STDOUT}")"
    DISTRIBUTED_TABLE_COUNT="$(read_key distributed_table_count "${LAST_STDOUT}")"
    PHYSICAL_TABLE_COUNT="$(read_key physical_table_count "${LAST_STDOUT}")"
    [[ "${ARTIFACT_SHA256}" =~ ^[0-9a-f]{64}$ ]] || fail_closed artifact_hash_receipt_invalid
    [[ "${SOURCE_INVENTORY_SHA256}" =~ ^[0-9a-f]{64}$ ]] || fail_closed inventory_hash_receipt_invalid
    valid_identity "${BACKUP_OPERATION_ID}" || fail_closed backup_operation_receipt_invalid
    if [ -n "${HOOK_ARTIFACT_SHA256}" ]; then
        [ "${HOOK_ARTIFACT_SHA256}" = "${ARTIFACT_SHA256}" ] \
            || fail_closed hook_artifact_sha256_mismatch
    fi
    if [ -n "${HOOK_OPERATION_ID}" ]; then
        [ "${HOOK_OPERATION_ID}" = "${BACKUP_OPERATION_ID}" ] \
            || fail_closed hook_backup_operation_id_mismatch
    fi

    run_captured clickhouse_image timeout --kill-after="${KILL_AFTER_SECONDS}" \
        "${COMMAND_TIMEOUT_SECONDS}" docker image inspect --format '{{.Id}}' "${CLICKHOUSE_IMAGE}" \
        || fail_closed clickhouse_local_image_missing
    image_line="$(tr -d '\r\n' < "${LAST_STDOUT}")"
    [ "${image_line}" = "${CLICKHOUSE_IMAGE_ID_EXPECTED}" ] \
        || fail_closed clickhouse_local_image_id_mismatch
    CLICKHOUSE_IMAGE_ID="${image_line}"

    run_captured zookeeper_image timeout --kill-after="${KILL_AFTER_SECONDS}" \
        "${COMMAND_TIMEOUT_SECONDS}" docker image inspect --format '{{.Id}}' "${ZOOKEEPER_IMAGE}" \
        || fail_closed zookeeper_local_image_missing
    image_line="$(tr -d '\r\n' < "${LAST_STDOUT}")"
    [ "${image_line}" = "${ZOOKEEPER_IMAGE_ID_EXPECTED}" ] \
        || fail_closed zookeeper_local_image_id_mismatch
    ZOOKEEPER_IMAGE_ID="${image_line}"

    BACKUP_CONFIG_SHA256="$(sha256_file "${BACKUP_DISK_CONFIG}")"
    CLUSTER_CONFIG_SHA256="$(sha256_file "${ISOLATED_CLUSTER_CONFIG}")"
    {
        printf 'identity_hash=%s\n' "${IDENTITY_HASH}"
        printf 'artifact_path=%s\n' "${ARTIFACT_PATH}"
        printf 'artifact_sha256=%s\n' "${ARTIFACT_SHA256}"
        printf 'manifest_path=%s\n' "${MANIFEST_PATH}"
        printf 'source_inventory_path=%s\n' "${SOURCE_INVENTORY_PATH}"
        printf 'source_inventory_sha256=%s\n' "${SOURCE_INVENTORY_SHA256}"
        printf 'backup_operation_id=%s\n' "${BACKUP_OPERATION_ID}"
        printf 'clickhouse_image=%s\n' "${CLICKHOUSE_IMAGE}"
        printf 'clickhouse_image_id=%s\n' "${CLICKHOUSE_IMAGE_ID}"
        printf 'zookeeper_image=%s\n' "${ZOOKEEPER_IMAGE}"
        printf 'zookeeper_image_id=%s\n' "${ZOOKEEPER_IMAGE_ID}"
        printf 'backup_config_sha256=%s\n' "${BACKUP_CONFIG_SHA256}"
        printf 'cluster_config_sha256=%s\n' "${CLUSTER_CONFIG_SHA256}"
        printf 'restore_replica=%s\n' "${RESTORE_REPLICA}"
        printf 'source_database_count=%s\n' "${SOURCE_DATABASE_COUNT}"
        printf 'source_table_count=%s\n' "${SOURCE_TABLE_COUNT}"
        printf 'replicated_table_count=%s\n' "${REPLICATED_TABLE_COUNT}"
        printf 'distributed_table_count=%s\n' "${DISTRIBUTED_TABLE_COUNT}"
        printf 'physical_table_count=%s\n' "${PHYSICAL_TABLE_COUNT}"
    } > "${PREFLIGHT_CANDIDATE}"
    emit_receipt sensor_source pass \
        "backup_${BACKUP_OPERATION_ID}_artifact_${ARTIFACT_SHA256}_inventory_${SOURCE_INVENTORY_SHA256}"
    emit_receipt normalized_asset_identity pass \
        "restore_${IDENTITY_HASH}_clickhouse_${CLICKHOUSE_IMAGE_ID}_zookeeper_${ZOOKEEPER_IMAGE_ID}"
    emit_receipt risk_policy pass \
        "risk_high_isolated_internal_network_fresh_keeper_fresh_data_volumes_no_production_attachment"
}

persist_check_contract() {
    if [ -e "${PREFLIGHT_FILE}" ]; then
        [ -f "${PREFLIGHT_FILE}" ] && [ ! -L "${PREFLIGHT_FILE}" ] \
            || fail_closed existing_preflight_contract_unsafe
        cmp -s "${PREFLIGHT_CANDIDATE}" "${PREFLIGHT_FILE}" \
            || fail_closed same_run_preflight_contract_drift
        rm -f "${PREFLIGHT_CANDIDATE}"
    else
        mv "${PREFLIGHT_CANDIDATE}" "${PREFLIGHT_FILE}"
    fi
    emit_receipt source_of_truth_diff pass \
        "exact_six_databases_tables_${SOURCE_TABLE_COUNT}_replicated_${REPLICATED_TABLE_COUNT}_distributed_${DISTRIBUTED_TABLE_COUNT}"
    emit_receipt ai_decision candidate \
        "isolated_keeper_clickhouse_restore_requires_same_run_apply"
    emit_receipt check check_pass_no_runtime_write \
        "local_artifact_config_and_exact_local_images_valid_no_container_volume_or_network_created"
    CHECK_PASS=1
    FINAL_REASON="check_pass_no_runtime_write"
}

require_check_contract() {
    [ -f "${PREFLIGHT_FILE}" ] && [ ! -L "${PREFLIGHT_FILE}" ] \
        || fail_closed same_run_check_contract_missing
    cmp -s "${PREFLIGHT_CANDIDATE}" "${PREFLIGHT_FILE}" \
        || fail_closed apply_preflight_contract_drift
    rm -f "${PREFLIGHT_CANDIDATE}"
    grep -F -q '"stage":"terminal","terminal":"check_pass_no_runtime_write"' "${RECEIPT_LOG}" \
        || fail_closed same_run_check_terminal_receipt_missing
    CHECK_PASS=1
    emit_receipt check pass apply_revalidation_matches_same_run_check_contract
}

ensure_resources_absent() {
    docker_daemon_ready || fail_closed docker_daemon_unavailable_before_restore_create
    exact_container_absent "${CLICKHOUSE_CONTAINER}" \
        || fail_closed isolated_clickhouse_container_name_collision_or_query_failed
    exact_container_absent "${ZOOKEEPER_CONTAINER}" \
        || fail_closed isolated_zookeeper_container_name_collision_or_query_failed
    exact_container_absent "${STAGING_CONTAINER}" \
        || fail_closed isolated_staging_container_name_collision_or_query_failed
    exact_volume_absent "${BACKUP_VOLUME}" \
        || fail_closed isolated_backup_volume_name_collision_or_query_failed
    exact_volume_absent "${CLICKHOUSE_VOLUME}" \
        || fail_closed isolated_clickhouse_volume_name_collision_or_query_failed
    exact_volume_absent "${ZOOKEEPER_DATA_VOLUME}" \
        || fail_closed isolated_zookeeper_data_volume_name_collision_or_query_failed
    exact_volume_absent "${ZOOKEEPER_LOG_VOLUME}" \
        || fail_closed isolated_zookeeper_log_volume_name_collision_or_query_failed
    exact_network_absent "${NETWORK_NAME}" \
        || fail_closed isolated_network_name_collision_or_query_failed
    emit_receipt source_of_truth_diff pass exact_ephemeral_resource_names_absent
}

docker_daemon_ready() {
    timeout --kill-after="${KILL_AFTER_SECONDS}" "${COMMAND_TIMEOUT_SECONDS}" \
        docker info --format '{{.ServerVersion}}' >/dev/null 2>&1
}

exact_container_absent() {
    local name="$1" observed
    observed="$(timeout --kill-after="${KILL_AFTER_SECONDS}" "${COMMAND_TIMEOUT_SECONDS}" \
        docker container ls -a --format '{{.Names}}')" || return 1
    ! grep -F -x -q -- "${name}" <<< "${observed}"
}

exact_volume_absent() {
    local name="$1" observed
    observed="$(timeout --kill-after="${KILL_AFTER_SECONDS}" "${COMMAND_TIMEOUT_SECONDS}" \
        docker volume ls -q)" || return 1
    ! grep -F -x -q -- "${name}" <<< "${observed}"
}

exact_network_absent() {
    local name="$1" observed
    observed="$(timeout --kill-after="${KILL_AFTER_SECONDS}" "${COMMAND_TIMEOUT_SECONDS}" \
        docker network ls --format '{{.Name}}')" || return 1
    ! grep -F -x -q -- "${name}" <<< "${observed}"
}

wait_for_container_running() {
    local label="$1"
    local container="$2"
    local deadline state
    deadline=$(( $(date +%s) + STARTUP_TIMEOUT_SECONDS ))
    while true; do
        if run_captured "${label}_running" timeout --kill-after="${KILL_AFTER_SECONDS}" \
            "${COMMAND_TIMEOUT_SECONDS}" docker inspect --format '{{.State.Running}}' "${container}"; then
            state="$(tr -d '\r\n' < "${LAST_STDOUT}")"
            [ "${state}" = "true" ] && return 0
        fi
        [ "$(date +%s)" -lt "${deadline}" ] || fail_closed "${label}_startup_timeout"
        sleep "${POLL_INTERVAL_SECONDS}"
    done
}

wait_for_clickhouse() {
    local deadline
    deadline=$(( $(date +%s) + STARTUP_TIMEOUT_SECONDS ))
    while true; do
        if run_sql clickhouse_ready 'SELECT 1'; then
            [ "$(tr -d '\r\n' < "${LAST_STDOUT}")" = "1" ] && return 0
        fi
        [ "$(date +%s)" -lt "${deadline}" ] || fail_closed clickhouse_startup_timeout
        sleep "${POLL_INTERVAL_SECONDS}"
    done
}

stage_artifact_into_backup_volume() {
    local staged_hash

    run_captured artifact_stage timeout --kill-after="${KILL_AFTER_SECONDS}" \
        "${COMMAND_TIMEOUT_SECONDS}" docker run --pull=never \
        --name "${STAGING_CONTAINER}" --network none --restart=no --user 0:0 \
        --label "com.awoooi.restore.identity=${IDENTITY_HASH}" \
        --env "EXPECTED_ARTIFACT_SHA256=${ARTIFACT_SHA256}" \
        --mount "type=bind,src=${ARTIFACT_PATH},dst=${STAGING_SOURCE_ARTIFACT_PATH},readonly" \
        --mount "type=volume,src=${BACKUP_VOLUME},dst=/backups" \
        --entrypoint /bin/sh "${CLICKHOUSE_IMAGE}" -ceu \
        'cp -- /tmp/awoooi-source-native-backup.zip /backups/native-backup.zip
         sync /backups/native-backup.zip
         set -- $(sha256sum /backups/native-backup.zip)
         [ "$1" = "$EXPECTED_ARTIFACT_SHA256" ]
         printf "%s\n" "$1"' \
        || fail_closed isolated_artifact_staging_failed
    STAGING_CONTAINER_CREATED=1
    staged_hash="$(tr -d '\r\n' < "${LAST_STDOUT}")"
    [[ "${staged_hash}" =~ ^[0-9a-f]{64}$ ]] \
        || fail_closed staged_artifact_hash_readback_invalid
    [ "${staged_hash}" = "${ARTIFACT_SHA256}" ] \
        || fail_closed staged_artifact_hash_mismatch
    STAGED_ARTIFACT_SHA256="${staged_hash}"

    run_captured staging_image_readback timeout --kill-after="${KILL_AFTER_SECONDS}" \
        "${COMMAND_TIMEOUT_SECONDS}" docker inspect --format '{{.Image}}' "${STAGING_CONTAINER}" \
        || fail_closed isolated_staging_image_readback_failed
    [ "$(tr -d '\r\n' < "${LAST_STDOUT}")" = "${CLICKHOUSE_IMAGE_ID_EXPECTED}" ] \
        || fail_closed isolated_staging_runtime_image_id_mismatch
    run_captured staging_network_readback timeout --kill-after="${KILL_AFTER_SECONDS}" \
        "${COMMAND_TIMEOUT_SECONDS}" docker inspect --format '{{.HostConfig.NetworkMode}}' \
        "${STAGING_CONTAINER}" \
        || fail_closed isolated_staging_network_readback_failed
    [ "$(tr -d '\r\n' < "${LAST_STDOUT}")" = none ] \
        || fail_closed isolated_staging_network_not_none
    ARTIFACT_STAGING_VERIFIED=1
    emit_receipt artifact_staging pass \
        "read_only_source_to_writable_owned_volume_sha256_${STAGED_ARTIFACT_SHA256}_network_none_exact_clickhouse_image"
}

create_isolated_runtime() {
    local clickhouse_artifact_hash

    # From this point onward cleanup inspects every exact resource name even if
    # a collision or Docker CLI timeout occurs before a *_CREATED flag is set.
    APPLY_RESOURCE_SCOPE=1
    ensure_resources_absent

    run_captured network_create timeout --kill-after="${KILL_AFTER_SECONDS}" \
        "${COMMAND_TIMEOUT_SECONDS}" docker network create --internal \
        --label "com.awoooi.restore.identity=${IDENTITY_HASH}" "${NETWORK_NAME}" \
        || fail_closed isolated_internal_network_create_failed
    NETWORK_CREATED=1
    EPHEMERAL_RUNTIME_CREATED=1

    run_captured backup_volume timeout --kill-after="${KILL_AFTER_SECONDS}" \
        "${COMMAND_TIMEOUT_SECONDS}" docker volume create \
        --label "com.awoooi.restore.identity=${IDENTITY_HASH}" "${BACKUP_VOLUME}" \
        || fail_closed isolated_backup_volume_create_failed
    BACKUP_VOLUME_CREATED=1
    run_captured clickhouse_volume timeout --kill-after="${KILL_AFTER_SECONDS}" \
        "${COMMAND_TIMEOUT_SECONDS}" docker volume create \
        --label "com.awoooi.restore.identity=${IDENTITY_HASH}" "${CLICKHOUSE_VOLUME}" \
        || fail_closed isolated_clickhouse_volume_create_failed
    CLICKHOUSE_VOLUME_CREATED=1
    run_captured zookeeper_data_volume timeout --kill-after="${KILL_AFTER_SECONDS}" \
        "${COMMAND_TIMEOUT_SECONDS}" docker volume create \
        --label "com.awoooi.restore.identity=${IDENTITY_HASH}" "${ZOOKEEPER_DATA_VOLUME}" \
        || fail_closed isolated_zookeeper_data_volume_create_failed
    ZOOKEEPER_DATA_VOLUME_CREATED=1
    run_captured zookeeper_log_volume timeout --kill-after="${KILL_AFTER_SECONDS}" \
        "${COMMAND_TIMEOUT_SECONDS}" docker volume create \
        --label "com.awoooi.restore.identity=${IDENTITY_HASH}" "${ZOOKEEPER_LOG_VOLUME}" \
        || fail_closed isolated_zookeeper_log_volume_create_failed
    ZOOKEEPER_LOG_VOLUME_CREATED=1

    stage_artifact_into_backup_volume

    run_captured zookeeper_run timeout --kill-after="${KILL_AFTER_SECONDS}" \
        "${COMMAND_TIMEOUT_SECONDS}" docker run --detach --pull=never \
        --name "${ZOOKEEPER_CONTAINER}" --network "${NETWORK_NAME}" \
        --network-alias restore-zookeeper --restart=no \
        --label "com.awoooi.restore.identity=${IDENTITY_HASH}" \
        --env "ALLOW_ANONYMOUS_LOGIN=yes" \
        --mount "type=volume,src=${ZOOKEEPER_DATA_VOLUME},dst=/data" \
        --mount "type=volume,src=${ZOOKEEPER_LOG_VOLUME},dst=/datalog" \
        "${ZOOKEEPER_IMAGE}" \
        || fail_closed isolated_zookeeper_start_failed
    ZOOKEEPER_CONTAINER_CREATED=1
    wait_for_container_running zookeeper "${ZOOKEEPER_CONTAINER}"

    run_captured clickhouse_run timeout --kill-after="${KILL_AFTER_SECONDS}" \
        "${COMMAND_TIMEOUT_SECONDS}" docker run --detach --pull=never \
        --name "${CLICKHOUSE_CONTAINER}" --network "${NETWORK_NAME}" \
        --network-alias restore-clickhouse --restart=no \
        --label "com.awoooi.restore.identity=${IDENTITY_HASH}" \
        --env "CLICKHOUSE_RESTORE_REPLICA=${RESTORE_REPLICA}" \
        --mount "type=volume,src=${CLICKHOUSE_VOLUME},dst=/var/lib/clickhouse" \
        --mount "type=volume,src=${BACKUP_VOLUME},dst=/backups" \
        --mount "type=bind,src=${BACKUP_DISK_CONFIG},dst=/etc/clickhouse-server/config.d/awoooi-backup-disk.xml,readonly" \
        --mount "type=bind,src=${ISOLATED_CLUSTER_CONFIG},dst=/etc/clickhouse-server/config.d/awoooi-restore-isolation.xml,readonly" \
        "${CLICKHOUSE_IMAGE}" \
        || fail_closed isolated_clickhouse_start_failed
    CLICKHOUSE_CONTAINER_CREATED=1
    wait_for_container_running clickhouse "${CLICKHOUSE_CONTAINER}"
    wait_for_clickhouse

    run_captured zookeeper_image_readback timeout --kill-after="${KILL_AFTER_SECONDS}" \
        "${COMMAND_TIMEOUT_SECONDS}" docker inspect --format '{{.Image}}' "${ZOOKEEPER_CONTAINER}" \
        || fail_closed isolated_zookeeper_image_readback_failed
    [ "$(tr -d '\r\n' < "${LAST_STDOUT}")" = "${ZOOKEEPER_IMAGE_ID_EXPECTED}" ] \
        || fail_closed isolated_zookeeper_runtime_image_id_mismatch
    run_captured clickhouse_image_readback timeout --kill-after="${KILL_AFTER_SECONDS}" \
        "${COMMAND_TIMEOUT_SECONDS}" docker inspect --format '{{.Image}}' "${CLICKHOUSE_CONTAINER}" \
        || fail_closed isolated_clickhouse_image_readback_failed
    [ "$(tr -d '\r\n' < "${LAST_STDOUT}")" = "${CLICKHOUSE_IMAGE_ID_EXPECTED}" ] \
        || fail_closed isolated_clickhouse_runtime_image_id_mismatch
    run_captured clickhouse_artifact_hash_readback timeout --kill-after="${KILL_AFTER_SECONDS}" \
        "${COMMAND_TIMEOUT_SECONDS}" docker exec "${CLICKHOUSE_CONTAINER}" \
        sha256sum "${CONTAINER_ARTIFACT_PATH}" \
        || fail_closed isolated_clickhouse_artifact_hash_readback_failed
    clickhouse_artifact_hash="$(awk 'NR == 1 {print $1}' "${LAST_STDOUT}")"
    [[ "${clickhouse_artifact_hash}" =~ ^[0-9a-f]{64}$ ]] \
        || fail_closed isolated_clickhouse_artifact_hash_readback_invalid
    CLICKHOUSE_ARTIFACT_SHA256="${clickhouse_artifact_hash}"
    [ "${CLICKHOUSE_ARTIFACT_SHA256}" = "${ARTIFACT_SHA256}" ] \
        || fail_closed isolated_clickhouse_artifact_hash_mismatch
    CLICKHOUSE_ARTIFACT_READBACK_VERIFIED=1
    emit_receipt execution created \
        "artifact_${ARTIFACT_SHA256}_staged_read_only_source_writable_backup_volume_internal_network_ephemeral_keeper_and_clickhouse_started_pull_never_exact_image_ids_no_published_ports"
}

verify_isolation_and_empty_target() {
    local observed

    run_captured network_internal timeout --kill-after="${KILL_AFTER_SECONDS}" \
        "${COMMAND_TIMEOUT_SECONDS}" docker network inspect --format '{{.Internal}}' "${NETWORK_NAME}" \
        || fail_closed isolated_network_inspect_failed
    [ "$(tr -d '\r\n' < "${LAST_STDOUT}")" = "true" ] \
        || fail_closed restore_network_not_internal
    for container in "${ZOOKEEPER_CONTAINER}" "${CLICKHOUSE_CONTAINER}"; do
        run_captured container_network timeout --kill-after="${KILL_AFTER_SECONDS}" \
            "${COMMAND_TIMEOUT_SECONDS}" docker inspect --format '{{.HostConfig.NetworkMode}}' "${container}" \
            || fail_closed isolated_container_network_inspect_failed
        [ "$(tr -d '\r\n' < "${LAST_STDOUT}")" = "${NETWORK_NAME}" ] \
            || fail_closed isolated_container_has_unexpected_network
    done

    run_sql disk_contract \
        "SELECT path FROM system.disks WHERE name={disk:String}" --param_disk "${BACKUP_DISK}" \
        || fail_closed isolated_backup_disk_missing
    [ "$(tr -d '\r\n' < "${LAST_STDOUT}")" = "/backups/" ] \
        || fail_closed isolated_backup_disk_path_mismatch

    run_sql macro_contract \
        'SELECT macro,substitution FROM system.macros ORDER BY macro' \
        || fail_closed isolated_macro_query_failed
    grep -F -x -q $'replica\t'"${RESTORE_REPLICA}" "${LAST_STDOUT}" \
        || fail_closed isolated_replica_macro_mismatch
    grep -F -x -q $'shard\t01' "${LAST_STDOUT}" \
        || fail_closed isolated_shard_macro_mismatch

    run_sql cluster_contract \
        "SELECT cluster,host_name,toString(port) FROM system.clusters WHERE cluster='cluster' ORDER BY shard_num,replica_num" \
        || fail_closed isolated_cluster_query_failed
    [ "$(tr -d '\r\n' < "${LAST_STDOUT}")" = $'cluster\trestore-clickhouse\t9000' ] \
        || fail_closed isolated_cluster_contract_mismatch

    run_sql keeper_contract "SELECT count() > 0 FROM system.zookeeper WHERE path='/'" \
        || fail_closed isolated_keeper_connection_failed
    [ "$(tr -d '\r\n' < "${LAST_STDOUT}")" = "1" ] \
        || fail_closed isolated_keeper_connection_not_ready

    run_sql empty_databases \
        "SELECT count() FROM system.databases WHERE name IN (${DATABASE_IN_SQL})" \
        || fail_closed isolated_database_empty_check_failed
    observed="$(tr -d '\r\n' < "${LAST_STDOUT}")"
    [ "${observed}" = "0" ] || fail_closed isolated_restore_target_databases_not_empty
    run_sql empty_tables \
        "SELECT count() FROM system.tables WHERE database IN (${DATABASE_IN_SQL})" \
        || fail_closed isolated_table_empty_check_failed
    observed="$(tr -d '\r\n' < "${LAST_STDOUT}")"
    [ "${observed}" = "0" ] || fail_closed isolated_restore_target_tables_not_empty
    emit_receipt independent_pre_verifier pass \
        "artifact_${STAGED_ARTIFACT_SHA256}_readback_internal_network_only_fresh_keeper_exact_macros_cluster_and_empty_signoz_target"
}

query_restore_status() {
    local line_count status_line restore_name error_hash
    run_sql restore_status \
        "SELECT name,toString(status),toString(num_files),toString(total_size),hex(SHA256(error)) FROM system.backups WHERE id={operation_id:String}" \
        --param_operation_id "${RESTORE_OPERATION_ID}" \
        || fail_closed restore_system_backups_query_failed
    line_count="$(awk 'NF { count++ } END { print count + 0 }' "${LAST_STDOUT}")"
    [ "${line_count}" -eq 1 ] || fail_closed restore_system_backups_terminal_not_unique
    status_line="$(tr -d '\r\n' < "${LAST_STDOUT}")"
    IFS=$'\t' read -r restore_name RESTORE_TERMINAL RESTORE_NUM_FILES RESTORE_TOTAL_SIZE error_hash <<< "${status_line}"
    [ "${restore_name}" = "Disk('${BACKUP_DISK}', '${CONTAINER_ARTIFACT_NAME}')" ] \
        || fail_closed restore_system_backups_artifact_mismatch
    [[ "${RESTORE_NUM_FILES}" =~ ^[0-9]+$ ]] && [[ "${RESTORE_TOTAL_SIZE}" =~ ^[0-9]+$ ]] \
        || fail_closed restore_system_backups_metrics_invalid
    [ "${error_hash}" = "${EMPTY_SHA256}" ] \
        || fail_closed restore_system_backups_error_present
}

restore_native_backup() {
    local submission_line submitted_id submitted_status deadline
    emit_receipt ai_decision candidate \
        "restore_exact_six_signoz_databases_to_empty_isolated_clickhouse_with_fresh_keeper"
    run_sql restore_submit \
        "RESTORE ${RESTORE_SCOPE_SQL} FROM Disk({disk:String},{artifact:String}) SETTINGS id='${RESTORE_OPERATION_ID}' ASYNC" \
        --param_disk "${BACKUP_DISK}" --param_artifact "${CONTAINER_ARTIFACT_NAME}" \
        || fail_closed native_restore_submission_failed
    submission_line="$(tr -d '\r\n' < "${LAST_STDOUT}")"
    IFS=$'\t' read -r submitted_id submitted_status <<< "${submission_line}"
    [ "${submitted_id}" = "${RESTORE_OPERATION_ID}" ] \
        || fail_closed native_restore_submission_id_mismatch
    case "${submitted_status}" in
        RESTORING|RESTORED) ;;
        *) fail_closed "native_restore_submission_status_${submitted_status:-empty}" ;;
    esac
    emit_receipt execution submitted \
        "restore_operation_${RESTORE_OPERATION_ID}_status_${submitted_status}"

    deadline=$(( $(date +%s) + RESTORE_TIMEOUT_SECONDS ))
    while true; do
        query_restore_status
        emit_receipt restore_status observed \
            "status_${RESTORE_TERMINAL}_files_${RESTORE_NUM_FILES}_total_${RESTORE_TOTAL_SIZE}"
        case "${RESTORE_TERMINAL}" in
            RESTORED) return 0 ;;
            RESTORING) ;;
            RESTORE_FAILED|RESTORE_CANCELLED) fail_closed "native_restore_terminal_${RESTORE_TERMINAL}" ;;
            *) fail_closed "native_restore_terminal_unexpected_${RESTORE_TERMINAL}" ;;
        esac
        [ "$(date +%s)" -lt "${deadline}" ] || fail_closed native_restore_poll_timeout
        sleep "${POLL_INTERVAL_SECONDS}"
    done
}

verify_restored_inventory_and_tables() {
    local restored_inventory compare_receipt check_list database table output step

    run_sql restored_inventory \
        "SELECT database,name,engine,toString(ifNull(total_rows,0)),toString(ifNull(total_bytes,0)),hex(SHA256(create_table_query)) FROM system.tables WHERE database IN (${DATABASE_IN_SQL}) ORDER BY database,name" \
        || fail_closed restored_table_inventory_query_failed
    restored_inventory="${LAST_STDOUT}"
    [ -s "${restored_inventory}" ] || fail_closed restored_table_inventory_empty

    run_captured inventory_compare python3 "${INVENTORY_HELPER}" compare \
        --source-inventory "${SOURCE_INVENTORY_PATH}" \
        --restored-inventory "${restored_inventory}" \
        || fail_closed restored_database_table_engine_manifest_mismatch
    compare_receipt="${LAST_STDOUT}"
    MANIFEST_PARITY="$(read_key manifest_parity "${compare_receipt}")"
    RESTORED_DATABASE_COUNT="$(read_key restored_database_count "${compare_receipt}")"
    RESTORED_TABLE_COUNT="$(read_key restored_table_count "${compare_receipt}")"
    PHYSICAL_TABLE_COUNT="$(read_key physical_table_count "${compare_receipt}")"
    CRITICAL_NONZERO_COUNT="$(read_key critical_nonzero_count "${compare_receipt}")"
    SOURCE_TOTAL_ROWS="$(read_key source_total_rows "${compare_receipt}")"
    RESTORED_TOTAL_ROWS="$(read_key restored_total_rows "${compare_receipt}")"
    ROW_DELTA="$(read_key row_delta "${compare_receipt}")"
    SOURCE_TOTAL_BYTES="$(read_key source_total_bytes "${compare_receipt}")"
    RESTORED_TOTAL_BYTES="$(read_key restored_total_bytes "${compare_receipt}")"
    BYTE_DELTA="$(read_key byte_delta "${compare_receipt}")"
    [ "${MANIFEST_PARITY}" = "1" ] || fail_closed restored_manifest_parity_not_proven

    run_captured physical_table_list python3 "${INVENTORY_HELPER}" check-list \
        --inventory "${restored_inventory}" \
        || fail_closed restored_physical_table_list_failed
    check_list="${LAST_STDOUT}"
    CHECK_TABLE_COUNT="$(awk 'NF { count++ } END { print count + 0 }' "${check_list}")"
    [ "${CHECK_TABLE_COUNT}" -eq "${PHYSICAL_TABLE_COUNT}" ] \
        || fail_closed restored_physical_table_count_mismatch

    while IFS=$'\t' read -r database table; do
        [ -n "${database}" ] && [ -n "${table}" ] || continue
        step="check_table_$(printf '%03d' "$((CHECK_TABLE_PASS_COUNT + 1))")"
        run_sql "${step}" \
            "CHECK TABLE \`${database}\`.\`${table}\` SETTINGS check_query_single_value_result=1" \
            || fail_closed "check_table_command_failed_${database}_${table}"
        output="$(tr -d '\r\n' < "${LAST_STDOUT}")"
        [ "${output}" = "1" ] || fail_closed "check_table_not_passed_${database}_${table}"
        CHECK_TABLE_PASS_COUNT=$((CHECK_TABLE_PASS_COUNT + 1))
    done < "${check_list}"
    [ "${CHECK_TABLE_PASS_COUNT}" -eq "${CHECK_TABLE_COUNT}" ] \
        || fail_closed check_table_pass_count_mismatch

    emit_receipt independent_post_verifier pass \
        "restore_RESTORED_manifest_parity_${MANIFEST_PARITY}_tables_${RESTORED_TABLE_COUNT}_check_tables_${CHECK_TABLE_PASS_COUNT}_critical_nonzero_${CRITICAL_NONZERO_COUNT}_row_delta_${ROW_DELTA}_byte_delta_${BYTE_DELTA}"
    APPLY_PASS=1
    FINAL_REASON="isolated_native_restore_verified"
}

capture_container_log() {
    local container="$1"
    local label="$2"
    local raw target
    raw="$(mktemp "${RECEIPT_DIR}/.${label}.XXXXXX")"
    target="${RECEIPT_DIR}/${INVOCATION_ID}-${label}.log"
    set +e
    timeout --kill-after="${KILL_AFTER_SECONDS}" "${COMMAND_TIMEOUT_SECONDS}" \
        docker logs --tail "${LOG_TAIL_LINES}" "${container}" >"${raw}" 2>&1
    set -e
    head -c "${MAX_LOG_BYTES}" "${raw}" > "${target}"
    rm -f "${raw}"
}

cleanup_container() {
    local container="$1"
    local label inspect_status remove_status final_status
    set +e
    label="$(timeout --kill-after="${KILL_AFTER_SECONDS}" "${COMMAND_TIMEOUT_SECONDS}" \
        docker container inspect --format '{{index .Config.Labels "com.awoooi.restore.identity"}}' \
        "${container}" 2>/dev/null)"
    inspect_status=$?
    set -e
    if [ "${inspect_status}" -ne 0 ]; then
        exact_container_absent "${container}"
        return $?
    fi
    [ "${label}" = "${IDENTITY_HASH}" ] || return 1
    capture_container_log "${container}" "$(basename "${container}")"
    set +e
    timeout --kill-after="${KILL_AFTER_SECONDS}" "${COMMAND_TIMEOUT_SECONDS}" \
        docker rm -f "${container}" >/dev/null 2>&1
    remove_status=$?
    exact_container_absent "${container}"
    final_status=$?
    set -e
    [ "${remove_status}" -eq 0 ] && [ "${final_status}" -eq 0 ] \
        && docker_daemon_ready
}

cleanup_volume() {
    local volume="$1"
    local label inspect_status remove_status final_status
    set +e
    label="$(timeout --kill-after="${KILL_AFTER_SECONDS}" "${COMMAND_TIMEOUT_SECONDS}" \
        docker volume inspect --format '{{index .Labels "com.awoooi.restore.identity"}}' \
        "${volume}" 2>/dev/null)"
    inspect_status=$?
    set -e
    if [ "${inspect_status}" -ne 0 ]; then
        exact_volume_absent "${volume}"
        return $?
    fi
    [ "${label}" = "${IDENTITY_HASH}" ] || return 1
    set +e
    timeout --kill-after="${KILL_AFTER_SECONDS}" "${COMMAND_TIMEOUT_SECONDS}" \
        docker volume rm "${volume}" >/dev/null 2>&1
    remove_status=$?
    exact_volume_absent "${volume}"
    final_status=$?
    set -e
    [ "${remove_status}" -eq 0 ] && [ "${final_status}" -eq 0 ] \
        && docker_daemon_ready
}

cleanup_network() {
    local label inspect_status remove_status final_status
    set +e
    label="$(timeout --kill-after="${KILL_AFTER_SECONDS}" "${COMMAND_TIMEOUT_SECONDS}" \
        docker network inspect --format '{{index .Labels "com.awoooi.restore.identity"}}' \
        "${NETWORK_NAME}" 2>/dev/null)"
    inspect_status=$?
    set -e
    if [ "${inspect_status}" -ne 0 ]; then
        exact_network_absent "${NETWORK_NAME}"
        return $?
    fi
    [ "${label}" = "${IDENTITY_HASH}" ] || return 1
    set +e
    timeout --kill-after="${KILL_AFTER_SECONDS}" "${COMMAND_TIMEOUT_SECONDS}" \
        docker network rm "${NETWORK_NAME}" >/dev/null 2>&1
    remove_status=$?
    exact_network_absent "${NETWORK_NAME}"
    final_status=$?
    set -e
    [ "${remove_status}" -eq 0 ] && [ "${final_status}" -eq 0 ] \
        && docker_daemon_ready
}

verify_all_resources_absent() {
    docker_daemon_ready || return 1
    exact_container_absent "${STAGING_CONTAINER}" \
        && exact_container_absent "${CLICKHOUSE_CONTAINER}" \
        && exact_container_absent "${ZOOKEEPER_CONTAINER}" \
        && exact_volume_absent "${BACKUP_VOLUME}" \
        && exact_volume_absent "${CLICKHOUSE_VOLUME}" \
        && exact_volume_absent "${ZOOKEEPER_DATA_VOLUME}" \
        && exact_volume_absent "${ZOOKEEPER_LOG_VOLUME}" \
        && exact_network_absent "${NETWORK_NAME}" \
        && docker_daemon_ready
}

write_status() {
    local terminal="$1"
    local exit_code="$2"
    local cleanup_json runtime_json staging_json clickhouse_hash_json check_json apply_json status_tmp
    cleanup_json=false
    runtime_json=false
    staging_json=false
    clickhouse_hash_json=false
    check_json=false
    apply_json=false
    [ "${CLEANUP_PASS}" -eq 1 ] && cleanup_json=true
    [ "${EPHEMERAL_RUNTIME_CREATED}" -eq 1 ] && runtime_json=true
    [ "${ARTIFACT_STAGING_VERIFIED}" -eq 1 ] && staging_json=true
    [ "${CLICKHOUSE_ARTIFACT_READBACK_VERIFIED}" -eq 1 ] && clickhouse_hash_json=true
    [ "${CHECK_PASS}" -eq 1 ] && check_json=true
    [ "${APPLY_PASS}" -eq 1 ] && apply_json=true
    status_tmp="${STATUS_FILE}.tmp.$$"
    printf '%s\n' \
        '{' \
        '  "schema_version":"clickhouse_native_restore_drill_status_v1",' \
        "  \"trace_id\":\"${TRACE_ID}\"," \
        "  \"run_id\":\"${RUN_ID}\"," \
        "  \"work_item_id\":\"${WORK_ITEM_ID}\"," \
        "  \"identity_hash\":\"${IDENTITY_HASH}\"," \
        "  \"mode\":\"${MODE}\"," \
        "  \"terminal\":\"${terminal}\"," \
        "  \"reason\":\"${FINAL_REASON}\"," \
        "  \"exit_code\":${exit_code}," \
        "  \"artifact_sha256\":\"${ARTIFACT_SHA256}\"," \
        "  \"staged_artifact_sha256\":\"${STAGED_ARTIFACT_SHA256}\"," \
        "  \"clickhouse_artifact_sha256\":\"${CLICKHOUSE_ARTIFACT_SHA256}\"," \
        "  \"source_inventory_sha256\":\"${SOURCE_INVENTORY_SHA256}\"," \
        "  \"clickhouse_image_id\":\"${CLICKHOUSE_IMAGE_ID}\"," \
        "  \"zookeeper_image_id\":\"${ZOOKEEPER_IMAGE_ID}\"," \
        '  "network_mode":"docker_internal_only",' \
        '  "staging_network_mode":"none",' \
        '  "production_network_attached":false,' \
        '  "production_restore_performed":false,' \
        '  "allow_non_empty_tables":false,' \
        '  "isolated_keeper":true,' \
        '  "artifact_source_mount":"read_only_staging_only",' \
        '  "backup_volume_mount":"writable_ephemeral",' \
        "  \"artifact_staging_verified\":${staging_json}," \
        "  \"clickhouse_artifact_readback_verified\":${clickhouse_hash_json}," \
        "  \"ephemeral_runtime_created\":${runtime_json}," \
        "  \"check_pass\":${check_json}," \
        "  \"apply_pass\":${apply_json}," \
        "  \"restore_terminal\":\"${RESTORE_TERMINAL}\"," \
        "  \"manifest_parity\":${MANIFEST_PARITY}," \
        "  \"source_database_count\":${SOURCE_DATABASE_COUNT}," \
        "  \"restored_database_count\":${RESTORED_DATABASE_COUNT}," \
        "  \"source_table_count\":${SOURCE_TABLE_COUNT}," \
        "  \"restored_table_count\":${RESTORED_TABLE_COUNT}," \
        "  \"check_table_count\":${CHECK_TABLE_COUNT}," \
        "  \"check_table_pass_count\":${CHECK_TABLE_PASS_COUNT}," \
        "  \"critical_nonzero_count\":${CRITICAL_NONZERO_COUNT}," \
        "  \"source_total_rows\":${SOURCE_TOTAL_ROWS}," \
        "  \"restored_total_rows\":${RESTORED_TOTAL_ROWS}," \
        "  \"row_delta\":${ROW_DELTA}," \
        "  \"source_total_bytes\":${SOURCE_TOTAL_BYTES}," \
        "  \"restored_total_bytes\":${RESTORED_TOTAL_BYTES}," \
        "  \"byte_delta\":${BYTE_DELTA}," \
        "  \"cleanup_verified\":${cleanup_json}," \
        '  "secret_values_collected":false' \
        '}' > "${status_tmp}"
    mv "${status_tmp}" "${STATUS_FILE}"
}

cleanup() {
    local exit_code=$?
    local terminal="failed"
    trap - EXIT HUP INT TERM
    set +e

    [ -z "${PREFLIGHT_CANDIDATE}" ] || rm -f "${PREFLIGHT_CANDIDATE}"
    if [ "${APPLY_RESOURCE_SCOPE}" -eq 1 ]; then
        cleanup_container "${CLICKHOUSE_CONTAINER}" || CLEANUP_PASS=0
        cleanup_container "${ZOOKEEPER_CONTAINER}" || CLEANUP_PASS=0
        cleanup_container "${STAGING_CONTAINER}" || CLEANUP_PASS=0
        cleanup_volume "${BACKUP_VOLUME}" || CLEANUP_PASS=0
        cleanup_volume "${CLICKHOUSE_VOLUME}" || CLEANUP_PASS=0
        cleanup_volume "${ZOOKEEPER_DATA_VOLUME}" || CLEANUP_PASS=0
        cleanup_volume "${ZOOKEEPER_LOG_VOLUME}" || CLEANUP_PASS=0
        cleanup_network || CLEANUP_PASS=0
        verify_all_resources_absent || CLEANUP_PASS=0
    fi
    [ -z "${LOCK_DIR}" ] || rmdir "${LOCK_DIR}" 2>/dev/null || CLEANUP_PASS=0

    if [ "${CLEANUP_PASS}" -ne 1 ]; then
        FINAL_REASON="ephemeral_resource_cleanup_failed"
        exit_code=1
    elif [ "${exit_code}" -eq 0 ] && [ "${MODE}" = check ] && [ "${CHECK_PASS}" -eq 1 ]; then
        terminal="check_pass_no_runtime_write"
    elif [ "${exit_code}" -eq 0 ] && [ "${MODE}" = apply ] && [ "${APPLY_PASS}" -eq 1 ]; then
        terminal="verified"
    fi
    emit_receipt cleanup "$([ "${CLEANUP_PASS}" -eq 1 ] && printf pass || printf failed)" \
        "staging_keeper_clickhouse_containers_backup_data_volumes_and_internal_network_absence_verified_${CLEANUP_PASS}"
    emit_receipt terminal "${terminal}" "${FINAL_REASON}"
    write_status "${terminal}" "${exit_code}"
    printf 'CLICKHOUSE_NATIVE_RESTORE_DRILL_TERMINAL trace_id=%s run_id=%s work_item_id=%s mode=%s terminal=%s cleanup=%s exit_code=%s\n' \
        "${TRACE_ID}" "${RUN_ID}" "${WORK_ITEM_ID}" "${MODE}" "${terminal}" \
        "${CLEANUP_PASS}" "${exit_code}"
    exit "${exit_code}"
}

on_signal() {
    local signal_number="$1"
    FINAL_REASON="signal_${signal_number}"
    emit_receipt signal failed "${FINAL_REASON}"
    exit "$((128 + signal_number))"
}

main() {
    parse_args "$@"
    validate_contract
    setup_receipts
    validate_preflight

    if [ "${MODE}" = check ]; then
        persist_check_contract
        log_info "isolated native RESTORE drill check passed without runtime mutation"
        return 0
    fi

    require_check_contract
    create_isolated_runtime
    verify_isolation_and_empty_target
    restore_native_backup
    verify_restored_inventory_and_tables
    log_info "isolated native RESTORE drill verified; ephemeral runtime will be removed"
}

main "$@"