#!/usr/bin/env bash
# =============================================================================
# P0-OBS-002 - SigNoz online backup canary and legacy maintenance lease verifier
#
# The default native-backup path never stops the collector and therefore does
# not depend on, read, or mutate the legacy monitor/cooldown contract.  The
# bounded lease path remains available only when an explicit legacy stop-mode
# canary is requested.  Neither mode reads or sources a secret/.env file.
# =============================================================================

set -euo pipefail

COLLECTOR_NAME="${SIGNOZ_CANARY_COLLECTOR_NAME:-signoz-otel-collector}"
BACKUP_SCRIPT="${SIGNOZ_CANARY_BACKUP_SCRIPT:-/backup/scripts/backup-signoz.sh}"
CLICKHOUSE_NATIVE_POST_VERIFY_HOOK="${CLICKHOUSE_NATIVE_POST_VERIFY_HOOK:-/backup/scripts/clickhouse-native-restore-drill.sh}"
CLICKHOUSE_RESTORE_INVENTORY_HELPER="${CLICKHOUSE_RESTORE_INVENTORY_HELPER:-/backup/scripts/clickhouse-restore-inventory.py}"
CLICKHOUSE_RESTORE_BACKUP_DISK_CONFIG="${CLICKHOUSE_RESTORE_BACKUP_DISK_CONFIG:-/backup/config/signoz/clickhouse/config.d/backup_disk.xml}"
CLICKHOUSE_RESTORE_CLUSTER_CONFIG="${CLICKHOUSE_RESTORE_CLUSTER_CONFIG:-/backup/config/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml}"
CLICKHOUSE_NATIVE_RECEIPT_ROOT="${CLICKHOUSE_NATIVE_RECEIPT_ROOT:-/backup/logs/clickhouse-native}"
CLICKHOUSE_RESTORE_RECEIPT_ROOT="${CLICKHOUSE_RESTORE_RECEIPT_ROOT:-/backup/logs/clickhouse-native-restore-drill}"
MONITOR_SCRIPT="${SIGNOZ_CANARY_MONITOR_SCRIPT:-/home/wooo/awoooi-ops/docker-health-monitor.sh}"
MONITOR_LOG="${SIGNOZ_CANARY_MONITOR_LOG:-/home/wooo/awoooi-ops/monitor.log}"
MONITOR_COOLDOWN_DIR="${SIGNOZ_CANARY_COOLDOWN_DIR:-/tmp/docker-health-monitor-cooldown}"
COOLDOWN_FILE="${MONITOR_COOLDOWN_DIR}/${COLLECTOR_NAME}.cooldown"
RECEIPT_ROOT="${SIGNOZ_CANARY_RECEIPT_ROOT:-/backup/logs/signoz-canary}"
LOCK_FILE="${SIGNOZ_CANARY_LOCK_FILE:-/tmp/awoooi-signoz-backup-canary.lock}"

CANARY_TIMEOUT_SECONDS="${SIGNOZ_CANARY_TIMEOUT_SECONDS:-12000}"
TIMEOUT_KILL_AFTER_SECONDS="${SIGNOZ_CANARY_KILL_AFTER_SECONDS:-60}"
DOCKER_TIMEOUT_SECONDS="${SIGNOZ_CANARY_DOCKER_TIMEOUT_SECONDS:-15}"
LEASE_MARGIN_SECONDS="${SIGNOZ_CANARY_LEASE_MARGIN_SECONDS:-60}"
MONITOR_CRON_INTERVAL_SECONDS="${SIGNOZ_CANARY_MONITOR_CRON_INTERVAL_SECONDS:-300}"
STATE_POLL_SECONDS="${SIGNOZ_CANARY_STATE_POLL_SECONDS:-1}"
EXPECT_COLLECTOR_STOP="${SIGNOZ_CANARY_EXPECT_COLLECTOR_STOP:-0}"

TRACE_ID="${TRACE_ID:-}"
RUN_ID="${RUN_ID:-}"
WORK_ITEM_ID="${WORK_ITEM_ID:-}"

RECEIPT_DIR=""
RECEIPT_LOG=""
BACKUP_OUTPUT=""
MONITOR_WINDOW=""
STATE_OBSERVATIONS=""
ROLLBACK_FILE=""
ROLLBACK_METADATA=""

RECEIPTS_READY=0
LEASE_ROLLBACK_READY=0
LEASE_MUTATION_STARTED=0
LEASE_ARMED=0
PRIOR_LEASE_PRESENT=0
PRIOR_LEASE_MODE=""
PRIOR_LEASE_UID=""
PRIOR_LEASE_GID=""
PRIOR_LEASE_HASH=""
PRIOR_LEASE_VALUE=""
LEASE_TMP=""
BACKUP_PID=""
OBSERVER_PID=""
CANARY_VERIFIED=0
CHECK_VERIFIED=0
MODE=""

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_poll_interval() {
    [[ "$1" =~ ^[0-9]+([.][0-9]+)?$ ]] && [[ "$1" != "0" ]] && [[ "$1" != "0.0" ]]
}

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","observed_at":"%s","stage":"%s","terminal":"%s","detail":"%s"}\n' \
        "${TRACE_ID}" "${RUN_ID}" "${WORK_ITEM_ID}" "${observed_at}" \
        "${stage}" "${terminal}" "${detail}" >> "${RECEIPT_LOG}"
}

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

require_command() {
    command -v "$1" >/dev/null 2>&1 || fail_closed "required_command_missing_${1}"
}

run_collector_docker() {
    timeout --kill-after="${TIMEOUT_KILL_AFTER_SECONDS}" \
        "${DOCKER_TIMEOUT_SECONDS}" docker "$@"
}

