#!/usr/bin/env bash

set -Eeuo pipefail
umask 077

PROGRAM=$(basename "$0")
BUNDLE_ROOT=${BUNDLE_ROOT:-/opt/zhct-deploy}
EVIDENCE_BASE=${EVIDENCE_BASE:-/var/backups/zhct-q131-132}
RUN_ID=
Q131_PASS_SHA=
ACTIVATION_RUN_ID=
SOURCE_RUN_ID=
TASK=
ACTION=
STARTED_AT=

usage() {
    cat <<'USAGE'
Usage:
  redis-q131-q132-role-control.sh --static-validate [--bundle-root PATH]
  redis-q131-q132-role-control.sh q131 preflight [--run-id ID]
  redis-q131-q132-role-control.sh q131 execute [--run-id ID]
  redis-q131-q132-role-control.sh q131 verify [--run-id ID]
  redis-q131-q132-role-control.sh q131 rollback --activation-run-id ID [--run-id ID]
  redis-q131-q132-role-control.sh q131 archive --source-run-id ID
  redis-q131-q132-role-control.sh q132 preflight --q131-pass-sha SHA256 [--run-id ID]
  redis-q131-q132-role-control.sh q132 execute --q131-pass-sha SHA256 [--run-id ID]
  redis-q131-q132-role-control.sh q132 verify --q131-pass-sha SHA256 [--run-id ID]
  redis-q131-q132-role-control.sh q132 rollback --activation-run-id ID [--run-id ID]
  redis-q131-q132-role-control.sh q132 archive --source-run-id ID

Options:
  --bundle-root PATH
  --evidence-base PATH
  --run-id ID
  --q131-pass-sha SHA256
  --activation-run-id ID
  --source-run-id ID

The script never provisions or prints secret values. Secret files must already
exist with the required ownership and mode.
USAGE
}

timestamp() {
    date '+%Y-%m-%dT%H:%M:%S%z'
}

log() {
    printf '%s [%s] %s\n' "$(timestamp)" "$PROGRAM" "$*"
}

die() {
    log "ERROR: $*"
    exit 1
}

require_command() {
    command -v "$1" >/dev/null 2>&1 || die "required command not found: $1"
}

require_root() {
    [[ $(id -u) -eq 0 ]] || die "remote actions require root"
}

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

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

sanitize_stream() {
    sed -E \
        -e 's/([Pp]assword|[Tt]oken|[Cc]ookie|[Ss]ecret)([[:space:]]*[:=][[:space:]]*)[^[:space:]",}]+/\1\2[REDACTED]/g' \
        -e 's/(requirepass|masterauth|auth-pass)([[:space:]]+).*/\1\2[REDACTED]/g'
}

expected_asset_checksums() {
    cat <<'EOF'
e3eae16aa5c87f43b56495fc71de378a947578cabe18b507e8fed01ac1f32025 common/bin/role-action.sh
4c06e4ad325de7241607b06d93c99c2f35f3422bfcbdbeeb43b530f2053f4bb6 common/lib/runtime.sh
e637ca657a897c6d0b689bbd6347f04516baf5c9f32350cd1e8a2071c211a0a5 common/scripts/activate-role.sh
ca5fa0a113739bcbe3ff51b2815bb50723cd8b1cf33dfb6d3f3f172caf2dabc6 common/scripts/post-clone.sh
a1c5bad3856b998c7e826cbbec0f70725d3bfb14eced4b9dbb2b88135b03cac5 common/scripts/rollback-role.sh
0e897cb18a4731e384f6beb2f0e5bf64398f4d5d5f6a5b20b202f37c1ecfdb42 common/scripts/verify-role.sh
a9195d2d8347ba1cde725ff3b9ca47b8e839a17972dfae1dceaadd0e475c03ea roles/redis/role.env
5c7ab26d6a10bc4d0fc362c77c918dde75b2599294e8c374c199b9d08c65bc97 roles/redis/hooks.sh
3a5790a52193017e2fdef76fc7c21893b21f14e51927d62314d48b079eba79b5 roles/redis/manifests/nodes.tsv
84d9f92fafd5acd250a2aba1e1a2f4084fb5066814f564b79c072a6afec767e1 roles/redis/manifests/ports.tsv
6eeb07767026b53b461a5d977ce8d0bc4bcb246d1f751e126fc752aa38503bde roles/redis/manifests/secrets.required.tsv
9bf844d68f6616355ef269cc6d901ccb8f822f4fc30d5bec72d9ff1c10c5eb0b roles/redis/manifests/systemd-units.tsv
7d6622ade8a2b42a4152911c71f4fd540114e00f3d49ab4a31d7cc36f745f5c0 roles/redis/templates/nodes/redis01/orchestrator.conf.json
7c85b7cc8fb97b64117c070d0bf61fd7310fd304aa1c25996068904920f27d22 roles/redis/templates/nodes/redis01/redis5.conf
f36db18a63c8d3d10ccfe785cc7227b55e6ea67b56e6b3f8301e6b69081e90cc roles/redis/templates/nodes/redis01/redis7.conf
3679608812afcc6c3e0044bfe30965c18b129c51de287ddbdd9153dc912aaec1 roles/redis/templates/nodes/redis01/sentinel5.conf
551d97e393a8e947a36cdca33e62b57ee441f1e92e0cedda04be6e5150f1bfc3 roles/redis/templates/nodes/redis01/sentinel7.conf
2b2ce57e346d901529760b4a265c73839384f7a1c0692113cea312cd8012b121 roles/redis/templates/nodes/redis02/orchestrator.conf.json
c79440cd6d43b5205902ec1f2e053e26a767a0b2e0550d112e25ca726c023127 roles/redis/templates/nodes/redis02/redis5.conf
e5ce8c72650df725b97cb2be4b2a1a3220896276bd9e3705d295c71ce4da7b38 roles/redis/templates/nodes/redis02/redis7.conf
447198c2b42f320410190bfe845dc7dfefc43d1fe89e5eed53aa5d5e9ff8bea2 roles/redis/templates/nodes/redis02/sentinel5.conf
0163851f139fdd6bac43f4ba725365ea091b6edc94a58ffb39801f28112c0868 roles/redis/templates/nodes/redis02/sentinel7.conf
b769b7f766e2eb62e5d24870c0162f6906ddb68daa72c6ae6eb9b9d489c6e6d5 roles/redis/templates/systemd/redis5.service
185bffed16998127dd508ac39c7bc9f60514b7ff90e7ff689a608de607544cd4 roles/redis/templates/systemd/redis7.service
f2b1feb9e87e058e17e3f7a36e405f83ad921484ec655fd04883bfe26d3fc113 roles/redis/templates/systemd/sentinel5.service
d503b2cccdd8102eca7445c98cc6b22d9d3f4f0503d158000e28b10041769368 roles/redis/templates/systemd/sentinel7.service
EOF
}

