#!/usr/bin/env bash
set -Eeuo pipefail
set +x
umask 077

readonly ROOT="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/../.." && pwd -P)"
readonly EXPECTED_CONTEXT=default
readonly EXPECTED_API_SERVER=https://127.0.0.1:6443
readonly EXPECTED_NODE=donghyeon-system-product-name
readonly PRODUCTION_INITIAL_SHA=79688d017d38eec9a6f100f8d0f784a5474e79802046ef1c2c11b30d170b0b0c
readonly PRODUCTION_POST_SHA=b1c3049206a1a88165ee672ae9aceac7945673a3bb9c3cf3670b7f0d56c3f291
readonly PRODUCTION_INITIAL_TARGET_COUNT=21
readonly PRODUCTION_POST_TARGET_COUNT=30
readonly LEDGER_HEADER=$'phase\towner\tapiVersion\tkind\tnamespace\tname\texisted\trestore-mode\tpayload-file\tpayload-sha256'
readonly MUTATION_HEADER=$'ordinal\tapiVersion\tkind\tnamespace\tname\tuid'
readonly ACCEPTANCE_SCHEMA=platform-observability-rules-alerts-v2
readonly TEST_MODE="${PLATFORM_OBSERVABILITY_ACCESS_TEST_MODE:-0}"
if [[ "$TEST_MODE" == 1 ]]; then
  readonly KUBECTL_BIN="${PLATFORM_OBSERVABILITY_ACCESS_KUBECTL_BIN:-}"
  readonly SUDO_BIN="${PLATFORM_OBSERVABILITY_ACCESS_SUDO_BIN:-}"
  readonly ENCRYPTION_SCRIPT="${PLATFORM_OBSERVABILITY_ACCESS_ENCRYPTION_SCRIPT:-}"
  readonly RESTORE_SCRIPT="${PLATFORM_OBSERVABILITY_ACCESS_RESTORE_SCRIPT:-}"
  readonly RECOVERY_SCRIPT="${PLATFORM_OBSERVABILITY_ACCESS_RECOVERY_SCRIPT:-}"
  readonly ROLLBACK_BASE="${PLATFORM_OBSERVABILITY_ACCESS_ROLLBACK_BASE:-}"
else
  readonly KUBECTL_BIN=/usr/local/bin/kubectl
  readonly SUDO_BIN=/usr/bin/sudo
  readonly ENCRYPTION_SCRIPT="$ROOT/scripts/validate/k3s-secret-encryption.sh"
  readonly RESTORE_SCRIPT="$ROOT/scripts/validate/k3s-secret-encryption-restore-evidence.sh"
  readonly RECOVERY_SCRIPT="$ROOT/scripts/bootstrap/create-observability-secrets.sh"
  readonly ROLLBACK_BASE=/var/lib/hyeonworks/platform-rollbacks
fi

# Source-only assertions bind the handoff to the same exact resource and policy
# contract used by the renderer. The renderer's executable main is guarded.
# shellcheck source=../validate/render-observability-access.sh
source "$ROOT/scripts/validate/render-observability-access.sh"

readonly -a RESOURCE_ROWS=(
  'platform-rules|v1|ConfigMap|observability|grafana-dashboard-https-endpoints|dashboards.yaml'
  'platform-rules|v1|ConfigMap|observability|grafana-dashboard-kubernetes-node|dashboards.yaml'
  'platform-rules|v1|ConfigMap|observability|grafana-dashboard-observability-backends|dashboards.yaml'
  'platform-rules|v1|ConfigMap|observability|grafana-dashboard-platform-services|dashboards.yaml'
  'platform-rules|v1|ConfigMap|observability|grafana-dashboard-workload-health|dashboards.yaml'
  'platform-rules|monitoring.coreos.com/v1|PrometheusRule|observability|platform-aistor-storage-quota|rules.yaml'
  'platform-rules|monitoring.coreos.com/v1|PrometheusRule|observability|platform-certificate-probes|rules.yaml'
  'platform-rules|monitoring.coreos.com/v1|PrometheusRule|observability|platform-observability-core|rules.yaml'
  'platform-rules|monitoring.coreos.com/v1|PrometheusRule|observability|platform-verified-services|rules.yaml'
  'platform-rules|monitoring.coreos.com/v1alpha1|AlertmanagerConfig|observability|platform-alertmanager|alertmanager.yaml'
  'kube-prometheus-stack|monitoring.coreos.com/v1|Alertmanager|observability|observability-core-kube-pr-alertmanager|alertmanager.yaml'
  'kube-prometheus-stack|networking.k8s.io/v1|NetworkPolicy|observability|observability-allow-alertmanager-public-https|alertmanager.yaml'
)

execute=false
rules_alerts=false
substrate=false
mode_count=0
verified_output_dir=''
verified_output_count=0
rollback_root=''
ledger_root=''
work=''
snapshot=''
mutation_started=false
rollback_running=false
failure_stage=preflight
rollback_ambiguous=false
api_proxy_pid=''
api_proxy_socket=''
expected_initial_sha=$PRODUCTION_INITIAL_SHA
expected_post_sha=$PRODUCTION_POST_SHA
target_initial_sha=''
post_substrate_sha=''
node_uid=''
slack_deployment_gate=''
slack_deployment_gate_first=''
slack_deployment_gate_value=''
slack_gate_approval_ref=''
slack_gate_accepted_by_uid=''
slack_gate_temp_root=''

declare -A HANDOFF_SHA=()
declare -A PRIOR_EXISTED=()
declare -A PRIOR_UID=()
declare -A SPLIT_FILE=()

fail() {
  printf 'ERROR: %s\n' "$*" >&2
  printf 'OBSERVABILITY_ACCESS_RULES_ALERTS_STAGE=%s\n' "$failure_stage" >&2
  printf 'OBSERVABILITY_ACCESS_RULES_ALERTS_RC=1\n' >&2
  exit 1
}

soft_fail() {
  printf 'ERROR: %s\n' "$*" >&2
  return 1
}

usage() {
  cat <<'USAGE'
Usage:
  PLATFORM_HELM_BIN=/home/donghyeon/.local/bin/helm \
    bash scripts/bootstrap/apply-observability-access.sh
  PLATFORM_HELM_BIN=/home/donghyeon/.local/bin/helm \
    bash scripts/bootstrap/apply-observability-access.sh --rules-alerts
  PLATFORM_HELM_BIN=/home/donghyeon/.local/bin/helm \
    bash scripts/bootstrap/apply-observability-access.sh \
      --execute --rules-alerts \
      --verified-output-dir /tmp/platform-observability-metrics.XXXXXX

Dry-run is source-only and performs no sudo, Kubernetes, network, encryption,
or recovery-evidence command. Execute accepts exactly the rules-alerts mode and
an already verified renderer handoff containing only the two inventories plus
dashboards.yaml, rules.yaml, and alertmanager.yaml. Substrate execution remains
outside this source-only implementation.
USAGE
}

parse_args() {
  while (( $# > 0 )); do
    case "$1" in
      --execute)
        [[ "$execute" == false ]] || return 2
        execute=true
        shift
        ;;
      --rules-alerts)
        rules_alerts=true
        mode_count=$((mode_count + 1))
        shift
        ;;
      --substrate)
        substrate=true
        mode_count=$((mode_count + 1))
        shift
        ;;
      --verified-output-dir)
        (( $# >= 2 )) || return 2
        verified_output_count=$((verified_output_count + 1))
        verified_output_dir=$2
        shift 2
        ;;
      -h|--help)
        usage
        exit 0
        ;;
      *)
        return 2
        ;;
    esac
  done
  (( mode_count <= 1 && verified_output_count <= 1 )) || return 2
  if [[ "$execute" == true ]]; then
    [[ "$rules_alerts" == true && "$substrate" == false && $mode_count == 1 &&
      $verified_output_count == 1 && -n "$verified_output_dir" ]] || return 2
  else
    [[ $verified_output_count == 0 ]] || return 2
  fi
}

print_dry_run() {
  printf 'OBSERVABILITY_ACCESS_EXECUTE_MODE=RULES_ALERTS_ONLY\n'
  printf 'OBSERVABILITY_ACCESS_TARGET_INITIAL_SHA256=%s\n' "$PRODUCTION_INITIAL_SHA"
  printf 'OBSERVABILITY_ACCESS_POST_SUBSTRATE_SHA256=%s\n' "$PRODUCTION_POST_SHA"
  printf 'OBSERVABILITY_ACCESS_HANDOFF=target-initial,post-substrate,dashboards.yaml,rules.yaml,alertmanager.yaml\n'
  printf 'OBSERVABILITY_ACCESS_MUTATION_SCOPE=ConfigMap,PrometheusRule,AlertmanagerConfig,Alertmanager,NetworkPolicy\n'
  printf 'OBSERVABILITY_ACCESS_FORBIDDEN_SCOPE=Secret,PVC,Grafana,Blackbox,Probe,Ingress,Host-Nginx\n'
  printf 'OBSERVABILITY_ACCESS_RUNBOOK_URL=REQUIRED_HTTPS_INPUT\n'
  printf 'OBSERVABILITY_ACCESS_SLACK_SECRET=REQUIRED_EXACT_SCHEMA_ONLY\n'
  printf 'OBSERVABILITY_ACCESS_DRY_RUN=PASS\n'
  # Backward-compatible concise names used by the operator checkpoint.
  printf 'TARGET_INITIAL_SHA256=%s\n' "$PRODUCTION_INITIAL_SHA"
  printf 'POST_SUBSTRATE_SHA256=%s\n' "$PRODUCTION_POST_SHA"
}

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

