#!/usr/bin/env bash
set -euo pipefail

script_dir=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
# shellcheck source=scripts/cluster/common.sh
source "$script_dir/common.sh"

usage() {
  cat <<'EOF'
Usage:
  scripts/cluster/run-production-acceptance.sh --execute dist/releases/<SHA>.json

This is a destructive production acceptance workflow. It pauses new live tasks,
changes Worker capacity, creates paid real canaries, and deletes one Worker Pod
during each recovery phase. A failed run deliberately keeps traffic in
validation mode and restores only the last stable capacity profile.

Required private environment values:
  AI_GATEWAY_ACCEPTANCE_ADMIN_TOKEN
  AI_GATEWAY_ACCEPTANCE_API_KEY
  AI_GATEWAY_ACCEPTANCE_API_KEY_ID
  AI_GATEWAY_ACCEPTANCE_USER_ID
  AI_GATEWAY_ACCEPTANCE_GATEWAYS
  AI_GATEWAY_ACCEPTANCE_REAL_IMAGE_URLS

Optional model overrides (otherwise selected from current production candidates):
  AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL
  AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL
  AI_GATEWAY_ACCEPTANCE_GATEWAY_TLS_SERVER_NAME
EOF
}

[[ ${1:-} == --execute && $# -eq 2 ]] || {
  usage >&2
  exit 64
}

release_manifest=$2
[[ -f $release_manifest && ! -L $release_manifest ]] || {
  echo 'release manifest must be a regular file' >&2
  exit 1
}

load_cluster_env
require_commands curl git go jq node openssl sed

: "${AI_GATEWAY_ACCEPTANCE_ADMIN_TOKEN:?}"
: "${AI_GATEWAY_ACCEPTANCE_API_KEY:?}"
: "${AI_GATEWAY_ACCEPTANCE_API_KEY_ID:?}"
: "${AI_GATEWAY_ACCEPTANCE_USER_ID:?}"
: "${AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL:=}"
: "${AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL:=}"
: "${AI_GATEWAY_ACCEPTANCE_GATEWAY_TLS_SERVER_NAME:=}"
: "${AI_GATEWAY_ACCEPTANCE_GATEWAYS:?}"
: "${AI_GATEWAY_ACCEPTANCE_REAL_IMAGE_URLS:?}"

node "$cluster_root/scripts/release-manifest.mjs" validate "$release_manifest" >/dev/null
release_sha=$(node "$cluster_root/scripts/release-manifest.mjs" get "$release_manifest" sourceSha)
api_image=$(node "$cluster_root/scripts/release-manifest.mjs" get "$release_manifest" images.api)
[[ $release_sha =~ ^[0-9a-f]{40}$ && $api_image =~ @sha256:[0-9a-f]{64}$ ]] || {
  echo 'release manifest is not full-SHA and digest pinned' >&2
  exit 1
}
api_digest=${api_image##*@}
worker_digest=$api_digest
[[ $(git -C "$cluster_root" rev-parse HEAD) == "$release_sha" ]] || {
  echo 'working copy HEAD must equal the release SHA' >&2
  exit 1
}
[[ -z $(git -C "$cluster_root" status --short) ]] || {
  echo 'production acceptance requires a clean release working copy' >&2
  exit 1
}

namespace=${AI_GATEWAY_K3S_NAMESPACE:-easyai}
remote_release_helper=${AI_GATEWAY_REMOTE_RELEASE_HELPER:-/usr/local/sbin/easyai-ai-gateway-cluster-release}
[[ $namespace =~ ^[a-z0-9-]+$ && $remote_release_helper =~ ^/[A-Za-z0-9._/-]+$ ]] || {
  echo 'invalid namespace or remote release helper' >&2
  exit 1
}
[[ $namespace == easyai ]] || {
  echo 'the production acceptance manifests currently require namespace easyai' >&2
  exit 1
}

temporary_root=$(mktemp -d)
chmod 0700 "$temporary_root"
current_manifest=$temporary_root/current.json
load_binary=$temporary_root/easyai-ai-gateway-acceptance-load
run_token=$(openssl rand -hex 32)
run_id=
report_root=
stable_profile=P24
active_profile=P24
failure_reason=
failure_recorded=false

cleanup() {
  local status=$?
  trap - EXIT
  if (( status != 0 )) && [[ -n $run_id && $failure_recorded != true ]]; then
    set +e
    apply_capacity_profile "$stable_profile" >/dev/null 2>&1
    mark_run_failed 'acceptance workflow exited unexpectedly'
    set -e
  fi
  rm -rf -- "$temporary_root"
  exit "$status"
}
trap cleanup EXIT

remote_kubectl() {
  local command_text
  printf -v command_text '%q ' "$@"
  cluster_ssh "$CLUSTER_NINGBO_HOST" "k3s kubectl $command_text"
}

database_query() {
  local sql=$1
  local primary
  primary=$(remote_kubectl get cluster easyai-postgres -n "$namespace" -o 'jsonpath={.status.currentPrimary}')
  [[ $primary =~ ^easyai-postgres-[0-9]+$ ]]
  remote_kubectl exec -n "$namespace" "$primary" -c postgres -- \
    psql -X -v ON_ERROR_STOP=1 -U postgres -d easyai_ai_gateway -At -c "$sql"
}

admin_request() {
  local method=$1
  local path=$2
  local body=${3:-}
  local output=$4
  local admin_base=${AI_GATEWAY_ACCEPTANCE_GATEWAYS%%,*}
  local status
  if [[ -n $body ]]; then
    status=$(curl -sS --max-time 30 -o "$output" -w '%{http_code}' \
      -X "$method" \
      -H "Authorization: Bearer $AI_GATEWAY_ACCEPTANCE_ADMIN_TOKEN" \
      -H 'Content-Type: application/json' \
      --data-binary "$body" \
      "$admin_base$path")
  else
    status=$(curl -sS --max-time 30 -o "$output" -w '%{http_code}' \
      -X "$method" \
      -H "Authorization: Bearer $AI_GATEWAY_ACCEPTANCE_ADMIN_TOKEN" \
      "$admin_base$path")
  fi
  [[ $status =~ ^2[0-9][0-9]$ ]] || {
    echo "acceptance control API failed: method=$method path=$path status=$status" >&2
    return 1
  }
}

verify_release_cas() {
  cluster_ssh "$CLUSTER_NINGBO_HOST" "$remote_release_helper status" >"$current_manifest"
  node "$cluster_root/scripts/release-manifest.mjs" validate "$current_manifest" >/dev/null
  local current_sha current_api
  current_sha=$(node "$cluster_root/scripts/release-manifest.mjs" get "$current_manifest" sourceSha)
  current_api=$(node "$cluster_root/scripts/release-manifest.mjs" get "$current_manifest" images.api)
  [[ $current_sha == "$release_sha" && $current_api == "$api_image" ]] || {
    echo "production release CAS mismatch: expected=$release_sha current=$current_sha" >&2
    return 1
  }
  for deployment in easyai-api-ningbo easyai-api-hongkong easyai-worker-ningbo easyai-worker-hongkong; do
    [[ $(remote_kubectl get deployment "$deployment" -n "$namespace" \
      -o 'jsonpath={.spec.template.spec.containers[0].image}') == "$api_image" ]]
  done
  [[ $(remote_kubectl get deployment easyai-worker-ningbo -n "$namespace" \
    -o 'jsonpath={.spec.replicas}') == 1 ]]
  [[ $(remote_kubectl get deployment easyai-worker-hongkong -n "$namespace" \
    -o 'jsonpath={.spec.replicas}') == 1 ]]
}

wait_for_existing_tasks_to_drain() {
  local deadline=$((SECONDS + 900))
  local active
  while (( SECONDS < deadline )); do
    active=$(database_query "SELECT count(*) FROM gateway_tasks WHERE acceptance_run_id IS NULL AND status IN ('queued','running');")
    if [[ $active == 0 ]]; then
      return 0
    fi
    echo "waiting_for_existing_tasks=$active"
    sleep 5
  done
  echo 'existing production tasks did not drain within 15 minutes' >&2
  return 1
}

select_acceptance_models() {
  if [[ -z $AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL ]]; then
    AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL=$(database_query "
SELECT b.invocation_name
FROM platform_models model
JOIN integration_platforms platform ON platform.id=model.platform_id
JOIN base_model_catalog b ON b.id=model.base_model_id
WHERE platform.status='enabled' AND platform.deleted_at IS NULL
  AND model.enabled=true
  AND lower(platform.provider) IN ('gemini','google-gemini','gemini-openai')
  AND model.model_type @> '[\"image_edit\"]'::jsonb
ORDER BY
  CASE WHEN lower(b.invocation_name) LIKE '%gemini%image%' THEN 0 ELSE 1 END,
  platform.priority, model.created_at
LIMIT 1;")
  fi
  if [[ -z $AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL ]]; then
    AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL=$(database_query "
SELECT b.invocation_name
FROM platform_models model
JOIN integration_platforms platform ON platform.id=model.platform_id
JOIN base_model_catalog b ON b.id=model.base_model_id
WHERE platform.status='enabled' AND platform.deleted_at IS NULL
  AND model.enabled=true
  AND model.model_type @> '[\"omni_video\"]'::jsonb
  AND COALESCE(
    NULLIF(model.capability_override #>> '{omni_video,max_images}','')::int,
    NULLIF(model.capabilities #>> '{omni_video,max_images}','')::int,
    NULLIF(b.capabilities #>> '{omni_video,max_images}','')::int,
    0
  ) >= 9
ORDER BY
  CASE
    WHEN lower(b.invocation_name) LIKE '%seedance%2%fast%' THEN 0
    WHEN lower(b.invocation_name) LIKE '%seedance%2%' THEN 1
    ELSE 2
  END,
  platform.priority, model.created_at
LIMIT 1;")
  fi
  [[ -n $AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL && -n $AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL ]] || {
    echo 'no enabled Gemini image-edit or omni-video max_images>=9 model was found' >&2
    return 1
  }
  export AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL
}

deploy_protocol_emulator() {
  sed "s|image: easyai-api|image: $api_image|" \
    "$cluster_root/deploy/kubernetes/acceptance/protocol-emulator.yaml" |
    cluster_ssh "$CLUSTER_NINGBO_HOST" 'k3s kubectl apply -f -' >/dev/null
  remote_kubectl rollout status deployment/easyai-acceptance-emulator \
    -n "$namespace" --timeout=300s
}

create_and_activate_run() {
  local create_response=$temporary_root/create-run.json
  local activate_response=$temporary_root/activate-run.json
  local traffic_response=$temporary_root/pre-activate-traffic.json
  local previous_run_response=$temporary_root/previous-run.json
  local abort_response=$temporary_root/abort-previous-run.json
  local body
  body=$(jq -cn \
    --arg releaseSha "$release_sha" \
    --arg apiDigest "$api_digest" \
    --arg workerDigest "$worker_digest" \
    --arg apiKeyID "$AI_GATEWAY_ACCEPTANCE_API_KEY_ID" \
    --arg userID "$AI_GATEWAY_ACCEPTANCE_USER_ID" \
    --arg token "$run_token" \
    --arg emulatorURL 'http://easyai-acceptance-emulator.easyai.svc.cluster.local:8090' \
    --arg callbackURL 'http://easyai-acceptance-emulator.easyai.svc.cluster.local:8090/callbacks' \
    '{
      releaseSha: $releaseSha,
      apiImageDigest: $apiDigest,
      workerImageDigest: $workerDigest,
      apiKeyId: $apiKeyID,
      userId: $userID,
      token: $token,
      emulatorBaseUrl: $emulatorURL,
      callbackUrl: $callbackURL,
      capacityProfile: "P24",
      config: {workloads: ["gemini_image_edit", "multi_reference_video"]}
    }')
  admin_request POST /api/admin/system/acceptance/runs "$body" "$create_response"
  run_id=$(jq -r '.id // empty' "$create_response")
  [[ $run_id =~ ^[0-9a-f-]{36}$ ]] || {
    echo 'acceptance control API returned an invalid Run ID' >&2
    return 1
  }
  report_root=${AI_GATEWAY_ACCEPTANCE_REPORT_DIR:-"$cluster_root/dist/acceptance/$run_id"}
  mkdir -p "$report_root"
  chmod 0700 "$report_root"

  admin_request GET /api/admin/system/acceptance/traffic-mode '' "$traffic_response"
  local traffic_mode previous_run_id previous_status
  traffic_mode=$(jq -r '.mode // empty' "$traffic_response")
  if [[ $traffic_mode == validation ]]; then
    previous_run_id=$(jq -r '.runId // empty' "$traffic_response")
    [[ $previous_run_id =~ ^[0-9a-f-]{36}$ && $previous_run_id != "$run_id" ]] || {
      echo 'validation mode does not reference a different previous Run' >&2
      return 1
    }
    [[ $(jq -r '.releaseSha // empty' "$traffic_response") == "$release_sha" &&
       $(jq -r '.apiImageDigest // empty' "$traffic_response") == "$api_digest" &&
       $(jq -r '.workerImageDigest // empty' "$traffic_response") == "$worker_digest" ]] || {
      echo 'refusing to replace validation mode for a different production release' >&2
      return 1
    }
    admin_request GET "/api/admin/system/acceptance/runs/$previous_run_id" '' \
      "$previous_run_response"
    previous_status=$(jq -r '.status // empty' "$previous_run_response")
    [[ $previous_status == failed ]] || {
      echo "refusing to replace previous acceptance Run in status $previous_status" >&2
      return 1
    }
    body=$(jq -cn \
      --argjson revision "$(jq -r '.revision' "$traffic_response")" \
      --arg releaseSha "$release_sha" \
      --arg apiDigest "$api_digest" \
      --arg workerDigest "$worker_digest" \
      '{
        revision:$revision,
        releaseSha:$releaseSha,
        apiImageDigest:$apiDigest,
        workerImageDigest:$workerDigest
      }')
    admin_request POST \
      "/api/admin/system/acceptance/runs/$previous_run_id/abort" \
      "$body" "$abort_response"
    [[ $(jq -r '.mode // empty' "$abort_response") == live ]] || {
      echo 'previous failed acceptance Run did not return traffic to live' >&2
      return 1
    }
  elif [[ $traffic_mode != live ]]; then
    echo "unsupported production traffic mode before acceptance: $traffic_mode" >&2
    return 1
  fi

  admin_request POST "/api/admin/system/acceptance/runs/$run_id/activate" '' "$activate_response"
  [[ $(jq -r '.mode' "$activate_response") == validation &&
     $(jq -r '.runId' "$activate_response") == "$run_id" ]]
}

mark_run_failed() {
  local reason=$1
  local response=$temporary_root/failed-run.json
  local body
  [[ -n $run_id ]] || return 0
  body=$(jq -cn --arg reason "$reason" \
    '{passed:false,failureReason:$reason,report:{queueFinal:null}}')
  if admin_request POST "/api/admin/system/acceptance/runs/$run_id/finish" "$body" "$response"; then
    failure_recorded=true
  fi
}

apply_capacity_profile() {
  local profile=$1
  local slots pool media global
  case $profile in
    P24) slots=24; pool=32; media=24; global=48 ;;
    P28) slots=28; pool=36; media=28; global=56 ;;
    P32) slots=32; pool=40; media=32; global=64 ;;
    *) return 1 ;;
  esac
  active_profile=$profile
  for site in ningbo hongkong; do
    remote_kubectl set env "deployment/easyai-worker-$site" -n "$namespace" \
      "AI_GATEWAY_ASYNC_WORKER_INSTANCE_HARD_LIMIT=$slots" \
      "AI_GATEWAY_ASYNC_WORKER_HARD_LIMIT=$global" \
      "AI_GATEWAY_DATABASE_MAX_CONNS=$pool" \
      "AI_GATEWAY_MEDIA_MATERIALIZATION_CONCURRENCY=$media" \
      "AI_GATEWAY_MEDIA_REQUEST_CONCURRENCY=$media" >/dev/null
    remote_kubectl rollout status "deployment/easyai-worker-$site" \
      -n "$namespace" --timeout=300s
    remote_kubectl set env "deployment/easyai-api-$site" -n "$namespace" \
      "AI_GATEWAY_ASYNC_WORKER_HARD_LIMIT=$global" >/dev/null
    remote_kubectl rollout status "deployment/easyai-api-$site" \
      -n "$namespace" --timeout=300s
  done
}

run_load_profile() {
  local profile=$1
  local report_path=$2
  (
    cd "$cluster_root/apps/api"
    AI_GATEWAY_ACCEPTANCE_PROFILE=$profile \
    AI_GATEWAY_ACCEPTANCE_GATEWAYS=$AI_GATEWAY_ACCEPTANCE_GATEWAYS \
    AI_GATEWAY_ACCEPTANCE_GATEWAY_TLS_SERVER_NAME=$AI_GATEWAY_ACCEPTANCE_GATEWAY_TLS_SERVER_NAME \
    AI_GATEWAY_ACCEPTANCE_EMULATOR_URL=http://easyai-acceptance-emulator.easyai.svc.cluster.local:8090 \
    AI_GATEWAY_ACCEPTANCE_API_KEY=$AI_GATEWAY_ACCEPTANCE_API_KEY \
    AI_GATEWAY_ACCEPTANCE_RUN_ID=$run_id \
    AI_GATEWAY_ACCEPTANCE_RUN_TOKEN=$run_token \
    AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL=$AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL \
    AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL=$AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL \
    AI_GATEWAY_ACCEPTANCE_REAL_IMAGE_URLS=$AI_GATEWAY_ACCEPTANCE_REAL_IMAGE_URLS \
      "$load_binary" -profile "$profile" -report "$report_path" >/dev/null
  )
}

kill_one_worker_during_recovery() {
  local deadline=$((SECONDS + 120))
  local running pod
  while (( SECONDS < deadline )); do
    running=$(database_query "SELECT count(*) FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid AND status='running' AND request::text LIKE '%acceptance-long-recovery%';")
    if (( running > 0 )); then
      break
    fi
    sleep 1
  done
  (( running > 0 )) || {
    echo 'recovery workload never reached a running Worker' >&2
    return 1
  }
  pod=$(remote_kubectl get pods -n "$namespace" \
    -l 'app.kubernetes.io/name=easyai-worker,easyai.io/site=hongkong' \
    -o 'jsonpath={.items[0].metadata.name}')
  [[ $pod =~ ^easyai-worker-hongkong-[a-z0-9-]+$ ]]
  remote_kubectl delete pod "$pod" -n "$namespace" --wait=false >/dev/null
  remote_kubectl rollout status deployment/easyai-worker-hongkong \
    -n "$namespace" --timeout=300s
}

run_recovery_profile() {
  local report_path=$1
  run_load_profile video-recovery "$report_path" &
  local load_pid=$!
  if ! kill_one_worker_during_recovery; then
    kill "$load_pid" >/dev/null 2>&1 || true
    wait "$load_pid" >/dev/null 2>&1 || true
    return 1
  fi
  wait "$load_pid"
}

verify_wireguard() {
  local source_host=$1
  local target_ip=$2
  local max_rtt=$3
  local label=$4
  local output loss average handshakes now latest age
  output=$(cluster_ssh "$source_host" "ping -q -c 10 -W 2 $target_ip")
  loss=$(awk -F',' '/packet loss/ {gsub(/[^0-9.]/, "", $3); print $3}' <<<"$output")
  average=$(awk -F'=' '/min\\/avg\\/max/ {split($2, values, \"/\"); gsub(/ /, "", values[2]); print values[2]}' <<<"$output")
  [[ -n $loss && -n $average ]]
  awk -v loss="$loss" -v average="$average" -v max="$max_rtt" \
    'BEGIN { exit !(loss < 1 && average < max) }'
  now=$(date +%s)
  latest=$(cluster_ssh "$source_host" "wg show all latest-handshakes | awk '{print \\$3}' | sort -nr | head -n 2")
  handshakes=$(wc -l <<<"$latest" | tr -d ' ')
  (( handshakes >= 2 ))
  while IFS= read -r timestamp; do
    [[ $timestamp =~ ^[0-9]+$ && $timestamp -gt 0 ]]
    age=$((now - timestamp))
    (( age < 180 ))
  done <<<"$latest"
  echo "wireguard=$label loss_percent=$loss average_rtt_ms=$average recent_handshakes=$handshakes"
}

verify_cluster_links() {
  verify_wireguard "$CLUSTER_NINGBO_HOST" 10.77.0.2 80 'ningbo_to_hongkong'
  verify_wireguard "$CLUSTER_HONGKONG_HOST" 10.77.0.1 80 'hongkong_to_ningbo'
  verify_wireguard "$CLUSTER_NINGBO_HOST" 10.77.0.3 300 'ningbo_to_los_angeles'
  verify_wireguard "$CLUSTER_LOS_ANGELES_HOST" 10.77.0.1 300 'los_angeles_to_ningbo'
  verify_wireguard "$CLUSTER_HONGKONG_HOST" 10.77.0.3 300 'hongkong_to_los_angeles'
  verify_wireguard "$CLUSTER_LOS_ANGELES_HOST" 10.77.0.2 300 'los_angeles_to_hongkong'
}

verify_control_plane_and_recent_logs() {
  local host ready_output pod error_count
  for host in "$CLUSTER_NINGBO_HOST" "$CLUSTER_HONGKONG_HOST" "$CLUSTER_LOS_ANGELES_HOST"; do
    ready_output=$(cluster_ssh "$host" "k3s kubectl get --raw='/readyz?verbose'")
    grep -Fq '[+]etcd ok' <<<"$ready_output"
  done
  while read -r pod; do
    error_count=$(remote_kubectl logs "$pod" -n "$namespace" --since=10m |
      awk 'BEGIN {IGNORECASE=1}
        /postgres_unavailable/ ||
        /postgres readiness.*(fail|unavailable|error)/ ||
        /leadership elector.*(error|fail)/ ||
        /concurrency lease.*(renew.*(error|fail)|lost)/ {count++}
        END {print count+0}')
    [[ $error_count == 0 ]]
  done < <(remote_kubectl get pods -n "$namespace" \
    -l 'app.kubernetes.io/name=easyai-worker' \
    -o jsonpath='{range .items[*]}{.metadata.name}{"\n"}{end}')
  while read -r pod; do
    error_count=$(remote_kubectl logs "$pod" -n "$namespace" -c postgres --since=10m |
      awk 'BEGIN {IGNORECASE=1}
        /apply request took too long/ ||
        /dial tcp.*6443/ ||
        /synchronous commit.*cancel/ ||
        /canceling statement due to user request/ ||
        /(liveness|readiness).*fail/ {count++}
        END {print count+0}')
    [[ $error_count == 0 ]]
  done < <(remote_kubectl get pods -n "$namespace" \
    -l 'cnpg.io/cluster=easyai-postgres' \
    -o jsonpath='{range .items[*]}{.metadata.name}{"\n"}{end}')
}

current_max_pod_memory_mib() {
  remote_kubectl top pods -n "$namespace" \
    -l 'app.kubernetes.io/part-of=easyai-ai-gateway' --no-headers |
    awk '{
      value=$3
      if (value ~ /Gi$/) {sub(/Gi$/, "", value); value*=1024}
      else if (value ~ /Mi$/) {sub(/Mi$/, "", value)}
      else if (value ~ /Ki$/) {sub(/Ki$/, "", value); value/=1024}
      if (value>max) max=value
    } END {printf "%.0f", max+0}'
}

wait_for_memory_recovery() {
  local baseline_mib=$1
  local deadline=$((SECONDS + 120))
  local current_mib
  while (( SECONDS < deadline )); do
    current_mib=$(current_max_pod_memory_mib)
    if (( current_mib <= baseline_mib + 256 )); then
      return 0
    fi
    sleep 5
  done
  echo "Pod memory did not return near baseline: baseline_mib=$baseline_mib current_mib=$current_mib" >&2
  return 1
}

verify_runtime_gates() {
  local expected_slots expected_global expected_pool
  case $active_profile in
    P24) expected_slots=24; expected_pool=32; expected_global=48 ;;
    P28) expected_slots=28; expected_pool=36; expected_global=56 ;;
    P32) expected_slots=32; expected_pool=40; expected_global=64 ;;
  esac
  [[ $(remote_kubectl get nodes -o json |
    jq '[.items[] | select(any(.status.conditions[]; .type=="Ready" and .status=="True"))] | length') == 3 ]]
  [[ $(remote_kubectl get nodes -o json |
    jq '[.items[] | select(any(.status.conditions[]; .type=="MemoryPressure" and .status=="True"))] | length') == 0 ]]
  [[ $(remote_kubectl get cluster easyai-postgres -n "$namespace" -o 'jsonpath={.status.readyInstances}') == 2 ]]
  local sync_state
  sync_state=$(database_query "SELECT COALESCE(string_agg(sync_state,','),'') FROM pg_stat_replication;")
  [[ ",$sync_state," == *,sync,* || ",$sync_state," == *,quorum,* ]]
  verify_control_plane_and_recent_logs

  local unhealthy_containers pod_memory
  unhealthy_containers=$(remote_kubectl get pods -n "$namespace" -o json |
    jq '[.items[]
      | select(.metadata.labels["app.kubernetes.io/name"] == "easyai-api" or
               .metadata.labels["app.kubernetes.io/name"] == "easyai-worker")
      | .status.containerStatuses[]?
      | select(.restartCount > 0 or .lastState.terminated.reason == "OOMKilled" or
               .state.terminated.reason == "OOMKilled")] | length')
  [[ $unhealthy_containers == 0 ]]
  while read -r _node _cpu _cpu_percent _memory memory_percent; do
    memory_percent=${memory_percent%\%}
    [[ $memory_percent =~ ^[0-9]+$ ]]
    (( memory_percent < 80 ))
  done < <(remote_kubectl top nodes --no-headers)
  while read -r _pod _cpu pod_memory; do
    case $pod_memory in
      *Gi) pod_memory=$(awk -v value="${pod_memory%Gi}" 'BEGIN { printf "%.0f", value * 1024 }') ;;
      *Mi) pod_memory=${pod_memory%Mi} ;;
      *Ki) pod_memory=$(awk -v value="${pod_memory%Ki}" 'BEGIN { printf "%.0f", value / 1024 }') ;;
      *) return 1 ;;
    esac
    [[ $pod_memory =~ ^[0-9]+$ ]]
    (( pod_memory < 1536 ))
  done < <(remote_kubectl top pods -n "$namespace" \
    -l 'app.kubernetes.io/part-of=easyai-ai-gateway' --no-headers)
  local pod_name pod_anon_mib
  while read -r pod_name; do
    pod_anon_mib=$(remote_kubectl exec -n "$namespace" "$pod_name" -- \
      cat /sys/fs/cgroup/memory.stat |
      awk '$1=="anon" {printf "%.0f", $2/1048576}')
    [[ $pod_anon_mib =~ ^[0-9]+$ ]]
    (( pod_anon_mib < 1536 ))
  done < <(remote_kubectl get pods -n "$namespace" -o json |
    jq -r '.items[]
      | select(.metadata.labels["app.kubernetes.io/name"] == "easyai-api" or
               .metadata.labels["app.kubernetes.io/name"] == "easyai-worker")
      | .metadata.name')

  local completion_deadline=$((SECONDS + 180))
  local incomplete side_effects
  while (( SECONDS < completion_deadline )); do
    incomplete=$(database_query "
SELECT count(*)
FROM gateway_tasks
WHERE acceptance_run_id='$run_id'::uuid
  AND (
    status <> 'succeeded'
    OR billing_status <> 'settled'
  );")
    side_effects=$(database_query "
SELECT
  (SELECT count(*)
   FROM settlement_outbox settlement
   WHERE settlement.task_id IN (
     SELECT id FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid
   )
     AND settlement.status <> 'completed')
  +
  (SELECT count(*)
   FROM gateway_tasks task
   WHERE task.acceptance_run_id='$run_id'::uuid
     AND NOT EXISTS (
       SELECT 1
       FROM gateway_task_callback_outbox callback
       WHERE callback.task_id=task.id
         AND callback.status='delivered'
     ));")
    if [[ $incomplete == 0 && $side_effects == 0 ]]; then
      break
    fi
    sleep 2
  done
  [[ $incomplete == 0 && $side_effects == 0 ]]

  local queue_state duplicates settlements wallet_duplicates missing_billings callbacks connections max_connections
  local active_workers stale_workers
  queue_state=$(database_query "SELECT count(*) FILTER (WHERE status='queued')||':'||count(*) FILTER (WHERE status='running') FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid;")
  [[ $queue_state == 0:0 ]]
  duplicates=$(database_query "SELECT count(*) FROM (SELECT remote_task_id FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid AND remote_task_id IS NOT NULL GROUP BY remote_task_id HAVING count(*)>1) duplicate;")
  [[ $duplicates == 0 ]]
  settlements=$(database_query "SELECT count(*) FROM (SELECT task_id,action FROM settlement_outbox WHERE task_id IN (SELECT id FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid) GROUP BY task_id,action HAVING count(*)>1) duplicate;")
  [[ $settlements == 0 ]]
  wallet_duplicates=$(database_query "
SELECT count(*)
FROM (
  SELECT reference_id,transaction_type
  FROM gateway_wallet_transactions
  WHERE reference_type='gateway_task'
    AND reference_id IN (
      SELECT id::text FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid
    )
  GROUP BY reference_id,transaction_type
  HAVING count(*)>1
) duplicate;")
  missing_billings=$(database_query "
SELECT count(*)
FROM gateway_tasks task
WHERE task.acceptance_run_id='$run_id'::uuid
  AND NOT EXISTS (
    SELECT 1
    FROM gateway_wallet_transactions transaction
    WHERE transaction.reference_type='gateway_task'
      AND transaction.reference_id=task.id::text
      AND transaction.transaction_type='task_billing'
  );")
  [[ $wallet_duplicates == 0 && $missing_billings == 0 ]]
  callbacks=$(database_query "
SELECT count(*)
FROM gateway_task_callback_outbox
WHERE task_id IN (
  SELECT id FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid
)
  AND status <> 'delivered';")
  [[ $callbacks == 0 ]]
  active_workers=$(database_query "
SELECT count(*)
FROM gateway_worker_instances
WHERE status='active'
  AND heartbeat_at > now()-interval '30 seconds';")
  stale_workers=$(database_query "
SELECT count(*)
FROM gateway_worker_instances
WHERE status='active'
  AND (
    heartbeat_at <= now()-interval '30 seconds'
    OR desired_capacity <> $expected_slots
    OR capacity_limit <> $expected_slots
    OR allocated_capacity > $expected_slots
  );")
  [[ $active_workers == 2 && $stale_workers == 0 ]]
  connections=$(database_query "SELECT count(*) FROM pg_stat_activity WHERE backend_type='client backend';")
  max_connections=$(database_query "SELECT setting::int FROM pg_settings WHERE name='max_connections';")
  (( connections * 4 < max_connections * 3 ))

  local pod metrics desired current global allocated active_instances pool_max acquired idle
  local canceled_acquires lease_failures lease_lost
  for _ in 1 2 3 4 5 6; do
    for site in ningbo hongkong; do
      pod=$(remote_kubectl get pods -n "$namespace" \
        -l "app.kubernetes.io/name=easyai-worker,easyai.io/site=$site" \
        -o 'jsonpath={.items[0].metadata.name}')
      metrics=$(remote_kubectl exec -n "$namespace" "$pod" -- wget -qO- http://127.0.0.1:8088/metrics)
      desired=$(awk '$1=="easyai_gateway_async_worker_desired_capacity" {print $2}' <<<"$metrics")
      current=$(awk '$1=="easyai_gateway_async_worker_capacity" {print $2}' <<<"$metrics")
      global=$(awk '$1=="easyai_gateway_worker_global_capacity" {print $2}' <<<"$metrics")
      allocated=$(awk '$1=="easyai_gateway_worker_allocated_capacity" {print $2}' <<<"$metrics")
      active_instances=$(awk '$1=="easyai_gateway_worker_active_instances" {print $2}' <<<"$metrics")
      pool_max=$(awk '$1=="easyai_gateway_postgres_pool_max_connections" {print $2}' <<<"$metrics")
      acquired=$(awk '$1=="easyai_gateway_postgres_pool_acquired_connections" {print $2}' <<<"$metrics")
      idle=$(awk '$1=="easyai_gateway_postgres_pool_idle_connections" {print $2}' <<<"$metrics")
      canceled_acquires=$(awk '$1=="easyai_gateway_postgres_pool_canceled_acquire_total" {print $2}' <<<"$metrics")
      lease_failures=$(awk 'index($1,"easyai_gateway_concurrency_lease_renewals_total{outcome=\"failure\"}")==1 {print $2}' <<<"$metrics")
      lease_lost=$(awk 'index($1,"easyai_gateway_concurrency_lease_renewals_total{outcome=\"lost\"}")==1 {print $2}' <<<"$metrics")
      [[ ${desired%.*} == "$expected_slots" && ${current%.*} == "$expected_slots" ]]
      [[ ${global%.*} == "$expected_global" && ${allocated%.*} -le "$expected_slots" ]]
      [[ ${active_instances%.*} == 2 ]]
      [[ ${pool_max%.*} == "$expected_pool" ]]
      [[ ${canceled_acquires%.*} == 0 && ${lease_failures%.*} == 0 && ${lease_lost%.*} == 0 ]]
      if [[ ${acquired%.*} == "$expected_pool" && ${idle%.*} == 0 ]]; then
        echo "Worker connection pool saturated for site $site" >&2
        return 1
      fi
    done
    sleep 5
  done
  verify_cluster_links
  curl -fsS --max-time 10 https://ai.51easyai.com/api/v1/healthz >/dev/null
  curl -fsS --max-time 10 https://ai.51easyai.com/api/v1/readyz >/dev/null
}

sample_pressure() {
  local output=$1
  local stop_file=$2
  local database_state node_memory_percent pod_memory_mib pool_state pod metrics
  echo 'timestamp,queued,running,oldest_wait_seconds,db_connections,db_max_connections,node_memory_percent,pod_memory_mib,pool_acquired_max,pool_max,pool_idle_min,empty_acquire_total,canceled_acquire_total,lease_failure_total,lease_lost_total,active_instances_min,allocated_capacity_max' >"$output"
  while [[ ! -f $stop_file ]]; do
    database_state=$(database_query "
SELECT
  count(*) FILTER (WHERE status='queued')||','||
  count(*) FILTER (WHERE status='running')||','||
  COALESCE(EXTRACT(EPOCH FROM (now()-(min(created_at) FILTER (WHERE status='queued')))),0)::bigint||','||
  (SELECT count(*) FROM pg_stat_activity WHERE backend_type='client backend')||','||
  (SELECT setting FROM pg_settings WHERE name='max_connections')
FROM gateway_tasks
WHERE acceptance_run_id='$run_id'::uuid;") || break
    node_memory_percent=$(remote_kubectl top nodes --no-headers |
      awk '{value=$5; sub(/%$/, "", value); if (value>max) max=value} END {print max+0}') || break
    pod_memory_mib=$(current_max_pod_memory_mib) || break
    pool_state=$(
      for site in ningbo hongkong; do
        pod=$(remote_kubectl get pods -n "$namespace" \
          -l "app.kubernetes.io/name=easyai-worker,easyai.io/site=$site" -o json |
          jq -r '[.items[]
            | select(any(.status.conditions[]?; .type=="Ready" and .status=="True"))
            | .metadata.name][0] // empty') || exit
        [[ -n $pod ]] || continue
        metrics=$(remote_kubectl exec -n "$namespace" "$pod" -- \
          wget -qO- http://127.0.0.1:8088/metrics) || exit
        awk '
          $1=="easyai_gateway_postgres_pool_acquired_connections" {acquired=$2}
          $1=="easyai_gateway_postgres_pool_max_connections" {pool_max=$2}
          $1=="easyai_gateway_postgres_pool_idle_connections" {idle=$2}
          $1=="easyai_gateway_postgres_pool_empty_acquire_total" {empty=$2}
          $1=="easyai_gateway_postgres_pool_canceled_acquire_total" {canceled=$2}
          index($1,"easyai_gateway_concurrency_lease_renewals_total{outcome=\"failure\"}")==1 {failure=$2}
          index($1,"easyai_gateway_concurrency_lease_renewals_total{outcome=\"lost\"}")==1 {lost=$2}
          $1=="easyai_gateway_worker_active_instances" {active=$2}
          $1=="easyai_gateway_worker_allocated_capacity" {allocated=$2}
          END {print acquired "," pool_max "," idle "," empty "," canceled "," failure "," lost "," active "," allocated}
        ' <<<"$metrics"
      done |
        awk -F',' '
          NR==1 {
            acquired=$1; pool_max=$2; idle=$3; empty=$4; canceled=$5
            failure=$6; lost=$7; active=$8; allocated=$9
            next
          }
          {
            if ($1>acquired) acquired=$1
            if ($2>pool_max) pool_max=$2
            if ($3<idle) idle=$3
            empty+=$4
            canceled+=$5
            failure+=$6
            lost+=$7
            if ($8<active) active=$8
            if ($9>allocated) allocated=$9
          }
          END {print acquired "," pool_max "," idle "," empty "," canceled "," failure "," lost "," active "," allocated}
        '
    ) || break
    printf '%s,%s,%s,%s,%s\n' \
      "$(date -u '+%Y-%m-%dT%H:%M:%SZ')" "$database_state" "$node_memory_percent" "$pod_memory_mib" "$pool_state" >>"$output"
    sleep 5
  done
}

verify_pressure_report() {
  local report_path=$1
  local expected_slots=$2
  awk -F',' '
    NR == 1 { next }
    {
      samples++
      if (samples == 1) {
        first_empty=$12
        first_canceled=$13
        first_lease_failure=$14
        first_lease_lost=$15
      }
      last_empty=$12
      last_canceled=$13
      last_lease_failure=$14
      last_lease_lost=$15
      if ($13 > first_canceled || $14 > first_lease_failure || $15 > first_lease_lost) {
        failed=1
      }
      if ($9 >= $10 && $11 == 0) {
        saturated++
      } else {
        saturated=0
      }
      if ($4 >= 900 || $6 <= 0 || $5 * 4 >= $6 * 3 || $7 >= 80 || $8 >= 1536 ||
          $10 <= 0 || saturated >= 6 || $16 < 1 || $16 > 2 || $17 > slots) {
        failed=1
      }
    }
    END {
      if (last_canceled > first_canceled ||
          last_lease_failure > first_lease_failure ||
          last_lease_lost > first_lease_lost ||
          (last_empty > first_empty && saturated >= 6)) {
        failed=1
      }
      exit !(samples > 0 && failed == 0)
    }
  ' slots="$expected_slots" "$report_path"
}

run_capacity_round() {
  local profile=$1
  local repetition=$2
  local prefix=$report_root/${profile,,}-$repetition
  local pressure_report=$prefix-pressure.csv
  local pressure_stop=$temporary_root/pressure-stop
  local pressure_pid status=0
  local baseline_memory_mib
  local expected_slots
  case $profile in
    P24) expected_slots=24 ;;
    P28) expected_slots=28 ;;
    P32) expected_slots=32 ;;
  esac
  baseline_memory_mib=$(current_max_pod_memory_mib)
  rm -f -- "$pressure_stop"
  sample_pressure "$pressure_report" "$pressure_stop" &
  pressure_pid=$!
  run_load_profile gemini-baseline "$prefix-gemini-baseline.json" || status=$?
  if (( status == 0 )); then run_load_profile gemini-large "$prefix-gemini-large.json" || status=$?; fi
  if (( status == 0 )); then run_load_profile gemini-peak "$prefix-gemini-peak.json" || status=$?; fi
  if (( status == 0 )); then run_load_profile video-throughput "$prefix-video-throughput.json" || status=$?; fi
  if (( status == 0 )); then run_recovery_profile "$prefix-video-recovery.json" || status=$?; fi
  touch "$pressure_stop"
  wait "$pressure_pid" || true
  (( status == 0 )) || return "$status"
  verify_pressure_report "$pressure_report" "$expected_slots"
  verify_runtime_gates
  wait_for_memory_recovery "$baseline_memory_mib"
}

finish_and_promote() {
  local emulator_report=$report_root/emulator-report.json
  local finish_response=$temporary_root/finish.json
  local mode_response=$temporary_root/mode.json
  remote_kubectl exec -n "$namespace" deployment/easyai-acceptance-emulator -- \
    wget -qO- http://127.0.0.1:8090/report >"$emulator_report"
  local task_count video_task_count gemini_task_count submissions duplicates callbacks
  local gemini_requests forced_tasks forced_requests verified_conversions compact_payloads
  task_count=$(database_query "SELECT count(*) FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid;")
  video_task_count=$(database_query "SELECT count(*) FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid AND kind='videos.generations' AND run_mode='acceptance';")
  gemini_task_count=$(database_query "SELECT count(*) FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid AND kind='images.edits' AND run_mode='acceptance';")
  forced_tasks=$(database_query "SELECT count(*) FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid AND request::text LIKE '%acceptance-force-conversion%';")
  compact_payloads=$(database_query "SELECT count(*) FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid AND kind='images.edits' AND (pg_column_size(request) >= 1048576 OR pg_column_size(result) >= 1048576);")
  submissions=$(jq -r '.videoSubmissions' "$emulator_report")
  duplicates=$(jq -r '.duplicateCallbacks' "$emulator_report")
  callbacks=$(jq -r '.callbackEvents' "$emulator_report")
  gemini_requests=$(jq -r '.geminiRequests' "$emulator_report")
  forced_requests=$(jq -r '.forcedConversionRequests' "$emulator_report")
  verified_conversions=$(jq -r '.verifiedConversions' "$emulator_report")
  [[ $submissions == "$video_task_count" && $gemini_requests == "$gemini_task_count" ]]
  [[ $forced_requests == "$forced_tasks" && $verified_conversions == "$forced_tasks" ]]
  [[ $duplicates == 0 && $compact_payloads == 0 ]]
  [[ $callbacks -gt 0 ]]
  admin_request GET /api/admin/system/acceptance/traffic-mode '' "$mode_response"
  verify_release_cas
  local promotion_body
  promotion_body=$(jq -cn \
    --argjson revision "$(jq -r '.revision' "$mode_response")" \
    --arg releaseSha "$release_sha" \
    --arg apiDigest "$api_digest" \
    --arg workerDigest "$worker_digest" \
    '{revision:$revision,releaseSha:$releaseSha,apiImageDigest:$apiDigest,workerImageDigest:$workerDigest}')
  jq -n \
    --arg runId "$run_id" \
    --arg releaseSha "$release_sha" \
    --arg apiImageDigest "$api_digest" \
    --arg workerImageDigest "$worker_digest" \
    --arg stableCapacityProfile "$stable_profile" \
    --argjson tasks "$task_count" \
    --argjson geminiTasks "$gemini_task_count" \
    --argjson videoTasks "$video_task_count" \
    --argjson upstreamSubmissions "$submissions" \
    --argjson callbackDeliveries "$callbacks" \
    '{
      runId:$runId,
      releaseSha:$releaseSha,
      apiImageDigest:$apiImageDigest,
      workerImageDigest:$workerImageDigest,
      stableCapacityProfile:$stableCapacityProfile,
      tasks:$tasks,
      geminiTasks:$geminiTasks,
      videoTasks:$videoTasks,
      upstreamSubmissions:$upstreamSubmissions,
      callbackDeliveries:$callbackDeliveries,
      promotionRequested:true,
      passed:true
    }' >"$report_root/summary.json"
  local finish_body
  finish_body=$(jq -cn \
    --argjson tasks "$task_count" \
    --argjson videos "$video_task_count" \
    --argjson submissions "$submissions" \
    --arg profile "$stable_profile" \
    '{passed:true,report:{tasks:$tasks,videos:$videos,upstreamSubmissions:$submissions,stableCapacityProfile:$profile,queueFinal:0}}')
  admin_request POST "/api/admin/system/acceptance/runs/$run_id/finish" "$finish_body" "$finish_response"
  admin_request POST "/api/admin/system/acceptance/runs/$run_id/promote" "$promotion_body" "$finish_response"
  [[ $(jq -r '.mode' "$finish_response") == live ]]
}

verify_release_cas
select_acceptance_models
deploy_protocol_emulator
create_and_activate_run
wait_for_existing_tasks_to_drain
(
  cd "$cluster_root/apps/api"
  go build -trimpath -o "$load_binary" ./cmd/acceptance-load
)

for profile in P24 P28 P32; do
  if ! apply_capacity_profile "$profile"; then
    failure_reason="failed to apply $profile"
    break
  fi
  profile_passed=true
  for repetition in 1 2 3; do
    if ! run_capacity_round "$profile" "$repetition"; then
      profile_passed=false
      failure_reason="$profile repetition $repetition failed"
      break
    fi
  done
  if [[ $profile_passed != true ]]; then
    apply_capacity_profile "$stable_profile" || true
    break
  fi
  stable_profile=$profile
done

if [[ -n $failure_reason ]]; then
  mark_run_failed "$failure_reason"
  echo "production_acceptance=FAIL run_id=$run_id reason=$failure_reason traffic_mode=validation restored_profile=$stable_profile" >&2
  exit 1
fi

if ! run_load_profile real-canary "$report_root/real-canary.json"; then
  failure_reason='real canary failed'
elif ! verify_runtime_gates; then
  failure_reason='post-canary runtime gates failed'
fi
if [[ -n $failure_reason ]]; then
  mark_run_failed "$failure_reason"
  echo "production_acceptance=FAIL run_id=$run_id reason=$failure_reason traffic_mode=validation restored_profile=$stable_profile" >&2
  exit 1
fi
finish_and_promote
remote_kubectl delete deployment easyai-acceptance-emulator -n "$namespace" --ignore-not-found >/dev/null ||
  echo 'warning: acceptance emulator Deployment cleanup failed' >&2
remote_kubectl delete service easyai-acceptance-emulator -n "$namespace" --ignore-not-found >/dev/null ||
  echo 'warning: acceptance emulator Service cleanup failed' >&2
echo "production_acceptance=PASS run_id=$run_id release=$release_sha stable_profile=$stable_profile traffic_mode=live"