verify_assets() {
    local manifest path
    manifest=$(mktemp)
    trap 'rm -f -- "$manifest"' RETURN
    while read -r sha path; do
        printf '%s  %s/%s\n' "$sha" "$BUNDLE_ROOT" "$path"
    done < <(expected_asset_checksums) >"$manifest"
    sha256sum -c "$manifest"
    trap - RETURN
    rm -f -- "$manifest"
}

static_validate() {
    local script_dir policy_root dryrun01 dryrun02
    require_command bash
    require_command grep
    require_command sha256sum
    script_dir=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
    if [[ -f "$script_dir/SHA256SUMS" ]]; then
        (cd "$script_dir" && sha256sum -c SHA256SUMS)
    fi
    verify_assets

    bash -n \
        "$BUNDLE_ROOT/common/bin/role-action.sh" \
        "$BUNDLE_ROOT/common/lib/runtime.sh" \
        "$BUNDLE_ROOT/common/scripts/activate-role.sh" \
        "$BUNDLE_ROOT/common/scripts/post-clone.sh" \
        "$BUNDLE_ROOT/common/scripts/rollback-role.sh" \
        "$BUNDLE_ROOT/common/scripts/verify-role.sh" \
        "$BUNDLE_ROOT/roles/redis/hooks.sh"

    grep -Fqx $'redis01\t10.2.203.112\tsqjcredis01\tredis-primary-and-raft-voter\ttemplate' \
        "$BUNDLE_ROOT/roles/redis/manifests/nodes.tsv"
    grep -Fqx $'redis02\t10.2.203.113\tsqjcredis02\tredis-replica-no-orchestrator\tredis01' \
        "$BUNDLE_ROOT/roles/redis/manifests/nodes.tsv"
    grep -Fqx $'orchestrator.service\tdisabled,inactive\tredis01=enabled,active;redis02=disabled,inactive\tthird MySQL HA Raft voter' \
        "$BUNDLE_ROOT/roles/redis/manifests/systemd-units.tsv"
    grep -Fqx 'replicaof 10.2.203.112 6379' \
        "$BUNDLE_ROOT/roles/redis/templates/nodes/redis02/redis5.conf"
    grep -Fqx 'replicaof 10.2.203.112 6387' \
        "$BUNDLE_ROOT/roles/redis/templates/nodes/redis02/redis7.conf"
    ! grep -Eq '^[[:space:]]*replicaof[[:space:]]' \
        "$BUNDLE_ROOT/roles/redis/templates/nodes/redis01/redis5.conf"
    ! grep -Eq '^[[:space:]]*replicaof[[:space:]]' \
        "$BUNDLE_ROOT/roles/redis/templates/nodes/redis01/redis7.conf"

    policy_root=$(prepare_policy_bundle)
    trap 'rm -rf -- "$policy_root"' RETURN
    dryrun01=$(ZHCT_STATIC_VALIDATE=1 \
        "$BUNDLE_ROOT/common/scripts/activate-role.sh" \
        --role redis --bundle-root "$policy_root" --node redis01 --dry-run 2>&1)
    printf '%s\n' "$dryrun01"
    dryrun02=$(ZHCT_STATIC_VALIDATE=1 \
        "$BUNDLE_ROOT/common/scripts/activate-role.sh" \
        --role redis --bundle-root "$policy_root" --node redis02 --dry-run 2>&1)
    printf '%s\n' "$dryrun02"
    local unit dryrun
    for dryrun in "$dryrun01" "$dryrun02"; do
        for unit in sentinel5 sentinel7 orchestrator; do
            grep -Fq "RUN: systemctl disable --now $unit.service" <<<"$dryrun"
        done
        ! grep -Eq 'RUN: systemctl (enable|start|restart) --now (sentinel5|sentinel7|orchestrator)\.service([[:space:]]|$)' <<<"$dryrun"
        ! grep -Eq 'RUN: systemctl (enable|start|restart) (sentinel5|sentinel7|orchestrator)\.service([[:space:]]|$)' <<<"$dryrun"
    done
    rm -rf -- "$policy_root"
    trap - RETURN

    printf 'Q131_Q132_STATIC_VALIDATE=PASS\n'
}

