#!/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"
cluster_root=${cluster_root:?}

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

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. It also grants the dedicated acceptance API Key
access to the selected production candidates. A failed run deliberately keeps
traffic in validation mode and restores only the last stable capacity profile.
Successful execute runs also remain in validation. Only the separate --promote
action performs the final CAS switch to live after manual confirmation.
The explicit --skip-local-acceptance form records both local stages as skipped
with a user-directed waiver. It never records them as passed and preserves all
online simulation, real canary, resource, consistency, and release CAS gates.

Required private environment values:
  AI_GATEWAY_ACCEPTANCE_ADMIN_TOKEN, or AI_GATEWAY_ONLINE_ACCOUNT/PASSWORD

The dedicated acceptance identity, API Key, two node gateways, synthetic real
canary images, and production snapshot transport are bootstrapped when their
AI_GATEWAY_ACCEPTANCE_* overrides are omitted.

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
  AI_GATEWAY_ACCEPTANCE_IDENTITY_SHARDS
  AI_GATEWAY_ACCEPTANCE_API_MEDIA_REQUEST_CONCURRENCY
EOF
}

if [[ ${1:-} == --promote ]]; then
  exec "$script_dir/promote-production-acceptance.sh" "$@"
fi

skip_local_acceptance=false
local_acceptance_report=
if [[ ${1:-} == --execute && ${3:-} == --local-report && $# -eq 4 ]]; then
  release_manifest=$2
  local_acceptance_report=$4
elif [[ ${1:-} == --execute && ${3:-} == --skip-local-acceptance && $# -eq 3 ]]; then
  release_manifest=$2
  skip_local_acceptance=true
else
  usage >&2
  exit 64
fi
[[ -f $release_manifest && ! -L $release_manifest ]] || {
  echo 'release manifest must be a regular file' >&2
  exit 1
}
if [[ $skip_local_acceptance != true ]]; then
  [[ -f $local_acceptance_report && ! -L $local_acceptance_report ]] || {
    echo 'local acceptance report must be a regular file' >&2
    exit 1
  }
fi

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_API_DATABASE_MAX_CONNS:=31}"
: "${AI_GATEWAY_ACCEPTANCE_WORKER_RIVER_MAX_CONNS:=8}"
: "${AI_GATEWAY_ACCEPTANCE_API_RIVER_MAX_CONNS:=4}"
: "${AI_GATEWAY_ACCEPTANCE_API_MEDIA_REQUEST_CONCURRENCY:=128}"
: "${AI_GATEWAY_ACCEPTANCE_IDENTITY_SHARDS:=32}"
: "${AI_GATEWAY_ACCEPTANCE_WORKER_MEMORY_REQUEST_MIB:=1536}"
: "${AI_GATEWAY_ACCEPTANCE_WORKER_CPU_REQUEST_MILLICORES:=500}"
: "${AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_NINGBO:=2}"
: "${AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_HONGKONG:=2}"
: "${AI_GATEWAY_ACCEPTANCE_SOAK_DURATION:=2h}"
: "${AI_GATEWAY_ACCEPTANCE_OVERLOAD_DURATION:=10m}"
: "${AI_GATEWAY_ACCEPTANCE_GATEWAY_TLS_SERVER_NAME:=}"
: "${AI_GATEWAY_ACCEPTANCE_GATEWAYS:=}"
: "${AI_GATEWAY_ACCEPTANCE_REAL_IMAGE_URLS:=}"
: "${AI_GATEWAY_ACCEPTANCE_SNAPSHOT_DATABASE_URL:=}"
: "${AI_GATEWAY_ONLINE_ACCOUNT:=}"
: "${AI_GATEWAY_ONLINE_PASSWORD:=}"
: "${AI_GATEWAY_ONLINE_BASE_URL:=}"
if [[ -z $AI_GATEWAY_ACCEPTANCE_GATEWAYS ]]; then
  AI_GATEWAY_ACCEPTANCE_GATEWAYS="https://${CLUSTER_NINGBO_HOST#root@},https://${CLUSTER_HONGKONG_HOST#root@}"
  : "${AI_GATEWAY_DEPLOY_DOMAIN:?}"
  AI_GATEWAY_ACCEPTANCE_GATEWAY_TLS_SERVER_NAME=${AI_GATEWAY_ACCEPTANCE_GATEWAY_TLS_SERVER_NAME:-$AI_GATEWAY_DEPLOY_DOMAIN}
fi
[[ $AI_GATEWAY_ACCEPTANCE_API_DATABASE_MAX_CONNS =~ ^[1-9][0-9]*$ &&
   $AI_GATEWAY_ACCEPTANCE_API_DATABASE_MAX_CONNS -le 256 ]] || {
  echo 'AI_GATEWAY_ACCEPTANCE_API_DATABASE_MAX_CONNS must be between 1 and 256' >&2
  exit 1
}
[[ $AI_GATEWAY_ACCEPTANCE_WORKER_RIVER_MAX_CONNS =~ ^[1-9][0-9]*$ &&
   $AI_GATEWAY_ACCEPTANCE_API_RIVER_MAX_CONNS =~ ^[1-9][0-9]*$ &&
   $AI_GATEWAY_ACCEPTANCE_WORKER_RIVER_MAX_CONNS -lt 28 &&
   $AI_GATEWAY_ACCEPTANCE_API_RIVER_MAX_CONNS -lt $((AI_GATEWAY_ACCEPTANCE_API_DATABASE_MAX_CONNS - 4)) ]] || {
  echo 'acceptance River database pool sizes must be positive integers' >&2
  exit 1
}
[[ $AI_GATEWAY_ACCEPTANCE_API_MEDIA_REQUEST_CONCURRENCY =~ ^[1-9][0-9]*$ &&
   $AI_GATEWAY_ACCEPTANCE_API_MEDIA_REQUEST_CONCURRENCY -le 1024 ]] || {
  echo 'AI_GATEWAY_ACCEPTANCE_API_MEDIA_REQUEST_CONCURRENCY must be between 1 and 1024' >&2
  exit 1
}
[[ $AI_GATEWAY_ACCEPTANCE_IDENTITY_SHARDS =~ ^[1-9][0-9]*$ &&
   $AI_GATEWAY_ACCEPTANCE_IDENTITY_SHARDS -le 128 ]] || {
  echo 'AI_GATEWAY_ACCEPTANCE_IDENTITY_SHARDS must be between 1 and 128' >&2
  exit 1
}
[[ $AI_GATEWAY_ACCEPTANCE_WORKER_MEMORY_REQUEST_MIB =~ ^[1-9][0-9]*$ &&
   $AI_GATEWAY_ACCEPTANCE_WORKER_MEMORY_REQUEST_MIB -ge 256 &&
   $AI_GATEWAY_ACCEPTANCE_WORKER_MEMORY_REQUEST_MIB -le 1900 ]] || {
  echo 'AI_GATEWAY_ACCEPTANCE_WORKER_MEMORY_REQUEST_MIB must be between 256 and 1900' >&2
  exit 1
}
[[ $AI_GATEWAY_ACCEPTANCE_WORKER_CPU_REQUEST_MILLICORES =~ ^[1-9][0-9]*$ &&
   $AI_GATEWAY_ACCEPTANCE_WORKER_CPU_REQUEST_MILLICORES -ge 100 &&
   $AI_GATEWAY_ACCEPTANCE_WORKER_CPU_REQUEST_MILLICORES -le 1900 ]] || {
  echo 'AI_GATEWAY_ACCEPTANCE_WORKER_CPU_REQUEST_MILLICORES must be between 100 and 1900' >&2
  exit 1
}
[[ $AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_NINGBO =~ ^[1-9][0-9]*$ &&
   $AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_HONGKONG =~ ^[1-9][0-9]*$ &&
   $AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_NINGBO -le 16 &&
   $AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_HONGKONG -le 16 ]] || {
  echo 'acceptance autoscaling replica caps must be between 1 and 16' >&2
  exit 1
}

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)
base_sha=$(node "$cluster_root/scripts/release-manifest.mjs" get "$release_manifest" baseReleaseSha)
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
if [[ $skip_local_acceptance != true ]]; then
  node "$cluster_root/scripts/acceptance/report.mjs" validate \
    --input "$local_acceptance_report" >/dev/null
  [[ $(jq -r '.release.sourceSha' "$local_acceptance_report") == "$release_sha" &&
     $(jq -r '.release.apiImageDigest' "$local_acceptance_report") == "$api_digest" &&
     $(jq -r '.stages.localNative.status' "$local_acceptance_report") == passed &&
     $(jq -r '.stages.amd64Artifact.status' "$local_acceptance_report") == passed &&
     $(jq -r '.stages.onlineSimulation.status' "$local_acceptance_report") == not-run &&
     $(jq -r '.promotion.ready' "$local_acceptance_report") == false ]] || {
    echo 'local acceptance report does not match this release or has invalid stage state' >&2
    exit 1
  }
fi
[[ $(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
run_token=$(openssl rand -hex 32)
run_id=
report_root=
stable_profile=P24
active_profile=P24
failure_reason=
failure_gate_id=
failure_recorded=false
acceptance_admin_base=
acceptance_group_id=
acceptance_participants_json='[]'
AI_GATEWAY_ACCEPTANCE_API_KEYS=$AI_GATEWAY_ACCEPTANCE_API_KEY
acceptance_load_binary=$temporary_root/easyai-ai-gateway-acceptance-load
acceptance_snapshot_binary=$temporary_root/easyai-ai-gateway-acceptance-snapshot
current_production_snapshot=$temporary_root/current-production-snapshot.json
local_snapshot_config_hash=
local_snapshot_sha256=
if [[ $skip_local_acceptance != true ]]; then
  local_snapshot_config_hash=$(jq -r '.snapshot.configHash' "$local_acceptance_report")
  local_snapshot_sha256=$(jq -r '.snapshot.sha256' "$local_acceptance_report")
fi
active_load_pid=
active_pressure_pid=
capacity_profiles_payload=
image_stable_throughput=
image_admitted_throughput=
video_stable_throughput=
video_admitted_throughput=
certified_max_replicas_ningbo=1
certified_max_replicas_hongkong=1

cleanup() {
  local status=$?
  trap - EXIT
  if [[ -n $active_load_pid ]]; then
    kill "$active_load_pid" >/dev/null 2>&1 || true
    wait "$active_load_pid" >/dev/null 2>&1 || true
  fi
  if [[ -n $active_pressure_pid ]]; then
    kill "$active_pressure_pid" >/dev/null 2>&1 || true
    wait "$active_pressure_pid" >/dev/null 2>&1 || true
  fi
  if (( status != 0 )) && [[ -n $run_id && $failure_recorded != true ]]; then
    set +e
    apply_capacity_profile "$stable_profile" >/dev/null 2>&1
    [[ -n $failure_gate_id ]] || failure_gate_id=workflow_unexpected_exit
    mark_run_failed "${failure_reason:-acceptance workflow exited unexpectedly}"
    set -e
  fi
  rm -rf -- "$temporary_root"
  exit "$status"
}
trap cleanup EXIT
trap 'failure_gate_id=workflow_interrupted; failure_reason="production acceptance interrupted by signal"; exit 130' HUP INT TERM

bootstrap_acceptance_admin_token() {
  if [[ -n $AI_GATEWAY_ACCEPTANCE_ADMIN_TOKEN ]]; then
    echo 'acceptance_admin_identity=PASS source=explicit_token'
    return 0
  fi
  [[ -n $AI_GATEWAY_ONLINE_ACCOUNT && -n $AI_GATEWAY_ONLINE_PASSWORD &&
     $AI_GATEWAY_ONLINE_BASE_URL == https://* ]] || {
    echo 'acceptance admin token or HTTPS online account credentials are required' >&2
    return 1
  }
  local response=$temporary_root/admin-login.json
  local body status
  body=$(jq -cn \
    --arg account "$AI_GATEWAY_ONLINE_ACCOUNT" \
    --arg password "$AI_GATEWAY_ONLINE_PASSWORD" \
    '{account:$account,password:$password}')
  status=$(curl -sS --max-time 15 -o "$response" -w '%{http_code}' \
    -H 'Content-Type: application/json' --data-binary "$body" \
    "${AI_GATEWAY_ONLINE_BASE_URL%/}/api/v1/auth/login")
  chmod 0600 "$response"
  [[ $status == 200 &&
     $(jq '(.user.role // []) as $roles | if ($roles|type)=="array" then ($roles|index("manager")) != null else $roles == "manager" end' "$response") == true ]] || {
    echo "acceptance Manager login failed: status=$status" >&2
    return 1
  }
  AI_GATEWAY_ACCEPTANCE_ADMIN_TOKEN=$(jq -r '.accessToken // empty' "$response")
  [[ -n $AI_GATEWAY_ACCEPTANCE_ADMIN_TOKEN ]] || {
    echo 'acceptance Manager login did not return an access token' >&2
    return 1
  }
  export AI_GATEWAY_ACCEPTANCE_ADMIN_TOKEN
  echo 'acceptance_admin_identity=PASS source=short_lived_manager_login'
}

bootstrap_acceptance_admin_token

(
  cd "$cluster_root/apps/api"
  env -u AI_GATEWAY_TEST_DATABASE_URL go build \
    -trimpath -o "$acceptance_load_binary" ./cmd/acceptance-load
  env -u AI_GATEWAY_TEST_DATABASE_URL go build \
    -trimpath -o "$acceptance_snapshot_binary" ./cmd/acceptance-snapshot
)
export_production_snapshot() {
  local output=$1
  if [[ -n $AI_GATEWAY_ACCEPTANCE_SNAPSHOT_DATABASE_URL ]]; then
    AI_GATEWAY_DATABASE_URL="$AI_GATEWAY_ACCEPTANCE_SNAPSHOT_DATABASE_URL" \
      "$acceptance_snapshot_binary" export \
        --release-sha "$release_sha" \
        --output "$output" >/dev/null
  else
    cluster_ssh "$CLUSTER_NINGBO_HOST" bash -s -- "$release_sha" >"$output" <<'REMOTE'
set -euo pipefail
release_sha=$1
namespace=easyai
pod=$(k3s kubectl get pods -n "$namespace" \
  -l 'app.kubernetes.io/name=easyai-api,easyai.io/site=ningbo' \
  --field-selector status.phase=Running \
  -o 'jsonpath={.items[0].metadata.name}')
[[ $pod == easyai-api-ningbo-* ]]
[[ $(k3s kubectl get pod "$pod" -n "$namespace" \
  -o 'jsonpath={.status.containerStatuses[0].ready}') == true ]]
snapshot=/tmp/easyai-production-acceptance-snapshot-$release_sha.json
cleanup() {
  k3s kubectl exec -n "$namespace" "$pod" -- rm -f -- "$snapshot" >/dev/null 2>&1 || true
}
trap cleanup EXIT
k3s kubectl exec -n "$namespace" "$pod" -- \
  /app/easyai-ai-gateway-acceptance-snapshot export \
    --release-sha "$release_sha" --output "$snapshot" >/dev/null
k3s kubectl exec -n "$namespace" "$pod" -- cat "$snapshot"
REMOTE
  fi
  chmod 0600 "$output"
  "$acceptance_snapshot_binary" validate --input "$output" >/dev/null
}

export_production_snapshot "$current_production_snapshot"
if [[ $skip_local_acceptance == true ]]; then
  local_acceptance_report=$temporary_root/online-waiver-base.json
  node "$cluster_root/scripts/acceptance/report.mjs" build-online-waiver \
    --release-sha "$release_sha" \
    --api-digest "$api_digest" \
    --snapshot "$current_production_snapshot" \
    --waiver-reason user_directed_online_only_acceptance \
    --output "$local_acceptance_report" >/dev/null
  local_snapshot_config_hash=$(jq -r '.snapshot.configHash' "$local_acceptance_report")
  local_snapshot_sha256=$(jq -r '.snapshot.sha256' "$local_acceptance_report")
else
  [[ $(jq -r '.source.configHash' "$current_production_snapshot") == "$local_snapshot_config_hash" ]] || {
    echo 'production candidate configuration changed after the local snapshot; rerun local acceptance' >&2
    exit 1
  }
fi

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"
}

bootstrap_acceptance_primary_identity() {
  if [[ -n $AI_GATEWAY_ACCEPTANCE_API_KEY || -n $AI_GATEWAY_ACCEPTANCE_API_KEY_ID ||
        -n $AI_GATEWAY_ACCEPTANCE_USER_ID ]]; then
    [[ -n $AI_GATEWAY_ACCEPTANCE_API_KEY &&
       $AI_GATEWAY_ACCEPTANCE_API_KEY_ID =~ ^[0-9a-f-]{36}$ &&
       $AI_GATEWAY_ACCEPTANCE_USER_ID =~ ^[0-9a-f-]{36}$ ]] || {
      echo 'explicit acceptance primary identity variables must be provided together' >&2
      return 1
    }
    echo 'acceptance_primary_identity=PASS source=explicit'
    return 0
  fi
  local material
  material=$(database_query "
WITH primary_user AS (
  INSERT INTO gateway_users (
    user_key, source, username, display_name, roles, auth_profile, metadata, status
  )
  VALUES (
    'production-acceptance-primary', 'gateway', 'production-acceptance-primary',
    'Production Acceptance Primary', '[\"user\"]'::jsonb, '{}'::jsonb,
    '{\"purpose\":\"production_acceptance\",\"isolated\":true}'::jsonb, 'active'
  )
  ON CONFLICT (user_key) DO UPDATE
  SET source=EXCLUDED.source,
      username=EXCLUDED.username,
      display_name=EXCLUDED.display_name,
      roles=EXCLUDED.roles,
      auth_profile=EXCLUDED.auth_profile,
      metadata=EXCLUDED.metadata,
      status='active',
      deleted_at=NULL,
      updated_at=now()
  RETURNING id
), existing_key AS (
  SELECT key.id, key.gateway_user_id, key.key_secret
  FROM gateway_api_keys key
  JOIN primary_user ON primary_user.id=key.gateway_user_id
  WHERE key.name='Production Acceptance Primary'
    AND key.status='active'
    AND key.deleted_at IS NULL
    AND COALESCE(key.key_secret, '') <> ''
  ORDER BY key.created_at
  LIMIT 1
), new_key_material AS (
  SELECT primary_user.id AS gateway_user_id,
         'sk-gw-' || encode(gen_random_bytes(32), 'hex') AS secret
  FROM primary_user
  WHERE NOT EXISTS (SELECT 1 FROM existing_key)
), inserted_key AS (
  INSERT INTO gateway_api_keys (
    gateway_user_id, user_id, key_prefix, key_secret, key_hash, name, scopes,
    rate_limit_policy, quota_policy, metadata, status
  )
  SELECT material.gateway_user_id,
         material.gateway_user_id::text,
         left(material.secret, 18),
         material.secret,
         crypt(material.secret, gen_salt('bf', 10)),
         'Production Acceptance Primary',
         '[\"chat\",\"embedding\",\"rerank\",\"image\",\"image_vectorize\",\"video\",\"video_enhance\",\"music\",\"audio\",\"voice_clone\"]'::jsonb,
         '{}'::jsonb,
         '{}'::jsonb,
         '{\"purpose\":\"production_acceptance\",\"isolated\":true}'::jsonb,
         'active'
  FROM new_key_material material
  RETURNING id, gateway_user_id, key_secret
), selected_key AS (
  SELECT * FROM existing_key
  UNION ALL
  SELECT * FROM inserted_key
), wallet AS (
  INSERT INTO gateway_wallet_accounts (
    gateway_user_id, user_id, currency, balance, total_recharged, metadata, status
  )
  SELECT primary_user.id,
         primary_user.id::text,
         'resource',
         1000000000::numeric,
         1000000000::numeric,
         '{\"purpose\":\"production_acceptance\",\"isolated\":true}'::jsonb,
         'active'
  FROM primary_user
  ON CONFLICT (gateway_user_id, currency) DO UPDATE
  SET balance=GREATEST(gateway_wallet_accounts.balance, EXCLUDED.balance),
      total_recharged=GREATEST(gateway_wallet_accounts.total_recharged, EXCLUDED.total_recharged),
      metadata=EXCLUDED.metadata,
      status='active',
      updated_at=now()
  RETURNING gateway_user_id
)
SELECT primary_user.id::text || E'\\t' || selected_key.id::text || E'\\t' || selected_key.key_secret
FROM primary_user
CROSS JOIN selected_key
CROSS JOIN (SELECT count(*) FROM wallet) wallet_write_barrier;")
  [[ $material == *$'\t'* ]] || {
    echo 'acceptance primary identity bootstrap returned an invalid payload' >&2
    return 1
  }
  IFS=$'\t' read -r AI_GATEWAY_ACCEPTANCE_USER_ID \
    AI_GATEWAY_ACCEPTANCE_API_KEY_ID AI_GATEWAY_ACCEPTANCE_API_KEY <<<"$material"
  [[ $AI_GATEWAY_ACCEPTANCE_USER_ID =~ ^[0-9a-f-]{36}$ &&
     $AI_GATEWAY_ACCEPTANCE_API_KEY_ID =~ ^[0-9a-f-]{36}$ &&
     $AI_GATEWAY_ACCEPTANCE_API_KEY =~ ^sk-gw-[0-9a-f]{64}$ ]] || {
    echo 'acceptance primary identity bootstrap failed validation' >&2
    return 1
  }
  AI_GATEWAY_ACCEPTANCE_API_KEYS=$AI_GATEWAY_ACCEPTANCE_API_KEY
  export AI_GATEWAY_ACCEPTANCE_USER_ID AI_GATEWAY_ACCEPTANCE_API_KEY_ID
  export AI_GATEWAY_ACCEPTANCE_API_KEY AI_GATEWAY_ACCEPTANCE_API_KEYS
  echo 'acceptance_primary_identity=PASS source=idempotent_database_bootstrap isolated=true'
}

ensure_acceptance_real_images() {
  if [[ -n $AI_GATEWAY_ACCEPTANCE_REAL_IMAGE_URLS ]]; then
    [[ $(awk -F',' '{print NF}' <<<"$AI_GATEWAY_ACCEPTANCE_REAL_IMAGE_URLS") -ge 3 ]] || {
      echo 'explicit acceptance real image URLs must contain at least three entries' >&2
      return 1
    }
    echo 'acceptance_real_images=PASS source=explicit count=3_or_more'
    return 0
  fi
  [[ $AI_GATEWAY_ONLINE_BASE_URL == https://* ]] || {
    echo 'HTTPS online base URL is required to upload synthetic real canary images' >&2
    return 1
  }
  local fixture_root=$temporary_root/real-canary-images
  local response url index
  local -a urls=()
  install -d -m 0700 "$fixture_root"
  node "$cluster_root/scripts/acceptance/generate-reference-images.mjs" "$fixture_root" >/dev/null
  for index in 1 2 3; do
    response=$fixture_root/upload-$index.json
    chmod 0600 "$fixture_root/production-acceptance-reference-$index.png"
    {
      printf 'silent\nshow-error\nfail-with-body\nmax-time = 120\n'
      printf 'header = "Authorization: Bearer %s"\n' "$AI_GATEWAY_ACCEPTANCE_API_KEY"
      printf 'form = "source=production-acceptance"\n'
      printf 'form = "file=@%s;type=image/png"\n' \
        "$fixture_root/production-acceptance-reference-$index.png"
      printf 'url = "%s/api/v1/files/upload"\n' "${AI_GATEWAY_ONLINE_BASE_URL%/}"
    } | curl --config - >"$response"
    chmod 0600 "$response"
    url=$(jq -r '.url // .data.url // empty' "$response")
    if [[ $url == /* ]]; then
      url="${AI_GATEWAY_ONLINE_BASE_URL%/}$url"
    fi
    [[ $url == https://* ]] || {
      echo "synthetic acceptance image upload $index did not return an HTTPS URL" >&2
      return 1
    }
    urls+=("$url")
  done
  AI_GATEWAY_ACCEPTANCE_REAL_IMAGE_URLS=$(IFS=,; printf '%s' "${urls[*]}")
  export AI_GATEWAY_ACCEPTANCE_REAL_IMAGE_URLS
  echo 'acceptance_real_images=PASS source=synthetic_gateway_upload count=3 dimensions=512x512'
}

admin_request() {
  local method=$1
  local path=$2
  local body=${3:-}
  local output=$4
  local status response payload token_encoded body_encoded has_body=false
  local service_ip service_port remote_script remote_command
  if [[ -z $acceptance_admin_base ]]; then
    service_ip=$(remote_kubectl get service api-ningbo-edge -n "$namespace" \
      -o 'jsonpath={.spec.clusterIP}')
    service_port=$(remote_kubectl get service api-ningbo-edge -n "$namespace" \
      -o 'jsonpath={.spec.ports[0].port}')
    [[ $service_ip =~ ^[0-9.]+$ && $service_port =~ ^[0-9]+$ ]] || {
      echo 'acceptance admin Service has an invalid cluster address' >&2
      return 1
    }
    acceptance_admin_base=http://$service_ip:$service_port
  fi
  token_encoded=$(printf '%s' "$AI_GATEWAY_ACCEPTANCE_ADMIN_TOKEN" |
    openssl base64 -A)
  body_encoded=$(printf '%s' "$body" | openssl base64 -A)
  [[ -n $body ]] && has_body=true
  # shellcheck disable=SC2016 # This script is quoted for execution on the remote node.
  remote_script='
set -euo pipefail
admin_base=$1
method=$2
path=$3
token=$(printf "%s" "$4" | base64 -d)
body=$(printf "%s" "$5" | base64 -d)
has_body=$6
if [[ $has_body == true ]]; then
  curl -sS --max-time 30 -w "\n%{http_code}" -X "$method" \
    -H "Authorization: Bearer $token" \
    -H "Content-Type: application/json" \
    --data-binary "$body" "$admin_base$path"
else
  curl -sS --max-time 30 -w "\n%{http_code}" -X "$method" \
    -H "Authorization: Bearer $token" "$admin_base$path"
fi
'
  printf -v remote_command 'bash -c %q -- %q %q %q %q %q %q' \
    "$remote_script" "$acceptance_admin_base" "$method" "$path" \
    "$token_encoded" "$body_encoded" "$has_body"
  if ! response=$(cluster_ssh "$CLUSTER_NINGBO_HOST" "$remote_command"); then
    echo "acceptance control API transport failed: method=$method path=$path" >&2
    return 1
  fi
  status=${response##*$'\n'}
  payload=${response%$'\n'*}
  printf '%s' "$payload" >"$output"
  [[ $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 ]]
}

verify_resource_preconditions() {
  local report=$temporary_root/resource-preconditions.csv
  local ningbo_nodes los_angeles_nodes invalid_mixed_workloads witness_workloads
  ningbo_nodes=$(remote_kubectl get nodes -l 'easyai.io/site=ningbo' \
    -o jsonpath='{range .items[*]}{.metadata.name}{"\n"}{end}')
  los_angeles_nodes=$(remote_kubectl get nodes -l 'easyai.io/site=los-angeles' \
    -o jsonpath='{range .items[*]}{.metadata.name}{"\n"}{end}')
  [[ -n $ningbo_nodes && -n $los_angeles_nodes ]]
  invalid_mixed_workloads=$(remote_kubectl get pods -A -o json |
    jq --argjson nodes "$(jq -Rn --arg value "$ningbo_nodes" '$value | split("\n") | map(select(length>0))')" '
      [.items[]
        | select(.spec.nodeName as $name | $nodes | index($name))
        | select(.metadata.name | test("staging|runner|(^|-)ci(-|$)"; "i"))
        | select(any(.spec.containers[];
            (.resources.requests.memory // "") == "" or
            (.resources.limits.memory // "") == "" or
            (.resources.requests.cpu // "") == "" or
            (.resources.limits.cpu // "") == ""))]
      | length')
  [[ $invalid_mixed_workloads == 0 ]] || {
    echo 'Ningbo staging/CI workloads must have explicit CPU and memory requests/limits before acceptance' >&2
    return 1
  }
  witness_workloads=$(remote_kubectl get pods -A -o json |
    jq --argjson nodes "$(jq -Rn --arg value "$los_angeles_nodes" '$value | split("\n") | map(select(length>0))')" '
      [.items[]
        | select(.spec.nodeName as $name | $nodes | index($name))
        | select(
            .metadata.labels["app.kubernetes.io/name"] == "easyai-worker" or
            .metadata.labels["app.kubernetes.io/name"] == "easyai-acceptance-emulator" or
            .metadata.labels["app.kubernetes.io/name"] == "easyai-capacity-controller"
          )]
      | length')
  [[ $witness_workloads == 0 ]] || {
    echo 'Los Angeles witness node is carrying a Worker, controller, or acceptance emulator' >&2
    return 1
  }
  echo 'timestamp,node,memory_percent' >"$report"
  local sample node memory_percent
  for sample in $(seq 1 90); do
    while read -r node; do
      [[ -n $node ]] || continue
      memory_percent=$(remote_kubectl top node "$node" --no-headers |
        awk '{value=$5; sub(/%$/, "", value); print value}')
      [[ $memory_percent =~ ^[0-9]+$ && $memory_percent -lt 65 ]] || {
        echo "Ningbo memory must remain below 65% for 15 minutes before acceptance: node=$node memory_percent=${memory_percent:-unknown}" >&2
        return 1
      }
      printf '%s,%s,%s\n' "$(date -u '+%Y-%m-%dT%H:%M:%SZ')" "$node" "$memory_percent" >>"$report"
    done <<<"$ningbo_nodes"
    (( sample == 90 )) || sleep 10
  done
  cp "$report" "$report_root/resource-preconditions.csv"
}

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
}

snapshot_pre_acceptance_capacity() {
  local output=$report_root/pre-acceptance-capacity.json
  remote_kubectl get \
    deployment/easyai-api-ningbo \
    deployment/easyai-api-hongkong \
    deployment/easyai-worker-ningbo \
    deployment/easyai-worker-hongkong \
    deployment/easyai-capacity-controller \
    configmap/easyai-ai-gateway-config \
    configmap/easyai-gateway-release \
    -n "$namespace" -o json |
    jq '
      .items |= map(
        del(
          .metadata.creationTimestamp,
          .metadata.generation,
          .metadata.managedFields,
          .metadata.resourceVersion,
          .metadata.uid,
          .status
        )
      )
    ' >"$output"
  chmod 0600 "$output"
  jq -e '.kind == "List" and (.items | length) == 7' "$output" >/dev/null
}

wait_for_terminal_acceptance_tasks_to_drain() {
  local deadline=$((SECONDS + 300))
  local state active side_effects connections max_connections
  while (( SECONDS < deadline )); do
    state=$(database_query "
SELECT
  (SELECT count(*)
   FROM gateway_tasks task
   JOIN gateway_acceptance_runs run ON run.id=task.acceptance_run_id
   WHERE run.status IN ('failed', 'aborted')
     AND task.status IN ('queued', 'running'))||','||
  (
    (SELECT count(*)
     FROM settlement_outbox settlement
     JOIN gateway_tasks task ON task.id=settlement.task_id
     JOIN gateway_acceptance_runs run ON run.id=task.acceptance_run_id
     WHERE run.status IN ('failed', 'aborted')
       AND settlement.status IN ('pending', 'processing', 'retryable_failed'))
    +
    (SELECT count(*)
     FROM gateway_task_callback_outbox callback
     JOIN gateway_tasks task ON task.id=callback.task_id
     JOIN gateway_acceptance_runs run ON run.id=task.acceptance_run_id
     WHERE run.status IN ('failed', 'aborted')
       AND callback.status <> 'delivered')
  )||','||
  (SELECT count(*) FROM pg_stat_activity WHERE backend_type='client backend')||','||
  (SELECT setting FROM pg_settings WHERE name='max_connections');")
    IFS=',' read -r active side_effects connections max_connections <<<"$state"
    if [[ $active == 0 && $side_effects == 0 ]] &&
      (( connections * 2 < max_connections )); then
      return 0
    fi
    echo "waiting_for_terminal_acceptance_tasks=$active side_effects=$side_effects database_connections=$connections"
    sleep 2
  done
  echo "terminal acceptance state did not drain within 5 minutes: active=$active side_effects=$side_effects database_connections=$connections" >&2
  return 1
}

ensure_acceptance_user_group() {
  local groups_response=$temporary_root/user-groups.json
  local group_response=$temporary_root/acceptance-user-group.json
  local users_response=$temporary_root/users.json
  local user_response=$temporary_root/acceptance-user.json
  local group_id group_body user_body isolation_state
  admin_request GET /api/admin/user-groups '' "$groups_response"
  group_id=$(jq -r '
    [.items[] | select(.groupKey == "production-acceptance") | .id]
    | if length == 1 then .[0] else empty end
  ' "$groups_response")
  group_body=$(jq -cn '{
    groupKey:"production-acceptance",
    name:"Production Acceptance",
    description:"Dedicated production-isomorphic acceptance traffic",
    source:"gateway",
    priority:1,
    rechargeDiscountPolicy:{},
    billingDiscountPolicy:{},
    rateLimitPolicy:{rules:[]},
    quotaPolicy:{},
    metadata:{purpose:"production_acceptance",isolated:true},
    status:"active"
  }')
  if [[ -z $group_id ]]; then
    [[ $(jq '[.items[] | select(.groupKey == "production-acceptance")] | length' \
      "$groups_response") == 0 ]] || {
      echo 'acceptance user group lookup was ambiguous' >&2
      return 1
    }
    admin_request POST /api/admin/user-groups "$group_body" "$group_response"
    group_id=$(jq -r '.id // empty' "$group_response")
  else
    [[ $group_id =~ ^[0-9a-f-]{36}$ ]] || {
      echo 'acceptance user group API returned an invalid group ID' >&2
      return 1
    }
    isolation_state=$(database_query "
SELECT
  (SELECT count(*)
   FROM gateway_users
   WHERE default_user_group_id='$group_id'::uuid
     AND id <> '$AI_GATEWAY_ACCEPTANCE_USER_ID'::uuid
     AND COALESCE(metadata->>'purpose','') <> 'production_acceptance_shard')
  ||':'||
  (SELECT count(*)
   FROM gateway_api_keys key
   JOIN gateway_users user_record ON user_record.id=key.gateway_user_id
   WHERE key.user_group_id='$group_id'::uuid
     AND key.gateway_user_id <> '$AI_GATEWAY_ACCEPTANCE_USER_ID'::uuid
     AND COALESCE(user_record.metadata->>'purpose','') <> 'production_acceptance_shard');")
    [[ $isolation_state == 0:0 ]] || {
      echo 'acceptance user group is referenced by non-acceptance identities' >&2
      return 1
    }
    admin_request PATCH "/api/admin/user-groups/$group_id" \
      "$group_body" "$group_response"
  fi
  [[ $group_id =~ ^[0-9a-f-]{36}$ ]] || {
    echo 'acceptance user group API returned an invalid group ID' >&2
    return 1
  }
  acceptance_group_id=$group_id

  admin_request GET /api/admin/users '' "$users_response"
  user_body=$(jq -c \
    --arg userID "$AI_GATEWAY_ACCEPTANCE_USER_ID" \
    --arg groupID "$group_id" '
      [.items[] | select(.id == $userID)]
      | if length != 1 then empty else .[0] end
      | {
          userKey:(.userKey // ""),
          source:(.source // "gateway"),
          externalUserId:(.externalUserId // ""),
          username:(.username // ""),
          displayName:(.displayName // ""),
          email:(.email // ""),
          phone:(.phone // ""),
          avatarUrl:(.avatarUrl // ""),
          gatewayTenantId:(.gatewayTenantId // ""),
          tenantId:(.tenantId // ""),
          tenantKey:(.tenantKey // ""),
          defaultUserGroupId:$groupID,
          roles:(.roles // []),
          authProfile:(.authProfile // {}),
          metadata:(.metadata // {}),
          status:(.status // "active")
        }
    ' "$users_response")
  [[ -n $user_body && $(jq -r '.username // empty' <<<"$user_body") != '' ]] || {
    echo 'dedicated acceptance user was not found through the control API' >&2
    return 1
  }
  admin_request PATCH "/api/admin/users/$AI_GATEWAY_ACCEPTANCE_USER_ID" \
    "$user_body" "$user_response"

  isolation_state=$(database_query "
SELECT
  (SELECT count(*)
   FROM gateway_users
   WHERE id='$AI_GATEWAY_ACCEPTANCE_USER_ID'::uuid
     AND default_user_group_id='$group_id'::uuid)
  ||':'||
  (SELECT count(*)
   FROM gateway_api_keys
   WHERE id='$AI_GATEWAY_ACCEPTANCE_API_KEY_ID'::uuid
     AND gateway_user_id='$AI_GATEWAY_ACCEPTANCE_USER_ID'::uuid
     AND COALESCE(user_group_id,'$group_id'::uuid)='$group_id'::uuid
     AND status='active'
     AND deleted_at IS NULL)
  ||':'||
  (SELECT count(*)
   FROM gateway_api_keys
   WHERE gateway_user_id='$AI_GATEWAY_ACCEPTANCE_USER_ID'::uuid
     AND status='active'
     AND deleted_at IS NULL)
  ||':'||
  (SELECT count(*)
   FROM gateway_user_groups
   WHERE id='$group_id'::uuid
     AND status='active'
     AND NOT EXISTS (
       SELECT 1
       FROM jsonb_array_elements(
         COALESCE(rate_limit_policy->'rules','[]'::jsonb)
       ) rule
       WHERE rule->>'metric' IN ('concurrent','queue_size')
     ))
  ||':'||
  (SELECT count(*)
   FROM gateway_users
   WHERE default_user_group_id='$group_id'::uuid
     AND id <> '$AI_GATEWAY_ACCEPTANCE_USER_ID'::uuid
     AND COALESCE(metadata->>'purpose','') <> 'production_acceptance_shard')
  ||':'||
  (SELECT count(*)
   FROM gateway_api_keys key
   JOIN gateway_users user_record ON user_record.id=key.gateway_user_id
   WHERE key.user_group_id='$group_id'::uuid
     AND key.gateway_user_id <> '$AI_GATEWAY_ACCEPTANCE_USER_ID'::uuid
     AND COALESCE(user_record.metadata->>'purpose','') <> 'production_acceptance_shard');")
  [[ $isolation_state == 1:1:1:1:0:0 ]] || {
    echo 'dedicated acceptance identity isolation verification failed' >&2
    return 1
  }
  echo 'acceptance_user_group=PASS rate_limit=production_candidates active_keys=1 isolated=true'
}

ensure_acceptance_identity_shards() {
  local identity_material key_count participant_count
  [[ $acceptance_group_id =~ ^[0-9a-f-]{36}$ ]] || {
    echo 'acceptance user group must be prepared before identity shards' >&2
    return 1
  }
  identity_material=$(database_query "
WITH primary_user AS (
  SELECT roles
  FROM gateway_users
  WHERE id='$AI_GATEWAY_ACCEPTANCE_USER_ID'::uuid
    AND status='active'
    AND deleted_at IS NULL
), desired AS (
  SELECT ordinal,
         'production-acceptance-shard-' || lpad(ordinal::text, 3, '0') AS user_key
  FROM generate_series(1, $((AI_GATEWAY_ACCEPTANCE_IDENTITY_SHARDS - 1))) ordinal
), shard_users AS (
  INSERT INTO gateway_users (
    user_key, source, username, display_name, default_user_group_id,
    roles, auth_profile, metadata, status
  )
  SELECT desired.user_key,
         'gateway',
         desired.user_key,
         'Production Acceptance Shard ' || lpad(desired.ordinal::text, 3, '0'),
         '$acceptance_group_id'::uuid,
         primary_user.roles,
         '{}'::jsonb,
         jsonb_build_object(
           'purpose', 'production_acceptance_shard',
           'ordinal', desired.ordinal,
           'isolated', true
         ),
         'active'
  FROM desired
  CROSS JOIN primary_user
  ON CONFLICT (user_key) DO UPDATE
  SET default_user_group_id=EXCLUDED.default_user_group_id,
      roles=EXCLUDED.roles,
      metadata=EXCLUDED.metadata,
      status='active',
      deleted_at=NULL,
      updated_at=now()
  RETURNING id, user_key
), primary_key AS (
  SELECT id, gateway_user_id, key_secret, scopes, rate_limit_policy, quota_policy
  FROM gateway_api_keys
  WHERE id='$AI_GATEWAY_ACCEPTANCE_API_KEY_ID'::uuid
    AND gateway_user_id='$AI_GATEWAY_ACCEPTANCE_USER_ID'::uuid
    AND status='active'
    AND deleted_at IS NULL
    AND COALESCE(key_secret, '') <> ''
), existing_keys AS (
  SELECT selected.id, selected.gateway_user_id, selected.key_secret
  FROM shard_users shard
  JOIN LATERAL (
    SELECT key.id, key.gateway_user_id, key.key_secret
    FROM gateway_api_keys key
    WHERE key.gateway_user_id=shard.id
      AND key.name='Production Acceptance Shard'
      AND key.status='active'
      AND key.deleted_at IS NULL
      AND COALESCE(key.key_secret, '') <> ''
    ORDER BY key.created_at
    LIMIT 1
  ) selected ON true
), new_key_material AS (
  SELECT shard.id AS gateway_user_id,
         'sk-gw-' || encode(gen_random_bytes(32), 'hex') AS secret
  FROM shard_users shard
  WHERE NOT EXISTS (
    SELECT 1 FROM existing_keys existing
    WHERE existing.gateway_user_id=shard.id
  )
), inserted_keys AS (
  INSERT INTO gateway_api_keys (
    gateway_user_id, user_id, key_prefix, key_secret, key_hash, name, scopes,
    user_group_id, rate_limit_policy, quota_policy, status
  )
  SELECT material.gateway_user_id,
         material.gateway_user_id::text,
         left(material.secret, 18),
         material.secret,
         crypt(material.secret, gen_salt('bf', 10)),
         'Production Acceptance Shard',
         template.scopes,
         '$acceptance_group_id'::uuid,
         template.rate_limit_policy,
         template.quota_policy,
         'active'
  FROM new_key_material material
  CROSS JOIN primary_key template
  RETURNING id, gateway_user_id, key_secret
), shard_keys AS (
  SELECT * FROM existing_keys
  UNION ALL
  SELECT * FROM inserted_keys
), wallets AS (
  INSERT INTO gateway_wallet_accounts (
    gateway_user_id, user_id, currency, balance, total_recharged, metadata, status
  )
  SELECT shard.id,
         shard.id::text,
         'resource',
         1000000000::numeric,
         1000000000::numeric,
         '{\"purpose\":\"production_acceptance_shard\",\"isolated\":true}'::jsonb,
         'active'
  FROM shard_users shard
  ON CONFLICT (gateway_user_id, currency) DO UPDATE
  SET balance=GREATEST(gateway_wallet_accounts.balance, EXCLUDED.balance),
      total_recharged=GREATEST(gateway_wallet_accounts.total_recharged, EXCLUDED.total_recharged),
      metadata=EXCLUDED.metadata,
      status='active',
      updated_at=now()
  RETURNING gateway_user_id
), participants AS (
  SELECT 0 AS ordinal, key.id, key.gateway_user_id, key.key_secret
  FROM primary_key key
  UNION ALL
  SELECT row_number() OVER (ORDER BY shard.user_key)::int,
         key.id,
         key.gateway_user_id,
         key.key_secret
  FROM shard_keys key
  JOIN shard_users shard ON shard.id=key.gateway_user_id
)
SELECT jsonb_agg(
         jsonb_build_object(
           'apiKeyId', participant.id,
           'userId', participant.gateway_user_id
         )
         ORDER BY participant.ordinal
       )::text
       || E'\\t' ||
       string_agg(participant.key_secret, ',' ORDER BY participant.ordinal)
FROM participants participant
CROSS JOIN (SELECT count(*) FROM wallets) wallet_write_barrier;")
  [[ $identity_material == *$'\t'* ]] || {
    echo 'acceptance identity shard provisioning returned an invalid payload' >&2
    return 1
  }
  acceptance_participants_json=${identity_material%%$'\t'*}
  AI_GATEWAY_ACCEPTANCE_API_KEYS=${identity_material#*$'\t'}
  participant_count=$(jq 'length' <<<"$acceptance_participants_json")
  key_count=$(awk -F',' '{print NF}' <<<"$AI_GATEWAY_ACCEPTANCE_API_KEYS")
  [[ $participant_count == "$AI_GATEWAY_ACCEPTANCE_IDENTITY_SHARDS" &&
     $key_count == "$AI_GATEWAY_ACCEPTANCE_IDENTITY_SHARDS" ]] || {
    echo 'acceptance identity shard count verification failed' >&2
    return 1
  }
  export AI_GATEWAY_ACCEPTANCE_API_KEYS
  echo "acceptance_identity_shards=PASS count=$participant_count isolated_wallets=true"
}

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) = 'gemini-3.1-flash-image' THEN 0
    WHEN lower(b.invocation_name) LIKE '%gemini%image%'
      AND lower(b.invocation_name) NOT LIKE '%preview%' THEN 1
    WHEN lower(b.invocation_name) LIKE '%gemini%image%' THEN 2
    ELSE 3
  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
  }
  [[ $AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL =~ ^[A-Za-z0-9._:/+-]+$ &&
     $AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL =~ ^[A-Za-z0-9._:/+-]+$ ]] || {
    echo 'acceptance model invocation names contain unsupported characters' >&2
    return 1
  }
  export AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL
}

ensure_acceptance_model_access() {
  local resources response body missing expected
  response=$temporary_root/model-access.json
  resources=$(database_query "
WITH selected_candidates AS (
  SELECT DISTINCT platform.id AS platform_id, model.id AS platform_model_id, model.base_model_id
  FROM platform_models model
  JOIN integration_platforms platform ON platform.id=model.platform_id
  JOIN base_model_catalog base_model ON base_model.id=model.base_model_id
  WHERE platform.status='enabled'
    AND platform.deleted_at IS NULL
    AND model.enabled=true
    AND base_model.invocation_name IN (
      '$AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL',
      '$AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL'
    )
), resources AS (
  SELECT 'platform'::text AS resource_type, platform_id AS resource_id
  FROM selected_candidates
  UNION
  SELECT 'platform_model', platform_model_id
  FROM selected_candidates
  UNION
  SELECT 'base_model', base_model_id
  FROM selected_candidates
  WHERE base_model_id IS NOT NULL
)
SELECT COALESCE(
  jsonb_agg(
    jsonb_build_object(
      'resourceType', resource_type,
      'resourceId', resource_id,
      'priority', 100,
      'minPermissionLevel', 0,
      'status', 'active'
    )
    ORDER BY resource_type, resource_id
  ),
  '[]'::jsonb
)::text
FROM resources;")
  expected=$(jq 'length' <<<"$resources")
  (( expected > 0 )) || {
    echo 'selected acceptance models did not resolve to production candidate resources' >&2
    return 1
  }
  body=$(jq -cn \
    --arg subjectID "$acceptance_group_id" \
    --argjson resources "$resources" \
    '{
      subjectType:"user_group",
      subjectId:$subjectID,
      effect:"allow",
      upsertResources:$resources,
      deleteResources:[]
    }')
  admin_request POST /api/admin/access-rules/batch "$body" "$response"
  missing=$(database_query "
WITH selected_candidates AS (
  SELECT DISTINCT platform.id AS platform_id, model.id AS platform_model_id, model.base_model_id
  FROM platform_models model
  JOIN integration_platforms platform ON platform.id=model.platform_id
  JOIN base_model_catalog base_model ON base_model.id=model.base_model_id
  WHERE platform.status='enabled'
    AND platform.deleted_at IS NULL
    AND model.enabled=true
    AND base_model.invocation_name IN (
      '$AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL',
      '$AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL'
    )
), resources AS (
  SELECT 'platform'::text AS resource_type, platform_id AS resource_id
  FROM selected_candidates
  UNION
  SELECT 'platform_model', platform_model_id
  FROM selected_candidates
  UNION
  SELECT 'base_model', base_model_id
  FROM selected_candidates
  WHERE base_model_id IS NOT NULL
)
SELECT count(*)
FROM resources resource
WHERE NOT EXISTS (
  SELECT 1
  FROM gateway_access_rules rule
  WHERE rule.subject_type='user_group'
    AND rule.subject_id='$acceptance_group_id'::uuid
    AND rule.effect='allow'
    AND rule.status='active'
    AND rule.resource_type=resource.resource_type
    AND rule.resource_id=resource.resource_id
);")
  [[ $missing == 0 ]] || {
    echo "acceptance user group is missing $missing selected candidate access rules" >&2
    return 1
  }
  echo "acceptance_model_access=PASS resources=$expected"
}

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
  remote_kubectl rollout status deployment/easyai-acceptance-callback-collector \
    -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-callback-collector.easyai.svc.cluster.local:8091/callbacks' \
    --argjson participants "$acceptance_participants_json" \
    '{
      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"],
        participants: $participants
      }
    }')
  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
  local traffic_release traffic_api_digest traffic_worker_digest
  local previous_release previous_api_digest previous_worker_digest
  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
    }
    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
    }
    traffic_release=$(jq -r '.releaseSha // empty' "$traffic_response")
    traffic_api_digest=$(jq -r '.apiImageDigest // empty' "$traffic_response")
    traffic_worker_digest=$(jq -r '.workerImageDigest // empty' "$traffic_response")
    previous_release=$(jq -r '.releaseSha // empty' "$previous_run_response")
    previous_api_digest=$(jq -r '.apiImageDigest // empty' "$previous_run_response")
    previous_worker_digest=$(jq -r '.workerImageDigest // empty' "$previous_run_response")
    [[ $traffic_release == "$previous_release" &&
       $traffic_api_digest == "$previous_api_digest" &&
       $traffic_worker_digest == "$previous_worker_digest" ]] || {
      echo 'previous failed Run does not match the active validation CAS fields' >&2
      return 1
    }
    if [[ $traffic_release == "$release_sha" ]]; then
      [[ $traffic_api_digest == "$api_digest" &&
         $traffic_worker_digest == "$worker_digest" ]] || {
        echo 'current-release validation mode has unexpected image digests' >&2
        return 1
      }
    elif [[ -z $base_sha || $traffic_release != "$base_sha" ]]; then
      echo 'refusing to replace validation mode outside the direct release base' >&2
      return 1
    fi
    body=$(jq -cn \
      --argjson revision "$(jq -r '.revision' "$traffic_response")" \
      --arg releaseSha "$traffic_release" \
      --arg apiDigest "$traffic_api_digest" \
      --arg workerDigest "$traffic_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
    }
    wait_for_terminal_acceptance_tasks_to_drain
  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
  local artifact_names
  artifact_names=$(find "$report_root" -maxdepth 1 -type f -exec basename {} \; 2>/dev/null |
    LC_ALL=C sort | jq -Rsc 'split("\n") | map(select(length > 0))')
  body=$(jq -cn \
    --arg reason "$reason" \
    --arg gateId "${failure_gate_id:-unknown_gate}" \
    --arg activeProfile "$active_profile" \
    --argjson artifacts "$artifact_names" \
    '{
      passed:false,
      failureReason:$reason,
      report:{
        gateId:$gateId,
        activeCapacityProfile:$activeProfile,
        queueFinal:null,
        completedArtifacts:$artifacts
      }
    }')
  jq -n \
    --arg runId "$run_id" \
    --arg releaseSha "$release_sha" \
    --arg apiImageDigest "$api_digest" \
    --arg snapshotConfigHash "$local_snapshot_config_hash" \
    --arg snapshotSha256 "$local_snapshot_sha256" \
    --arg stableCapacityProfile "$stable_profile" \
    --arg gateId "${failure_gate_id:-unknown_gate}" \
    --arg reason "$reason" \
    --argjson artifacts "$artifact_names" \
    '{
      runId:$runId,
      releaseSha:$releaseSha,
      apiImageDigest:$apiImageDigest,
      snapshotConfigHash:$snapshotConfigHash,
      snapshotSha256:$snapshotSha256,
      stableCapacityProfile:$stableCapacityProfile,
      tasks:null,
      geminiTasks:null,
      videoTasks:null,
      realCanaryPassed:false,
      failureReason:$reason,
      completedArtifacts:$artifacts,
      gates:[
        {
          id:$gateId,
          passed:false,
          detail:$reason
        }
      ],
      passed:false
    }' >"$report_root/summary.partial.json"
  chmod 0600 "$report_root/summary.partial.json"
  node "$cluster_root/scripts/acceptance/report.mjs" merge-production \
    --local-report "$local_acceptance_report" \
    --production-summary "$report_root/summary.partial.json" \
    --output "$report_root/acceptance-report.partial.json" >/dev/null || true
  if admin_request POST "/api/admin/system/acceptance/runs/$run_id/finish" "$body" "$response"; then
    failure_recorded=true
  fi
}

fail_acceptance_gate() {
  failure_gate_id=$1
  failure_reason=$2
  echo "acceptance gate failed: gate_id=$failure_gate_id reason=$failure_reason" >&2
  return 1
}

apply_capacity_profile() {
  local profile=$1
  local slots pool media global
  local api_pool=$AI_GATEWAY_ACCEPTANCE_API_DATABASE_MAX_CONNS
  local min_idle=4
  local max_idle_seconds=30
  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
  local command_text capacity_output
  printf -v command_text \
    'AI_GATEWAY_WORKER_REPLICAS_NINGBO=1 AI_GATEWAY_WORKER_REPLICAS_HONGKONG=1 AI_GATEWAY_WORKER_MIN_REPLICAS_NINGBO=1 AI_GATEWAY_WORKER_MIN_REPLICAS_HONGKONG=1 AI_GATEWAY_WORKER_MAX_REPLICAS_NINGBO=1 AI_GATEWAY_WORKER_MAX_REPLICAS_HONGKONG=1 AI_GATEWAY_WORKER_AUTOSCALING_ENABLED=false AI_GATEWAY_ASYNC_WORKER_INSTANCE_HARD_LIMIT=%q AI_GATEWAY_ASYNC_WORKER_HARD_LIMIT=%q AI_GATEWAY_ASYNC_WORKER_GLOBAL_HARD_LIMIT=%q AI_GATEWAY_DATABASE_MAX_CONNS=%q AI_GATEWAY_API_DATABASE_MAX_CONNS=%q AI_GATEWAY_DATABASE_CRITICAL_MAX_CONNS=4 AI_GATEWAY_DATABASE_RIVER_MAX_CONNS=%q AI_GATEWAY_API_DATABASE_RIVER_MAX_CONNS=%q AI_GATEWAY_DATABASE_MIN_IDLE_CONNS=%q AI_GATEWAY_DATABASE_MAX_CONN_IDLE_SECONDS=%q AI_GATEWAY_MEDIA_MATERIALIZATION_CONCURRENCY=%q AI_GATEWAY_MEDIA_REQUEST_CONCURRENCY=%q AI_GATEWAY_WORKER_TARGET_OUTSTANDING_PER_REPLICA=%q AI_GATEWAY_WORKER_MEMORY_REQUEST_MIB=%q AI_GATEWAY_WORKER_CPU_REQUEST_MILLICORES=%q %q capacity' \
    "$slots" "$global" "$global" "$pool" "$api_pool" \
    "$AI_GATEWAY_ACCEPTANCE_WORKER_RIVER_MAX_CONNS" "$AI_GATEWAY_ACCEPTANCE_API_RIVER_MAX_CONNS" \
    "$min_idle" "$max_idle_seconds" \
    "$media" "$AI_GATEWAY_ACCEPTANCE_API_MEDIA_REQUEST_CONCURRENCY" "$((slots * 2))" \
    "$AI_GATEWAY_ACCEPTANCE_WORKER_MEMORY_REQUEST_MIB" \
    "$AI_GATEWAY_ACCEPTANCE_WORKER_CPU_REQUEST_MILLICORES" \
    "$remote_release_helper"
  capacity_output=$(cluster_ssh "$CLUSTER_NINGBO_HOST" "$command_text")
  grep -Fq 'production_capacity=PASS' <<<"$capacity_output"
  active_profile=$profile
}

capacity_controller_status() {
  local pod status
  while read -r pod; do
    [[ -n $pod ]] || continue
    if status=$(remote_kubectl exec -n "$namespace" "$pod" -- \
      wget -qO- http://127.0.0.1:8088/status 2>/dev/null) &&
      jq -e '.leader == true and (.lastError // "") == ""' >/dev/null <<<"$status"; then
      printf '%s\n' "$status"
      return 0
    fi
  done < <(remote_kubectl get pods -n "$namespace" \
    -l 'app.kubernetes.io/name=easyai-capacity-controller' \
    -o 'jsonpath={range .items[*]}{.metadata.name}{"\n"}{end}')
  return 1
}

apply_autoscaling_profile() {
  local profile=$1
  local max_replicas_ningbo=${2:-$AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_NINGBO}
  local max_replicas_hongkong=${3:-$AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_HONGKONG}
  local slots pool media global
  [[ $max_replicas_ningbo =~ ^[1-9][0-9]*$ &&
     $max_replicas_hongkong =~ ^[1-9][0-9]*$ ]] || return 1
  (( max_replicas_ningbo <= AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_NINGBO &&
     max_replicas_hongkong <= AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_HONGKONG )) || return 1
  case $profile in
    P24) slots=24; pool=32 ;;
    P28) slots=28; pool=36 ;;
    P32) slots=32; pool=40 ;;
    *) return 1 ;;
  esac
  media=$slots
  global=$((slots * (max_replicas_ningbo + max_replicas_hongkong)))
  local command_text capacity_output
  printf -v command_text \
    'AI_GATEWAY_WORKER_REPLICAS_NINGBO=1 AI_GATEWAY_WORKER_REPLICAS_HONGKONG=1 AI_GATEWAY_WORKER_MIN_REPLICAS_NINGBO=1 AI_GATEWAY_WORKER_MIN_REPLICAS_HONGKONG=1 AI_GATEWAY_WORKER_MAX_REPLICAS_NINGBO=%q AI_GATEWAY_WORKER_MAX_REPLICAS_HONGKONG=%q AI_GATEWAY_WORKER_AUTOSCALING_ENABLED=true AI_GATEWAY_WORKER_SCALE_UP_WINDOW_SECONDS=20 AI_GATEWAY_WORKER_SCALE_DOWN_STABILIZATION_SECONDS=600 AI_GATEWAY_WORKER_DRAIN_TIMEOUT_SECONDS=600 AI_GATEWAY_ASYNC_WORKER_INSTANCE_HARD_LIMIT=%q AI_GATEWAY_ASYNC_WORKER_HARD_LIMIT=%q AI_GATEWAY_ASYNC_WORKER_GLOBAL_HARD_LIMIT=%q AI_GATEWAY_DATABASE_MAX_CONNS=%q AI_GATEWAY_API_DATABASE_MAX_CONNS=%q AI_GATEWAY_DATABASE_CRITICAL_MAX_CONNS=4 AI_GATEWAY_DATABASE_RIVER_MAX_CONNS=%q AI_GATEWAY_API_DATABASE_RIVER_MAX_CONNS=%q AI_GATEWAY_DATABASE_MIN_IDLE_CONNS=4 AI_GATEWAY_DATABASE_MAX_CONN_IDLE_SECONDS=30 AI_GATEWAY_MEDIA_MATERIALIZATION_CONCURRENCY=%q AI_GATEWAY_MEDIA_REQUEST_CONCURRENCY=%q AI_GATEWAY_WORKER_TARGET_OUTSTANDING_PER_REPLICA=%q AI_GATEWAY_WORKER_MEMORY_REQUEST_MIB=%q AI_GATEWAY_WORKER_CPU_REQUEST_MILLICORES=%q %q capacity' \
    "$max_replicas_ningbo" \
    "$max_replicas_hongkong" \
    "$slots" "$global" "$global" "$pool" \
    "$AI_GATEWAY_ACCEPTANCE_API_DATABASE_MAX_CONNS" \
    "$AI_GATEWAY_ACCEPTANCE_WORKER_RIVER_MAX_CONNS" \
    "$AI_GATEWAY_ACCEPTANCE_API_RIVER_MAX_CONNS" \
    "$media" "$AI_GATEWAY_ACCEPTANCE_API_MEDIA_REQUEST_CONCURRENCY" "$((slots * 2))" \
    "$AI_GATEWAY_ACCEPTANCE_WORKER_MEMORY_REQUEST_MIB" \
    "$AI_GATEWAY_ACCEPTANCE_WORKER_CPU_REQUEST_MILLICORES" \
    "$remote_release_helper"
  capacity_output=$(cluster_ssh "$CLUSTER_NINGBO_HOST" "$command_text")
  grep -Fq 'production_capacity=PASS' <<<"$capacity_output"
}

autoscaling_phase_observed_max=2

run_autoscaling_topology_phase() {
  local phase=$1
  local target_ningbo=$2
  local target_hongkong=$3
  local observations=$4
  local load_report=$report_root/autoscaling-"$phase"-load.json
  local status deadline load_pid load_status=0
  local ningbo_replicas hongkong_replicas current_total
  local reached=false drained=false resource_max

  run_load_profile video-throughput "$load_report" &
  load_pid=$!
  active_load_pid=$load_pid
  deadline=$((SECONDS + 420))
  while (( SECONDS < deadline )); do
    status=$(capacity_controller_status || true)
    if [[ -n $status ]]; then
      ningbo_replicas=$(remote_kubectl get deployment easyai-worker-ningbo -n "$namespace" \
        -o 'jsonpath={.spec.replicas}')
      hongkong_replicas=$(remote_kubectl get deployment easyai-worker-hongkong -n "$namespace" \
        -o 'jsonpath={.spec.replicas}')
      current_total=$((ningbo_replicas + hongkong_replicas))
      (( current_total > autoscaling_phase_observed_max )) &&
        autoscaling_phase_observed_max=$current_total
      resource_max=$(jq '[.plan.sites[].resourceMax] | add // 0' <<<"$status")
      printf '%s,%s,%s,%s,%s,%s,%s,%s\n' \
        "$(date -u '+%Y-%m-%dT%H:%M:%SZ')" "$phase" \
        "$ningbo_replicas" "$hongkong_replicas" \
        "$(jq -r '.plan.rawDesired // 0' <<<"$status")" \
        "$(jq -r '.plan.desiredTotal // 0' <<<"$status")" \
        "$(jq -r '.plan.frozenReason // ""' <<<"$status")" "$resource_max" >>"$observations"
      if (( ningbo_replicas == target_ningbo && hongkong_replicas == target_hongkong )); then
        reached=true
        break
      fi
    fi
    kill -0 "$load_pid" >/dev/null 2>&1 || break
    sleep 5
  done
  wait "$load_pid" || load_status=$?
  active_load_pid=
  (( load_status == 0 )) || return "$load_status"
  [[ $reached == true ]] || {
    echo "capacity controller did not reach topology $target_ningbo+$target_hongkong during $phase" >&2
    return 1
  }

  deadline=$((SECONDS + 780))
  while (( SECONDS < deadline )); do
    ningbo_replicas=$(remote_kubectl get deployment easyai-worker-ningbo -n "$namespace" \
      -o 'jsonpath={.spec.replicas}')
    hongkong_replicas=$(remote_kubectl get deployment easyai-worker-hongkong -n "$namespace" \
      -o 'jsonpath={.spec.replicas}')
    status=$(capacity_controller_status || true)
    resource_max=$(jq '[.plan.sites[].resourceMax] | add // 0' <<<"${status:-{}}")
    printf '%s,%s,%s,%s,%s,%s,%s,%s\n' \
      "$(date -u '+%Y-%m-%dT%H:%M:%SZ')" "$phase-drain" \
      "$ningbo_replicas" "$hongkong_replicas" \
      "$(jq -r '.plan.rawDesired // 0' <<<"${status:-{}}")" \
      "$(jq -r '.plan.desiredTotal // 0' <<<"${status:-{}}")" \
      "$(jq -r '.plan.frozenReason // ""' <<<"${status:-{}}")" "$resource_max" >>"$observations"
    if (( ningbo_replicas == 1 && hongkong_replicas == 1 )); then
      drained=true
      break
    fi
    sleep 10
  done
  [[ $drained == true ]] || {
    echo "capacity controller did not safely drain $phase to topology 1+1" >&2
    return 1
  }
  local unsafe_residue
  unsafe_residue=$(database_query "
SELECT count(*)
FROM gateway_worker_instances worker
WHERE worker.status IN ('draining','stale')
  AND (
    EXISTS (
      SELECT 1 FROM gateway_tasks task
      WHERE task.locked_by=worker.instance_id AND task.status='running'
    )
    OR EXISTS (
      SELECT 1
      FROM gateway_tasks task
      JOIN gateway_concurrency_leases lease ON lease.task_id=task.id
      WHERE task.locked_by=worker.instance_id AND lease.released_at IS NULL
    )
  );")
  [[ $unsafe_residue == 0 ]] || {
    echo "capacity controller left task or lease residue after draining $phase" >&2
    return 1
  }
}

run_autoscaling_acceptance() {
  local report=$report_root/autoscaling.json
  local observations=$report_root/autoscaling-observations.csv
  local status deadline resource_max_hongkong resource_max_ningbo
  local tested_1_plus_2=false tested_2_plus_2=false
  local requested_hongkong_max=$AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_HONGKONG
  local requested_ningbo_max=$AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_NINGBO
  (( requested_hongkong_max > 2 )) && requested_hongkong_max=2
  (( requested_ningbo_max > 2 )) && requested_ningbo_max=2
  autoscaling_phase_observed_max=2
  echo 'timestamp,phase,ningbo_replicas,hongkong_replicas,raw_desired,desired_total,frozen_reason,resource_max_total' >"$observations"

  apply_autoscaling_profile "$stable_profile" 1 "$requested_hongkong_max"
  deadline=$((SECONDS + 180))
  while (( SECONDS < deadline )); do
    status=$(capacity_controller_status || true)
    [[ -n $status ]] && break
    sleep 2
  done
  [[ -n ${status:-} ]] || {
    echo 'capacity controller did not elect a healthy leader' >&2
    return 1
  }
  resource_max_hongkong=$(jq -r \
    '[.plan.sites[] | select(.site == "hongkong") | .resourceMax][0] // 0' <<<"$status")
  if (( requested_hongkong_max >= 2 && resource_max_hongkong >= 2 )); then
    run_autoscaling_topology_phase 1-plus-2 1 2 "$observations"
    tested_1_plus_2=true
    certified_max_replicas_hongkong=2
  fi

  if [[ $tested_1_plus_2 == true && $requested_ningbo_max -ge 2 ]]; then
    apply_autoscaling_profile "$stable_profile" 2 2
    deadline=$((SECONDS + 180))
    status=
    while (( SECONDS < deadline )); do
      status=$(capacity_controller_status || true)
      [[ -n $status ]] && break
      sleep 2
    done
    [[ -n $status ]]
    resource_max_ningbo=$(jq -r \
      '[.plan.sites[] | select(.site == "ningbo") | .resourceMax][0] // 0' <<<"$status")
    resource_max_hongkong=$(jq -r \
      '[.plan.sites[] | select(.site == "hongkong") | .resourceMax][0] // 0' <<<"$status")
    if (( resource_max_ningbo >= 2 && resource_max_hongkong >= 2 )); then
      run_autoscaling_topology_phase 2-plus-2 2 2 "$observations"
      tested_2_plus_2=true
      certified_max_replicas_ningbo=2
      certified_max_replicas_hongkong=2
    fi
  fi

  jq -n \
    --arg profile "$stable_profile" \
    --argjson testedOnePlusTwo "$tested_1_plus_2" \
    --argjson testedTwoPlusTwo "$tested_2_plus_2" \
    --argjson observedMaxReplicas "$autoscaling_phase_observed_max" \
    '{
      passed:true,
      drained:true,
      capacityProfile:$profile,
      topologies:{
        "1+1":true,
        "1+2":$testedOnePlusTwo,
        "2+2":$testedTwoPlusTwo
      },
      observedMaxReplicas:$observedMaxReplicas
    }' >"$report"
  apply_capacity_profile "$stable_profile"
  verify_runtime_gates
}

calculate_capacity_profiles() {
  local profile_slug instance_slots max_running config_hash
  local -a image_reports video_reports
  profile_slug=$(printf '%s' "$stable_profile" | tr '[:upper:]' '[:lower:]')
  image_reports=("$report_root"/"$profile_slug"-*-gemini-*.json)
  video_reports=("$report_root"/"$profile_slug"-*-video-*.json)
  [[ -f ${image_reports[0]} && -f ${video_reports[0]} ]]
  image_stable_throughput=$(jq -s '
    [.[] | .phases[]?
      | select(.completed > 0 and .durationMs > 0)
      | (.completed * 1000 / .durationMs)] | min
  ' "${image_reports[@]}")
  video_stable_throughput=$(jq -s '
    [.[] | .phases[]?
      | select(.completed > 0 and .durationMs > 0)
      | (.completed * 1000 / .durationMs)] | min
  ' "${video_reports[@]}")
  image_admitted_throughput=$(awk -v value="$image_stable_throughput" 'BEGIN {printf "%.6f", value * 0.8}')
  video_admitted_throughput=$(awk -v value="$video_stable_throughput" 'BEGIN {printf "%.6f", value * 0.8}')
  awk -v image="$image_admitted_throughput" -v video="$video_admitted_throughput" \
    'BEGIN {exit !(image > 0 && video > 0)}'
  case $stable_profile in
    P24) instance_slots=24 ;;
    P28) instance_slots=28 ;;
    P32) instance_slots=32 ;;
  esac
  max_running=$((instance_slots * (
    certified_max_replicas_ningbo + certified_max_replicas_hongkong
  )))
  config_hash=$(remote_kubectl get configmap easyai-gateway-release -n "$namespace" \
    -o 'jsonpath={.data.configHash}')
  [[ $config_hash =~ ^[0-9a-f]{64}$ ]]
  capacity_profiles_payload=$report_root/capacity-profiles.json
  jq -n \
    --arg configHash "$config_hash" \
    --arg capacityProfile "$stable_profile" \
    --arg imageModel "$AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL" \
    --arg videoModel "$AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL" \
    --argjson imageStable "$image_stable_throughput" \
    --argjson imageAdmitted "$image_admitted_throughput" \
    --argjson videoStable "$video_stable_throughput" \
    --argjson videoAdmitted "$video_admitted_throughput" \
    --argjson maxRunning "$max_running" \
    '{
      configHash:$configHash,
      capacityProfiles:[
        {
          kind:"image",
          model:$imageModel,
          stableThroughputPerSecond:$imageStable,
          admittedThroughputPerSecond:$imageAdmitted,
          maxRunning:$maxRunning,
          maxQueued:([($imageAdmitted * 60 | floor), $maxRunning] | max),
          maxPredictedWaitSeconds:60,
          metadata:{
            capacityProfile:$capacityProfile,
            repetitions:3,
            loadRunner:"external"
          }
        },
        {
          kind:"video",
          model:$videoModel,
          stableThroughputPerSecond:$videoStable,
          admittedThroughputPerSecond:$videoAdmitted,
          maxRunning:$maxRunning,
          maxQueued:([($videoAdmitted * 120 | floor), $maxRunning] | max),
          maxPredictedWaitSeconds:120,
          metadata:{
            capacityProfile:$capacityProfile,
            repetitions:3,
            loadRunner:"external"
          }
        }
      ]
    }' >"$capacity_profiles_payload"
}

stage_capacity_profiles() {
  local response=$temporary_root/staged-capacity-profiles.json
  calculate_capacity_profiles
  admin_request POST \
    "/api/admin/system/acceptance/runs/$run_id/capacity-profiles" \
    "$(cat "$capacity_profiles_payload")" \
    "$response"
  [[ $(jq -r '.status' "$response") == running ]]
}

run_mixed_load_profile() {
  local profile=$1
  local duration=$2
  local image_rate=$3
  local video_rate=$4
  local report_path=$5
  local timeout_value=$6
  local load_stdout=$temporary_root/load-$profile-stdout.json
  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_KEYS="$AI_GATEWAY_ACCEPTANCE_API_KEYS" \
  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" \
    "$acceptance_load_binary" \
      -profile "$profile" \
      -duration "$duration" \
      -image-rate "$image_rate" \
      -video-rate "$video_rate" \
      -timeout "$timeout_value" \
      -report "$report_path" >"$load_stdout"
  jq -e '.secretSafe == true and .passed == true' "$report_path" >/dev/null
}

run_mixed_soak_and_overload() {
  local historical_mix image_count video_count total_count image_weight video_weight total_rate
  local soak_image_rate soak_video_rate overload_image_rate overload_video_rate
  historical_mix=$(database_query "
SELECT
  count(*) FILTER (WHERE kind IN ('images.generations','images.edits','images.vectorize'))||','||
  count(*) FILTER (WHERE kind IN ('videos.generations','videos.upscales'))
FROM gateway_tasks
WHERE acceptance_run_id IS NULL
  AND created_at >= now()-interval '7 days';")
  IFS=',' read -r image_count video_count <<<"$historical_mix"
  total_count=$((image_count + video_count))
  if (( total_count < 100 || image_count == 0 || video_count == 0 )); then
    image_weight=0.5
    video_weight=0.5
  else
    image_weight=$(awk -v image="$image_count" -v total="$total_count" 'BEGIN {printf "%.6f", image/total}')
    video_weight=$(awk -v video="$video_count" -v total="$total_count" 'BEGIN {printf "%.6f", video/total}')
  fi
  total_rate=$(awk \
    -v image="$image_admitted_throughput" \
    -v video="$video_admitted_throughput" \
    -v image_weight="$image_weight" \
    -v video_weight="$video_weight" '
    BEGIN {
      image_capacity=image/image_weight
      video_capacity=video/video_weight
      if (image_capacity<video_capacity) print image_capacity
      else print video_capacity
    }')
  soak_image_rate=$(awk -v total="$total_rate" -v weight="$image_weight" 'BEGIN {printf "%.6f", total*weight}')
  soak_video_rate=$(awk -v total="$total_rate" -v weight="$video_weight" 'BEGIN {printf "%.6f", total*weight}')
  overload_image_rate=$(awk -v value="$soak_image_rate" 'BEGIN {printf "%.6f", value*1.2}')
  overload_video_rate=$(awk -v value="$soak_video_rate" 'BEGIN {printf "%.6f", value*1.2}')
  jq -n \
    --argjson imageCount "$image_count" \
    --argjson videoCount "$video_count" \
    --argjson imageWeight "$image_weight" \
    --argjson videoWeight "$video_weight" \
    --argjson soakImageRate "$soak_image_rate" \
    --argjson soakVideoRate "$soak_video_rate" \
    --argjson overloadImageRate "$overload_image_rate" \
    --argjson overloadVideoRate "$overload_video_rate" \
    '{
      historicalWindowDays:7,
      historicalImageTasks:$imageCount,
      historicalVideoTasks:$videoCount,
      imageWeight:$imageWeight,
      videoWeight:$videoWeight,
      soakImageRatePerSecond:$soakImageRate,
      soakVideoRatePerSecond:$soakVideoRate,
      overloadImageRatePerSecond:$overloadImageRate,
      overloadVideoRatePerSecond:$overloadVideoRate
    }' >"$report_root/mixed-workload-profile.json"

  local pressure_report=$report_root/mixed-soak-pressure.csv
  local pressure_stop=$temporary_root/mixed-pressure-stop
  local pressure_pid pressure_status=0 status=0
  local expected_slots
  case $stable_profile in
    P24) expected_slots=24 ;;
    P28) expected_slots=28 ;;
    P32) expected_slots=32 ;;
  esac
  rm -f -- "$pressure_stop"
  sample_pressure "$pressure_report" "$pressure_stop" &
  pressure_pid=$!
  active_pressure_pid=$pressure_pid
  run_mixed_load_profile \
    mixed-soak "$AI_GATEWAY_ACCEPTANCE_SOAK_DURATION" \
    "$soak_image_rate" "$soak_video_rate" \
    "$report_root/mixed-soak.json" 3h || status=$?
  if (( status == 0 )); then
    run_mixed_load_profile \
      mixed-overload "$AI_GATEWAY_ACCEPTANCE_OVERLOAD_DURATION" \
      "$overload_image_rate" "$overload_video_rate" \
      "$report_root/mixed-overload.json" 30m || status=$?
  fi
  touch "$pressure_stop"
  wait "$pressure_pid" || pressure_status=$?
  active_pressure_pid=
  (( status == 0 && pressure_status == 0 )) || {
    (( status != 0 )) && return "$status"
    return "$pressure_status"
  }
  jq -e '.phases[0].throttled == 0 and .phases[0].unexpected5xx == 0' \
    "$report_root/mixed-soak.json" >/dev/null
  jq -e '.phases[0].throttled > 0 and .phases[0].unexpected5xx == 0' \
    "$report_root/mixed-overload.json" >/dev/null
  verify_pressure_report "$pressure_report" "$expected_slots"
  verify_runtime_gates
}

activate_certified_autoscaling() {
  local deadline status resource_max
  apply_autoscaling_profile \
    "$stable_profile" \
    "$certified_max_replicas_ningbo" \
    "$certified_max_replicas_hongkong"
  deadline=$((SECONDS + 180))
  while (( SECONDS < deadline )); do
    status=$(capacity_controller_status || true)
    if [[ -n $status ]]; then
      resource_max=$(jq '[.plan.sites[].resourceMax] | add // 0' <<<"$status")
      if (( resource_max >= 2 )); then
        break
      fi
    fi
    sleep 2
  done
  [[ -n ${status:-} && ${resource_max:-0} -ge 2 ]]
  certified_max_replicas_ningbo=$(jq -r \
    '[.plan.sites[] | select(.site == "ningbo") | .resourceMax][0] // 0' <<<"$status")
  certified_max_replicas_hongkong=$(jq -r \
    '[.plan.sites[] | select(.site == "hongkong") | .resourceMax][0] // 0' <<<"$status")
  (( certified_max_replicas_ningbo >= 1 && certified_max_replicas_hongkong >= 1 ))
  jq -n \
    --arg profile "$stable_profile" \
    --argjson ningbo "$certified_max_replicas_ningbo" \
    --argjson hongkong "$certified_max_replicas_hongkong" \
    --argjson total "$resource_max" \
    '{
      capacityProfile:$profile,
      sites:{ningbo:{resourceMax:$ningbo},hongkong:{resourceMax:$hongkong}},
      resourceMaxTotal:$total
    }' >"$report_root/certified-site-capacity.json"
  [[ $(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 ]]
  stage_capacity_profiles
  verify_runtime_gates
}

run_load_profile() {
  local profile=$1
  local report_path=$2
  [[ $profile =~ ^[a-z-]+$ ]] || return 1
  local load_stdout=$temporary_root/load-$profile-stdout.json
  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_KEYS="$AI_GATEWAY_ACCEPTANCE_API_KEYS" \
  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" \
    "$acceptance_load_binary" -profile "$profile" -report "$report_path" \
      >"$load_stdout"
  jq -e '.secretSafe == true and .passed == true' "$report_path" >/dev/null
}

kill_one_worker_during_recovery() {
  local deadline=$((SECONDS + 120))
  local owner pod site task_id
  while (( SECONDS < deadline )); do
    owner=$(database_query "
SELECT task.id::text||','||worker.pod_name||','||worker.site
FROM gateway_tasks task
JOIN gateway_worker_instances worker ON worker.instance_id=task.locked_by
WHERE task.acceptance_run_id='$run_id'::uuid
  AND task.status='running'
  AND task.request::text LIKE '%acceptance-long-recovery%'
  AND worker.status='active'
  AND worker.heartbeat_at > now()-interval '30 seconds'
ORDER BY task.updated_at ASC, task.id ASC
LIMIT 1;")
    if [[ -n $owner ]]; then
      break
    fi
    sleep 1
  done
  [[ -n $owner ]] || {
    echo 'recovery workload never obtained a live Worker owner' >&2
    return 1
  }
  IFS=',' read -r task_id pod site <<<"$owner"
  [[ $task_id =~ ^[0-9a-f-]{36}$ ]]
  [[ $site == ningbo || $site == hongkong ]]
  [[ $pod =~ ^easyai-worker-$site-[a-z0-9-]+$ ]]
  [[ $(remote_kubectl get pod "$pod" -n "$namespace" \
    -o 'jsonpath={.metadata.uid}') != "" ]]
  printf 'worker_failure_target_task=%s pod=%s site=%s\n' "$task_id" "$pod" "$site"
  remote_kubectl delete pod "$pod" -n "$namespace" --wait=false >/dev/null
  remote_kubectl rollout status "deployment/easyai-worker-$site" \
    -n "$namespace" --timeout=300s
}

run_recovery_profile() {
  local report_path=$1
  run_load_profile video-recovery "$report_path" &
  local load_pid=$!
  active_load_pid=$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"
  active_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 -i 0.2 -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 workload 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
  for workload in api worker; do
    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-$workload" \
      -o jsonpath='{range .items[*]}{.metadata.name}{"\n"}{end}')
  done
  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}'
}

current_max_worker_resources() {
  remote_kubectl top pods -n "$namespace" \
    -l 'app.kubernetes.io/name=easyai-worker' --no-headers |
    awk '{
      cpu=$2
      memory=$3
      if (cpu ~ /n$/) {sub(/n$/, "", cpu); cpu/=1000000}
      else if (cpu ~ /u$/) {sub(/u$/, "", cpu); cpu/=1000}
      else if (cpu ~ /m$/) {sub(/m$/, "", cpu)}
      else {cpu*=1000}
      if (memory ~ /Gi$/) {sub(/Gi$/, "", memory); memory*=1024}
      else if (memory ~ /Mi$/) {sub(/Mi$/, "", memory)}
      else if (memory ~ /Ki$/) {sub(/Ki$/, "", memory); memory/=1024}
      if (cpu>max_cpu) max_cpu=cpu
      if (memory>max_memory) max_memory=memory
    } END {printf "%.0f,%.0f", max_memory+0, max_cpu+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 expected_worker_execution_pool expected_api_execution_pool
  local expected_critical_pool=4
  local expected_api_pool=$AI_GATEWAY_ACCEPTANCE_API_DATABASE_MAX_CONNS
  local expected_worker_river_pool=$AI_GATEWAY_ACCEPTANCE_WORKER_RIVER_MAX_CONNS
  local expected_api_river_pool=$AI_GATEWAY_ACCEPTANCE_API_RIVER_MAX_CONNS
  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
  expected_worker_execution_pool=$((expected_pool - expected_critical_pool - expected_worker_river_pool))
  expected_api_execution_pool=$((expected_api_pool - expected_critical_pool - expected_api_river_pool))
  [[ $(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 critical_pool_max critical_acquired critical_idle
  local river_pool_max river_acquired river_idle
  local lease_failures lease_lost
  for site in ningbo hongkong; do
    pod=$(remote_kubectl get pods -n "$namespace" \
      -l "app.kubernetes.io/name=easyai-api,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)
    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")
    critical_pool_max=$(awk '$1=="easyai_gateway_postgres_critical_pool_max_connections" {print $2}' <<<"$metrics")
    river_pool_max=$(awk '$1=="easyai_gateway_postgres_river_pool_max_connections" {print $2}' <<<"$metrics")
    river_acquired=$(awk '$1=="easyai_gateway_postgres_river_pool_acquired_connections" {print $2}' <<<"$metrics")
    river_idle=$(awk '$1=="easyai_gateway_postgres_river_pool_idle_connections" {print $2}' <<<"$metrics")
    [[ ${pool_max%.*} == "$expected_api_execution_pool" ]]
    [[ ${critical_pool_max%.*} == "$expected_critical_pool" ]]
    [[ ${river_pool_max%.*} == "$expected_api_river_pool" ]]
    if [[ ${acquired%.*} == "$expected_api_execution_pool" && ${idle%.*} == 0 ]]; then
      echo "API connection pool saturated for site $site" >&2
      return 1
    fi
    if [[ ${river_acquired%.*} == "$expected_api_river_pool" && ${river_idle%.*} == 0 ]]; then
      echo "API River connection pool saturated for site $site" >&2
      return 1
    fi
  done
  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")
      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_global" && ${current%.*} == "$expected_slots" ]]
      [[ ${global%.*} == "$expected_global" && ${allocated%.*} -le "$expected_slots" ]]
      [[ ${active_instances%.*} == 2 ]]
      [[ ${pool_max%.*} == "$expected_worker_execution_pool" ]]
      critical_pool_max=$(awk '$1=="easyai_gateway_postgres_critical_pool_max_connections" {print $2}' <<<"$metrics")
      critical_acquired=$(awk '$1=="easyai_gateway_postgres_critical_pool_acquired_connections" {print $2}' <<<"$metrics")
      critical_idle=$(awk '$1=="easyai_gateway_postgres_critical_pool_idle_connections" {print $2}' <<<"$metrics")
      river_pool_max=$(awk '$1=="easyai_gateway_postgres_river_pool_max_connections" {print $2}' <<<"$metrics")
      river_acquired=$(awk '$1=="easyai_gateway_postgres_river_pool_acquired_connections" {print $2}' <<<"$metrics")
      river_idle=$(awk '$1=="easyai_gateway_postgres_river_pool_idle_connections" {print $2}' <<<"$metrics")
      [[ ${critical_pool_max%.*} == "$expected_critical_pool" ]]
      [[ ${river_pool_max%.*} == "$expected_worker_river_pool" ]]
      [[ ${lease_failures%.*} == 0 && ${lease_lost%.*} == 0 ]]
      if [[ ${acquired%.*} == "$expected_worker_execution_pool" && ${idle%.*} == 0 ]]; then
        echo "Worker connection pool saturated for site $site" >&2
        return 1
      fi
      if [[ ${critical_acquired%.*} == "$expected_critical_pool" && ${critical_idle%.*} == 0 ]]; then
        echo "Worker critical connection pool saturated for site $site" >&2
        return 1
      fi
      if [[ ${river_acquired%.*} == "$expected_worker_river_pool" && ${river_idle%.*} == 0 ]]; then
        echo "Worker River 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 worker_resources pool_state pod metrics
  local consecutive_failures=0
  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,critical_pool_acquired_max,critical_pool_max,critical_pool_idle_min,critical_empty_acquire_total,critical_canceled_acquire_total,river_pool_acquired_max,river_pool_max,river_pool_idle_min,river_empty_acquire_total,river_canceled_acquire_total,worker_memory_mib,worker_cpu_millicores' >"$output"
  while [[ ! -f $stop_file ]]; do
    if ! 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;"); then
      database_state=
    fi
    if ! node_memory_percent=$(remote_kubectl top nodes --no-headers 2>/dev/null |
      awk '{value=$5; sub(/%$/, "", value); if (value>max) max=value} END {print max+0}'); then
      node_memory_percent=
    fi
    if ! pod_memory_mib=$(current_max_pod_memory_mib 2>/dev/null); then
      pod_memory_mib=
    fi
    if ! worker_resources=$(current_max_worker_resources 2>/dev/null); then
      worker_resources=
    fi
    if ! pool_state=$(
      for workload in api worker; do
        for site in ningbo hongkong; do
          pod=$(remote_kubectl get pods -n "$namespace" \
            -l "app.kubernetes.io/name=easyai-$workload,easyai.io/site=$site" -o json |
            jq -r '[.items[]
              | select(.metadata.deletionTimestamp == null)
              | 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 -v workload="$workload" '
            $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}
            $1=="easyai_gateway_postgres_critical_pool_acquired_connections" {critical_acquired=$2}
            $1=="easyai_gateway_postgres_critical_pool_max_connections" {critical_max=$2}
            $1=="easyai_gateway_postgres_critical_pool_idle_connections" {critical_idle=$2}
            $1=="easyai_gateway_postgres_critical_pool_empty_acquire_total" {critical_empty=$2}
            $1=="easyai_gateway_postgres_critical_pool_canceled_acquire_total" {critical_canceled=$2}
            $1=="easyai_gateway_postgres_river_pool_acquired_connections" {river_acquired=$2}
            $1=="easyai_gateway_postgres_river_pool_max_connections" {river_max=$2}
            $1=="easyai_gateway_postgres_river_pool_idle_connections" {river_idle=$2}
            $1=="easyai_gateway_postgres_river_pool_empty_acquire_total" {river_empty=$2}
            $1=="easyai_gateway_postgres_river_pool_canceled_acquire_total" {river_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 {
              if (workload=="api") {
                failure=0
                lost=0
                active=999999
                allocated=0
              }
              print acquired "," pool_max "," idle "," empty "," canceled "," failure "," lost "," active "," allocated "," critical_acquired "," critical_max "," critical_idle "," critical_empty "," critical_canceled "," river_acquired "," river_max "," river_idle "," river_empty "," river_canceled
            }
          ' <<<"$metrics"
        done
      done |
        awk -F',' '
          NR==1 {
            acquired=$1; pool_max=$2; idle=$3; empty=$4; canceled=$5
            failure=$6; lost=$7; active=$8; allocated=$9
            critical_acquired=$10; critical_max=$11; critical_idle=$12
            critical_empty=$13; critical_canceled=$14
            river_acquired=$15; river_max=$16; river_idle=$17
            river_empty=$18; river_canceled=$19
            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
            if ($10>critical_acquired) critical_acquired=$10
            if ($11>critical_max) critical_max=$11
            if ($12<critical_idle) critical_idle=$12
            critical_empty+=$13
            critical_canceled+=$14
            if ($15>river_acquired) river_acquired=$15
            if ($16>river_max) river_max=$16
            if ($17<river_idle) river_idle=$17
            river_empty+=$18
            river_canceled+=$19
          }
          END {
            if (critical_idle==999999) critical_idle=0
            print acquired "," pool_max "," idle "," empty "," canceled "," failure "," lost "," active "," allocated "," critical_acquired "," critical_max "," critical_idle "," critical_empty "," critical_canceled "," river_acquired "," river_max "," river_idle "," river_empty "," river_canceled
          }
        '
    ) 2>/dev/null; then
      pool_state=
    fi
    if [[ -z $database_state || -z $node_memory_percent ||
          -z $pod_memory_mib || -z $worker_resources || -z $pool_state ]]; then
      ((consecutive_failures += 1))
      if (( consecutive_failures >= 3 )); then
        echo 'pressure sampling failed three consecutive times' >&2
        return 1
      fi
      sleep 2
      continue
    fi
    consecutive_failures=0
    printf '%s,%s,%s,%s,%s,%s\n' \
      "$(date -u '+%Y-%m-%dT%H:%M:%SZ')" "$database_state" "$node_memory_percent" \
      "$pod_memory_mib" "$pool_state" "$worker_resources" >>"$output"
    sleep 5
  done
}

verify_pressure_report() {
  local report_path=$1
  local expected_slots=$2
  awk -F',' '
    NR == 1 { next }
    {
      samples++
      if ($14 > 0 || $15 > 0) {
        failed=1
      }
      if ($9 >= $10 && $11 == 0) {
        saturated++
      } else {
        saturated=0
      }
      if ($18 >= $19 && $20 == 0) {
        critical_saturated++
      } else {
        critical_saturated=0
      }
      if ($23 >= $24 && $25 == 0) {
        river_saturated++
      } else {
        river_saturated=0
      }
      if (samples > 1 && $12 > previous_empty) {
        empty_growth++
      } else {
        empty_growth=0
      }
      if (samples > 1 && $21 > previous_critical_empty) {
        critical_empty_growth++
      } else {
        critical_empty_growth=0
      }
      if (samples > 1 && $26 > previous_river_empty) {
        river_empty_growth++
      } else {
        river_empty_growth=0
      }
      if (samples > 1 && ($13 > previous_canceled ||
          $22 > previous_critical_canceled || $27 > previous_river_canceled)) {
        failed=1
      }
      previous_empty=$12
      previous_canceled=$13
      previous_critical_empty=$21
      previous_critical_canceled=$22
      previous_river_empty=$26
      previous_river_canceled=$27
      if ($4 >= 900 || $6 <= 0 || $5 * 4 >= $6 * 3 || $7 >= 80 || $8 >= 1536 ||
          $10 <= 0 || $28 >= 1536 || saturated >= 6 || critical_saturated >= 6 ||
          river_saturated >= 6 || empty_growth >= 6 || critical_empty_growth >= 6 ||
          river_empty_growth >= 6 || $16 < 1 || $16 > 2 || $17 > slots ||
          $19 != 4 || $24 <= 0) {
        failed=1
      }
    }
    END {
      exit !(samples > 0 && failed == 0)
    }
  ' slots="$expected_slots" "$report_path"
}

certify_worker_resource_requests() {
  local profile_slug
  profile_slug=$(printf '%s' "$stable_profile" | tr '[:upper:]' '[:lower:]')
  local memory_values=$temporary_root/worker-memory-values
  local cpu_values=$temporary_root/worker-cpu-values
  awk -F',' 'FNR > 1 && $28 ~ /^[0-9]+([.][0-9]+)?$/ {print $28}' \
    "$report_root"/"$profile_slug"-*-pressure.csv | sort -n >"$memory_values"
  awk -F',' 'FNR > 1 && $29 ~ /^[0-9]+([.][0-9]+)?$/ {print $29}' \
    "$report_root"/"$profile_slug"-*-pressure.csv | sort -n >"$cpu_values"
  local memory_count cpu_count memory_rank cpu_rank memory_p99 cpu_p99
  memory_count=$(wc -l <"$memory_values" | tr -d ' ')
  cpu_count=$(wc -l <"$cpu_values" | tr -d ' ')
  (( memory_count > 0 && cpu_count > 0 ))
  memory_rank=$(( (memory_count * 99 + 99) / 100 ))
  cpu_rank=$(( (cpu_count * 99 + 99) / 100 ))
  memory_p99=$(sed -n "${memory_rank}p" "$memory_values")
  cpu_p99=$(sed -n "${cpu_rank}p" "$cpu_values")
  AI_GATEWAY_ACCEPTANCE_WORKER_MEMORY_REQUEST_MIB=$(awk -v value="$memory_p99" '
    BEGIN {
      request=int(value*1.2+0.999)
      if (request<256) request=256
      print request
    }')
  AI_GATEWAY_ACCEPTANCE_WORKER_CPU_REQUEST_MILLICORES=$(awk -v value="$cpu_p99" '
    BEGIN {
      request=int(value*1.2+0.999)
      if (request<100) request=100
      print request
    }')
  (( AI_GATEWAY_ACCEPTANCE_WORKER_MEMORY_REQUEST_MIB <= 1900 &&
     AI_GATEWAY_ACCEPTANCE_WORKER_CPU_REQUEST_MILLICORES <= 1900 )) || {
    echo "certified Worker request exceeds the 2 GiB/2 CPU Pod budget: memory_mib=$AI_GATEWAY_ACCEPTANCE_WORKER_MEMORY_REQUEST_MIB cpu_millicores=$AI_GATEWAY_ACCEPTANCE_WORKER_CPU_REQUEST_MILLICORES" >&2
    return 1
  }
  jq -n \
    --arg capacityProfile "$stable_profile" \
    --argjson memoryP99MiB "$memory_p99" \
    --argjson cpuP99Millicores "$cpu_p99" \
    --argjson memoryRequestMiB "$AI_GATEWAY_ACCEPTANCE_WORKER_MEMORY_REQUEST_MIB" \
    --argjson cpuRequestMillicores "$AI_GATEWAY_ACCEPTANCE_WORKER_CPU_REQUEST_MILLICORES" \
    '{
      capacityProfile:$capacityProfile,
      workerMemoryP99MiB:$memoryP99MiB,
      workerCPUP99Millicores:$cpuP99Millicores,
      workerMemoryRequestMiB:$memoryRequestMiB,
      workerCPURequestMillicores:$cpuRequestMillicores,
      safetyMultiplier:1.2
    }' >"$report_root/certified-worker-resources.json"
}

run_capacity_round() {
  local profile=$1
  local repetition=$2
  local profile_slug
  profile_slug=$(printf '%s' "$profile" | tr '[:upper:]' '[:lower:]')
  local prefix=$report_root/$profile_slug-$repetition
  local pressure_report=$prefix-pressure.csv
  local pressure_stop=$temporary_root/pressure-stop
  local pressure_pid status=0 pressure_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=$!
  active_pressure_pid=$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" || pressure_status=$?
  active_pressure_pid=
  (( status == 0 && pressure_status == 0 )) || {
    (( status != 0 )) && return "$status"
    return "$pressure_status"
  }
  verify_pressure_report "$pressure_report" "$expected_slots"
  verify_runtime_gates
  wait_for_memory_recovery "$baseline_memory_mib"
}

finish_and_report() {
  local emulator_report=$report_root/emulator-report.json
  local callback_report=$report_root/callback-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"
  remote_kubectl exec -n "$namespace" deployment/easyai-acceptance-callback-collector -- \
    wget -qO- http://127.0.0.1:8091/report >"$callback_report"
  local task_count video_task_count gemini_task_count submissions duplicates callbacks
  local missing_idempotency duplicate_submissions
  local gemini_requests forced_tasks forced_requests verified_conversions compact_payloads
  local gemini_invalid video_invalid reference_count_3 reference_count_6 reference_count_9
  local reference_role_count first_frame_role_count last_frame_role_count
  local unique_image_hashes unique_video_tasks
  local duplicate_remote_tasks duplicate_terminal_events duplicate_settlements
  local duplicate_wallet_transactions duplicate_callback_rows pending_settlements pending_callbacks
  local missing_billings
  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 '.duplicates' "$callback_report")
  callbacks=$(jq -r '.deliveries' "$callback_report")
  [[ $(jq -r '.invalid' "$callback_report") == 0 ]] ||
    fail_acceptance_gate callback_protocol 'callback collector observed an invalid event'
  gemini_requests=$(jq -r '.geminiRequests' "$emulator_report")
  forced_requests=$(jq -r '.forcedConversionRequests' "$emulator_report")
  verified_conversions=$(jq -r '.verifiedConversions' "$emulator_report")
  missing_idempotency=$(jq -r '.missingIdempotencyKeys' "$emulator_report")
  duplicate_submissions=$(jq -r '.duplicateSubmissionAttempts' "$emulator_report")
  gemini_invalid=$(jq -r '.geminiInvalid // 0' "$emulator_report")
  video_invalid=$(jq -r '.videoInvalid // 0' "$emulator_report")
  reference_count_3=$(jq -r '.videoReferenceCounts["3"] // 0' "$emulator_report")
  reference_count_6=$(jq -r '.videoReferenceCounts["6"] // 0' "$emulator_report")
  reference_count_9=$(jq -r '.videoReferenceCounts["9"] // 0' "$emulator_report")
  reference_role_count=$(jq -r '.videoRoleCounts.reference_image // 0' "$emulator_report")
  first_frame_role_count=$(jq -r '.videoRoleCounts.first_frame // 0' "$emulator_report")
  last_frame_role_count=$(jq -r '.videoRoleCounts.last_frame // 0' "$emulator_report")
  unique_image_hashes=$(jq -r '.uniqueImageHashes // 0' "$emulator_report")
  unique_video_tasks=$(jq -r '.uniqueVideoTasks // 0' "$emulator_report")
  [[ $submissions == "$video_task_count" && $gemini_requests == "$gemini_task_count" ]] ||
    fail_acceptance_gate protocol_task_counts 'emulator and database task counts differ'
  [[ $forced_requests == "$forced_tasks" && $verified_conversions == "$forced_tasks" ]] ||
    fail_acceptance_gate media_conversion_counts 'forced image conversion counts differ'
  [[ $duplicates == 0 && $duplicate_submissions == 0 &&
     $missing_idempotency == 0 && $compact_payloads == 0 ]] ||
    fail_acceptance_gate idempotency_and_materialization \
      'duplicate submission, missing idempotency key, or inline database media detected'
  [[ $gemini_invalid == 0 && $video_invalid == 0 ]] ||
    fail_acceptance_gate provider_protocol_validity 'Gemini or Volces protocol validation failed'
  [[ $reference_count_3 -gt 0 && $reference_count_6 -gt 0 && $reference_count_9 -gt 0 ]] ||
    fail_acceptance_gate video_reference_distribution '3/6/9 reference-image coverage is incomplete'
  [[ $((reference_count_3 + reference_count_6 + reference_count_9)) -eq $submissions ]] ||
    fail_acceptance_gate video_reference_accounting 'reference-image distribution does not cover every video'
  [[ $reference_role_count -gt 0 && $first_frame_role_count -gt 0 && $last_frame_role_count -gt 0 ]] ||
    fail_acceptance_gate video_reference_roles 'reference, first-frame, and last-frame roles were not all observed'
  [[ $unique_image_hashes -ge 12 && $unique_video_tasks -eq $submissions ]] ||
    fail_acceptance_gate media_content_uniqueness 'media hashes or unique remote video tasks are incomplete'
  [[ $callbacks -gt 0 ]] ||
    fail_acceptance_gate callback_delivery 'no acceptance callback was delivered'
  duplicate_remote_tasks=$(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;")
  duplicate_terminal_events=$(database_query "
SELECT count(*)
FROM (
  SELECT task_id,event_type
  FROM gateway_task_events
  WHERE task_id IN (
    SELECT id FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid
  )
    AND event_type IN ('task.completed','task.failed','task.cancelled')
  GROUP BY task_id,event_type
  HAVING count(*) > 1
) duplicate;")
  duplicate_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;")
  duplicate_wallet_transactions=$(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;")
  duplicate_callback_rows=$(database_query "
SELECT count(*)
FROM (
  SELECT task_id,seq,callback_url
  FROM gateway_task_callback_outbox
  WHERE task_id IN (
    SELECT id FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid
  )
  GROUP BY task_id,seq,callback_url
  HAVING count(*) > 1
) duplicate;")
  pending_settlements=$(database_query "
SELECT count(*)
FROM settlement_outbox
WHERE task_id IN (
  SELECT id FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid
)
  AND status <> 'completed';")
  pending_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';")
  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'
  );")
  [[ $duplicate_remote_tasks == 0 && $duplicate_terminal_events == 0 ]] ||
    fail_acceptance_gate remote_and_terminal_uniqueness 'duplicate remote tasks or terminal events detected'
  [[ $duplicate_settlements == 0 && $duplicate_wallet_transactions == 0 ]] ||
    fail_acceptance_gate billing_uniqueness 'duplicate settlements or wallet transactions detected'
  [[ $duplicate_callback_rows == 0 && $pending_settlements == 0 ]] ||
    fail_acceptance_gate outbox_completion 'duplicate callbacks or pending settlements remain'
  [[ $pending_callbacks == 0 && $missing_billings == 0 ]] ||
    fail_acceptance_gate billing_and_callback_completion 'pending callbacks or missing billings remain'
  admin_request GET /api/admin/system/acceptance/traffic-mode '' "$mode_response"
  verify_release_cas
  local config_hash
  [[ -n $capacity_profiles_payload && -f $capacity_profiles_payload ]] ||
    fail_acceptance_gate capacity_profile_artifact 'staged capacity profile artifact is missing'
  config_hash=$(jq -r '.configHash' "$capacity_profiles_payload")
  [[ $(remote_kubectl get configmap easyai-gateway-release -n "$namespace" \
    -o 'jsonpath={.data.configHash}') == "$config_hash" ]] ||
    fail_acceptance_gate capacity_profile_config_cas 'staged capacity profile config hash changed'
  [[ $(jq -r '.mode' "$mode_response") == validation &&
     $(jq -r '.runId' "$mode_response") == "$run_id" ]] ||
    fail_acceptance_gate validation_mode_cas 'traffic mode changed before acceptance completion'
  jq -n \
    --arg runId "$run_id" \
    --arg releaseSha "$release_sha" \
    --arg apiImageDigest "$api_digest" \
    --arg workerImageDigest "$worker_digest" \
    --arg stableCapacityProfile "$stable_profile" \
    --arg configHash "$config_hash" \
    --arg snapshotConfigHash "$local_snapshot_config_hash" \
    --arg snapshotSha256 "$local_snapshot_sha256" \
    --argjson imageStableThroughput "$image_stable_throughput" \
    --argjson imageAdmittedThroughput "$image_admitted_throughput" \
    --argjson videoStableThroughput "$video_stable_throughput" \
    --argjson videoAdmittedThroughput "$video_admitted_throughput" \
    --argjson tasks "$task_count" \
    --argjson geminiTasks "$gemini_task_count" \
    --argjson videoTasks "$video_task_count" \
    --argjson upstreamSubmissions "$submissions" \
    --argjson callbackDeliveries "$callbacks" \
    --argjson missingIdempotencyKeys "$missing_idempotency" \
    --argjson duplicateSubmissionAttempts "$duplicate_submissions" \
    --argjson geminiInvalid "$gemini_invalid" \
    --argjson videoInvalid "$video_invalid" \
    --argjson referenceCount3 "$reference_count_3" \
    --argjson referenceCount6 "$reference_count_6" \
    --argjson referenceCount9 "$reference_count_9" \
    --argjson referenceRoleCount "$reference_role_count" \
    --argjson firstFrameRoleCount "$first_frame_role_count" \
    --argjson lastFrameRoleCount "$last_frame_role_count" \
    --argjson uniqueImageHashes "$unique_image_hashes" \
    --argjson uniqueVideoTasks "$unique_video_tasks" \
    --argjson duplicateRemoteTasks "$duplicate_remote_tasks" \
    --argjson duplicateTerminalEvents "$duplicate_terminal_events" \
    --argjson duplicateSettlements "$duplicate_settlements" \
    --argjson duplicateWalletTransactions "$duplicate_wallet_transactions" \
    --argjson duplicateCallbackRows "$duplicate_callback_rows" \
    --argjson pendingSettlements "$pending_settlements" \
    --argjson pendingCallbacks "$pending_callbacks" \
    --argjson missingBillings "$missing_billings" \
    '{
      runId:$runId,
      releaseSha:$releaseSha,
      apiImageDigest:$apiImageDigest,
      workerImageDigest:$workerImageDigest,
      snapshotConfigHash:$snapshotConfigHash,
      snapshotSha256:$snapshotSha256,
      configHash:$configHash,
      stableCapacityProfile:$stableCapacityProfile,
      imageStableThroughputPerSecond:$imageStableThroughput,
      imageAdmittedThroughputPerSecond:$imageAdmittedThroughput,
      videoStableThroughputPerSecond:$videoStableThroughput,
      videoAdmittedThroughputPerSecond:$videoAdmittedThroughput,
      tasks:$tasks,
      geminiTasks:$geminiTasks,
      videoTasks:$videoTasks,
      upstreamSubmissions:$upstreamSubmissions,
      callbackDeliveries:$callbackDeliveries,
      missingIdempotencyKeys:$missingIdempotencyKeys,
      duplicateSubmissionAttempts:$duplicateSubmissionAttempts,
      protocolValidation:{
        geminiInvalid:$geminiInvalid,
        videoInvalid:$videoInvalid,
        videoReferenceCounts:{
          "3":$referenceCount3,
          "6":$referenceCount6,
          "9":$referenceCount9
        },
        videoRoleCounts:{
          reference_image:$referenceRoleCount,
          first_frame:$firstFrameRoleCount,
          last_frame:$lastFrameRoleCount
        },
        uniqueImageHashes:$uniqueImageHashes,
        uniqueVideoTasks:$uniqueVideoTasks
      },
      databaseConsistency:{
        duplicateRemoteTasks:$duplicateRemoteTasks,
        duplicateTerminalEvents:$duplicateTerminalEvents,
        duplicateSettlements:$duplicateSettlements,
        duplicateWalletTransactions:$duplicateWalletTransactions,
        duplicateCallbackRows:$duplicateCallbackRows,
        pendingSettlements:$pendingSettlements,
        pendingCallbacks:$pendingCallbacks,
        missingBillings:$missingBillings
      },
      realCanaryPassed:true,
      certifiedProfile:{
        name:$stableCapacityProfile,
        configHash:$configHash,
        imageStableThroughputPerSecond:$imageStableThroughput,
        imageAdmittedThroughputPerSecond:$imageAdmittedThroughput,
        videoStableThroughputPerSecond:$videoStableThroughput,
        videoAdmittedThroughputPerSecond:$videoAdmittedThroughput
      },
      gates:[
        {id:"protocol_and_media_integrity",passed:true},
        {id:"database_exactly_once",passed:true},
        {id:"runtime_resources",passed:true},
        {id:"real_media_canary",passed:true}
      ],
      promotionRequested:false,
      awaitingManualPromotion: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"
  local certification_command certification_output instance_slots
  case $stable_profile in
    P24) instance_slots=24 ;;
    P28) instance_slots=28 ;;
    P32) instance_slots=32 ;;
  esac
  printf -v certification_command '%q certify %q %q %q %q %q %q' \
    "$remote_release_helper" \
    "$run_id" \
    "$stable_profile" \
    "$config_hash" \
    "$instance_slots" \
    "$certified_max_replicas_ningbo" \
    "$certified_max_replicas_hongkong"
  certification_output=$(cluster_ssh "$CLUSTER_NINGBO_HOST" "$certification_command")
  grep -Fq 'production_certification=PASS' <<<"$certification_output"
  node "$cluster_root/scripts/acceptance/report.mjs" merge-production \
    --local-report "$local_acceptance_report" \
    --production-summary "$report_root/summary.json" \
    --output "$report_root/acceptance-report.json" >/dev/null
  node "$cluster_root/scripts/acceptance/report.mjs" validate \
    --input "$report_root/acceptance-report.json" >/dev/null
  [[ $(jq -r '.promotion.ready' "$report_root/acceptance-report.json") == true ]]
}

verify_release_cas
bootstrap_acceptance_primary_identity
ensure_acceptance_user_group
ensure_acceptance_identity_shards
ensure_acceptance_real_images
select_acceptance_models
ensure_acceptance_model_access
create_and_activate_run
wait_for_existing_tasks_to_drain
snapshot_pre_acceptance_capacity
if ! verify_resource_preconditions; then
  failure_gate_id=resource_preconditions
  failure_reason='production resource preconditions failed'
  mark_run_failed "$failure_reason"
  exit 1
fi
deploy_protocol_emulator

for profile in P24 P28 P32; do
  if ! apply_capacity_profile "$profile"; then
    failure_gate_id=capacity_profile_apply
    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_gate_id=capacity_round
      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
  apply_capacity_profile "$stable_profile" || true
  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 ! certify_worker_resource_requests; then
  failure_gate_id=worker_resource_certification
  failure_reason='worker p99 resource certification failed'
elif ! apply_capacity_profile "$stable_profile"; then
  failure_gate_id=certified_resource_apply
  failure_reason='failed to apply certified Worker resource requests'
elif ! verify_runtime_gates; then
  failure_gate_id=certified_resource_runtime
  failure_reason='certified Worker resource rollout failed runtime gates'
elif ! run_autoscaling_acceptance; then
  failure_gate_id=worker_autoscaling
  failure_reason='Worker autoscaling, drain, or resource-budget acceptance failed'
elif ! stage_capacity_profiles; then
  failure_gate_id=capacity_profile_staging
  failure_reason='failed to stage certified capacity profiles for validation traffic'
elif ! run_mixed_soak_and_overload; then
  failure_gate_id=mixed_soak_overload
  failure_reason='80% mixed soak or 120% active-throttling acceptance failed'
elif ! activate_certified_autoscaling; then
  failure_gate_id=certified_autoscaling_activation
  failure_reason='failed to activate and CAS the certified elastic Worker profile'
fi
if [[ -n $failure_reason ]]; then
  apply_capacity_profile "$stable_profile" || true
  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_gate_id=real_media_canary
  failure_reason='real canary failed'
elif ! verify_runtime_gates; then
  failure_gate_id=post_canary_runtime
  failure_reason='post-canary runtime gates failed'
fi
if [[ -n $failure_reason ]]; then
  apply_capacity_profile "$stable_profile" || true
  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
failure_gate_id=release_certification_or_report
failure_reason='release certification record or final acceptance report failed'
finish_and_report
failure_gate_id=
failure_reason=
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
remote_kubectl delete deployment easyai-acceptance-callback-collector -n "$namespace" --ignore-not-found >/dev/null ||
  echo 'warning: acceptance callback collector Deployment cleanup failed' >&2
remote_kubectl delete service easyai-acceptance-callback-collector -n "$namespace" --ignore-not-found >/dev/null ||
  echo 'warning: acceptance callback collector Service cleanup failed' >&2
echo "production_acceptance=PASS run_id=$run_id release=$release_sha stable_profile=$stable_profile traffic_mode=validation promotion=awaiting_manual_confirmation report=$report_root/acceptance-report.json"