validate_test_mode() {
  local variable fixture_root fixture_parent fixture_bin command_path label expected_path
  [[ "$TEST_MODE" == 0 || "$TEST_MODE" == 1 ]] || fail 'invalid test-mode flag'
  (( EUID != 0 )) || fail 'whole-script root execution is forbidden; use narrow sudo from the invoking user'
  if [[ "$TEST_MODE" == 1 ]]; then
    [[ "$SUDO_BIN" != /usr/bin/sudo ]] || fail 'test command boundary is unsafe: real /usr/bin/sudo is forbidden'
    fixture_root="$(dirname -- "$ROLLBACK_BASE")"
    fixture_parent="$(dirname -- "$fixture_root")"
    fixture_bin="$fixture_root/bin"
    [[ "$fixture_parent" == /tmp/platform-observability-access-test.?????? &&
      "$fixture_root" == "$fixture_parent"/* && "$fixture_root" != "$fixture_parent"/ &&
      "$ROLLBACK_BASE" == "$fixture_root/rollbacks" &&
      "$(readlink -f -- "$fixture_parent" 2>/dev/null)" == "$fixture_parent" &&
      "$(readlink -f -- "$fixture_root" 2>/dev/null)" == "$fixture_root" &&
      "$(readlink -f -- "$ROLLBACK_BASE" 2>/dev/null)" == "$ROLLBACK_BASE" ]] ||
      fail 'test command fixture root is not exact and canonical'
    local directory metadata mode_value
    for directory in "$fixture_parent" "$fixture_root" "$fixture_bin" "$ROLLBACK_BASE"; do
      [[ -d "$directory" && ! -L "$directory" ]] || fail 'test command fixture directory is unsafe'
      metadata="$(stat -c '%F|%u|%a' -- "$directory")" || fail 'cannot stat test command fixture directory'
      IFS='|' read -r type owner mode <<<"$metadata"
      mode_value=$((8#$mode))
      [[ "$type" == directory && "$owner" == "$(id -u)" &&
        $((mode_value & 0022)) == 0 && $((mode_value & 07000)) == 0 ]] ||
        fail 'test command fixture directory owner or mode is unsafe'
    done
    local -a boundary_names=(kubectl sudo encryption restore recovery)
    local -a boundary_paths=("$KUBECTL_BIN" "$SUDO_BIN" "$ENCRYPTION_SCRIPT" "$RESTORE_SCRIPT" \
      "$RECOVERY_SCRIPT")
    local index resolved type owner mode
    for index in "${!boundary_names[@]}"; do
      label=${boundary_names[$index]}
      command_path=${boundary_paths[$index]}
      expected_path="$fixture_bin/$label"
      resolved="$(readlink -f -- "$command_path" 2>/dev/null)" ||
        fail "test command boundary is unsafe: $label does not resolve"
      [[ "$command_path" == "$expected_path" && "$resolved" == "$expected_path" &&
        -f "$command_path" && ! -L "$command_path" && -x "$command_path" ]] ||
        fail "test command boundary is unsafe: $label escapes the exact fixture bin"
      metadata="$(stat -c '%F|%u|%a' -- "$command_path")" ||
        fail "test command boundary is unsafe: cannot stat $label"
      IFS='|' read -r type owner mode <<<"$metadata"
      mode_value=$((8#$mode))
      [[ "$type" == 'regular file' && "$owner" == "$(id -u)" &&
        $((mode_value & 0022)) == 0 && $((mode_value & 07000)) == 0 ]] ||
        fail "test command boundary is unsafe: $label owner or mode"
    done
    if [[ -n "$verified_output_dir" ]]; then
      [[ "$verified_output_dir" == "$fixture_root/metrics" &&
        "$(readlink -f -- "$verified_output_dir" 2>/dev/null)" == "$verified_output_dir" ]] ||
        fail 'test verified output escapes the exact fixture root'
      expected_initial_sha="${PLATFORM_TEST_EXPECTED_INITIAL_SHA:-}"
      expected_post_sha="${PLATFORM_TEST_EXPECTED_POST_SHA:-}"
      is_sha256 "$expected_initial_sha" && is_sha256 "$expected_post_sha" ||
        fail 'test inventory hashes are malformed'
    fi
  else
    for variable in ${!PLATFORM_OBSERVABILITY_ACCESS_@}; do
      fail "production rejects observability access override: $variable"
    done
    [[ "$KUBECTL_BIN" == /usr/local/bin/kubectl && "$SUDO_BIN" == /usr/bin/sudo &&
      "$ENCRYPTION_SCRIPT" == "$ROOT/scripts/validate/k3s-secret-encryption.sh" &&
      "$RESTORE_SCRIPT" == "$ROOT/scripts/validate/k3s-secret-encryption-restore-evidence.sh" &&
      "$RECOVERY_SCRIPT" == "$ROOT/scripts/bootstrap/create-observability-secrets.sh" &&
      "$ROLLBACK_BASE" == /var/lib/hyeonworks/platform-rollbacks ]] ||
      fail 'production observability access constants are not exact'
    [[ -f "$KUBECTL_BIN" && ! -L "$KUBECTL_BIN" && -x "$KUBECTL_BIN" &&
      -f "$SUDO_BIN" && ! -L "$SUDO_BIN" && -x "$SUDO_BIN" ]] ||
      fail 'production command boundary is unsafe'
    for command_path in "$ENCRYPTION_SCRIPT" "$RESTORE_SCRIPT" "$RECOVERY_SCRIPT"; do
      [[ -f "$command_path" && ! -L "$command_path" ]] ||
        fail "required production gate script is missing or unsafe: $command_path"
    done
    local required
    for required in base64 curl date find gzip jq mktemp python3 sha256sum stat tac; do
      command -v "$required" >/dev/null 2>&1 || fail "required command is unavailable: $required"
    done
  fi
}

kubectl_bounded() {
  "$KUBECTL_BIN" --request-timeout=10s "$@"
}

run_encryption_gates() {
  /usr/bin/bash "$ENCRYPTION_SCRIPT" --expect-reencrypted || fail 'Secret encryption evidence gate failed'
  /usr/bin/bash "$RESTORE_SCRIPT" --check || fail 'Secret restore evidence gate failed'
}

validate_context() {
  local context api node_json
  context="$(kubectl_bounded config current-context)" || return 1
  [[ "$context" == "$EXPECTED_CONTEXT" ]] || fail "Kubernetes context changed: $context"
  api="$(kubectl_bounded config view --minify --output=jsonpath='{.clusters[0].cluster.server}')" || return 1
  [[ "$api" == "$EXPECTED_API_SERVER" ]] || fail "Kubernetes API server changed: $api"
  node_json="$(kubectl_bounded get node "$EXPECTED_NODE" --output=json)" || return 1
  node_uid="$(python3 -c '
import json, sys
item=json.load(sys.stdin)
ready=any(x.get("type")=="Ready" and x.get("status")=="True" for x in item.get("status",{}).get("conditions",[]))
meta=item.get("metadata",{})
if meta.get("name") != sys.argv[1] or not meta.get("uid") or not ready:
    raise SystemExit(1)
print(meta["uid"])
' "$EXPECTED_NODE" <<<"$node_json")" || fail 'target node identity or Ready condition changed'
}

validate_inventory_phase() {
  local phase=$1 expected_sha=$2 expected_target_count=$3 directory checksum_line declared filename extra actual entries current_uid captured captured_epoch now_epoch age
  [[ "$expected_target_count" =~ ^[1-9][0-9]*$ ]] ||
    fail "inventory expected target count is invalid: $phase"
  current_uid="$(id -u)" || return 1
  directory="$verified_output_dir/$phase"
  [[ -d "$directory" && ! -L "$directory" && "$(readlink -f -- "$directory")" == "$directory" ]] ||
    fail "inventory directory is unsafe: $phase"
  [[ "$(stat -c '%u:%a:%h' -- "$directory")" == "$current_uid:700:2" ]] ||
    fail "inventory directory metadata changed: $phase"
  entries="$(find "$directory" -mindepth 1 -maxdepth 1 -printf '%f\n' | sort)" || return 1
  [[ "$entries" == $'inventory.json\ninventory.sha256' ]] || fail "inventory entry set changed: $phase"
  local file
  for file in inventory.json inventory.sha256; do
    [[ -f "$directory/$file" && ! -L "$directory/$file" &&
      "$(stat -c '%u:%a:%h' -- "$directory/$file")" == "$current_uid:600:1" ]] ||
      fail "inventory file metadata changed: $phase/$file"
  done
  IFS=' ' read -r declared filename extra <"$directory/inventory.sha256"
  [[ "$declared" =~ ^[0-9a-f]{64}$ && "$filename" == inventory.json && -z "${extra:-}" ]] ||
    fail "inventory checksum schema changed: $phase"
  actual="$(sha256sum -- "$directory/inventory.json" | awk '{print $1}')" || return 1
  [[ "$actual" == "$declared" && "$actual" == "$expected_sha" ]] ||
    fail "inventory hash changed: $phase"
  jq -e --arg phase "$phase" --argjson expected_target_count "$expected_target_count" '
    .schema == "platform-observability-metric-inventory/v1" and
    .phase == $phase and
    .context == "default" and .api_server == "https://127.0.0.1:6443" and
    (.captured_at_utc | type == "string") and
    (.targets | type == "array" and length > 0 and length == $expected_target_count) and
    all(.targets[];
      .health == "up" and .last_error == "" and
      (.metrics | type == "array" and length > 0) and
      all(.metrics[];
        (.name | type == "string" and length > 0) and
        (.label_names | type == "array") and
        all(.label_names[]; type == "string")))
  ' "$directory/inventory.json" >/dev/null || fail "inventory semantic contract changed: $phase"
  captured="$(jq -r '.captured_at_utc' "$directory/inventory.json")" || return 1
  [[ "$captured" =~ ^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$ ]] ||
    fail "inventory capture timestamp is malformed: $phase"
  captured_epoch="$(date -u -d "$captured" +%s 2>/dev/null)" ||
    fail "inventory capture timestamp is invalid: $phase"
  now_epoch="$(date -u +%s)" || return 1
  age=$((now_epoch - captured_epoch))
  (( age >= -300 )) || fail "inventory capture timestamp is too far in the future: $phase"
  printf '%s' "$actual"
}

validate_handoff_root() {
  local current_uid entries file validation_reason expected_test_root
  current_uid="$(id -u)" || return 1
  if [[ "$TEST_MODE" == 1 ]]; then
    expected_test_root="$(dirname -- "$ROLLBACK_BASE")/metrics"
    [[ "$verified_output_dir" == "$expected_test_root" ]] ||
      fail 'test verified output is not confined to the exact fixture root'
  else
    [[ "$verified_output_dir" =~ ^/tmp/platform-observability-metrics\.[A-Za-z0-9]{6}$ ]] ||
      fail 'verified output path is not an exact /tmp metric handoff'
    [[ -d /tmp && ! -L /tmp && "$(readlink -f -- /tmp)" == /tmp &&
      "$(stat -c '%u:%a' -- /tmp)" == 0:1777 ]] || fail '/tmp lineage is unsafe'
  fi
  [[ -d "$verified_output_dir" && ! -L "$verified_output_dir" &&
    "$(readlink -f -- "$verified_output_dir")" == "$verified_output_dir" &&
    "$(stat -c '%u:%a:%h' -- "$verified_output_dir")" == "$current_uid:700:4" ]] ||
    fail 'verified output root metadata changed'
  entries="$(find "$verified_output_dir" -mindepth 1 -maxdepth 1 -printf '%f\n' | sort)" || return 1
  [[ "$entries" == $'alertmanager.yaml\ndashboards.yaml\npost-substrate\nrules.yaml\ntarget-initial' ]] ||
    fail 'verified renderer handoff must contain only two inventories and three rules-alerts manifests'
  target_initial_sha="$(validate_inventory_phase target-initial "$expected_initial_sha" \
    "$PRODUCTION_INITIAL_TARGET_COUNT")" || return 1
  post_substrate_sha="$(validate_inventory_phase post-substrate "$expected_post_sha" \
    "$PRODUCTION_POST_TARGET_COUNT")" || return 1
  for file in dashboards.yaml rules.yaml alertmanager.yaml; do
    [[ -f "$verified_output_dir/$file" && ! -L "$verified_output_dir/$file" &&
      "$(stat -c '%u:%a:%h' -- "$verified_output_dir/$file")" == "$current_uid:600:1" ]] ||
      fail "renderer handoff file metadata changed: $file"
    HANDOFF_SHA[$file]="$(sha256sum -- "$verified_output_dir/$file" | awk '{print $1}')" || return 1
  done
  if ! validation_reason="$(python3 - "$verified_output_dir" <<'PY'
import pathlib
import re
import sys
import yaml

root=pathlib.Path(sys.argv[1])
def load(name):
    with (root/name).open(encoding="utf-8") as stream:
        return [item for item in yaml.safe_load_all(stream) if item is not None]
def identity(item):
    meta=item.get("metadata") or {}
    return item.get("apiVersion"), item.get("kind"), meta.get("namespace"), meta.get("name")
expected_dashboards={
    ("v1","ConfigMap","observability",name) for name in (
        "grafana-dashboard-https-endpoints", "grafana-dashboard-kubernetes-node",
        "grafana-dashboard-observability-backends", "grafana-dashboard-platform-services",
        "grafana-dashboard-workload-health",
    )
}
expected_rules={
    ("monitoring.coreos.com/v1","PrometheusRule","observability",name) for name in (
        "platform-aistor-storage-quota", "platform-certificate-probes",
        "platform-observability-core", "platform-verified-services",
    )
}
expected_alertmanager={
    ("monitoring.coreos.com/v1","Alertmanager","observability","observability-core-kube-pr-alertmanager"),
    ("monitoring.coreos.com/v1alpha1","AlertmanagerConfig","observability","platform-alertmanager"),
    ("networking.k8s.io/v1","NetworkPolicy","observability","observability-allow-alertmanager-public-https"),
}
dashboards, rules, alertmanager = load("dashboards.yaml"), load("rules.yaml"), load("alertmanager.yaml")
groups=((dashboards, expected_dashboards), (rules, expected_rules), (alertmanager, expected_alertmanager))
for items, expected in groups:
    actual=[identity(item) for item in items]
    if len(actual) != len(set(actual)) or set(actual) != expected:
        print("INVALID_RESOURCE_SET")
        raise SystemExit(1)
for item in dashboards:
    if ((item.get("metadata") or {}).get("labels") or {}).get("grafana_dashboard") != "1":
        print("INVALID_DASHBOARD_CONTRACT")
        raise SystemExit(1)
runbook=re.compile(r"^https://[^/\s]+/.+", re.ASCII)
for item in rules:
    for group in (item.get("spec") or {}).get("groups") or []:
        for rule in group.get("rules") or []:
            if "alert" in rule:
                value=(rule.get("annotations") or {}).get("runbook_url")
                if not isinstance(value,str) or not runbook.fullmatch(value) or re.search(r"placeholder|example|todo|tbd",value,re.I):
                    print("BLOCKED_RUNBOOK_URL")
                    raise SystemExit(1)
am=next(item for item in alertmanager if item.get("kind")=="Alertmanager")
if (((am.get("spec") or {}).get("alertmanagerConfiguration") or {}).get("name")) != "platform-alertmanager":
    print("INVALID_ALERTMANAGER_REFERENCE")
    raise SystemExit(1)
config=next(item for item in alertmanager if item.get("kind")=="AlertmanagerConfig")
receivers=(config.get("spec") or {}).get("receivers") or []
slack_receiver=next((receiver for receiver in receivers if receiver.get("name")=="platform-slack"), {})
slacks=(slack_receiver.get("slackConfigs") or [])
if len(slacks)!=1 or slacks[0].get("apiURL") != {"name":"alertmanager-slack-webhook","key":"url"} or slacks[0].get("sendResolved") is not True:
    print("INVALID_SLACK_REFERENCE")
    raise SystemExit(1)
print("PASS")
PY
)"; then
    if [[ "$validation_reason" == BLOCKED_RUNBOOK_URL ]]; then
      printf 'OBSERVABILITY_ACCESS_RULES_ALERTS_READINESS=BLOCKED_RUNBOOK_URL\n' >&2
      fail 'every alert requires the operator-supplied HTTPS runbook URL; guessed URLs are forbidden'
    else
      fail "renderer handoff contract failed: ${validation_reason:-unknown}"
    fi
  fi
  [[ "$validation_reason" == PASS ]] || fail 'renderer handoff validation did not complete'
  assert_access_rules_alerts_contract \
    "$verified_output_dir/dashboards.yaml" \
    "$verified_output_dir/rules.yaml" \
    "$verified_output_dir/alertmanager.yaml" ||
    fail 'renderer rules-alerts assertion library rejected the handoff'
  assert_access_scope_contract "$verified_output_dir/rules.yaml" ||
    fail 'renderer scope assertion library rejected the handoff'
  assert_no_credentials "$verified_output_dir/dashboards.yaml" \
    "$verified_output_dir/rules.yaml" "$verified_output_dir/alertmanager.yaml" ||
    fail 'renderer handoff contains a credential payload'
}

verify_handoff_unchanged() {
  local file actual
  validate_inventory_phase target-initial "$target_initial_sha" \
    "$PRODUCTION_INITIAL_TARGET_COUNT" >/dev/null || return 1
  validate_inventory_phase post-substrate "$post_substrate_sha" \
    "$PRODUCTION_POST_TARGET_COUNT" >/dev/null || return 1
  for file in dashboards.yaml rules.yaml alertmanager.yaml; do
    [[ -f "$verified_output_dir/$file" && ! -L "$verified_output_dir/$file" ]] || return 1
    actual="$(sha256sum -- "$verified_output_dir/$file" | awk '{print $1}')" || return 1
    [[ "$actual" == "${HANDOFF_SHA[$file]}" ]] || fail "verified renderer handoff changed: $file"
  done
}

validate_slack_secret_schema() {
  local secret_json
  secret_json="$(kubectl_bounded get secret alertmanager-slack-webhook \
    --namespace observability --output=json)" || fail 'Slack Secret is absent or unreadable'
  python3 -c '
import json, sys
item=json.load(sys.stdin)
meta=item.get("metadata") or {}
if item.get("apiVersion")!="v1" or item.get("kind")!="Secret" or item.get("type")!="Opaque":
    raise SystemExit(1)
if meta.get("namespace")!="observability" or meta.get("name")!="alertmanager-slack-webhook":
    raise SystemExit(1)
if sorted((item.get("data") or {}).keys()) != ["url"]:
    raise SystemExit(1)
' <<<"$secret_json" || fail 'Slack Secret schema or key set is not exact'
}

capture_slack_deployment_gate() {
  local out_var=$1 label=$2 output producer_rc verdict
  [[ "$out_var" =~ ^[a-zA-Z_][a-zA-Z0-9_]*$ && "$label" =~ ^[a-z0-9-]+$ ]] ||
    fail 'Slack deployment gate capture arguments are invalid'
  if [[ -z "$slack_gate_temp_root" ]]; then
    slack_gate_temp_root="$(mktemp -d /tmp/platform-observability-slack-gate.XXXXXX)" ||
      fail 'cannot create Slack deployment gate temporary root'
    chmod 0700 "$slack_gate_temp_root" || fail 'cannot secure Slack deployment gate temporary root'
  fi
  [[ "$slack_gate_temp_root" =~ ^/tmp/platform-observability-slack-gate\.[A-Za-z0-9]{6}$ &&
    -d "$slack_gate_temp_root" && ! -L "$slack_gate_temp_root" &&
    "$(readlink -f -- "$slack_gate_temp_root")" == "$slack_gate_temp_root" &&
    "$(stat -c '%u:%a' -- "$slack_gate_temp_root")" == "$(id -u):700" ]] ||
    fail 'Slack deployment gate temporary root is unsafe'
  output="$slack_gate_temp_root/$label.stdout"
  [[ ! -e "$output" && ! -L "$output" ]] || fail 'Slack deployment gate output already exists'
  umask 077
  if /usr/bin/bash "$RECOVERY_SCRIPT" --check-slack-deployment-evidence >"$output"; then
    producer_rc=0
  else
    producer_rc=$?
  fi
  chmod 0600 "$output" || fail 'cannot secure Slack deployment gate output'
  (( producer_rc == 0 )) || fail 'Slack deployment gate is unavailable'
  verdict="$(python3 - "$output" <<'PY'
import pathlib, sys
raw=pathlib.Path(sys.argv[1]).read_bytes()
accepted={
    b"SLACK_DEPLOYMENT_GATE=RECOVERY\n": "SLACK_DEPLOYMENT_GATE=RECOVERY",
    b"SLACK_DEPLOYMENT_GATE=RISK_ACCEPTED\n": "SLACK_DEPLOYMENT_GATE=RISK_ACCEPTED",
}
value=accepted.get(raw)
if value is None:
    raise SystemExit(1)
print(value)
PY
)" || fail 'Slack deployment gate verdict is malformed'
  printf -v "$out_var" '%s' "$verdict"
}

map_slack_deployment_gate_acceptance() {
  case "$slack_deployment_gate" in
    SLACK_DEPLOYMENT_GATE=RECOVERY)
      slack_deployment_gate_value=RECOVERY
      slack_gate_approval_ref=strict-recovery-evidence-v1
      slack_gate_accepted_by_uid=not-applicable
      ;;
    SLACK_DEPLOYMENT_GATE=RISK_ACCEPTED)
      slack_deployment_gate_value=RISK_ACCEPTED
      slack_gate_approval_ref=2026-08-14-observability-slack-recovery-risk-acceptance-design
      slack_gate_accepted_by_uid=1000
      ;;
    *)
      return 1
      ;;
  esac
}

require_matching_slack_deployment_gates() {
  [[ "$slack_deployment_gate" == "$slack_deployment_gate_first" ]] ||
    fail 'Slack deployment gate verdict changed before first apply'
}

run_production_preflight() {
  kubectl_bounded wait --namespace observability --for=condition=Available \
    deployment/grafana deployment/blackbox-exporter --timeout=60s >/dev/null || return 1
  kubectl_bounded wait --namespace observability --for=condition=Available \
    prometheus/observability-core-kube-pr-prometheus \
    alertmanager/observability-core-kube-pr-alertmanager --timeout=60s >/dev/null || return 1
  kubectl_bounded rollout status --namespace observability \
    statefulset/prometheus-observability-core-kube-pr-prometheus --timeout=60s >/dev/null || return 1
  local probes
  probes="$(kubectl_bounded get probe --namespace observability --output=json)" || return 1
  jq -e '
    [.items[].metadata.name] | sort ==
    ["platform-private-edge","platform-private-internal","platform-public-edge"]
  ' <<<"$probes" >/dev/null || return 1
  kubectl_bounded get servicemonitor aistor-bucket-usage --namespace object-storage --output=name >/dev/null || return 1
  kubectl_bounded get podmonitor platform-postgres --namespace platform-data --output=name >/dev/null || return 1
}

run_preflight() {
  run_production_preflight
}

server_dry_run_handoff() {
  local file
  for file in dashboards.yaml rules.yaml alertmanager.yaml; do
    kubectl_bounded apply --dry-run=server -f "$verified_output_dir/$file" >/dev/null ||
      fail "server dry-run rejected renderer handoff: $file"
  done
}

validate_root_metadata() {
  local path=$1 expected_mode=$2 expected_type=$3 metadata type owner mode
  "$SUDO_BIN" -n test ! -L "$path" || return 1
  metadata="$($SUDO_BIN -n stat -c '%F|%U:%G|%a' -- "$path")" || return 1
  IFS='|' read -r type owner mode <<<"$metadata"
  [[ "$type" == "$expected_type" && "$mode" == "$expected_mode" ]] || return 1
  [[ "$TEST_MODE" == 1 || "$owner" == root:root ]] || return 1
}

validate_rollback_root() {
  local id=${PLATFORM_OBSERVABILITY_ROLLBACK_ID:-}
  [[ "$id" =~ ^[0-9]{8}T[0-9]{6}Z$ ]] || fail 'PLATFORM_OBSERVABILITY_ROLLBACK_ID format is invalid'
  rollback_root="$ROLLBACK_BASE/observability-$id"
  "$SUDO_BIN" -n test -d "$rollback_root" || fail 'rollback root is absent'
  [[ "$($SUDO_BIN -n readlink -f -- "$rollback_root")" == "$rollback_root" ]] ||
    fail 'rollback root lineage contains a symlink'
  validate_root_metadata "$rollback_root" 700 directory || fail 'rollback root metadata is unsafe'
  ledger_root="$rollback_root/access-rules-alerts"
}

confirm_apply() {
  local answer=''
  printf 'Type APPLY default to apply only observability rules-alerts: '
  if [[ "$TEST_MODE" == 1 ]]; then
    answer=${PLATFORM_OBSERVABILITY_ACCESS_CONFIRMATION:-}
    printf '%s\n' "$answer"
  else
    [[ -t 0 ]] || fail '--execute requires an interactive terminal'
    read -r answer
  fi
  [[ "$answer" == APPLY ]] || fail 'cancelled'
}

resource_for_kind() {
  case "$1" in
    ConfigMap) printf 'configmap' ;;
    PrometheusRule) printf 'prometheusrule' ;;
    AlertmanagerConfig) printf 'alertmanagerconfig' ;;
    Alertmanager) printf 'alertmanager' ;;
    NetworkPolicy) printf 'networkpolicy' ;;
    *) return 1 ;;
  esac
}

object_key() {
  printf '%s|%s|%s' "$1" "$2" "$3"
}

get_object_json() {
  local kind=$1 namespace=$2 name=$3 resource output rc
  resource="$(resource_for_kind "$kind")" || return 2
  if output="$(kubectl_bounded get "$resource" "$name" --namespace "$namespace" \
    --ignore-not-found --output=json)"; then
    rc=0
  else
    rc=$?
  fi
  (( rc == 0 )) || return 2
  printf '%s' "$output"
}

object_uid_from_json() {
  python3 -c '
import json, sys
item=json.load(sys.stdin)
value=(item.get("metadata") or {}).get("uid")
if not isinstance(value,str) or not value:
    raise SystemExit(1)
print(value)
'
}

prepare_split_files() {
  local ordinal=0 row owner api kind namespace name source key output
  mkdir -m 0700 "$work/apply"
  for row in "${RESOURCE_ROWS[@]}"; do
    ordinal=$((ordinal + 1))
    IFS='|' read -r owner api kind namespace name source <<<"$row"
    output="$work/apply/$(printf '%02d' "$ordinal")-$kind-$name.yaml"
    python3 - "$verified_output_dir/$source" "$api" "$kind" "$namespace" "$name" "$output" <<'PY'
import pathlib, sys, yaml
source, api, kind, namespace, name, output=sys.argv[1:]
with open(source,encoding="utf-8") as stream:
    matches=[item for item in yaml.safe_load_all(stream) if item and item.get("apiVersion")==api and item.get("kind")==kind and (item.get("metadata") or {}).get("namespace")==namespace and (item.get("metadata") or {}).get("name")==name]
if len(matches)!=1:
    raise SystemExit(1)
with open(output,"x",encoding="utf-8") as stream:
    yaml.safe_dump(matches[0],stream,explicit_start=True,sort_keys=False)
PY
    chmod 0600 "$output"
    key="$(object_key "$kind" "$namespace" "$name")"
    SPLIT_FILE[$key]=$output
  done
}

sanitize_prior_object() {
  local input=$1 output=$2
  python3 - "$input" "$output" <<'PY'
import json, sys
item=json.load(open(sys.argv[1],encoding="utf-8"))
item.pop("status",None)
meta=item.setdefault("metadata",{})
for key in ("managedFields","resourceVersion","uid","generation","creationTimestamp","deletionTimestamp","deletionGracePeriodSeconds","selfLink"):
    meta.pop(key,None)
annotations=meta.get("annotations") or {}
annotations.pop("kubectl.kubernetes.io/last-applied-configuration",None)
if annotations:
    meta["annotations"]=annotations
else:
    meta.pop("annotations",None)
with open(sys.argv[2],"x",encoding="utf-8") as stream:
    json.dump(item,stream,separators=(",",":"),sort_keys=True)
    stream.write("\n")
PY
  chmod 0600 "$output"
}

validate_local_ledger() {
  local ledger=$1 payload_root=$2 line_count expected_count ordinal=0 row
  [[ -f "$ledger" && ! -L "$ledger" && "$(stat -c %a "$ledger")" == 600 ]] || return 1
  [[ "$(head -n1 "$ledger")" == "$LEDGER_HEADER" ]] || return 1
  line_count="$(wc -l <"$ledger" | tr -d ' ')"
  expected_count=$((${#RESOURCE_ROWS[@]} + 1))
  [[ "$line_count" == "$expected_count" ]] || return 1
  while IFS=$'\t' read -r phase owner api kind namespace name existed restore payload_file payload_sha extra; do
    (( ordinal += 1 ))
    row=${RESOURCE_ROWS[$((ordinal - 1))]}
    local expected_owner expected_api expected_kind expected_namespace expected_name source
    IFS='|' read -r expected_owner expected_api expected_kind expected_namespace expected_name source <<<"$row"
    [[ "$phase" == access-rules-alerts && "$owner" == "$expected_owner" && "$api" == "$expected_api" &&
      "$kind" == "$expected_kind" && "$namespace" == "$expected_namespace" && "$name" == "$expected_name" &&
      "$restore" == apply && -z "${extra:-}" ]] || return 1
    [[ "$existed" == true || "$existed" == false ]] || return 1
    if [[ "$existed" == true ]]; then
      [[ "$payload_file" =~ ^payloads/prior-[0-9]{2}\.json$ && "$payload_sha" =~ ^[0-9a-f]{64}$ ]] || return 1
      [[ -f "$payload_root/${payload_file#payloads/}" && ! -L "$payload_root/${payload_file#payloads/}" ]] || return 1
      [[ "$(sha256sum "$payload_root/${payload_file#payloads/}" | awk '{print $1}')" == "$payload_sha" ]] || return 1
    else
      [[ "$payload_file" == - && "$payload_sha" == - ]] || return 1
    fi
  done < <(tail -n +2 "$ledger")
  (( ordinal == ${#RESOURCE_ROWS[@]} ))
}

capture_snapshot() {
  local ledger="$snapshot/objects.tsv" payload_root="$snapshot/payloads"
  local ordinal=0 row owner api kind namespace name source key live live_file uid payload_rel payload sha
  mkdir -m 0700 "$payload_root"
  printf '%s\n' "$LEDGER_HEADER" >"$ledger"
  chmod 0600 "$ledger"
  for row in "${RESOURCE_ROWS[@]}"; do
    ordinal=$((ordinal + 1))
    IFS='|' read -r owner api kind namespace name source <<<"$row"
    key="$(object_key "$kind" "$namespace" "$name")"
    live="$(get_object_json "$kind" "$namespace" "$name")" || fail "cannot capture current object: $kind/$name"
    if [[ -n "$live" ]]; then
      live_file="$snapshot/live-$(printf '%02d' "$ordinal").json"
      printf '%s\n' "$live" >"$live_file"
      chmod 0600 "$live_file"
      python3 - "$live_file" "$api" "$kind" "$namespace" "$name" <<'PY'
import json,sys
item=json.load(open(sys.argv[1]))
meta=item.get("metadata") or {}
if (item.get("apiVersion"),item.get("kind"),meta.get("namespace"),meta.get("name")) != tuple(sys.argv[2:]):
    raise SystemExit(1)
if not meta.get("uid"):
    raise SystemExit(1)
PY
      uid="$(object_uid_from_json <"$live_file")" || return 1
      PRIOR_EXISTED[$key]=true
      PRIOR_UID[$key]=$uid
      payload_rel="payloads/prior-$(printf '%02d' "$ordinal").json"
      payload="$snapshot/$payload_rel"
      sanitize_prior_object "$live_file" "$payload"
      sha="$(sha256sum -- "$payload" | awk '{print $1}')"
      printf 'access-rules-alerts\t%s\t%s\t%s\t%s\t%s\ttrue\tapply\t%s\t%s\n' \
        "$owner" "$api" "$kind" "$namespace" "$name" "$payload_rel" "$sha" >>"$ledger"
    else
      PRIOR_EXISTED[$key]=false
      PRIOR_UID[$key]='-'
      printf 'access-rules-alerts\t%s\t%s\t%s\t%s\t%s\tfalse\tapply\t-\t-\n' \
        "$owner" "$api" "$kind" "$namespace" "$name" >>"$ledger"
    fi
  done
  validate_local_ledger "$ledger" "$payload_root" || fail 'captured rollback ledger is invalid'
}

ensure_root_dir() {
  local path=$1
  if "$SUDO_BIN" -n test -e "$path"; then
    validate_root_metadata "$path" 700 directory || return 1
  else
    "$SUDO_BIN" -n install -d -o root -g root -m 0700 "$path" || return 1
    validate_root_metadata "$path" 700 directory || return 1
  fi
}

install_snapshot() {
  local root_payload="$ledger_root/payloads" payload
  ensure_root_dir "$ledger_root" || fail 'rules-alerts ledger directory is unsafe'
  ensure_root_dir "$root_payload" || fail 'rules-alerts payload directory is unsafe'
  for payload in objects.tsv mutations.tsv acceptance.env; do
    ! "$SUDO_BIN" -n test -e "$ledger_root/$payload" || fail "rules-alerts ledger already exists: $payload"
  done
  [[ -z "$($SUDO_BIN -n find "$root_payload" -mindepth 1 -maxdepth 1 -print -quit)" ]] ||
    fail 'rules-alerts payload directory is not empty'
  for payload in "$snapshot"/payloads/*.json; do
    [[ -e "$payload" ]] || continue
    root_atomic_install "$payload" "$root_payload/${payload##*/}" create || return 1
  done
  root_atomic_install "$snapshot/objects.tsv" "$ledger_root/objects.tsv" create || return 1
  printf '%s\n' "$MUTATION_HEADER" >"$snapshot/mutations.tsv"
  chmod 0600 "$snapshot/mutations.tsv"
  root_atomic_install "$snapshot/mutations.tsv" "$ledger_root/mutations.tsv" create || return 1
  validate_root_metadata "$ledger_root/objects.tsv" 600 'regular file' || return 1
  validate_root_metadata "$ledger_root/mutations.tsv" 600 'regular file' || return 1
}

root_atomic_install() {
  local source=$1 destination=$2 mode=$3 temporary
  temporary="${destination}.next-$$"
  [[ "$mode" == create || "$mode" == replace ]] || return 2
  ! "$SUDO_BIN" -n test -e "$temporary" || return 1
  if [[ "$mode" == create ]]; then
    ! "$SUDO_BIN" -n test -e "$destination" || return 1
  else
    "$SUDO_BIN" -n test -f "$destination" || return 1
    "$SUDO_BIN" -n test ! -L "$destination" || return 1
  fi
  "$SUDO_BIN" -n install -o root -g root -m 0600 "$source" "$temporary" || return 1
  validate_root_metadata "$temporary" 600 'regular file' || return 1
  if [[ "$mode" == create ]]; then
    ! "$SUDO_BIN" -n test -e "$destination" || return 1
  fi
  "$SUDO_BIN" -n mv -T -- "$temporary" "$destination" || return 1
  validate_root_metadata "$destination" 600 'regular file'
}

verify_live_preconditions() {
  local row owner api kind namespace name source key live uid
  for row in "${RESOURCE_ROWS[@]}"; do
    IFS='|' read -r owner api kind namespace name source <<<"$row"
    key="$(object_key "$kind" "$namespace" "$name")"
    live="$(get_object_json "$kind" "$namespace" "$name")" || return 1
    if [[ "${PRIOR_EXISTED[$key]}" == true ]]; then
      [[ -n "$live" ]] || fail "existing object disappeared after snapshot: $kind/$name"
      uid="$(object_uid_from_json <<<"$live")" || return 1
      [[ "$uid" == "${PRIOR_UID[$key]}" ]] || fail "existing object UID changed after snapshot: $kind/$name"
    else
      [[ -z "$live" ]] || fail "new object appeared after snapshot: $kind/$name"
    fi
  done
}

record_mutation() {
  local ordinal=$1 api=$2 kind=$3 namespace=$4 name=$5 uid=$6
  printf '%s\t%s\t%s\t%s\t%s\t%s\n' "$ordinal" "$api" "$kind" "$namespace" "$name" "$uid" \
    >>"$snapshot/mutations.tsv"
  root_atomic_install "$snapshot/mutations.tsv" "$ledger_root/mutations.tsv" replace
}

apply_resources() {
  local ordinal=0 row owner api kind namespace name source key live uid
  for row in "${RESOURCE_ROWS[@]}"; do
    ordinal=$((ordinal + 1))
    IFS='|' read -r owner api kind namespace name source <<<"$row"
    key="$(object_key "$kind" "$namespace" "$name")"
    live="$(get_object_json "$kind" "$namespace" "$name")" || return 1
    if [[ "${PRIOR_EXISTED[$key]}" == true ]]; then
      [[ -n "$live" && "$(object_uid_from_json <<<"$live")" == "${PRIOR_UID[$key]}" ]] || {
        soft_fail "forward ownership precondition changed: $kind/$name"
        return 1
      }
    else
      [[ -z "$live" ]] || {
        soft_fail "forward absence precondition changed: $kind/$name"
        return 1
      }
    fi
    failure_stage="apply-$kind-$name"
    # A non-zero client result cannot prove that the API server made no change.
    # Arm rollback before the request and retain ambiguity unless the resulting
    # UID is observed and durably recorded below.
    mutation_started=true
    rollback_ambiguous=true
    if ! kubectl_bounded apply -f "${SPLIT_FILE[$key]}" >/dev/null; then
      return 1
    fi
    live="$(get_object_json "$kind" "$namespace" "$name")" || {
      rollback_ambiguous=true
      return 1
    }
    [[ -n "$live" ]] || {
      rollback_ambiguous=true
      return 1
    }
    uid="$(object_uid_from_json <<<"$live")" || {
      rollback_ambiguous=true
      return 1
    }
    if [[ "${PRIOR_EXISTED[$key]}" == true && "$uid" != "${PRIOR_UID[$key]}" ]]; then
      rollback_ambiguous=true
      return 1
    fi
    record_mutation "$ordinal" "$api" "$kind" "$namespace" "$name" "$uid" || {
      rollback_ambiguous=true
      return 1
    }
    rollback_ambiguous=false
  done
}

validate_authoritative_ledger_for_rollback() {
  local copy="$work/root-ledger.tsv" payload_copy_root="$work/root-payloads" line payload_rel payload_sha
  mkdir -m 0700 "$payload_copy_root" || return 1
  validate_root_metadata "$ledger_root" 700 directory || return 1
  validate_root_metadata "$ledger_root/payloads" 700 directory || return 1
  validate_root_metadata "$ledger_root/objects.tsv" 600 'regular file' || return 1
  validate_root_metadata "$ledger_root/mutations.tsv" 600 'regular file' || return 1
  "$SUDO_BIN" -n cat -- "$ledger_root/objects.tsv" >"$copy" || return 1
  chmod 0600 "$copy" || return 1
  while IFS=$'\t' read -r phase owner api kind namespace name existed restore payload_rel payload_sha; do
    [[ "$existed" == true ]] || continue
    validate_root_metadata "$ledger_root/$payload_rel" 600 'regular file' || return 1
    "$SUDO_BIN" -n cat -- "$ledger_root/$payload_rel" >"$payload_copy_root/${payload_rel##*/}" || return 1
    chmod 0600 "$payload_copy_root/${payload_rel##*/}" || return 1
  done < <(tail -n +2 "$copy")
  validate_local_ledger "$copy" "$payload_copy_root" || return 1
  printf '%s' "$copy"
}

load_mutation_uids() {
  local copy=$1
  declare -gA MUTATED_UID=()
  local ordinal api kind namespace name uid extra key
  [[ "$(head -n1 "$copy")" == "$MUTATION_HEADER" ]] || return 1
  while IFS=$'\t' read -r ordinal api kind namespace name uid extra; do
    [[ "$ordinal" =~ ^[0-9]+$ && "$api" != '' && "$kind" != '' && "$namespace" == observability &&
      "$name" != '' && "$uid" != '' && -z "${extra:-}" ]] || return 1
    key="$(object_key "$kind" "$namespace" "$name")"
    [[ -z "${MUTATED_UID[$key]+x}" ]] || return 1
    MUTATED_UID[$key]=$uid
  done < <(tail -n +2 "$copy")
}

restore_owned_object() {
  local kind=$1 namespace=$2 name=$3 expected_uid=$4 live_json=$5 prior_payload=$6
  local conditional="$work/conditional-restore-$kind-$name.json"
  python3 - "$live_json" "$prior_payload" "$expected_uid" "$conditional" <<'PY'
import json, sys
live=json.loads(sys.argv[1])
prior=json.load(open(sys.argv[2],encoding="utf-8"))
expected=sys.argv[3]
output=sys.argv[4]
live_meta=live.get("metadata") or {}
if live_meta.get("uid") != expected or not live_meta.get("resourceVersion"):
    raise SystemExit(1)
meta=prior.setdefault("metadata",{})
meta["uid"]=expected
meta["resourceVersion"]=live_meta["resourceVersion"]
with open(output,"x",encoding="utf-8") as stream:
    json.dump(prior,stream,separators=(",",":"),sort_keys=True)
    stream.write("\n")
PY
  chmod 0600 "$conditional" || return 1
  kubectl_bounded replace -f "$conditional" >/dev/null
}

delete_api_path() {
  local kind=$1 namespace=$2 name=$3
  [[ "$namespace" =~ ^[a-z0-9]([-a-z0-9]*[a-z0-9])?$ &&
    "$name" =~ ^[a-z0-9]([-a-z0-9.]*[a-z0-9])?$ ]] || return 1
  case "$kind" in
    ConfigMap) printf '/api/v1/namespaces/%s/configmaps/%s' "$namespace" "$name" ;;
    PrometheusRule) printf '/apis/monitoring.coreos.com/v1/namespaces/%s/prometheusrules/%s' "$namespace" "$name" ;;
    AlertmanagerConfig) printf '/apis/monitoring.coreos.com/v1alpha1/namespaces/%s/alertmanagerconfigs/%s' "$namespace" "$name" ;;
    Alertmanager) printf '/apis/monitoring.coreos.com/v1/namespaces/%s/alertmanagers/%s' "$namespace" "$name" ;;
    NetworkPolicy) printf '/apis/networking.k8s.io/v1/namespaces/%s/networkpolicies/%s' "$namespace" "$name" ;;
    *) return 1 ;;
  esac
}