set_task_policy() {
    case "$TASK" in
        q131)
            EXPECTED_HOST=sqjcredis01
            EXPECTED_IP=10.2.203.112
            NODE_KEY=redis01
            EXPECTED_ACTIVE_UNITS=(redis5 redis7)
            EXPECTED_INACTIVE_UNITS=(sentinel5 sentinel7 orchestrator)
            EXPECTED_PORTS=(6379 6387)
            ;;
        q132)
            EXPECTED_HOST=sqjcredis02
            EXPECTED_IP=10.2.203.113
            NODE_KEY=redis02
            EXPECTED_ACTIVE_UNITS=(redis5 redis7)
            EXPECTED_INACTIVE_UNITS=(sentinel5 sentinel7 orchestrator)
            EXPECTED_PORTS=(6379 6387)
            ;;
        *)
            die "task must be q131 or q132"
            ;;
    esac
}

assert_target_identity() {
    local actual_host
    actual_host=$(hostname -s)
    [[ "$actual_host" == "$EXPECTED_HOST" ]] ||
        die "hostname $actual_host does not match $EXPECTED_HOST"
    ip -o -4 addr show scope global |
        awk '{split($4,a,"/"); print a[1]}' |
        grep -Fxq "$EXPECTED_IP" ||
        die "expected IP $EXPECTED_IP is not configured"
    if ip -o -4 addr show scope global |
        awk '{split($4,a,"/"); print a[1]}' |
        grep -Eq '^10\.2\.203\.(115|116|117)$'
    then
        die "VIP is present before its authorized task"
    fi
}

private_host_key_count() {
    find /etc/ssh -maxdepth 1 -type f -name 'ssh_host_*_key' | wc -l
}

node_identity_state() {
    local machine_ready=0 keys_ready=0 env_ready=0
    [[ -s /etc/machine-id ]] && machine_ready=1
    [[ $(private_host_key_count) -eq 3 ]] && keys_ready=1
    if [[ -r /etc/zhct/node.env ]] &&
        grep -Fxq 'ROLE=redis' /etc/zhct/node.env &&
        grep -Fxq "NODE_KEY=$NODE_KEY" /etc/zhct/node.env &&
        grep -Fxq "NODE_IP=$EXPECTED_IP" /etc/zhct/node.env
    then
        env_ready=1
    fi

    if [[ "$machine_ready$keys_ready$env_ready" == 111 ]]; then
        printf 'READY\n'
        return
    fi
    if [[ "$TASK" == q131 && "$machine_ready$keys_ready$env_ready" == 000 ]] &&
        [[ ! -e /etc/zhct/node.env ]]
    then
        printf 'SEALED\n'
        return
    fi
    printf 'PARTIAL\n'
}

assert_node_identity_ready() {
    local state
    state=$(node_identity_state)
    [[ "$state" == READY ]] || die "node identity state is $state, expected READY"
    grep -Fxq "NODE_NAME=$EXPECTED_HOST" /etc/zhct/node.env ||
        die "node.env NODE_NAME does not match $EXPECTED_HOST"
}

assert_q132_sequence_gate() {
    [[ "$TASK" != q132 ]] && return
    valid_sha256 "$Q131_PASS_SHA" ||
        die "Q132 requires --q131-pass-sha with the verified Q131 result/archive SHA256"
}

assert_secret_metadata() {
    local file mode
    local files=(
        /etc/zhct/secrets/redis5.password
        /etc/zhct/secrets/redis7.password
    )
    for file in "${files[@]}"; do
        [[ -s "$file" ]] || die "secret file is missing or empty: $file"
        [[ $(stat -c '%U:%G' "$file") == root:root ]] ||
            die "secret owner is not root:root: $file"
        mode=$(stat -c '%a' "$file")
        [[ "$mode" == 600 || "$mode" == 400 ]] ||
            die "secret mode is not 0600 or 0400: $file"
        stat -c 'PRESENT_NONEMPTY|%A|%U:%G|%s|%n' "$file"
    done
}