require_native_restore_contract() {
    local label path

    for label in \
        CLICKHOUSE_NATIVE_POST_VERIFY_HOOK \
        CLICKHOUSE_RESTORE_INVENTORY_HELPER \
        CLICKHOUSE_RESTORE_BACKUP_DISK_CONFIG \
        CLICKHOUSE_RESTORE_CLUSTER_CONFIG; do
        path="${!label}"
        case "${path}" in
            /*) ;;
            *) fail_closed "native_restore_contract_${label}_path_not_absolute" ;;
        esac
        [ -f "${path}" ] && [ -r "${path}" ] && [ ! -L "${path}" ] \
            || fail_closed "native_restore_contract_${label}_missing_unreadable_or_symlink"
    done

    [ -x "${CLICKHOUSE_NATIVE_POST_VERIFY_HOOK}" ] \
        || fail_closed "native_restore_contract_post_verify_hook_not_executable"
    [ -x "${CLICKHOUSE_RESTORE_INVENTORY_HELPER}" ] \
        || fail_closed "native_restore_contract_inventory_helper_not_executable"
}

acquire_canary_lock() {
    [ ! -L "${LOCK_FILE}" ] || fail_closed "canary_lock_file_symlink_unsafe"
    if [ -e "${LOCK_FILE}" ]; then
        [ -f "${LOCK_FILE}" ] || fail_closed "canary_lock_path_not_regular_file"
    fi
    exec 9>>"${LOCK_FILE}"
    flock -n 9 || fail_closed "another_signoz_canary_holds_lock"
}

require_native_receipt_contract() {
    local label path run_dir

    for label in CLICKHOUSE_NATIVE_RECEIPT_ROOT CLICKHOUSE_RESTORE_RECEIPT_ROOT; do
        path="${!label}"
        case "${path}" in
            /*) ;;
            *) fail_closed "native_receipt_contract_${label}_path_not_absolute" ;;
        esac
        [[ "${path}" != *$'\n'* ]] \
            || fail_closed "native_receipt_contract_${label}_path_contains_newline"
        [ ! -L "${path}" ] \
            || fail_closed "native_receipt_contract_${label}_root_symlink_unsafe"
        if [ -e "${path}" ]; then
            [ -d "${path}" ] && [ -r "${path}" ] && [ -w "${path}" ] \
                || fail_closed "native_receipt_contract_${label}_root_not_readable_writable_directory"
        fi
        run_dir="${path}/${RUN_ID}"
        [ ! -e "${run_dir}" ] && [ ! -L "${run_dir}" ] \
            || fail_closed "native_receipt_contract_${label}_stale_same_run_path_present"
    done

    [ "${CLICKHOUSE_NATIVE_RECEIPT_ROOT}" != "${CLICKHOUSE_RESTORE_RECEIPT_ROOT}" ] \
        || fail_closed "native_and_restore_receipt_roots_must_be_distinct"
}

verify_same_run_native_restore_receipts() {
    local native_receipts restore_receipts restore_status

    native_receipts="${CLICKHOUSE_NATIVE_RECEIPT_ROOT}/${RUN_ID}/receipts.jsonl"
    restore_receipts="${CLICKHOUSE_RESTORE_RECEIPT_ROOT}/${RUN_ID}/receipts.jsonl"
    restore_status="${CLICKHOUSE_RESTORE_RECEIPT_ROOT}/${RUN_ID}/status.json"

    for path in "${native_receipts}" "${restore_receipts}" "${restore_status}"; do
        [ -f "${path}" ] && [ -r "${path}" ] && [ ! -L "${path}" ] \
            || fail_closed "same_run_native_restore_receipt_missing_unreadable_or_symlink"
    done

    python3 - "${native_receipts}" "${restore_receipts}" "${restore_status}" \
        "${TRACE_ID}" "${RUN_ID}" "${WORK_ITEM_ID}" <<'PY' \
        || fail_closed "same_run_native_restore_receipt_validation_failed"
import json
import os
import sys


def reject(message: str) -> None:
    print(f"receipt_verifier:{message}", file=sys.stderr)
    raise SystemExit(1)


native_path, restore_path, status_path, trace_id, run_id, work_item_id = sys.argv[1:]
identity = {
    "trace_id": trace_id,
    "run_id": run_id,
    "work_item_id": work_item_id,
}


def read_jsonl(path: str, label: str) -> list[dict[str, object]]:
    if os.path.islink(path) or not os.path.isfile(path):
        reject(f"{label}_unsafe")
    records: list[dict[str, object]] = []
    try:
        with open(path, encoding="utf-8") as source:
            for line_number, raw_line in enumerate(source, start=1):
                if not raw_line.strip():
                    continue
                value = json.loads(raw_line)
                if not isinstance(value, dict):
                    reject(f"{label}_line_{line_number}_not_object")
                for key, expected in identity.items():
                    if value.get(key) != expected:
                        reject(f"{label}_line_{line_number}_{key}_mismatch")
                invocation_id = value.get("invocation_id")
                if not isinstance(invocation_id, str) or not invocation_id:
                    reject(f"{label}_line_{line_number}_invocation_id_invalid")
                records.append(value)
    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
        reject(f"{label}_malformed_{type(exc).__name__}")
    if not records:
        reject(f"{label}_empty")
    return records


def mode_records(
    records: list[dict[str, object]], mode: str, label: str
) -> list[dict[str, object]]:
    selected = [
        record
        for record in records
        if str(record.get("invocation_id", "")).startswith(f"{mode}-")
    ]
    if not selected:
        reject(f"{label}_{mode}_invocation_missing")
    return selected


def require_exact(
    records: list[dict[str, object]],
    *,
    label: str,
    stage: str,
    terminal: str,
    detail_prefix: str | None = None,
) -> None:
    matches = [
        record
        for record in records
        if record.get("stage") == stage
        and record.get("terminal") == terminal
        and (
            detail_prefix is None
            or str(record.get("detail", "")).startswith(detail_prefix)
        )
    ]
    if len(matches) != 1:
        reject(f"{label}_{stage}_{terminal}_count_{len(matches)}")


native_records = read_jsonl(native_path, "native_receipts")
native_check = mode_records(native_records, "check", "native")
native_apply = mode_records(native_records, "apply", "native")
require_exact(
    native_check,
    label="native_check",
    stage="terminal",
    terminal="check_pass_no_write",
)
require_exact(
    native_apply,
    label="native_apply",
    stage="terminal",
    terminal="pass",
)
require_exact(
    native_apply,
    label="native_apply",
    stage="independent_verifier",
    terminal="pass",
)
require_exact(
    native_apply,
    label="native_apply_hook_check",
    stage="command",
    terminal="pass",
    detail_prefix="step_post_verify_hook_check_exit_0_",
)
require_exact(
    native_apply,
    label="native_apply_hook",
    stage="command",
    terminal="pass",
    detail_prefix="step_post_verify_hook_exit_0_",
)

restore_records = read_jsonl(restore_path, "restore_receipts")
restore_check = mode_records(restore_records, "check", "restore")
restore_apply = mode_records(restore_records, "apply", "restore")
require_exact(
    restore_check,
    label="restore_check",
    stage="terminal",
    terminal="check_pass_no_runtime_write",
)
require_exact(
    restore_apply,
    label="restore_apply",
    stage="terminal",
    terminal="verified",
)
require_exact(
    restore_apply,
    label="restore_apply",
    stage="independent_post_verifier",
    terminal="pass",
)
require_exact(
    restore_apply,
    label="restore_apply",
    stage="cleanup",
    terminal="pass",
)

if os.path.islink(status_path) or not os.path.isfile(status_path):
    reject("restore_status_unsafe")
try:
    with open(status_path, encoding="utf-8") as source:
        status = json.load(source)
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
    reject(f"restore_status_malformed_{type(exc).__name__}")
if not isinstance(status, dict):
    reject("restore_status_not_object")
for key, expected in identity.items():
    if status.get(key) != expected:
        reject(f"restore_status_{key}_mismatch")

expected_strings = {
    "schema_version": "clickhouse_native_restore_drill_status_v1",
    "mode": "apply",
    "terminal": "verified",
    "restore_terminal": "RESTORED",
    "network_mode": "docker_internal_only",
    "staging_network_mode": "none",
}
for key, expected in expected_strings.items():
    if not isinstance(status.get(key), str) or status.get(key) != expected:
        reject(f"restore_status_{key}_invalid")
for key, expected in (("exit_code", 0), ("manifest_parity", 1)):
    if type(status.get(key)) is not int or status.get(key) != expected:
        reject(f"restore_status_{key}_invalid")
for key in (
    "check_pass",
    "apply_pass",
    "cleanup_verified",
    "artifact_staging_verified",
    "clickhouse_artifact_readback_verified",
    "ephemeral_runtime_created",
    "isolated_keeper",
):
    if status.get(key) is not True:
        reject(f"restore_status_{key}_not_true")
for key in (
    "production_network_attached",
    "production_restore_performed",
    "allow_non_empty_tables",
):
    if status.get(key) is not False:
        reject(f"restore_status_{key}_not_false")


def positive_integer(key: str) -> int:
    value = status.get(key)
    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
        reject(f"restore_status_{key}_not_positive_integer")
    return value


source_database_count = positive_integer("source_database_count")
restored_database_count = positive_integer("restored_database_count")
if source_database_count != restored_database_count:
    reject("restore_status_database_count_mismatch")
source_table_count = positive_integer("source_table_count")
restored_table_count = positive_integer("restored_table_count")
if source_table_count != restored_table_count:
    reject("restore_status_table_count_mismatch")
positive_integer("critical_nonzero_count")
check_count = status.get("check_table_count")
check_pass_count = status.get("check_table_pass_count")
if (
    isinstance(check_count, bool)
    or not isinstance(check_count, int)
    or isinstance(check_pass_count, bool)
    or not isinstance(check_pass_count, int)
    or check_count <= 0
    or check_count != check_pass_count
):
    reject("restore_status_check_table_counts_invalid")
PY

    emit_receipt "native_restore_receipt_verifier" "pass" \
        "same_run_machine_readable_native_check_apply_hook_and_isolated_restore_status_verified"
}

require_monitor_fragment() {
    local fragment="$1"
    local count
    count="$(grep -F -c -- "${fragment}" "${MONITOR_SCRIPT}" || true)"
    [ "${count}" -eq 1 ] || fail_closed "monitor_contract_ambiguous_fragment_count_${count}"
}

require_monitor_function_fragment() {
    local function_name="$1"
    local fragment="$2"
    local count

    count="$(awk -v header="${function_name}() {" -v needle="${fragment}" '
        $0 == header {
            inside = 1
            depth = 0
        }
        inside {
            original = $0
            if (index(original, needle)) {
                count++
            }
            open_line = original
            close_line = original
            depth += gsub(/\{/, "", open_line)
            depth -= gsub(/\}/, "", close_line)
            if (depth == 0) {
                print count + 0
                printed = 1
                exit
            }
        }
        END {
            if (!printed) {
                print count + 0
            }
        }
    ' "${MONITOR_SCRIPT}")"
    [ "${count}" -eq 1 ] \
        || fail_closed "monitor_${function_name}_contract_ambiguous_fragment_count_${count}"
}

collector_running_state() {
    local running
    if ! running="$(run_collector_docker inspect --format '{{.State.Running}}' "${COLLECTOR_NAME}" 2>/dev/null)"; then
        return 1
    fi
    case "${running}" in
        true) printf '%s\n' "running" ;;
        false) printf '%s\n' "stopped" ;;
        *) return 1 ;;
    esac
}

collector_runtime_metadata() {
    run_collector_docker inspect --format $'{{.Id}}\t{{.State.Running}}\t{{.State.StartedAt}}\t{{.RestartCount}}\t{{if (index .State "Health")}}{{(index .State "Health").Status}}{{else}}none{{end}}' \
        "${COLLECTOR_NAME}" 2>/dev/null
}

listener_is_up() {
    local port="$1"
    ss -H -lnt | awk -v port="${port}" '
        $4 ~ (":" port "$") { found = 1 }
        END { exit(found ? 0 : 1) }
    '
}

verify_collector_runtime() {
    local expected_id="$1"
    local expected_started_at="$2"
    local expected_restart_count="$3"
    local expected_health="$4"
    local metadata actual_id actual_running actual_started_at actual_restart_count actual_health

    metadata="$(collector_runtime_metadata)" || return 1
    IFS=$'\t' read -r actual_id actual_running actual_started_at actual_restart_count actual_health <<< "${metadata}"
    [ "${actual_id}" = "${expected_id}" ] || return 1
    [ "${actual_running}" = "true" ] || return 1
    [ "${actual_started_at}" = "${expected_started_at}" ] || return 1
    [ "${actual_restart_count}" = "${expected_restart_count}" ] || return 1
    [ "${actual_health}" = "${expected_health}" ] || return 1
    listener_is_up 4317 || return 1
    listener_is_up 4318 || return 1
}

restore_lease() {
    local restore_tmp restore_hash current_value

    [ "${LEASE_MUTATION_STARTED}" -eq 1 ] || return 0
    [ "${LEASE_ROLLBACK_READY}" -eq 1 ] || return 1

    if [ "${PRIOR_LEASE_PRESENT}" -eq 1 ]; then
        [ -f "${ROLLBACK_FILE}" ] || return 1
        restore_tmp="$(mktemp "${MONITOR_COOLDOWN_DIR}/.${COLLECTOR_NAME}.restore.${RUN_ID}.XXXXXX")" || return 1
        cp -p "${ROLLBACK_FILE}" "${restore_tmp}" || return 1
        chmod "${PRIOR_LEASE_MODE}" "${restore_tmp}" || return 1
        mv -f "${restore_tmp}" "${COOLDOWN_FILE}" || return 1
        restore_tmp=""

        restore_hash="$(sha256sum "${COOLDOWN_FILE}" | awk '{print $1}')" || return 1
        [ "${restore_hash}" = "${PRIOR_LEASE_HASH}" ] || return 1
        [ "$(stat -c '%a' "${COOLDOWN_FILE}")" = "${PRIOR_LEASE_MODE}" ] || return 1
        [ "$(stat -c '%u' "${COOLDOWN_FILE}")" = "${PRIOR_LEASE_UID}" ] || return 1
        [ "$(stat -c '%g' "${COOLDOWN_FILE}")" = "${PRIOR_LEASE_GID}" ] || return 1
        current_value="$(tr -d '\r\n' < "${COOLDOWN_FILE}")"
        [ "${current_value}" = "${PRIOR_LEASE_VALUE}" ] || return 1
    else
        rm -f "${COOLDOWN_FILE}" || return 1
        [ ! -e "${COOLDOWN_FILE}" ] || return 1
    fi

    LEASE_ARMED=0
    LEASE_MUTATION_STARTED=0
    return 0
}

terminate_children() {
    if [ -n "${BACKUP_PID}" ] && kill -0 "${BACKUP_PID}" 2>/dev/null; then
        kill -TERM "${BACKUP_PID}" 2>/dev/null || true
        wait "${BACKUP_PID}" 2>/dev/null || true
    fi
    if [ -n "${OBSERVER_PID}" ] && kill -0 "${OBSERVER_PID}" 2>/dev/null; then
        kill -TERM "${OBSERVER_PID}" 2>/dev/null || true
        wait "${OBSERVER_PID}" 2>/dev/null || true
    fi
}

cleanup() {
    local exit_code=$?
    local lease_terminal="not_armed"
    local overall_terminal="failed"

    trap - EXIT HUP INT TERM
    set +e
    terminate_children
    rm -f "${LEASE_TMP}" 2>/dev/null || true

    if [ "${LEASE_MUTATION_STARTED}" -eq 1 ]; then
        if restore_lease; then
            lease_terminal="restored"
        else
            lease_terminal="restore_failed"
            exit_code=91
        fi
    elif [ "${MODE}" = "check" ] && [ "${CHECK_VERIFIED}" -eq 1 ]; then
        lease_terminal="no_write"
    elif [ "${MODE}" = "apply" ] && [ "${EXPECT_COLLECTOR_STOP}" = "0" ]; then
        lease_terminal="no_write"
    fi

    if [ "${exit_code}" -eq 0 ] && [ "${MODE}" = "check" ] && [ "${CHECK_VERIFIED}" -eq 1 ]; then
        overall_terminal="check_pass_no_write"
    elif [ "${exit_code}" -eq 0 ] && [ "${CANARY_VERIFIED}" -eq 1 ] && [ "${lease_terminal}" != "restore_failed" ]; then
        overall_terminal="partial_degraded"
    fi

    emit_receipt "rollback" "${lease_terminal}" "cooldown_lease_${lease_terminal}"
    if [ "${overall_terminal}" = "partial_degraded" ]; then
        emit_receipt "runtime_operation" "pass" \
            "clickhouse_native_backup_and_isolated_restore_runtime_verifier_pass"
        emit_receipt "closure_writeback" "partial_degraded" \
            "local_durable_receipt_ack_incident_not_applicable_telegram_not_sent_km_rag_writeback_pending_offsite_false_sqlite_application_consistent_export_pending"
    elif [ "${overall_terminal}" = "check_pass_no_write" ]; then
        emit_receipt "closure_writeback" "not_applicable" \
            "check_mode_no_runtime_change_local_durable_receipt_ack"
    else
        emit_receipt "closure_writeback" "deferred" \
            "failed_terminal_requires_followup_no_completion_claim"
    fi
    emit_receipt "terminal" "${overall_terminal}" \
        "backup_canary_${overall_terminal}_clickhouse_native_scope_only_offsite_false_sqlite_pending"
    printf 'SIGNOZ_BACKUP_CANARY_TERMINAL trace_id=%s run_id=%s work_item_id=%s terminal=%s lease=%s exit_code=%s\n' \
        "${TRACE_ID}" "${RUN_ID}" "${WORK_ITEM_ID}" "${overall_terminal}" \
        "${lease_terminal}" "${exit_code}"
    exit "${exit_code}"
}

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

setup_receipts() {
    valid_identity "${TRACE_ID}" || { log_error "TRACE_ID is required and must be path-safe"; exit 2; }
    valid_identity "${RUN_ID}" || { log_error "RUN_ID is required and must be path-safe"; exit 2; }
    valid_identity "${WORK_ITEM_ID}" || { log_error "WORK_ITEM_ID is required and must be path-safe"; exit 2; }

    [ ! -L "${RECEIPT_ROOT}" ] || { log_error "receipt root must not be a symlink"; exit 2; }
    mkdir -p "${RECEIPT_ROOT}"
    RECEIPT_DIR="${RECEIPT_ROOT}/${RUN_ID}"
    [ ! -e "${RECEIPT_DIR}" ] || { log_error "receipt directory already exists: ${RECEIPT_DIR}"; exit 2; }
    mkdir -m 700 "${RECEIPT_DIR}"

    RECEIPT_LOG="${RECEIPT_DIR}/receipts.jsonl"
    BACKUP_OUTPUT="${RECEIPT_DIR}/backup-output.log"
    MONITOR_WINDOW="${RECEIPT_DIR}/monitor-window.log"
    STATE_OBSERVATIONS="${RECEIPT_DIR}/collector-state.tsv"
    ROLLBACK_FILE="${RECEIPT_DIR}/cooldown.rollback"
    ROLLBACK_METADATA="${RECEIPT_DIR}/cooldown.rollback.json"
    : > "${RECEIPT_LOG}"
    RECEIPTS_READY=1

    trap cleanup EXIT
    trap 'on_signal 1' HUP
    trap 'on_signal 2' INT
    trap 'on_signal 15' TERM
}

verify_monitor_contract() {
    local cron_output cron_count cron_line exclude_count exclude_line

    [ -f "${MONITOR_SCRIPT}" ] && [ -r "${MONITOR_SCRIPT}" ] && [ ! -L "${MONITOR_SCRIPT}" ] \
        || fail_closed "monitor_script_missing_unreadable_or_symlink"
    [ -f "${MONITOR_LOG}" ] && [ -r "${MONITOR_LOG}" ] && [ ! -L "${MONITOR_LOG}" ] \
        || fail_closed "monitor_log_missing_unreadable_or_symlink"
    [ -d "${MONITOR_COOLDOWN_DIR}" ] && [ -w "${MONITOR_COOLDOWN_DIR}" ] && [ ! -L "${MONITOR_COOLDOWN_DIR}" ] \
        || fail_closed "monitor_cooldown_directory_missing_unwritable_or_symlink"

    require_monitor_fragment "COOLDOWN_DIR:=${MONITOR_COOLDOWN_DIR}"
    require_monitor_fragment 'local cooldown_file="${COOLDOWN_DIR}/${container}.cooldown"'
    require_monitor_fragment 'last_sent=$(cat "$cooldown_file")'
    # The live monitor has another cooldown path with the same elapsed
    # calculation. Bind this assertion to is_in_cooldown instead of requiring
    # a globally unique string.
    require_monitor_function_fragment is_in_cooldown 'elapsed=$(( now - last_sent ))'
    require_monitor_fragment 'if (( elapsed < ACTION_COOLDOWN_SECONDS )); then'
    require_monitor_fragment 'COOLDOWN: ${container}'
    require_monitor_fragment 'is_in_cooldown "$container_name" && continue'
    require_monitor_fragment 'set_cooldown "$container_name"'
    require_monitor_fragment 'log "AUTO_REPAIR: docker restart ${container}"'
    require_monitor_fragment 'if docker restart "$container"'

    exclude_count="$(grep -F -c 'EXCLUDE_CONTAINERS:=' "${MONITOR_SCRIPT}" || true)"
    [ "${exclude_count}" -eq 1 ] || fail_closed "monitor_exclude_contract_ambiguous"
    exclude_line="$(grep -F 'EXCLUDE_CONTAINERS:=' "${MONITOR_SCRIPT}")"
    [[ "${exclude_line}" != *"${COLLECTOR_NAME}"* ]] || fail_closed "collector_already_excluded_monitor_contract_changed"
    [ "$(grep -F -c "${COLLECTOR_NAME}" "${MONITOR_SCRIPT}" || true)" -eq 0 ] \
        || fail_closed "collector_literal_present_monitor_contract_changed"

    cron_output="$(crontab -l 2>/dev/null)" || fail_closed "monitor_crontab_unreadable"
    cron_count="$(printf '%s\n' "${cron_output}" | awk -v script="${MONITOR_SCRIPT}" '
        $0 !~ /^[[:space:]]*#/ && index($0, script) { count++ }
        END { print count + 0 }
    ')"
    [ "${cron_count}" -eq 1 ] || fail_closed "monitor_cron_contract_ambiguous_count_${cron_count}"
    cron_line="$(printf '%s\n' "${cron_output}" | awk -v script="${MONITOR_SCRIPT}" '
        $0 !~ /^[[:space:]]*#/ && index($0, script) { print }
    ')"
    [[ "${cron_line}" == "*/5 * * * * "* ]] || fail_closed "monitor_cron_schedule_not_every_five_minutes"
    [[ "${cron_line}" == *">> ${MONITOR_LOG}"* ]] || fail_closed "monitor_cron_log_route_mismatch"
    [[ "${cron_line}" != *"COOLDOWN_DIR="* ]] || fail_closed "monitor_cron_cooldown_override_ambiguous"

}