ensure_api_proxy() {
  if [[ -n "$api_proxy_pid" ]] && kill -0 "$api_proxy_pid" 2>/dev/null && [[ -S "$api_proxy_socket" ]]; then
    return 0
  fi
  api_proxy_socket="$work/kube-api-proxy.sock"
  "$KUBECTL_BIN" --request-timeout=10s proxy --unix-socket="$api_proxy_socket" \
    --accept-paths='^/(api|apis)/.*$' \
    --reject-methods='^(POST|PUT|PATCH)$' \
    >"$work/kube-api-proxy.log" 2>&1 &
  api_proxy_pid=$!
  local attempt
  for ((attempt=1; attempt<=50; attempt++)); do
    [[ -S "$api_proxy_socket" ]] && return 0
    kill -0 "$api_proxy_pid" 2>/dev/null || return 1
    sleep 0.1
  done
  return 1
}

delete_owned_object() {
  local kind=$1 namespace=$2 name=$3 expected_uid=$4 api_path body response status live uid attempt
  ensure_api_proxy || return 1
  api_path="$(delete_api_path "$kind" "$namespace" "$name")" || return 1
  body="$work/delete-options-$kind-$name.json"
  response="$work/delete-response-$kind-$name.json"
  python3 - "$expected_uid" "$body" <<'PY'
import json,sys
uid=sys.argv[1]
if not uid or any(character.isspace() for character in uid):
    raise SystemExit(1)
with open(sys.argv[2],"x",encoding="utf-8") as stream:
    json.dump({"apiVersion":"v1","kind":"DeleteOptions","preconditions":{"uid":uid},"propagationPolicy":"Background"},stream,separators=(",",":"))
    stream.write("\n")
PY
  chmod 0600 "$body" || return 1
  if ! status="$(curl --silent --show-error --connect-timeout 1 --max-time 1 \
    --output "$response" --write-out '%{http_code}' \
    --unix-socket "$api_proxy_socket" \
    --request DELETE --header 'Content-Type: application/json' \
    --data-binary "@$body" "http://localhost$api_path")"; then
    return 1
  fi
  [[ "$status" == 200 || "$status" == 202 ]] || return 1
  chmod 0600 "$response" || return 1
  for ((attempt=1; attempt<=30; attempt++)); do
    live="$(get_object_json "$kind" "$namespace" "$name")" || return 1
    [[ -z "$live" ]] && return 0
    uid="$(object_uid_from_json <<<"$live")" || return 1
    [[ "$uid" == "$expected_uid" ]] || return 1
    (( attempt < 30 )) && sleep 1
  done
  return 1
}