assert_config_identity() {
    grep -Fqx "bind $EXPECTED_IP" /etc/redis/redis5.conf
    grep -Fqx "bind $EXPECTED_IP" /etc/redis/redis7.conf
    grep -Fqx "bind $EXPECTED_IP" /var/lib/redis-sentinel5/sentinel.conf
    grep -Fqx "bind $EXPECTED_IP" /var/lib/redis-sentinel7/sentinel.conf

    if [[ "$TASK" == q131 ]]; then
        ! grep -Eq '^[[:space:]]*replicaof[[:space:]]' /etc/redis/redis5.conf
        ! grep -Eq '^[[:space:]]*replicaof[[:space:]]' /etc/redis/redis7.conf
        python3 - /etc/orchestrator.conf.json <<'PY'
import json
import pathlib
import sys

data = json.loads(pathlib.Path(sys.argv[1]).read_text(encoding="utf-8"))
assert data["ListenAddress"] == "10.2.203.112:3000"
assert data["RaftEnabled"] is True
assert data["RaftBind"] == "10.2.203.112"
assert data["DefaultRaftPort"] == 10008
PY
    else
        grep -Fqx 'replicaof 10.2.203.112 6379' /etc/redis/redis5.conf
        grep -Fqx 'replicaof 10.2.203.112 6387' /etc/redis/redis7.conf
        python3 - /etc/orchestrator.conf.json <<'PY'
import json
import pathlib
import sys

data = json.loads(pathlib.Path(sys.argv[1]).read_text(encoding="utf-8"))
assert data["ListenAddress"] == "127.0.0.1:3000"
assert data["RaftEnabled"] is False
PY
    fi
}

assert_binaries() {
    /opt/redis/5.0.14/bin/redis-server --version | grep -Fq 'v=5.0.14'
    /opt/redis/7.2.14/bin/redis-server --version | grep -Fq 'v=7.2.14'
    [[ -x /opt/redis/5.0.14/bin/redis-sentinel ]]
    [[ -x /opt/redis/7.2.14/bin/redis-sentinel ]]
    rpm -q jq oniguruma percona-orchestrator \
        percona-orchestrator-cli percona-orchestrator-client
}

unit_state_table() {
    local unit
    for unit in redis5 redis7 sentinel5 sentinel7 orchestrator; do
        printf '%s.service|' "$unit"
        systemctl show "$unit.service" \
            -p LoadState -p ActiveState -p UnitFileState -p Result \
            --value 2>/dev/null |
            paste -sd '|' - || true
    done
}

assert_units_initially_inactive() {
    local unit
    for unit in redis5 redis7 sentinel5 sentinel7 orchestrator; do
        if systemctl is-active --quiet "$unit.service"; then
            die "unit is already active before execute: $unit.service"
        fi
    done
}

assert_units_final_policy() {
    local unit
    for unit in "${EXPECTED_ACTIVE_UNITS[@]}"; do
        [[ $(systemctl is-enabled "$unit.service") == enabled ]] ||
            die "$unit.service is not enabled"
        [[ $(systemctl is-active "$unit.service") == active ]] ||
            die "$unit.service is not active"
    done
    for unit in "${EXPECTED_INACTIVE_UNITS[@]}"; do
        [[ $(systemctl is-enabled "$unit.service" 2>/dev/null || true) == disabled ]] ||
            die "$unit.service is not disabled"
        [[ $(systemctl is-active "$unit.service" 2>/dev/null || true) == inactive ]] ||
            die "$unit.service is not inactive"
    done
}

assert_expected_listeners() {
    local port
    for port in "${EXPECTED_PORTS[@]}"; do
        ss -lntH |
            awk -v endpoint="$EXPECTED_IP:$port" '$4 == endpoint {found=1} END {exit !found}' ||
            die "expected listener not found: $EXPECTED_IP:$port"
    done
    if ss -lntH | awk '$4 ~ /:(26379|26387|3000|10008)$/ {found=1} END {exit !found}'
    then
        die "$TASK has an unexpected Sentinel or Orchestrator listener"
    fi
}

prepare_policy_bundle() {
    local policy_root source_root tmp_base=/var/tmp
    source_root=$(cd "$BUNDLE_ROOT" && pwd)
    if [[ ! -d "$tmp_base" || ! -w "$tmp_base" ]]; then
        tmp_base=${TMPDIR:-/tmp}
    fi
    policy_root=$(mktemp -d "$tmp_base/zhct-redis-q131-132-policy.XXXXXX")
    install -d -m 0700 "$policy_root/roles/redis/manifests"
    ln -s "$source_root/roles/redis/role.env" "$policy_root/roles/redis/role.env"
    ln -s "$source_root/roles/redis/templates" "$policy_root/roles/redis/templates"
    ln -s "$source_root/roles/redis/manifests/nodes.tsv" \
        "$policy_root/roles/redis/manifests/nodes.tsv"
    cat >"$policy_root/roles/redis/manifests/systemd-units.tsv" <<'EOF'
unit	template_state	node_states	purpose
redis5.service	disabled,inactive	redis01=enabled,active;redis02=enabled,active	Redis5 front settlement
redis7.service	disabled,inactive	redis01=enabled,active;redis02=enabled,active	Redis7 inventory
sentinel5.service	disabled,inactive	redis01=disabled,inactive;redis02=disabled,inactive	Redis5 Sentinel remains off until Q-230
sentinel7.service	disabled,inactive	redis01=disabled,inactive;redis02=disabled,inactive	Redis7 Sentinel remains off until Q-231
orchestrator.service	disabled,inactive	redis01=disabled,inactive;redis02=disabled,inactive	Orchestrator remains off until Q-210 after Q-200
EOF
    cat >"$policy_root/roles/redis/hooks.sh" <<'EOF'
#!/usr/bin/env bash
role_activate() {
    if [[ "$DRY_RUN" == "1" ]]; then
        log 'DRY-RUN: would inject Redis server secrets and keep Sentinel/Orchestrator disabled/inactive'
        return
    fi
    materialize_secret /etc/redis/redis5.conf __REDIS5_PASSWORD__ /etc/zhct/secrets/redis5.password
    materialize_secret /etc/redis/redis7.conf __REDIS7_PASSWORD__ /etc/zhct/secrets/redis7.password
    for config in /etc/redis/redis5.conf /etc/redis/redis7.conf; do
        grep -Eq '__[A-Z0-9_]+__' "$config" && die "secret placeholder has not been injected: $config"
    done
}
EOF
    chmod 0700 "$policy_root/roles/redis/hooks.sh"
    printf '%s\n' "$policy_root"
}