inspect_prior_lease() {
    local current_uid current_gid
    current_uid="$(id -u)"
    current_gid="$(id -g)"

    if [ -e "${COOLDOWN_FILE}" ]; then
        [ -f "${COOLDOWN_FILE}" ] && [ ! -L "${COOLDOWN_FILE}" ] && [ -r "${COOLDOWN_FILE}" ] && [ -w "${COOLDOWN_FILE}" ] \
            || fail_closed "existing_cooldown_receipt_unsafe"
        PRIOR_LEASE_PRESENT=1
        PRIOR_LEASE_MODE="$(stat -c '%a' "${COOLDOWN_FILE}")"
        PRIOR_LEASE_UID="$(stat -c '%u' "${COOLDOWN_FILE}")"
        PRIOR_LEASE_GID="$(stat -c '%g' "${COOLDOWN_FILE}")"
        [ "${PRIOR_LEASE_UID}" = "${current_uid}" ] && [ "${PRIOR_LEASE_GID}" = "${current_gid}" ] \
            || fail_closed "existing_cooldown_receipt_owner_mismatch"
        PRIOR_LEASE_VALUE="$(tr -d '\r\n' < "${COOLDOWN_FILE}")"
        [[ "${PRIOR_LEASE_VALUE}" =~ ^[0-9]+$ ]] || fail_closed "existing_cooldown_receipt_not_epoch"
        PRIOR_LEASE_HASH="$(sha256sum "${COOLDOWN_FILE}" | awk '{print $1}')"
    else
        PRIOR_LEASE_PRESENT=0
        PRIOR_LEASE_MODE="664"
        PRIOR_LEASE_UID="${current_uid}"
        PRIOR_LEASE_GID="${current_gid}"
        PRIOR_LEASE_HASH="absent"
        PRIOR_LEASE_VALUE="absent"
    fi

}