rollback_resources() {
  rollback_running=true
  local authority mutation_copy payload_copy_root row phase owner api kind namespace name existed restore payload_rel payload_sha
  local key live uid resource payload action_failed=false
  authority="$(validate_authoritative_ledger_for_rollback)" || {
    rollback_ambiguous=true
    return 1
  }
  mutation_copy="$work/root-mutations.tsv"
  "$SUDO_BIN" -n cat -- "$ledger_root/mutations.tsv" >"$mutation_copy" || {
    rollback_ambiguous=true
    return 1
  }
  chmod 0600 "$mutation_copy" || {
    rollback_ambiguous=true
    return 1
  }
  load_mutation_uids "$mutation_copy" || {
    rollback_ambiguous=true
    return 1
  }
  payload_copy_root="$work/root-payloads"
  mapfile -t rollback_rows < <(tail -n +2 "$authority" | tac) || {
    rollback_ambiguous=true
    return 1
  }
  for row in "${rollback_rows[@]}"; do
    IFS=$'\t' read -r phase owner api kind namespace name existed restore payload_rel payload_sha <<<"$row"
    key="$(object_key "$kind" "$namespace" "$name")"
    [[ -n "${MUTATED_UID[$key]+x}" ]] || continue
    live="$(get_object_json "$kind" "$namespace" "$name")" || {
      rollback_ambiguous=true
      action_failed=true
      continue
    }
    if [[ -z "$live" ]]; then
      if [[ "$existed" == true ]]; then
        rollback_ambiguous=true
        action_failed=true
      fi
      continue
    fi
    uid="$(object_uid_from_json <<<"$live")" || {
      rollback_ambiguous=true
      action_failed=true
      continue
    }
    if [[ "$uid" != "${MUTATED_UID[$key]}" ]]; then
      rollback_ambiguous=true
      action_failed=true
      continue
    fi
    if [[ "$existed" == false ]]; then
      if ! delete_owned_object "$kind" "$namespace" "$name" "${MUTATED_UID[$key]}"; then
        rollback_ambiguous=true
        action_failed=true
        continue
      fi
      live="$(get_object_json "$kind" "$namespace" "$name")" || {
        rollback_ambiguous=true
        action_failed=true
        continue
      }
      [[ -z "$live" ]] || {
        rollback_ambiguous=true
        action_failed=true
      }
    else
      payload="$payload_copy_root/${payload_rel##*/}"
      if ! restore_owned_object "$kind" "$namespace" "$name" \
        "${MUTATED_UID[$key]}" "$live" "$payload"; then
        rollback_ambiguous=true
        action_failed=true
        continue
      fi
      live="$(get_object_json "$kind" "$namespace" "$name")" || {
        rollback_ambiguous=true
        action_failed=true
        continue
      }
      uid="$(object_uid_from_json <<<"$live")" || {
        rollback_ambiguous=true
        action_failed=true
        continue
      }
      [[ "$uid" == "${PRIOR_UID[$key]}" ]] || {
        rollback_ambiguous=true
        action_failed=true
      }
    fi
  done
  [[ "$action_failed" == false && "$rollback_ambiguous" == false ]]
}