activate_with_q131_q132_policy() {
    local activation_id=$1 policy_root activate_rc
    policy_root=$(prepare_policy_bundle)
    set +e
    RUN_ID="$activation_id" \
        "$BUNDLE_ROOT/common/scripts/activate-role.sh" \
        --role redis \
        --bundle-root "$policy_root"
    activate_rc=$?
    set -e
    rm -rf -- "$policy_root"
    return "$activate_rc"
}

redis_cli() {
    local version=$1 port=$2 secret_file=$3
    shift 3
    REDISCLI_AUTH=$(tr -d '\r\n' <"$secret_file") \
        "/opt/redis/$version/bin/redis-cli" \
        -h "$EXPECTED_IP" -p "$port" --no-auth-warning "$@"
}

redis_runtime_checks() {
    local replication5 replication7
    redis_cli 5.0.14 6379 /etc/zhct/secrets/redis5.password PING
    redis_cli 7.2.14 6387 /etc/zhct/secrets/redis7.password PING

    if [[ "$TASK" == q131 ]]; then
        redis_cli 5.0.14 6379 /etc/zhct/secrets/redis5.password ROLE |
            sed -n '1,3p'
        redis_cli 7.2.14 6387 /etc/zhct/secrets/redis7.password ROLE |
            sed -n '1,3p'
    else
        replication5=$(redis_cli 5.0.14 6379 \
            /etc/zhct/secrets/redis5.password INFO replication)
        replication7=$(redis_cli 7.2.14 6387 \
            /etc/zhct/secrets/redis7.password INFO replication)
        printf '%s\n' "$replication5" |
            grep -E '^(role|master_host|master_port|master_link_status|master_sync_in_progress):'
        printf '%s\n' "$replication7" |
            grep -E '^(role|master_host|master_port|master_link_status|master_sync_in_progress):'
        grep -Fqx 'role:slave' <<<"$replication5"
        grep -Fqx 'master_host:10.2.203.112' <<<"$replication5"
        grep -Fqx 'master_port:6379' <<<"$replication5"
        grep -Fqx 'role:slave' <<<"$replication7"
        grep -Fqx 'master_host:10.2.203.112' <<<"$replication7"
        grep -Fqx 'master_port:6387' <<<"$replication7"
    fi

}

safe_config_summary() {
    local file
    for file in /etc/redis/redis5.conf /etc/redis/redis7.conf; do
        printf 'CONFIG=%s\n' "$file"
        grep -E '^(bind|port|protected-mode|dir|appendonly|replicaof|replica-read-only|maxmemory-policy)[[:space:]]' "$file"
        sha256sum "$file"
    done
    for file in /var/lib/redis-sentinel5/sentinel.conf \
        /var/lib/redis-sentinel7/sentinel.conf
    do
        printf 'CONFIG=%s\n' "$file"
        grep -E '^(bind|port|protected-mode|dir|sentinel monitor|sentinel down-after-milliseconds|sentinel failover-timeout|sentinel parallel-syncs)[[:space:]]' "$file"
        sha256sum "$file"
    done
    python3 - /etc/orchestrator.conf.json <<'PY'
import json
import pathlib
import sys

path = pathlib.Path(sys.argv[1])
data = json.loads(path.read_text(encoding="utf-8"))
allowed = [
    "ListenAddress", "BackendDB", "SQLite3DataFile", "RaftEnabled",
    "RaftDataDir", "RaftBind", "DefaultRaftPort", "RaftNodes",
]
for key in allowed:
    if key in data:
        print(f"{key}={json.dumps(data[key], ensure_ascii=True, separators=(',', ':'))}")
PY
    sha256sum /etc/orchestrator.conf.json
}