capture_prior_lease() {
    local current_hash current_value

    if [ "${PRIOR_LEASE_PRESENT}" -eq 1 ]; then
        [ -f "${COOLDOWN_FILE}" ] && [ ! -L "${COOLDOWN_FILE}" ] && [ -r "${COOLDOWN_FILE}" ] && [ -w "${COOLDOWN_FILE}" ] \
            || fail_closed "cooldown_changed_between_check_and_rollback_capture"
        current_hash="$(sha256sum "${COOLDOWN_FILE}" | awk '{print $1}')"
        current_value="$(tr -d '\r\n' < "${COOLDOWN_FILE}")"
        [ "${current_hash}" = "${PRIOR_LEASE_HASH}" ] \
            && [ "$(stat -c '%a' "${COOLDOWN_FILE}")" = "${PRIOR_LEASE_MODE}" ] \
            && [ "$(stat -c '%u' "${COOLDOWN_FILE}")" = "${PRIOR_LEASE_UID}" ] \
            && [ "$(stat -c '%g' "${COOLDOWN_FILE}")" = "${PRIOR_LEASE_GID}" ] \
            && [ "${current_value}" = "${PRIOR_LEASE_VALUE}" ] \
            || fail_closed "cooldown_changed_between_check_and_rollback_capture"
        cp -p "${COOLDOWN_FILE}" "${ROLLBACK_FILE}"
        [ "$(sha256sum "${ROLLBACK_FILE}" | awk '{print $1}')" = "${PRIOR_LEASE_HASH}" ] \
            || fail_closed "cooldown_changed_between_check_and_rollback_capture"
    else
        [ ! -e "${COOLDOWN_FILE}" ] \
            || fail_closed "cooldown_appeared_between_check_and_rollback_capture"
    fi

    printf '{"trace_id":"%s","run_id":"%s","work_item_id":"%s","prior_present":%s,"prior_mode":"%s","prior_uid":"%s","prior_gid":"%s","prior_hash":"%s","prior_value":"%s"}\n' \
        "${TRACE_ID}" "${RUN_ID}" "${WORK_ITEM_ID}" "${PRIOR_LEASE_PRESENT}" \
        "${PRIOR_LEASE_MODE}" "${PRIOR_LEASE_UID}" "${PRIOR_LEASE_GID}" \
        "${PRIOR_LEASE_HASH}" "${PRIOR_LEASE_VALUE}" > "${ROLLBACK_METADATA}"
    LEASE_ROLLBACK_READY=1
    emit_receipt "rollback_capture" "pass" "prior_cooldown_receipt_captured"
}