run_production_postcheck_once() {
  local prometheus_ready alertmanager_ready rules receivers dashboard_hashes
  local rules_file="$work/prometheus-rules-api.json"
  local receivers_file="$work/alertmanager-receivers-api.json"
  local dashboard_hashes_file="$work/grafana-dashboard-sidecar.sha256"
  kubectl_bounded wait --namespace observability --for=condition=Available \
    prometheus/observability-core-kube-pr-prometheus \
    alertmanager/observability-core-kube-pr-alertmanager --timeout=10s >/dev/null || return 1
  kubectl_bounded rollout status --namespace observability \
    statefulset/prometheus-observability-core-kube-pr-prometheus --timeout=10s >/dev/null || return 1
  kubectl_bounded rollout status --namespace observability \
    statefulset/alertmanager-observability-core-kube-pr-alertmanager --timeout=10s >/dev/null || return 1
  prometheus_ready="$(kubectl_bounded get \
    --raw=/api/v1/namespaces/observability/services/http:observability-core-kube-pr-prometheus:9090/proxy/-/ready)" || return 1
  [[ "$prometheus_ready" == *'Prometheus Server is Ready.'* ]] || return 1
  alertmanager_ready="$(kubectl_bounded get \
    --raw=/api/v1/namespaces/observability/services/http:observability-core-kube-pr-alertmanager:9093/proxy/-/ready)" || return 1
  [[ "$alertmanager_ready" == *'OK'* ]] || return 1
  rules="$(kubectl_bounded get \
    --raw=/api/v1/namespaces/observability/services/http:observability-core-kube-pr-prometheus:9090/proxy/api/v1/rules)" || return 1
  receivers="$(kubectl_bounded get \
    --raw=/api/v1/namespaces/observability/services/http:observability-core-kube-pr-alertmanager:9093/proxy/api/v2/receivers)" || return 1
  dashboard_hashes="$(kubectl_bounded exec --namespace observability deployment/grafana --container grafana -- \
    /bin/sh -ceu 'for file in https-endpoints.json kubernetes-node.json observability-backends.json platform-services.json workload-health.json; do test -f "/tmp/dashboards/$file"; sha256sum "/tmp/dashboards/$file"; done')" || return 1
  printf '%s\n' "$rules" >"$rules_file"
  printf '%s\n' "$receivers" >"$receivers_file"
  printf '%s\n' "$dashboard_hashes" >"$dashboard_hashes_file"
  chmod 0600 "$rules_file" "$receivers_file" "$dashboard_hashes_file"
  python3 - "$verified_output_dir/rules.yaml" "$rules_file" \
    "$verified_output_dir/dashboards.yaml" "$dashboard_hashes_file" "$receivers_file" <<'PY'
import collections
import hashlib
import json
import pathlib
import re
import sys
import yaml

rules_yaml, rules_api, dashboards_yaml, dashboard_hashes, receivers_api = map(pathlib.Path, sys.argv[1:])

def documents(path):
    with path.open(encoding="utf-8") as stream:
        return [item for item in yaml.safe_load_all(stream) if item]

desired_rules = collections.Counter()
desired_groups = set()
for item in documents(rules_yaml):
    for group in (item.get("spec") or {}).get("groups") or []:
        group_name = group.get("name")
        if not isinstance(group_name, str) or not group_name:
            raise SystemExit(1)
        desired_groups.add(group_name)
        for rule in group.get("rules") or []:
            alert = rule.get("alert")
            record = rule.get("record")
            if bool(alert) == bool(record):
                raise SystemExit(1)
            name = alert if alert else record
            if not isinstance(name, str) or not name:
                raise SystemExit(1)
            desired_rules[(group_name, name, "alerting" if alert else "recording")] += 1
if not desired_rules:
    raise SystemExit(1)

payload = json.loads(rules_api.read_text(encoding="utf-8"))
if payload.get("status") != "success":
    raise SystemExit(1)
accepted = collections.Counter()
for group in (payload.get("data") or {}).get("groups") or []:
    group_name = group.get("name")
    if group_name not in desired_groups:
        continue
    for rule in group.get("rules") or []:
        if rule.get("health") != "ok" or rule.get("lastError") not in (None, ""):
            raise SystemExit(1)
        accepted[(group_name, rule.get("name"), rule.get("type"))] += 1
if accepted != desired_rules:
    raise SystemExit(1)

expected_dashboard_names = {
    "grafana-dashboard-https-endpoints": "https-endpoints.json",
    "grafana-dashboard-kubernetes-node": "kubernetes-node.json",
    "grafana-dashboard-observability-backends": "observability-backends.json",
    "grafana-dashboard-platform-services": "platform-services.json",
    "grafana-dashboard-workload-health": "workload-health.json",
}
expected_hashes = {}
dashboard_items = documents(dashboards_yaml)
if len(dashboard_items) != len(expected_dashboard_names):
    raise SystemExit(1)
for item in dashboard_items:
    metadata = item.get("metadata") or {}
    name = metadata.get("name")
    filename = expected_dashboard_names.get(name)
    data = item.get("data") or {}
    if not filename or set(data) != {filename} or not isinstance(data[filename], str):
        raise SystemExit(1)
    expected_hashes[filename] = hashlib.sha256(data[filename].encode()).hexdigest()

actual_hashes = {}
line_pattern = re.compile(r"^([0-9a-f]{64})\s+/tmp/dashboards/([^/\s]+)$")
for line in dashboard_hashes.read_text(encoding="utf-8").splitlines():
    match = line_pattern.fullmatch(line)
    if not match or match.group(2) in actual_hashes:
        raise SystemExit(1)
    actual_hashes[match.group(2)] = match.group(1)
if actual_hashes != expected_hashes:
    raise SystemExit(1)

receiver_payload = json.loads(receivers_api.read_text(encoding="utf-8"))
expected_receivers = {
    "observability/platform-alertmanager/platform-null",
    "observability/platform-alertmanager/platform-slack",
}
if (
    not isinstance(receiver_payload, list)
    or len(receiver_payload) != 2
    or any(
        not isinstance(item, dict)
        or set(item) != {"labels", "name"}
        or item.get("labels") != {"name": item.get("name")}
        for item in receiver_payload
    )
    or {item["name"] for item in receiver_payload} != expected_receivers
):
    raise SystemExit(1)
PY
}