snapshot_state() {
    printf '%s\n' '=== IDENTITY ==='
    date --iso-8601=seconds
    hostname -s
    ip -o -4 addr show scope global
    if [[ -r /etc/zhct/node.env ]]; then
        grep -E '^(ROLE|NODE_KEY|NODE_IP|NODE_NAME|APPLIED_AT)=' /etc/zhct/node.env
    else
        echo 'NODE_ENV=ABSENT'
    fi
    stat -c '%s|%A|%U:%G|%n' /etc/machine-id 2>&1 || true
    find /etc/ssh -maxdepth 1 -type f -name 'ssh_host_*_key' \
        -printf '%f|%s\n' | sort

    printf '%s\n' '=== ASSETS ==='
    verify_assets
    printf '%s\n' '=== BINARIES ==='
    assert_binaries
    printf '%s\n' '=== UNITS ==='
    unit_state_table
    printf '%s\n' '=== LISTENERS ==='
    ss -lntupH |
        grep -E ':(6379|6387|26379|26387|3000|10008)([[:space:]]|$)' ||
        echo NONE
    printf '%s\n' '=== FAILED UNITS ==='
    systemctl --failed --no-legend --plain || true
    printf '%s\n' '=== MEMORY PARAMETERS ==='
    sysctl vm.overcommit_memory
    cat /sys/kernel/mm/transparent_hugepage/enabled
    printf '%s\n' '=== CAPACITY ==='
    df -hT /
    free -h
    printf '%s\n' '=== DNF LATEST ==='
    dnf history list 2>/dev/null | sed -n '1,8p'
}

capture_logs() {
    local unit
    for unit in redis5 redis7 sentinel5 sentinel7 orchestrator; do
        printf '=== JOURNAL %s.service ===\n' "$unit"
        journalctl -u "$unit.service" --since '-15 min' --no-pager -n 160 |
            sanitize_stream
    done
}

write_result() {
    local state=$1 detail=$2 result_file=$3 command_rc=${4:-0}
    {
        printf 'task=%s\n' "$TASK"
        printf 'run_id=%s\n' "$RUN_ID"
        printf 'host=%s\n' "$EXPECTED_HOST"
        printf 'ip=%s\n' "$EXPECTED_IP"
        printf 'action=%s\n' "$ACTION"
        printf 'started_at=%s\n' "$STARTED_AT"
        printf 'finished_at=%s\n' "$(timestamp)"
        printf 'command_rc=%s\n' "$command_rc"
        printf 'state=%s\n' "$state"
        printf 'detail=%s\n' "$detail"
        printf 'scope=LOCAL_ROLE_ONLY\n'
        printf 'cross_host_ha=NOT_ACCEPTED_BEFORE_Q200_Q230_Q231\n'
        printf 'vip=NOT_ENABLED\n'
        printf 'sentinel5=DISABLED_INACTIVE_REQUIRED_UNTIL_Q230\n'
        printf 'sentinel7=DISABLED_INACTIVE_REQUIRED_UNTIL_Q231\n'
        if [[ "$TASK" == q132 ]]; then
            printf 'q131_pass_sha=%s\n' "$Q131_PASS_SHA"
            printf 'orchestrator=DISABLED_INACTIVE_REQUIRED\n'
        else
            printf 'orchestrator=DISABLED_INACTIVE_REQUIRED_UNTIL_Q210\n'
        fi
    } >"$result_file"
    chmod 0600 "$result_file"
}

archive_evidence() {
    local evidence_dir=$1 archive archive_sha receipt
    archive="/root/redis-${TASK}-evidence-$(basename "$evidence_dir").tar.gz"
    [[ ! -e "$archive" ]] || die "archive already exists: $archive"
    tar -C "$(dirname "$evidence_dir")" -czf "$archive" "$(basename "$evidence_dir")"
    chmod 0600 "$archive"
    archive_sha="${archive}.sha256"
    (
        cd "$(dirname "$archive")"
        sha256sum "$(basename "$archive")" >"$(basename "$archive_sha")"
    )
    chmod 0600 "$archive_sha"
    receipt="${archive}.receipt.env"
    {
        printf 'task=%s\n' "$TASK"
        printf 'source_run_id=%s\n' "$(basename "$evidence_dir")"
        printf 'target_hostname=%s\n' "$EXPECTED_HOST"
        printf 'target_ip=%s\n' "$EXPECTED_IP"
        printf 'archive_created_at=%s\n' "$(timestamp)"
        printf 'archive=%s\n' "$archive"
        printf 'archive_sha256=%s\n' "$(awk '{print $1}' "$archive_sha")"
        printf 'rc=0\n'
    } >"$receipt"
    chmod 0600 "$receipt"
    cat "$archive_sha"
    stat -c 'ARCHIVE|%A|%U:%G|%s|%n' "$archive"
    stat -c 'ARCHIVE_RECEIPT|%A|%U:%G|%s|%n' "$archive_sha" "$receipt"
}

stop_disable_role_units() {
    systemctl disable --now \
        redis5.service redis7.service sentinel5.service sentinel7.service \
        orchestrator.service
}

rollback_activation() {
    local activation_id=$1
    if [[ -d "/var/backups/zhct-deploy/$activation_id" ]]; then
        RUN_ID="${RUN_ID}-rollback" \
            "$BUNDLE_ROOT/common/scripts/rollback-role.sh" \
            --role redis \
            --bundle-root "$BUNDLE_ROOT" \
            --backup-id "$activation_id"
    else
        log "activation backup is absent; stopping and disabling role units"
        stop_disable_role_units
    fi
}

