diff --git a/.github/workflows/build-test.yml b/.github/workflows/build-test.yml
index 4c1623f3a..0c2de3118 100644
--- a/.github/workflows/build-test.yml
+++ b/.github/workflows/build-test.yml
@@ -141,3 +141,14 @@ jobs:
         run: |
           bash tools/ci/test-check-gazelle
           bash tools/ci/check-gazelle
+
+  openbao-bootstrap:
+    name: openbao bootstrap
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+
+      - name: Run OpenBao bootstrap tests
+        run: bash deploy/helm/openbao/tests/bootstrap/test-deploy-bootstrap.sh
diff --git a/deploy/helm/openbao/deploy.sh b/deploy/helm/openbao/deploy.sh
index 328884017..33ba686f5 100755
--- a/deploy/helm/openbao/deploy.sh
+++ b/deploy/helm/openbao/deploy.sh
@@ -131,6 +131,37 @@ initialize_cluster() {
     done
     log_info "All OpenBao pods are ready"
 
+    # Idempotency guard. This script runs as a post-install Job whose pod is
+    # retried by the Kubernetes Job controller (backoffLimit defaults to 6, so
+    # up to 7 attempts within a single helm install). If an earlier attempt
+    # already ran `bao operator init` (for example it initialized the cluster
+    # and then failed later during the raft bootstrap), re-running it returns
+    # HTTP 400 "Vault is already initialized" and the attempt bails out, so the
+    # bootstrap can never finish. Treat an already-initialized cluster as
+    # success and reuse the stored secrets, letting the retry pick up where the
+    # previous attempt stopped.
+    local init_status=$(kubectl exec ${statefulset}-0 -c openbao -n ${namespace} -- \
+        bao status -format=json 2>/dev/null | jq -r '.initialized')
+    if [ "${init_status}" = "true" ]; then
+        log_info "OpenBao cluster already initialized; skipping 'bao operator init'"
+        # "Secret exists" is not a sufficient check: the unseal secret is
+        # pre-created empty and only patched with the real key later, so
+        # present-but-empty is a reachable state. These helpers return "" both
+        # when the secret is missing and when its value is empty. Without this
+        # check we would fall through to `bao operator unseal ""` and report a
+        # far less actionable error. An initialized cluster whose keys were
+        # never persisted cannot be recovered.
+        local stored_unseal_key=$(get_unseal_key "${namespace}" "${statefulset}")
+        local stored_root_token=$(get_root_token "${namespace}" "${statefulset}")
+        if [ -z "${stored_unseal_key}" ] || [ -z "${stored_root_token}" ]; then
+            log_error "Cluster is initialized but its stored unseal key and/or root token are missing or empty."
+            log_error "The generated keys are unrecoverable. Run ./cleanup.sh and reinstall."
+            return 1
+        fi
+        log_success "Reusing unseal key and root token from '${statefulset}-unseal' and '${statefulset}-root-token'"
+        return 0
+    fi
+
     log_info "Initializing OpenBao cluster"
     local init_output=$(kubectl exec ${statefulset}-0 -c openbao -n ${namespace} -- \
         bao operator init \
@@ -169,6 +200,83 @@ get_unseal_key() {
     kubectl get secret ${statefulset}-unseal -n ${namespace} -o jsonpath='{.data.unseal_key}' | base64 -d
 }
 
+# Wait until the bootstrap node (pod 0) has won leader election and can serve
+# raft challenges. After being unsealed the node must become the active leader
+# before any peer joins; joining earlier returns HTTP 500 "failed to join raft
+# cluster: failed to get raft challenge". This replaces a fixed `sleep 5` that
+# raced leader election.
+wait_for_active_leader() {
+    local namespace=$1
+    local statefulset=$2
+    local timeout=${3:-120}
+    local pod="${statefulset}-0"
+    local end=$((SECONDS + timeout))
+
+    log_info "Waiting for ${pod} to become the active Raft leader (timeout: ${timeout}s)..."
+    while true; do
+        local status_json=$(kubectl exec "${pod}" -c openbao -n "${namespace}" -- \
+            bao status -format=json 2>/dev/null)
+        # Read these with plain accessors, never `.field // empty`: jq's `//`
+        # treats boolean false the same as null, so `.sealed // empty` yields
+        # "" for an unsealed node and this loop would never see sealed=false.
+        local initialized=$(echo "${status_json}" | jq -r '.initialized')
+        local sealed=$(echo "${status_json}" | jq -r '.sealed')
+        local ha_mode=$(echo "${status_json}" | jq -r '.ha_mode')
+        if [ "${initialized}" = "true" ] && [ "${sealed}" = "false" ] && [ "${ha_mode}" = "active" ]; then
+            log_success "${pod} is unsealed and active (ha_mode=active)"
+            return 0
+        fi
+        if [ $SECONDS -gt $end ]; then
+            log_error "Timeout waiting for ${pod} to become active leader (initialized=${initialized:-?}, sealed=${sealed:-?}, ha_mode=${ha_mode:-?})"
+            return 1
+        fi
+        log_info "  ${pod} not ready yet (sealed=${sealed:-?}, ha_mode=${ha_mode:-?}); retrying in 3s..."
+        sleep 3
+    done
+}
+
+# True when the pod is already unsealed, and therefore already a working member
+# of the raft cluster. Lets a retry skip peers a previous attempt finished.
+is_pod_unsealed() {
+    local namespace=$1
+    local pod=$2
+    local sealed=$(kubectl exec "${pod}" -c openbao -n "${namespace}" -- \
+        bao status -format=json 2>/dev/null | jq -r '.sealed')
+    [ "${sealed}" = "false" ]
+}
+
+# Join a peer to the raft cluster, retrying the transient "failed to get raft
+# challenge" 500 that can still occur for a few seconds after the leader goes
+# active. Treats an already-joined node as success so retries are idempotent.
+raft_join_with_retry() {
+    local namespace=$1
+    local statefulset=$2
+    local pod=$3
+    local attempts=${4:-12}
+    local i=1
+    # Declare before assigning: with `local out=$(...)` the `local` builtin is
+    # the executed command, so $? would capture the assignment status (always 0)
+    # rather than the kubectl/bao exit code, masking real join failures.
+    local out rc
+    while [ $i -le $attempts ]; do
+        out=$(kubectl exec "${pod}" -c openbao -n "${namespace}" -- \
+            bao operator raft join "http://${statefulset}-0.${statefulset}-internal:8200" 2>&1)
+        rc=$?
+        echo "${out}"
+        if [ $rc -eq 0 ]; then
+            return 0
+        fi
+        if echo "${out}" | grep -qi "already"; then
+            log_info "${pod} is already a member of the Raft cluster"
+            return 0
+        fi
+        log_warn "raft join attempt ${i}/${attempts} for ${pod} failed; retrying in 5s..."
+        sleep 5
+        i=$((i + 1))
+    done
+    return 1
+}
+
 # Step 4: Unseal the cluster
 unseal_cluster() {
     local namespace=$1
@@ -177,30 +285,49 @@ unseal_cluster() {
 
     log_section "Unsealing OpenBao cluster"
 
-    # First unseal the primary node (pod 0)
+    # get_unseal_key returns "" when the secret is missing or still holds the
+    # empty placeholder. initialize_cluster rejects that state before we get
+    # here, but check anyway so this never degrades into
+    # `bao operator unseal ""` and its low-signal error.
+    if [ -z "${unseal_key}" ]; then
+        log_error "No unseal key stored in secret '${statefulset}-unseal'; cannot unseal the cluster."
+        log_error "If the cluster is already initialized the generated keys are unrecoverable. Run ./cleanup.sh and reinstall."
+        return 1
+    fi
+
+    # First unseal the primary node (pod 0). Unsealing an already-unsealed node
+    # is a no-op that exits 0, so this is safe to re-run.
     log_info "Unsealing primary pod ${statefulset}-0"
-    if ! kubectl exec ${statefulset}-0 -n ${namespace} -- \
+    if ! kubectl exec ${statefulset}-0 -c openbao -n ${namespace} -- \
         bao operator unseal ${unseal_key}; then
         log_error "Failed to unseal primary pod ${statefulset}-0"
         return 1
     fi
 
-    # Wait a moment for the primary to be ready
-    sleep 5
+    # Wait for the primary to win leader election and start serving raft
+    # challenges before any peer joins.
+    if ! wait_for_active_leader "${namespace}" "${statefulset}"; then
+        return 1
+    fi
 
     # Join and unseal remaining pods
     for i in {1..2}; do
-        log_info "Joining pod ${statefulset}-${i} to Raft cluster"
-        if ! kubectl exec ${statefulset}-${i} -c openbao -n ${namespace} -- \
-            bao operator raft join http://${statefulset}-0.${statefulset}-internal:8200; then
-            log_error "Failed to join pod ${statefulset}-${i} to Raft cluster"
+        local pod="${statefulset}-${i}"
+        if is_pod_unsealed "${namespace}" "${pod}"; then
+            log_info "Pod ${pod} already unsealed and joined; skipping"
+            continue
+        fi
+
+        log_info "Joining pod ${pod} to Raft cluster"
+        if ! raft_join_with_retry "${namespace}" "${statefulset}" "${pod}"; then
+            log_error "Failed to join pod ${pod} to Raft cluster"
             return 1
         fi
 
-        log_info "Unsealing pod ${statefulset}-${i}"
-        if ! kubectl exec ${statefulset}-${i} -c openbao -n ${namespace} -- \
+        log_info "Unsealing pod ${pod}"
+        if ! kubectl exec ${pod} -c openbao -n ${namespace} -- \
             bao operator unseal ${unseal_key}; then
-            log_error "Failed to unseal pod ${statefulset}-${i}"
+            log_error "Failed to unseal pod ${pod}"
             return 1
         fi
     done
diff --git a/deploy/helm/openbao/helm/scripts/deploy.sh b/deploy/helm/openbao/helm/scripts/deploy.sh
index c4fe17b62..85832d195 100755
--- a/deploy/helm/openbao/helm/scripts/deploy.sh
+++ b/deploy/helm/openbao/helm/scripts/deploy.sh
@@ -187,6 +187,37 @@ initialize_cluster() {
     done
     log_info "All OpenBao pods are ready"
 
+    # Idempotency guard. This script runs as a post-install Job whose pod is
+    # retried by the Kubernetes Job controller (backoffLimit defaults to 6, so
+    # up to 7 attempts within a single helm install). If an earlier attempt
+    # already ran `bao operator init` (for example it initialized the cluster
+    # and then failed later during the raft bootstrap), re-running it returns
+    # HTTP 400 "Vault is already initialized" and the attempt bails out, so the
+    # bootstrap can never finish. Treat an already-initialized cluster as
+    # success and reuse the stored secrets, letting the retry pick up where the
+    # previous attempt stopped.
+    local init_status=$(kubectl exec ${statefulset}-0 -c openbao -n ${namespace} -- \
+        bao status -format=json 2>/dev/null | jq -r '.initialized')
+    if [ "${init_status}" = "true" ]; then
+        log_info "OpenBao cluster already initialized; skipping 'bao operator init'"
+        # "Secret exists" is not a sufficient check: the unseal secret is
+        # pre-created empty and only patched with the real key later, so
+        # present-but-empty is a reachable state. These helpers return "" both
+        # when the secret is missing and when its value is empty. Without this
+        # check we would fall through to `bao operator unseal ""` and report a
+        # far less actionable error. An initialized cluster whose keys were
+        # never persisted cannot be recovered.
+        local stored_unseal_key=$(get_unseal_key "${namespace}" "${statefulset}")
+        local stored_root_token=$(get_root_token "${namespace}" "${statefulset}")
+        if [ -z "${stored_unseal_key}" ] || [ -z "${stored_root_token}" ]; then
+            log_error "Cluster is initialized but its stored unseal key and/or root token are missing or empty."
+            log_error "The generated keys are unrecoverable. Run ./cleanup.sh and reinstall."
+            return 1
+        fi
+        log_success "Reusing unseal key and root token from '${statefulset}-unseal' and '${statefulset}-root-token'"
+        return 0
+    fi
+
     log_info "Initializing OpenBao cluster"
     local init_output=$(kubectl exec ${statefulset}-0 -c openbao -n ${namespace} -- \
         bao operator init \
@@ -225,6 +256,83 @@ get_unseal_key() {
     kubectl get secret ${statefulset}-unseal -n ${namespace} -o jsonpath='{.data.unseal_key}' | base64 -d
 }
 
+# Wait until the bootstrap node (pod 0) has won leader election and can serve
+# raft challenges. After being unsealed the node must become the active leader
+# before any peer joins; joining earlier returns HTTP 500 "failed to join raft
+# cluster: failed to get raft challenge". This replaces a fixed `sleep 5` that
+# raced leader election.
+wait_for_active_leader() {
+    local namespace=$1
+    local statefulset=$2
+    local timeout=${3:-120}
+    local pod="${statefulset}-0"
+    local end=$((SECONDS + timeout))
+
+    log_info "Waiting for ${pod} to become the active Raft leader (timeout: ${timeout}s)..."
+    while true; do
+        local status_json=$(kubectl exec "${pod}" -c openbao -n "${namespace}" -- \
+            bao status -format=json 2>/dev/null)
+        # Read these with plain accessors, never `.field // empty`: jq's `//`
+        # treats boolean false the same as null, so `.sealed // empty` yields
+        # "" for an unsealed node and this loop would never see sealed=false.
+        local initialized=$(echo "${status_json}" | jq -r '.initialized')
+        local sealed=$(echo "${status_json}" | jq -r '.sealed')
+        local ha_mode=$(echo "${status_json}" | jq -r '.ha_mode')
+        if [ "${initialized}" = "true" ] && [ "${sealed}" = "false" ] && [ "${ha_mode}" = "active" ]; then
+            log_success "${pod} is unsealed and active (ha_mode=active)"
+            return 0
+        fi
+        if [ $SECONDS -gt $end ]; then
+            log_error "Timeout waiting for ${pod} to become active leader (initialized=${initialized:-?}, sealed=${sealed:-?}, ha_mode=${ha_mode:-?})"
+            return 1
+        fi
+        log_info "  ${pod} not ready yet (sealed=${sealed:-?}, ha_mode=${ha_mode:-?}); retrying in 3s..."
+        sleep 3
+    done
+}
+
+# True when the pod is already unsealed, and therefore already a working member
+# of the raft cluster. Lets a retry skip peers a previous attempt finished.
+is_pod_unsealed() {
+    local namespace=$1
+    local pod=$2
+    local sealed=$(kubectl exec "${pod}" -c openbao -n "${namespace}" -- \
+        bao status -format=json 2>/dev/null | jq -r '.sealed')
+    [ "${sealed}" = "false" ]
+}
+
+# Join a peer to the raft cluster, retrying the transient "failed to get raft
+# challenge" 500 that can still occur for a few seconds after the leader goes
+# active. Treats an already-joined node as success so retries are idempotent.
+raft_join_with_retry() {
+    local namespace=$1
+    local statefulset=$2
+    local pod=$3
+    local attempts=${4:-12}
+    local i=1
+    # Declare before assigning: with `local out=$(...)` the `local` builtin is
+    # the executed command, so $? would capture the assignment status (always 0)
+    # rather than the kubectl/bao exit code, masking real join failures.
+    local out rc
+    while [ $i -le $attempts ]; do
+        out=$(kubectl exec "${pod}" -c openbao -n "${namespace}" -- \
+            bao operator raft join "http://${statefulset}-0.${statefulset}-internal:8200" 2>&1)
+        rc=$?
+        echo "${out}"
+        if [ $rc -eq 0 ]; then
+            return 0
+        fi
+        if echo "${out}" | grep -qi "already"; then
+            log_info "${pod} is already a member of the Raft cluster"
+            return 0
+        fi
+        log_warn "raft join attempt ${i}/${attempts} for ${pod} failed; retrying in 5s..."
+        sleep 5
+        i=$((i + 1))
+    done
+    return 1
+}
+
 # Step 4: Unseal the cluster
 unseal_cluster() {
     local namespace=$1
@@ -233,30 +341,49 @@ unseal_cluster() {
 
     log_section "Unsealing OpenBao cluster"
 
-    # First unseal the primary node (pod 0)
+    # get_unseal_key returns "" when the secret is missing or still holds the
+    # empty placeholder. initialize_cluster rejects that state before we get
+    # here, but check anyway so this never degrades into
+    # `bao operator unseal ""` and its low-signal error.
+    if [ -z "${unseal_key}" ]; then
+        log_error "No unseal key stored in secret '${statefulset}-unseal'; cannot unseal the cluster."
+        log_error "If the cluster is already initialized the generated keys are unrecoverable. Run ./cleanup.sh and reinstall."
+        return 1
+    fi
+
+    # First unseal the primary node (pod 0). Unsealing an already-unsealed node
+    # is a no-op that exits 0, so this is safe to re-run.
     log_info "Unsealing primary pod ${statefulset}-0"
-    if ! kubectl exec ${statefulset}-0 -n ${namespace} -- \
+    if ! kubectl exec ${statefulset}-0 -c openbao -n ${namespace} -- \
         bao operator unseal ${unseal_key}; then
         log_error "Failed to unseal primary pod ${statefulset}-0"
         return 1
     fi
 
-    # Wait a moment for the primary to be ready
-    sleep 5
+    # Wait for the primary to win leader election and start serving raft
+    # challenges before any peer joins.
+    if ! wait_for_active_leader "${namespace}" "${statefulset}"; then
+        return 1
+    fi
 
     # Join and unseal remaining pods
     for i in {1..2}; do
-        log_info "Joining pod ${statefulset}-${i} to Raft cluster"
-        if ! kubectl exec ${statefulset}-${i} -c openbao -n ${namespace} -- \
-            bao operator raft join http://${statefulset}-0.${statefulset}-internal:8200; then
-            log_error "Failed to join pod ${statefulset}-${i} to Raft cluster"
+        local pod="${statefulset}-${i}"
+        if is_pod_unsealed "${namespace}" "${pod}"; then
+            log_info "Pod ${pod} already unsealed and joined; skipping"
+            continue
+        fi
+
+        log_info "Joining pod ${pod} to Raft cluster"
+        if ! raft_join_with_retry "${namespace}" "${statefulset}" "${pod}"; then
+            log_error "Failed to join pod ${pod} to Raft cluster"
             return 1
         fi
 
-        log_info "Unsealing pod ${statefulset}-${i}"
-        if ! kubectl exec ${statefulset}-${i} -c openbao -n ${namespace} -- \
+        log_info "Unsealing pod ${pod}"
+        if ! kubectl exec ${pod} -c openbao -n ${namespace} -- \
             bao operator unseal ${unseal_key}; then
-            log_error "Failed to unseal pod ${statefulset}-${i}"
+            log_error "Failed to unseal pod ${pod}"
             return 1
         fi
     done
diff --git a/deploy/helm/openbao/tests/bootstrap/test-deploy-bootstrap.sh b/deploy/helm/openbao/tests/bootstrap/test-deploy-bootstrap.sh
new file mode 100755
index 000000000..cb7c87a0f
--- /dev/null
+++ b/deploy/helm/openbao/tests/bootstrap/test-deploy-bootstrap.sh
@@ -0,0 +1,289 @@
+#!/usr/bin/env bash
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Behavioral test for the OpenBao bootstrap functions in deploy.sh.
+#
+# The post-install hook Job retries its own pod (backoffLimit defaults to 6, so
+# up to 7 attempts per helm install), which means initialize_cluster and
+# unseal_cluster both have to survive being re-run against a cluster an earlier
+# attempt already changed. A test that only walked the clean-install path would
+# not notice a regression here, so each case drives the real functions against a
+# stub cluster and asserts the retry-specific behavior directly: no second
+# `bao operator init`, no peer join before the primary reports ha_mode=active,
+# a transient raft challenge 500 retried instead of fatal, and already-joined
+# peers left alone.
+#
+# Both deploy.sh copies carry the same bootstrap logic and are expected to stay
+# in sync, so the whole suite runs against each.
+#
+# Requires bash, jq, and coreutils. No cluster, no network.
+# Run: deploy/helm/openbao/tests/bootstrap/test-deploy-bootstrap.sh
+set -euo pipefail
+
+script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+openbao_dir="$(cd "${script_dir}/../.." && pwd)"
+
+namespace="vault-system"
+statefulset="openbao-server"
+primary="${statefulset}-0"
+
+fail=0
+workdir="$(mktemp -d)"
+trap 'rm -rf "${workdir}"' EXIT
+
+# Sourcing deploy.sh outright would run the installer, so lift out the block
+# that is nothing but function definitions: everything from the first helper
+# down to the line where the main flow begins.
+extract_functions() {
+    local src="$1" log_sh="$2" out="$3"
+    # The bootstrap functions log through the helpers each copy sources at
+    # startup, so bring those along. log.sh has no trailing newline, hence the
+    # explicit separator.
+    cat "${log_sh}" >"${out}"
+    printf '\n' >>"${out}"
+    sed -n '/^# Helper function to get root token$/,/^log_section "Deploying OpenBao cluster/p' "${src}" |
+        sed '$d' >>"${out}"
+    if ! grep -q '^unseal_cluster() {' "${out}" || ! grep -q '^log_warn()' "${out}"; then
+        printf 'FAIL: could not extract functions from %s\n' "${src}"
+        exit 1
+    fi
+}
+
+# A stub kubectl backed by files under $STATE. It records every invocation so a
+# test can assert both what the bootstrap did and the order it did it in.
+stub_bin="${workdir}/bin"
+mkdir -p "${stub_bin}"
+cat >"${stub_bin}/kubectl" <<'STUB'
+#!/usr/bin/env bash
+set -uo pipefail
+printf '%s\n' "$*" >>"${STATE}/calls"
+
+state() { cat "${STATE}/$1" 2>/dev/null || true; }
+
+verb="${1:-}"
+shift || true
+
+case "${verb}" in
+get)
+    case "${1:-}" in
+    pod) printf 'true' ;;
+    secret)
+        # Real kubectl emits the base64 from the secret; callers pipe it
+        # through `base64 -d`. An absent or empty value must decode to "".
+        case "${2:-}" in
+        *-unseal) printf '%s' "$(state unseal_key)" | base64 | tr -d '\n' ;;
+        *-root-token) printf '%s' "$(state root_token)" | base64 | tr -d '\n' ;;
+        esac
+        ;;
+    esac
+    ;;
+patch | create) ;; # recorded above; nothing else to model
+exec)
+    pod="${1:-}"
+    while [ "$#" -gt 0 ] && [ "${1}" != "--" ]; do shift; done
+    shift || true
+    [ "${1:-}" = "bao" ] || exit 0
+    shift
+    case "${1:-}:${2:-}" in
+    status:*)
+        sealed="$(state "sealed.${pod}")"
+        initialized="$(state initialized)"
+        ha_mode="standby"
+        if [ "${pod}" = "${STATE_PRIMARY}" ]; then
+            # Report standby until the configured number of polls has elapsed,
+            # so a caller that joins peers on a fixed sleep is caught.
+            polls="$(state "polls.${pod}")"
+            polls=$((${polls:-0} + 1))
+            printf '%s' "${polls}" >"${STATE}/polls.${pod}"
+            [ "${polls}" -gt "$(state active_after)" ] && ha_mode="active"
+        fi
+        printf '{"initialized":%s,"sealed":%s,"ha_mode":"%s"}\n' \
+            "${initialized:-false}" "${sealed:-true}" "${ha_mode}"
+        ;;
+    operator:init)
+        if [ "$(state initialized)" = "true" ]; then
+            echo "Error initializing: Error making API request." >&2
+            echo "Code: 400. Errors:" >&2
+            echo "* Vault is already initialized" >&2
+            exit 2
+        fi
+        printf 'true' >"${STATE}/initialized"
+        printf 'test-unseal-key' >"${STATE}/unseal_key"
+        printf 'test-root-token' >"${STATE}/root_token"
+        printf '{"unseal_keys_b64":["test-unseal-key"],"root_token":"test-root-token"}\n'
+        ;;
+    operator:unseal)
+        if [ -z "${3:-}" ]; then
+            echo "Error unsealing: no key supplied" >&2
+            exit 2
+        fi
+        printf 'false' >"${STATE}/sealed.${pod}"
+        ;;
+    operator:raft)
+        remaining="$(state join_fail_remaining)"
+        if [ "${remaining:-0}" -gt 0 ]; then
+            printf '%s' "$((remaining - 1))" >"${STATE}/join_fail_remaining"
+            echo "Error joining the node to the Raft cluster: Error making API request." >&2
+            echo "Code: 500. Errors:" >&2
+            echo "* failed to join raft cluster: failed to get raft challenge" >&2
+            exit 2
+        fi
+        printf 'true' >"${STATE}/joined.${pod}"
+        ;;
+    esac
+    ;;
+esac
+exit 0
+STUB
+chmod +x "${stub_bin}/kubectl"
+
+# A cluster that is installed but not yet initialized, with every pod sealed.
+new_state() {
+    local dir
+    dir="$(mktemp -d -p "${workdir}")"
+    printf 'false' >"${dir}/initialized"
+    printf '0' >"${dir}/active_after"
+    printf '0' >"${dir}/join_fail_remaining"
+    : >"${dir}/calls"
+    printf '%s' "${dir}"
+}
+
+# A cluster a previous Job attempt already initialized and stored keys for.
+bootstrapped_state() {
+    local dir
+    dir="$(new_state)"
+    printf 'true' >"${dir}/initialized"
+    printf 'test-unseal-key' >"${dir}/unseal_key"
+    printf 'test-root-token' >"${dir}/root_token"
+    printf '%s' "${dir}"
+}
+
+out=""
+rc=0
+run_fn() { # run_fn <funcs-file> <state-dir> <function> [args...]
+    local funcs="$1" state="$2"
+    shift 2
+    set +e
+    out="$(
+        export STATE="${state}" STATE_PRIMARY="${primary}" PATH="${stub_bin}:${PATH}"
+        # shellcheck disable=SC1090
+        source "${funcs}"
+        # Advance the clock instead of actually waiting. This keeps the polling
+        # loops instant while still letting their $SECONDS timeouts fire, so a
+        # regression that never satisfies a wait condition fails the test
+        # rather than hanging it.
+        sleep() { SECONDS=$((SECONDS + ${1:-0})); }
+        "$@" 2>&1
+    )"
+    rc=$?
+    set -e
+}
+
+expect_rc() { # expect_rc <desc> <want> <got>
+    if [ "$3" != "$2" ]; then
+        printf 'FAIL: %s (want rc %s, got %s)\n%s\n' "$1" "$2" "$3" "${out}"
+        fail=1
+    fi
+}
+
+expect_calls() { # expect_calls <desc> <state> <pattern> <want-count>
+    local got
+    got="$(grep -c -- "$3" "$2/calls" || true)"
+    if [ "${got}" != "$4" ]; then
+        printf 'FAIL: %s (want %s calls matching "%s", got %s)\n' "$1" "$4" "$3" "${got}"
+        fail=1
+    fi
+}
+
+expect_output() { # expect_output <desc> <substring>
+    case "${out}" in
+    *"$2"*) ;;
+    *)
+        printf 'FAIL: %s (output did not contain "%s")\n%s\n' "$1" "$2" "${out}"
+        fail=1
+        ;;
+    esac
+}
+
+run_suite() {
+    local label="$1" src="$2" log_sh="$3" funcs state join_line polls
+    funcs="${workdir}/funcs-${label}.sh"
+    extract_functions "${src}" "${log_sh}" "${funcs}"
+    printf -- '--- %s\n' "${src#"${openbao_dir}/"}"
+
+    # A clean install initializes exactly once.
+    state="$(new_state)"
+    run_fn "${funcs}" "${state}" initialize_cluster "${namespace}" "${statefulset}"
+    expect_rc "${label}: clean install initializes" 0 "${rc}"
+    expect_calls "${label}: clean install runs bao operator init" "${state}" "bao operator init" 1
+
+    # Re-running against that same cluster is the Job's own retry. It must skip
+    # init and reuse the stored keys rather than fail on "already initialized".
+    run_fn "${funcs}" "${state}" initialize_cluster "${namespace}" "${statefulset}"
+    expect_rc "${label}: retry after a completed init succeeds" 0 "${rc}"
+    expect_calls "${label}: retry does not re-run bao operator init" "${state}" "bao operator init" 1
+    expect_output "${label}: retry reports the skip" "already initialized"
+
+    # Initialized, but the generated keys were never persisted. Unrecoverable,
+    # and it has to say so instead of unsealing with an empty key.
+    state="$(new_state)"
+    printf 'true' >"${state}/initialized"
+    run_fn "${funcs}" "${state}" initialize_cluster "${namespace}" "${statefulset}"
+    expect_rc "${label}: initialized without stored keys fails" 1 "${rc}"
+    expect_output "${label}: unrecoverable keys are called out" "unrecoverable"
+    expect_calls "${label}: no unseal is attempted with an empty key" "${state}" "bao operator unseal" 0
+
+    # An unseal secret that is missing or still empty must fail with an
+    # actionable error rather than running `bao operator unseal ""`.
+    state="$(new_state)"
+    printf 'true' >"${state}/initialized"
+    run_fn "${funcs}" "${state}" unseal_cluster "${namespace}" "${statefulset}"
+    expect_rc "${label}: unseal without a stored key fails" 1 "${rc}"
+    expect_output "${label}: the missing unseal key is named" "No unseal key stored"
+    expect_calls "${label}: no unseal is attempted without a key" "${state}" "bao operator unseal" 0
+
+    # No peer may join before the primary reports ha_mode=active. Hold the
+    # primary in standby for three polls; a fixed sleep would join immediately.
+    state="$(bootstrapped_state)"
+    printf '3' >"${state}/active_after"
+    run_fn "${funcs}" "${state}" unseal_cluster "${namespace}" "${statefulset}"
+    expect_rc "${label}: unseal succeeds once the leader goes active" 0 "${rc}"
+    join_line="$(grep -n 'bao operator raft join' "${state}/calls" | head -1 | cut -d: -f1 || true)"
+    if [ -z "${join_line}" ]; then
+        printf 'FAIL: %s: expected a raft join call\n' "${label}"
+        fail=1
+    else
+        polls="$(head -n "$((join_line - 1))" "${state}/calls" | grep -c "${primary} .* bao status" || true)"
+        if [ "${polls}" -lt 4 ]; then
+            printf 'FAIL: %s: peer joined after only %s leader polls, expected to wait for ha_mode=active\n' \
+                "${label}" "${polls}"
+            fail=1
+        fi
+    fi
+
+    # The transient challenge 500 right after leader election is retried.
+    state="$(bootstrapped_state)"
+    printf '2' >"${state}/join_fail_remaining"
+    run_fn "${funcs}" "${state}" unseal_cluster "${namespace}" "${statefulset}"
+    expect_rc "${label}: a transient raft challenge 500 is retried" 0 "${rc}"
+    # Two failures then success for the first peer, one call for the second.
+    expect_calls "${label}: join retried before succeeding" "${state}" "bao operator raft join" 4
+
+    # Peers a previous attempt already unsealed are left alone.
+    state="$(bootstrapped_state)"
+    printf 'false' >"${state}/sealed.${statefulset}-1"
+    printf 'false' >"${state}/sealed.${statefulset}-2"
+    run_fn "${funcs}" "${state}" unseal_cluster "${namespace}" "${statefulset}"
+    expect_rc "${label}: rerun with peers already joined succeeds" 0 "${rc}"
+    expect_calls "${label}: already-joined peers are not rejoined" "${state}" "bao operator raft join" 0
+}
+
+run_suite "helm" "${openbao_dir}/helm/scripts/deploy.sh" "${openbao_dir}/helm/scripts/log.sh"
+run_suite "standalone" "${openbao_dir}/deploy.sh" "${openbao_dir}/utils/log.sh"
+
+if [ "${fail}" -ne 0 ]; then
+    printf '\nFAILED\n'
+    exit 1
+fi
+printf '\nPASS\n'