run_postcheck() {
  local attempt max_attempts=30 delay=2
  if [[ "$TEST_MODE" == 1 ]]; then
    max_attempts=1
    delay=0
  fi
  for ((attempt=1; attempt<=max_attempts; attempt++)); do
    if run_production_postcheck_once; then
      return 0
    fi
    (( attempt < max_attempts )) && sleep "$delay"
  done
  return 1
}

validate_acceptance() {
  local path=$1 expected=$2
  "$SUDO_BIN" -n cmp -s -- "$expected" "$path"
}

write_acceptance() {
  local accepted_at marker="$snapshot/acceptance.env"
  accepted_at="$(date -u +%Y-%m-%dT%H:%M:%SZ)" || return 1
  cat >"$marker" <<EOF
schema=$ACCEPTANCE_SCHEMA
rollback_id=${PLATFORM_OBSERVABILITY_ROLLBACK_ID}
target_initial_sha256=$target_initial_sha
post_substrate_sha256=$post_substrate_sha
slack_deployment_gate=$slack_deployment_gate_value
slack_gate_approval_ref=$slack_gate_approval_ref
slack_gate_accepted_by_uid=$slack_gate_accepted_by_uid
accepted_at_utc=$accepted_at
state=accepted
EOF
  chmod 0600 "$marker"
  root_atomic_install "$marker" "$ledger_root/acceptance.env" create || return 1
  validate_root_metadata "$ledger_root/acceptance.env" 600 'regular file' || return 1
  validate_acceptance "$ledger_root/acceptance.env" "$marker"
}