arm_lease() {
    local now lease_epoch lease_hash
    now="$(date +%s)"
    lease_epoch=$((now + CANARY_TIMEOUT_SECONDS + LEASE_MARGIN_SECONDS))
    LEASE_TMP="$(mktemp "${MONITOR_COOLDOWN_DIR}/.${COLLECTOR_NAME}.lease.${RUN_ID}.XXXXXX")"
    printf '%s\n' "${lease_epoch}" > "${LEASE_TMP}"
    chmod "${PRIOR_LEASE_MODE}" "${LEASE_TMP}"

    LEASE_MUTATION_STARTED=1
    mv -f "${LEASE_TMP}" "${COOLDOWN_FILE}"
    LEASE_TMP=""
    LEASE_ARMED=1

    [ "$(tr -d '\r\n' < "${COOLDOWN_FILE}")" = "${lease_epoch}" ] \
        || fail_closed "cooldown_lease_epoch_readback_mismatch"
    [ "$(stat -c '%u' "${COOLDOWN_FILE}")" = "$(id -u)" ] \
        || fail_closed "cooldown_lease_owner_readback_mismatch"
    [ "$(stat -c '%g' "${COOLDOWN_FILE}")" = "$(id -g)" ] \
        || fail_closed "cooldown_lease_group_readback_mismatch"
    lease_hash="$(sha256sum "${COOLDOWN_FILE}" | awk '{print $1}')"
    emit_receipt "lease" "armed" "future_epoch_${lease_epoch}_hash_${lease_hash}"
}