verify_remote_state() {
    assert_target_identity
    assert_node_identity_ready
    assert_config_identity
    assert_units_final_policy
    assert_expected_listeners
    redis_runtime_checks
    safe_config_summary
}

run_preflight() {
    assert_target_identity
    assert_q132_sequence_gate
    verify_assets
    assert_binaries

    local state
    state=$(node_identity_state)
    if [[ "$TASK" == q131 ]]; then
        [[ "$state" == SEALED || "$state" == READY ]] ||
            die "Redis01 identity is partial; refuse post-clone or activation"
    else
        [[ "$state" == READY ]] ||
            die "Redis02 post-clone identity is not READY"
        assert_config_identity
    fi
    assert_units_initially_inactive
    printf 'PREFLIGHT_NODE_IDENTITY=%s\n' "$state"
    printf '%s_PREFLIGHT=PASS\n' "${TASK^^}"
}

run_execute() {
    local evidence_dir=$1 result_file=$2 activation_id="${RUN_ID}-activate"
    assert_target_identity
    assert_q132_sequence_gate
    verify_assets
    assert_binaries
    assert_units_initially_inactive

    local state
    state=$(node_identity_state)
    if [[ "$TASK" == q131 && "$state" == SEALED ]]; then
        log "Redis01 sealed identity confirmed; executing required post-clone"
        set +e
        RUN_ID="${RUN_ID}-postclone" \
            "$BUNDLE_ROOT/common/scripts/post-clone.sh" \
            --role redis \
            --bundle-root "$BUNDLE_ROOT" \
            --node redis01 \
            --expected-ip 10.2.203.112
        local postclone_rc=$?
        set -e
        if [[ "$postclone_rc" -ne 0 ]]; then
            write_result FAIL "post_clone_rc_$postclone_rc" "$result_file" "$postclone_rc"
            capture_logs
            return "$postclone_rc"
        fi
    elif [[ "$state" != READY ]]; then
        write_result FAIL "identity_state_$state" "$result_file" 1
        return 1
    fi

    assert_node_identity_ready
    assert_config_identity
    assert_secret_metadata
    snapshot_state

    set +e
    activate_with_q131_q132_policy "$activation_id"
    local activate_rc=$?
    set -e
    if [[ "$activate_rc" -ne 0 ]]; then
        log "activation failed rc=$activate_rc; entering bounded activation rollback"
        rollback_activation "$activation_id" || true
        write_result FAIL "activate_rc_$activate_rc" "$result_file" "$activate_rc"
        capture_logs
        return "$activate_rc"
    fi

    set +e
    verify_remote_state
    local verify_rc=$?
    set -e
    if [[ "$verify_rc" -ne 0 ]]; then
        log "post-activation verification failed rc=$verify_rc; entering bounded activation rollback"
        rollback_activation "$activation_id" || true
        write_result FAIL "verify_rc_$verify_rc" "$result_file" "$verify_rc"
        capture_logs
        return "$verify_rc"
    fi

    capture_logs
    write_result PASS ROLE_ACTIVATED_LOCAL_ONLY "$result_file" 0
    printf '%s_ROLE_ACTIVATION=PASS_LOCAL_ONLY\n' "${TASK^^}"
}

run_verify() {
    local evidence_dir=$1 result_file=$2
    assert_q132_sequence_gate
    snapshot_state
    verify_remote_state
    capture_logs
    write_result PASS ROLE_VERIFIED_LOCAL_ONLY "$result_file" 0
    printf '%s_ROLE_VERIFY=PASS_LOCAL_ONLY\n' "${TASK^^}"
}

assert_rollback_state() {
    local unit
    for unit in redis5 redis7 sentinel5 sentinel7 orchestrator; do
        [[ $(systemctl is-enabled "$unit.service" 2>/dev/null || true) == disabled ]] ||
            die "rollback left $unit.service enabled"
        [[ $(systemctl is-active "$unit.service" 2>/dev/null || true) == inactive ]] ||
            die "rollback left $unit.service active"
    done
    if ss -lntH |
        awk '$4 ~ /:(6379|6387|26379|26387|3000|10008)$/ {found=1} END {exit !found}'
    then
        die "rollback left a target listener"
    fi
    assert_node_identity_ready
}

run_rollback() {
    local evidence_dir=$1 result_file=$2
    valid_id "$ACTIVATION_RUN_ID" || die "invalid or missing --activation-run-id"
    assert_target_identity
    rollback_activation "$ACTIVATION_RUN_ID"
    assert_rollback_state
    snapshot_state
    capture_logs
    write_result PASS ACTIVATION_ROLLED_BACK_IDENTITY_RETAINED "$result_file" 0
    printf '%s_ACTIVATION_ROLLBACK=PASS_IDENTITY_RETAINED\n' "${TASK^^}"
}

if [[ ${1:-} == --static-validate ]]; then
    shift
    while (($#)); do
        case "$1" in
            --bundle-root) BUNDLE_ROOT=${2:-}; shift 2 ;;
            -h|--help) usage; exit 0 ;;
            *) die "unknown static option: $1" ;;
        esac
    done
    static_validate
    exit 0