cleanup() {
  local rc=$?
  trap - EXIT HUP INT TERM
  if [[ -n "$api_proxy_pid" ]] && kill -0 "$api_proxy_pid" 2>/dev/null; then
    kill "$api_proxy_pid" 2>/dev/null || true
    wait "$api_proxy_pid" 2>/dev/null || true
  fi
  case "$work" in
    /tmp/platform-observability-access-apply.??????) rm -rf -- "$work" ;;
  esac
  case "$slack_gate_temp_root" in
    /tmp/platform-observability-slack-gate.??????)
      [[ -d "$slack_gate_temp_root" && ! -L "$slack_gate_temp_root" &&
        "$(readlink -f -- "$slack_gate_temp_root")" == "$slack_gate_temp_root" &&
        "$(stat -c '%u:%a' -- "$slack_gate_temp_root")" == "$(id -u):700" ]] &&
        rm -rf -- "$slack_gate_temp_root"
      ;;
  esac
  exit "$rc"
}

on_signal() {
  local rc=$1
  trap - HUP INT TERM
  set +e
  if [[ "$mutation_started" == true && "$rollback_running" == false ]]; then
    rollback_resources
    if [[ $? == 0 ]]; then
      printf 'OBSERVABILITY_ACCESS_RULES_ALERTS_ROLLBACK=PASS\n' >&2
    else
      printf 'OBSERVABILITY_ACCESS_RULES_ALERTS_ROLLBACK=AMBIGUOUS\nMANUAL_RECOVERY_REQUIRED=YES\n' >&2
    fi
  fi
  exit "$rc"
}