observe_collector() {
    local epoch state
    while kill -0 "${BACKUP_PID}" 2>/dev/null; do
        epoch="$(date +%s)"
        if state="$(collector_running_state)"; then
            printf '%s\t%s\n' "${epoch}" "${state}" >> "${STATE_OBSERVATIONS}"
        else
            printf '%s\tunknown\n' "${epoch}" >> "${STATE_OBSERVATIONS}"
        fi
        sleep "${STATE_POLL_SECONDS}"
    done

    epoch="$(date +%s)"
    if state="$(collector_running_state)"; then
        printf '%s\t%s\n' "${epoch}" "${state}" >> "${STATE_OBSERVATIONS}"
    else
        printf '%s\tunknown\n' "${epoch}" >> "${STATE_OBSERVATIONS}"
    fi
}

verify_monitor_window() {
    local log_identity_before="$1"
    local log_size_before="$2"
    local log_identity_after log_size_after start_byte
    local auto_repair_count detected_count cooldown_count
    local stop_epoch restore_epoch next_cron_epoch cron_crossed=0

    log_identity_after="$(stat -c '%d:%i' "${MONITOR_LOG}")"
    log_size_after="$(stat -c '%s' "${MONITOR_LOG}")"
    [ "${log_identity_after}" = "${log_identity_before}" ] \
        || fail_closed "monitor_log_rotated_during_canary"
    [ "${log_size_after}" -ge "${log_size_before}" ] \
        || fail_closed "monitor_log_shrank_during_canary"

    start_byte=$((log_size_before + 1))
    tail -c "+${start_byte}" "${MONITOR_LOG}" > "${MONITOR_WINDOW}"
    auto_repair_count="$(grep -a -F -c "AUTO_REPAIR: docker restart ${COLLECTOR_NAME}" "${MONITOR_WINDOW}" || true)"
    detected_count="$(grep -a -F -c "DETECTED: ${COLLECTOR_NAME} state=exited" "${MONITOR_WINDOW}" || true)"
    cooldown_count="$(grep -a -F -c "COOLDOWN: ${COLLECTOR_NAME}" "${MONITOR_WINDOW}" || true)"

    [ "${auto_repair_count}" -eq 0 ] || fail_closed "monitor_auto_repair_contaminated_canary"
    [ "$(awk '$2 == "unknown" { count++ } END { print count + 0 }' "${STATE_OBSERVATIONS}")" -eq 0 ] \
        || fail_closed "collector_state_observer_ambiguous"

    if [ "${EXPECT_COLLECTOR_STOP}" = "0" ]; then
        [ "$(awk '$2 != "running" { count++ } END { print count + 0 }' "${STATE_OBSERVATIONS}")" -eq 0 ] \
            || fail_closed "online_native_backup_collector_continuity_failed"
        [ "${detected_count}" -eq 0 ] \
            || fail_closed "online_native_backup_monitor_detected_collector"
        emit_receipt "monitor_verifier" "pass" \
            "online_native_collector_continuous_auto_repair_0_detected_0"
        return 0
    fi

    stop_epoch="$(awk '$2 == "stopped" { print $1; exit }' "${STATE_OBSERVATIONS}")"
    [ -n "${stop_epoch}" ] || fail_closed "collector_stop_transition_not_observed"
    restore_epoch="$(awk '
        $2 == "stopped" { stopped_seen = 1; next }
        stopped_seen && $2 == "running" { print $1; exit }
    ' "${STATE_OBSERVATIONS}")"
    [ -n "${restore_epoch}" ] || fail_closed "collector_restore_transition_not_observed"

    next_cron_epoch=$(( ((stop_epoch / MONITOR_CRON_INTERVAL_SECONDS) + 1) * MONITOR_CRON_INTERVAL_SECONDS ))
    if [ "${next_cron_epoch}" -le "${restore_epoch}" ]; then
        cron_crossed=1
    fi

    if [ "${detected_count}" -gt 0 ]; then
        [ "${cooldown_count}" -ge "${detected_count}" ] \
            || fail_closed "monitor_detected_without_matching_cooldown"
    fi
    if [ "${cron_crossed}" -eq 1 ]; then
        [ "${detected_count}" -gt 0 ] && [ "${cooldown_count}" -gt 0 ] \
            || fail_closed "cron_crossed_without_collector_cooldown_receipt"
    fi

    emit_receipt "monitor_verifier" "pass" \
        "auto_repair_${auto_repair_count}_detected_${detected_count}_cooldown_${cooldown_count}_cron_crossed_${cron_crossed}"
}