fi

TASK=${1:-}
ACTION=${2:-}
[[ -n "$TASK" && -n "$ACTION" ]] || {
    usage
    exit 2
}
shift 2

while (($#)); do
    case "$1" in
        --bundle-root) BUNDLE_ROOT=${2:-}; shift 2 ;;
        --evidence-base) EVIDENCE_BASE=${2:-}; shift 2 ;;
        --run-id) RUN_ID=${2:-}; shift 2 ;;
        --q131-pass-sha) Q131_PASS_SHA=${2:-}; shift 2 ;;
        --activation-run-id) ACTIVATION_RUN_ID=${2:-}; shift 2 ;;
        --source-run-id) SOURCE_RUN_ID=${2:-}; shift 2 ;;
        -h|--help) usage; exit 0 ;;
        *) die "unknown option: $1" ;;
    esac
done

case "$ACTION" in
    preflight|execute|verify|rollback|archive) ;;
    *) die "action must be preflight, execute, verify, rollback or archive" ;;
esac

set_task_policy
require_root
for command in awk bash cut date dnf find grep hostname ip journalctl \
    paste python3 rpm sed sha256sum ss stat systemctl tar
do
    require_command "$command"
done

if [[ "$ACTION" == archive ]]; then
    valid_id "$SOURCE_RUN_ID" || die "invalid or missing --source-run-id"
    assert_target_identity
    SOURCE_EVIDENCE_DIR="$EVIDENCE_BASE/$TASK/$SOURCE_RUN_ID"
    [[ -d "$SOURCE_EVIDENCE_DIR" ]] ||
        die "source evidence directory not found: $SOURCE_EVIDENCE_DIR"
    [[ -s "$SOURCE_EVIDENCE_DIR/stdout-stderr.log" ]] ||
        die "source stdout/stderr log is missing"
    [[ -s "$SOURCE_EVIDENCE_DIR/execution-receipt.env" ]] ||
        die "source execution receipt is missing"
    [[ -s "$SOURCE_EVIDENCE_DIR/result.env" ]] ||
        die "source result is missing"
    archive_evidence "$SOURCE_EVIDENCE_DIR"
    printf '%s_EVIDENCE_ARCHIVE=PASS\n' "${TASK^^}"
    exit 0
fi

if [[ -z "$RUN_ID" ]]; then
    RUN_ID="${TASK}-${ACTION}-$(date +%Y%m%dT%H%M%S%z)"
fi
valid_id "$RUN_ID" || die "invalid --run-id"

EVIDENCE_DIR="$EVIDENCE_BASE/$TASK/$RUN_ID"
[[ ! -e "$EVIDENCE_DIR" ]] || die "evidence directory already exists: $EVIDENCE_DIR"
install -d -m 0700 -o root -g root "$EVIDENCE_DIR"
COMMAND_LOG="$EVIDENCE_DIR/stdout-stderr.log"
RESULT_FILE="$EVIDENCE_DIR/result.env"
RECEIPT_FILE="$EVIDENCE_DIR/execution-receipt.env"
STARTED_AT=$(timestamp)
exec > >(sanitize_stream | tee -a "$COMMAND_LOG") 2>&1

finalize_receipt() {
    local rc=$1
    trap - EXIT
    if [[ ! -s "$RESULT_FILE" ]]; then
        write_result FAIL "unhandled_exit_rc_$rc" "$RESULT_FILE" "$rc"
    fi
    {
        printf 'task=%s\n' "$TASK"
        printf 'run_id=%s\n' "$RUN_ID"
        printf 'action=%s\n' "$ACTION"
        printf 'target_hostname=%s\n' "$EXPECTED_HOST"
        printf 'target_ip=%s\n' "$EXPECTED_IP"
        printf 'started_at=%s\n' "$STARTED_AT"
        printf 'finished_at=%s\n' "$(timestamp)"
        printf 'rc=%s\n' "$rc"
        printf 'stdout_stderr=%s\n' "$COMMAND_LOG"
    } >"$RECEIPT_FILE"
    chmod 0600 "$RECEIPT_FILE"
    printf 'FINAL_RC=%s\n' "$rc"
    exit "$rc"
}
trap 'finalize_receipt $?' EXIT

log "task=$TASK action=$ACTION run_id=$RUN_ID host=$EXPECTED_HOST ip=$EXPECTED_IP"
if [[ "$TASK" == q132 ]]; then
    log "q131_pass_sha=$Q131_PASS_SHA"
fi

case "$ACTION" in
    preflight)
        snapshot_state
        run_preflight
        write_result PASS PREFLIGHT_ONLY_NO_ACTIVATION "$RESULT_FILE" 0
        ;;
    execute)
        run_execute "$EVIDENCE_DIR" "$RESULT_FILE"
        ;;
    verify)
        run_verify "$EVIDENCE_DIR" "$RESULT_FILE"
        ;;
    rollback)
        run_rollback "$EVIDENCE_DIR" "$RESULT_FILE"
        ;;
esac