on_error() {
  local rc=$1
  trap - ERR
  printf 'OBSERVABILITY_ACCESS_RULES_ALERTS_STAGE=%s\n' "$failure_stage" >&2
  printf 'OBSERVABILITY_ACCESS_RULES_ALERTS_RC=%s\n' "$rc" >&2
  exit "$rc"
}

main() {
  if ! parse_args "$@"; then
    usage >&2
    return 2
  fi
  validate_test_mode
  if [[ "$execute" == false ]]; then
    print_dry_run
    return 0
  fi
  trap cleanup EXIT
  trap 'on_signal 129' HUP
  trap 'on_signal 130' INT
  trap 'on_signal 143' TERM

  # Refresh sudo before validators that perform narrow root evidence reads.
  # This is authorization only; no object, file, or host state is mutated.
  "$SUDO_BIN" -v
  "$SUDO_BIN" -n true
  failure_stage=encryption-pre
  run_encryption_gates
  failure_stage=context
  validate_context
  failure_stage=handoff
  validate_handoff_root
  failure_stage=slack-secret
  validate_slack_secret_schema
  failure_stage=slack-deployment-gate
  capture_slack_deployment_gate slack_deployment_gate_first pre-prompt
  failure_stage=substrate-target-health
  run_preflight || fail 'substrate or target health preflight failed'
  failure_stage=server-dry-run
  server_dry_run_handoff

  validate_rollback_root
  confirm_apply

  work="$(mktemp -d /tmp/platform-observability-access-apply.XXXXXX)"
  chmod 0700 "$work"
  snapshot="$work/snapshot"
  mkdir -m 0700 "$snapshot"
  trap cleanup EXIT
  trap 'on_signal 129' HUP
  trap 'on_signal 130' INT
  trap 'on_signal 143' TERM

  failure_stage=renderer-split
  prepare_split_files
  failure_stage=rollback-ledger
  capture_snapshot
  install_snapshot
  failure_stage=encryption-last
  run_encryption_gates
  failure_stage=last-gate
  verify_handoff_unchanged
  verify_live_preconditions
  failure_stage=slack-secret-last
  validate_slack_secret_schema
  failure_stage=slack-deployment-gate-last
  capture_slack_deployment_gate slack_deployment_gate last
  require_matching_slack_deployment_gates
  map_slack_deployment_gate_acceptance || fail 'Slack deployment gate verdict is invalid'

  if ! apply_resources; then
    if [[ "$mutation_started" == true ]]; then
      if rollback_resources; then
        printf 'OBSERVABILITY_ACCESS_RULES_ALERTS_ROLLBACK=PASS\n' >&2
      else
        printf 'OBSERVABILITY_ACCESS_RULES_ALERTS_ROLLBACK=AMBIGUOUS\n' >&2
        printf 'MANUAL_RECOVERY_REQUIRED=YES\n' >&2
      fi
    fi
    printf 'OBSERVABILITY_ACCESS_RULES_ALERTS_FAILED_AT=%s\n' "$failure_stage" >&2
    return 1
  fi

  failure_stage=reconcile
  if ! run_postcheck; then
    if rollback_resources; then
      printf 'OBSERVABILITY_ACCESS_RULES_ALERTS_ROLLBACK=PASS\n' >&2
    else
      printf 'OBSERVABILITY_ACCESS_RULES_ALERTS_ROLLBACK=AMBIGUOUS\n' >&2
      printf 'MANUAL_RECOVERY_REQUIRED=YES\n' >&2
    fi
    printf 'OBSERVABILITY_ACCESS_RULES_ALERTS_FAILED_AT=%s\n' "$failure_stage" >&2
    return 1
  fi

  failure_stage=acceptance
  write_acceptance || {
    printf 'OBSERVABILITY_ACCESS_RULES_ALERTS_STATE=APPLIED_ACCEPTANCE_UNRECORDED\n' >&2
    printf 'MANUAL_RECOVERY_REQUIRED=YES\n' >&2
    return 1
  }
  printf 'OBSERVABILITY_ACCESS_RULES_ALERTS_TARGET_INITIAL_SHA256=%s\n' "$target_initial_sha"
  printf 'OBSERVABILITY_ACCESS_RULES_ALERTS_POST_SUBSTRATE_SHA256=%s\n' "$post_substrate_sha"
  printf 'OBSERVABILITY_ACCESS_RULES_ALERTS=PASS\n'
}

if [[ "${BASH_SOURCE[0]}" == "$0" ]]; then
  trap 'on_error "$?"' ERR
  main "$@"
fi