verify_online_collector_window() {
    [ "$(awk '$2 == "unknown" { count++ } END { print count + 0 }' "${STATE_OBSERVATIONS}")" -eq 0 ] \
        || fail_closed "collector_state_observer_ambiguous"
    [ "$(awk '$2 != "running" { count++ } END { print count + 0 }' "${STATE_OBSERVATIONS}")" -eq 0 ] \
        || fail_closed "online_native_backup_collector_continuity_failed"
    emit_receipt "monitor_verifier" "pass" \
        "online_native_collector_continuous_monitor_contract_not_applicable"
}

main() {
    local container_id_before container_started_at_before container_restart_count_before
    local container_health_before collector_metadata_before
    local monitor_log_identity="" monitor_log_size=""
    local source_diff_detail ai_decision_detail
    local backup_status native_restore_contract_detail="not_applicable"

    setup_receipts

    [ "$#" -eq 1 ] || fail_closed "exactly_one_mode_required_use_check_or_apply"
    case "$1" in
        --check) MODE="check" ;;
        --apply) MODE="apply" ;;
        *) fail_closed "unsupported_mode_use_check_or_apply" ;;
    esac

    # Hold the canary lock before inspecting or hashing any mutable backup or
    # restore toolchain path.  This closes the check/apply substitution window.
    require_command flock
    acquire_canary_lock

    for command_name in awk date docker env grep kill pgrep python3 sha256sum sleep ss timeout tr; do
        require_command "${command_name}"
    done
    valid_positive_integer "${CANARY_TIMEOUT_SECONDS}" || fail_closed "invalid_canary_timeout"
    valid_positive_integer "${TIMEOUT_KILL_AFTER_SECONDS}" || fail_closed "invalid_timeout_kill_after"
    valid_positive_integer "${DOCKER_TIMEOUT_SECONDS}" || fail_closed "invalid_docker_timeout"
    valid_poll_interval "${STATE_POLL_SECONDS}" || fail_closed "invalid_state_poll_interval"
    case "${EXPECT_COLLECTOR_STOP}" in
        0|1) ;;
        *) fail_closed "invalid_expect_collector_stop_mode" ;;
    esac

    [ -f "${BACKUP_SCRIPT}" ] && [ -x "${BACKUP_SCRIPT}" ] && [ ! -L "${BACKUP_SCRIPT}" ] \
        || fail_closed "backup_script_missing_unexecutable_or_symlink"
    if [ "${EXPECT_COLLECTOR_STOP}" = "0" ]; then
        require_native_restore_contract
        require_native_receipt_contract
        native_restore_contract_detail="hook_$(sha256sum "${CLICKHOUSE_NATIVE_POST_VERIFY_HOOK}" | awk '{print $1}')_inventory_$(sha256sum "${CLICKHOUSE_RESTORE_INVENTORY_HELPER}" | awk '{print $1}')_disk_config_$(sha256sum "${CLICKHOUSE_RESTORE_BACKUP_DISK_CONFIG}" | awk '{print $1}')_cluster_config_$(sha256sum "${CLICKHOUSE_RESTORE_CLUSTER_CONFIG}" | awk '{print $1}')"
    else
        for command_name in cp crontab id mktemp mv rm stat tail; do
            require_command "${command_name}"
        done
        valid_positive_integer "${LEASE_MARGIN_SECONDS}" || fail_closed "invalid_lease_margin"
        valid_positive_integer "${MONITOR_CRON_INTERVAL_SECONDS}" || fail_closed "invalid_monitor_interval"
        [ "${MONITOR_CRON_INTERVAL_SECONDS}" -le 300 ] || fail_closed "monitor_interval_exceeds_cron_contract"
        verify_monitor_contract
    fi

    if pgrep -af '(^|/|[[:space:]])backup-signoz[.]sh([[:space:]]|$)' >/dev/null 2>&1; then
        fail_closed "another_signoz_backup_is_running"
    fi

    collector_metadata_before="$(collector_runtime_metadata)" \
        || fail_closed "collector_runtime_metadata_unreadable"
    IFS=$'\t' read -r container_id_before _container_running_before \
        container_started_at_before container_restart_count_before container_health_before \
        <<< "${collector_metadata_before}"
    [ -n "${container_id_before}" ] \
        && [ -n "${container_started_at_before}" ] \
        && [[ "${container_restart_count_before}" =~ ^[0-9]+$ ]] \
        && [[ "${container_health_before}" =~ ^(healthy|none)$ ]] \
        || fail_closed "collector_runtime_metadata_invalid"
    verify_collector_runtime "${container_id_before}" "${container_started_at_before}" \
        "${container_restart_count_before}" "${container_health_before}" \
        || fail_closed "collector_preflight_runtime_failed"

    if [ "${EXPECT_COLLECTOR_STOP}" = "1" ]; then
        monitor_log_identity="$(stat -c '%d:%i' "${MONITOR_LOG}")"
        monitor_log_size="$(stat -c '%s' "${MONITOR_LOG}")"
        inspect_prior_lease
        emit_receipt "sensor_source" "pass" \
            "monitor_hash_$(sha256sum "${MONITOR_SCRIPT}" | awk '{print $1}')_backup_hash_$(sha256sum "${BACKUP_SCRIPT}" | awk '{print $1}')_collector_${container_id_before}_started_${container_started_at_before}_restarts_${container_restart_count_before}_health_${container_health_before}"
        source_diff_detail="legacy_monitor_and_cron_contract_exact_cooldown_prior_present_${PRIOR_LEASE_PRESENT}_hash_${PRIOR_LEASE_HASH}"
        ai_decision_detail="arm_exact_container_future_epoch_run_legacy_stop_backup_then_restore"
    else
        PRIOR_LEASE_PRESENT=0
        PRIOR_LEASE_HASH="not_applicable"
        emit_receipt "sensor_source" "pass" \
            "online_backup_hash_$(sha256sum "${BACKUP_SCRIPT}" | awk '{print $1}')_${native_restore_contract_detail}_collector_${container_id_before}_started_${container_started_at_before}_restarts_${container_restart_count_before}_health_${container_health_before}"
        source_diff_detail="online_native_backup_isolated_restore_contract_bound_monitor_and_cooldown_contract_not_applicable"
        ai_decision_detail="run_online_native_backup_with_mandatory_isolated_restore_verifier_without_collector_or_monitor_mutation"
    fi
    emit_receipt "normalized_asset_identity" "pass" \
        "asset_container_${COLLECTOR_NAME}_runtime_id_${container_id_before}_cluster_host_docker"
    emit_receipt "source_of_truth_diff" "pass" "${source_diff_detail}"
    emit_receipt "ai_decision" "candidate" "${ai_decision_detail}"
    emit_receipt "risk_policy" "pass" \
        "risk_medium_bounded_timeout_no_retention_cleanup_runtime_metadata_mandatory_isolated_restore_independent_verifier"
    emit_receipt "check" "pass" \
        "mode_${MODE}_expect_stop_${EXPECT_COLLECTOR_STOP}_${native_restore_contract_detail}_collector_same_identity_started_restart_health_running_listeners_lock_clear_backup_absent"

    if [ "${MODE}" = "check" ]; then
        CHECK_VERIFIED=1
        log_info "SigNoz backup canary check passed without arming lease or launching backup"
        return 0
    fi

    if [ "${EXPECT_COLLECTOR_STOP}" = "1" ]; then
        capture_prior_lease
        arm_lease
    else
        emit_receipt "lease" "not_required" \
            "online_native_backup_has_no_collector_stop_or_cooldown_write"
    fi

    : > "${STATE_OBSERVATIONS}"
    set +e
    timeout --kill-after="${TIMEOUT_KILL_AFTER_SECONDS}" "${CANARY_TIMEOUT_SECONDS}" \
        env TRACE_ID="${TRACE_ID}" RUN_ID="${RUN_ID}" WORK_ITEM_ID="${WORK_ITEM_ID}" \
        CLICKHOUSE_NATIVE_POST_VERIFY_HOOK="${CLICKHOUSE_NATIVE_POST_VERIFY_HOOK}" \
        CLICKHOUSE_RESTORE_INVENTORY_HELPER="${CLICKHOUSE_RESTORE_INVENTORY_HELPER}" \
        CLICKHOUSE_RESTORE_BACKUP_DISK_CONFIG="${CLICKHOUSE_RESTORE_BACKUP_DISK_CONFIG}" \
        CLICKHOUSE_RESTORE_CLUSTER_CONFIG="${CLICKHOUSE_RESTORE_CLUSTER_CONFIG}" \
        CLICKHOUSE_NATIVE_RECEIPT_ROOT="${CLICKHOUSE_NATIVE_RECEIPT_ROOT}" \
        CLICKHOUSE_RESTORE_RECEIPT_ROOT="${CLICKHOUSE_RESTORE_RECEIPT_ROOT}" \
        BACKUP_SKIP_RETENTION_CLEANUP=1 "${BACKUP_SCRIPT}" \
        > "${BACKUP_OUTPUT}" 2>&1 &
    BACKUP_PID=$!
    observe_collector &
    OBSERVER_PID=$!
    wait "${BACKUP_PID}"
    backup_status=$?
    BACKUP_PID=""
    wait "${OBSERVER_PID}"
    OBSERVER_PID=""
    set -e

    emit_receipt "execution" "$([ "${backup_status}" -eq 0 ] && printf pass || printf failed)" \
        "backup_exit_${backup_status}"
    [ "${backup_status}" -eq 0 ] || fail_closed "backup_terminal_nonzero_${backup_status}"
    grep -a -F -q '略過 SignOz retention cleanup (BACKUP_SKIP_RETENTION_CLEANUP=1)' "${BACKUP_OUTPUT}" \
        || fail_closed "retention_skip_receipt_missing"
    if [ "${EXPECT_COLLECTOR_STOP}" = "1" ]; then
        grep -a -F -q '========== SignOz 備份完成 (' "${BACKUP_OUTPUT}" \
            || fail_closed "backup_success_terminal_missing"
    else
        grep -a -F -q '========== SigNoz ClickHouse native 備份完成 (' "${BACKUP_OUTPUT}" \
            || fail_closed "native_backup_success_terminal_missing"
        verify_same_run_native_restore_receipts
    fi

    if [ "${EXPECT_COLLECTOR_STOP}" = "1" ]; then
        verify_monitor_window "${monitor_log_identity}" "${monitor_log_size}"
    else
        verify_online_collector_window
    fi
    verify_collector_runtime "${container_id_before}" "${container_started_at_before}" \
        "${container_restart_count_before}" "${container_health_before}" \
        || fail_closed "collector_post_backup_runtime_failed"
    emit_receipt "post_verifier" "pass" \
        "collector_running_same_identity_started_at_restart_count_health_listeners_4317_4318"

    CANARY_VERIFIED=1
    log_info "SigNoz backup canary verified; cooldown lease will now roll back"
}

main "$@"