feat(cluster): separate operator authority from data-plane access

Every destructive /cluster/* verb sat behind the SAME bearer as /items and
/search, so any application key could remove a member, force a partition, or
transfer a shard. There was no way to hand out a client credential without also
handing out the ability to destroy the cluster.

Adds TIDAL_ADMIN_KEY (and TIDAL_ADMIN_KEY_FILE, rotatable without restart like
the others). /cluster/promote, /cluster/partition, /cluster/heal,
/cluster/members/remove, /cluster/reseed and /cluster/shards/{id}/{replicas,
transfer} move into their own router subtree behind an admin gate; the data
bearer now gets 403 there - authenticated but not authorized, distinct from the
401 for a bad token.

Three things this had to get right:

* The admin key must ALSO authenticate. A request carries one Authorization
  header, so if the admin key did not satisfy the bearer gate, an operator
  presenting it would be 401'd before the admin gate ran and the verbs would be
  reachable by nobody. Caught while writing the test, not after.

* A verified sibling node token clears the gate too. Nodes relay operator verbs
  to the leader/target carrying whatever credential the caller sent, and the
  legacy fan-out promote uses the internal marker, so requiring the admin key on
  that hop would partition the control plane.

* The peer-callable verbs stay on the plain bearer. /cluster/catchup (self-heal
  nudge), /cluster/join + /cluster/members (seed-join) and the
  /cluster/reconcile* pair are dialled node-to-node, so gating them would break
  replication and joining.

Absent admin key = previous behavior exactly, plus a startup WARN naming the
exposure, so this is safe to upgrade into. The k8s secret mount is optional:true
because without that a deployment lacking the key would fail to MOUNT and never
start.

Also closes the /cluster/status hole this exposed: it and /cluster/status/local
reported leader identity, membership, term and per-shard applied/lag/commit
seqnos from the UNAUTHENTICATED probe group. They are protected now, which is
what k8s/cluster/networkpolicy.yaml deferred to rather than working around at the
network layer.

And fixes a latent bug found on the way: seed-join discovery, reseed discovery
and the self-heal catch-up nudge read std::env::var("TIDAL_API_KEY") directly,
which yields nothing on a *_FILE-only deployment - the node would dial an
authenticated peer with no credential. They use security::bearer_from_env() now,
which honours both shapes.

Verified: 5 new unit tests; two multi-process runbook tests on real 3-process
clusters (data bearer 403 on promote / 204 on signals, admin key 200 on status
and through the gate on heal; bare /cluster/status 401, 200 with the bearer).
That the authenticated cluster converges at all is the load-bearing assertion -
if moving status behind auth had broken leader discovery, startup would hang.
Full unit suites green (2101 + 162), reseed e2e green, clippy clean.

docs/guides/server-deployment.md

@@ -80,6 +80,8 @@ For everything cluster-specific (topology file, leader promotion, partition/heal
 |----------|---------|---------|
 | `TIDAL_API_KEY` | unset | Bearer token for data routes. **If unset, the server is UNAUTHENTICATED** and logs a WARN at startup. See [section 4](#4-authentication). |
 | `TIDAL_API_KEY_FILE` | unset | Path whose CONTENT is the bearer token (m11p7); takes precedence over `TIDAL_API_KEY`. The FILE form rotates **without a restart** (a credential poller re-reads it). |
+| `TIDAL_ADMIN_KEY` | unset | OPERATOR credential gating the destructive `/cluster/*` verbs. **If unset, those verbs accept the data bearer** and the server logs a WARN. See [section 4](#4-authentication). |
+| `TIDAL_ADMIN_KEY_FILE` | unset | Path whose CONTENT is the operator key; takes precedence over `TIDAL_ADMIN_KEY` and rotates without a restart. |
 | `TIDAL_CONFIG` | unset | Config directory (backs `--config-dir`). |
 | `PORT` | unset | Listen port/address (backs `--listen`); bare port → `0.0.0.0:PORT`. |
 | `TIDAL_SERVER_LOG` | `info` | `tracing` env-filter directive, e.g. `TIDAL_SERVER_LOG=tidal_server=debug,info`. |
@@ -255,6 +257,54 @@ curl -H "Authorization: Bearer $TIDAL_API_KEY" \
      "http://localhost:9400/feed?user_id=42&profile=for_you&limit=20"
 ```
 
+### Two credential tiers (cluster mode)
+
+A cluster has **two** credentials, because one is not enough: every `/cluster/*`
+mutation used to sit behind the same bearer as `/items` and `/search`, so any
+application key could remove a member, force a partition, or transfer a shard.
+
+| Credential | Env | Grants |
+|---|---|---|
+| Data bearer | `TIDAL_API_KEY` / `TIDAL_API_KEY_FILE` | The data routes, `/cluster/status`, and the peer-callable verbs (`/cluster/catchup`, `/cluster/join`, `/cluster/members`, `/cluster/reconcile*`). |
+| Operator key | `TIDAL_ADMIN_KEY` / `TIDAL_ADMIN_KEY_FILE` | Everything above **plus** the destructive verbs: `/cluster/promote`, `/cluster/partition`, `/cluster/heal`, `/cluster/members/remove`, `/cluster/reseed`, `/cluster/shards/{id}/replicas`, `/cluster/shards/{id}/transfer`. |
+
+Presenting the data bearer to a destructive verb returns **`403`** (authenticated,
+not authorized) — distinct from the `401` a missing/invalid token returns.
+
+The operator key is a **superset** credential: it also satisfies the
+authentication gate, because a request carries a single `Authorization` header —
+if it did not authenticate, an operator presenting it would be rejected `401`
+before the authorization gate ran. Keep it off application hosts.
+
+A **verified sibling node token** also clears the gate. That is load-bearing, not
+a convenience: nodes relay operator verbs to the leader/target carrying whatever
+credential the caller sent, so enabling an operator key can never partition the
+control plane.
+
+> **WARNING — one key means every client is an operator.** If `TIDAL_ADMIN_KEY` is
+> unset, the destructive verbs accept the data bearer and the server logs a
+> startup WARN naming the exposure. Behavior is unchanged from before the split,
+> so this is safe to upgrade into, but set the key before handing a bearer to
+> anything you do not operate.
+
+```bash
+export TIDAL_API_KEY="$(openssl rand -hex 32)"
+export TIDAL_ADMIN_KEY="$(openssl rand -hex 32)"
+
+# 403: the data key is not an operator credential
+curl -X POST -H "Authorization: Bearer $TIDAL_API_KEY" \
+     -d '{"region":"eu-west"}' http://localhost:9500/cluster/promote
+
+# accepted
+curl -X POST -H "Authorization: Bearer $TIDAL_ADMIN_KEY" \
+     -d '{"region":"eu-west"}' http://localhost:9500/cluster/promote
+```
+
+`/cluster/status` and `/cluster/status/local` require a credential too. They
+report leader identity, membership, term, and per-shard applied/lag/commit
+seqnos — reconnaissance, not a probe — so they are **not** in the open group with
+`/health`.
+
 ---
 
 ## 5. The Served OpenAPI Reference (`GET /openapi.json`)

k8s/cluster/ingress.yaml

@@ -4,14 +4,15 @@
 # it out of kustomization.yaml for an internal-only deployment.
 #
 # WHAT THIS DELIBERATELY DOES NOT PUBLISH
-#   * `/cluster/*` - every mutating admin verb (`promote`, `partition`, `heal`,
+#   * `/cluster/*` - the destructive admin verbs (`promote`, `partition`, `heal`,
-#     `members/remove`, `join`, `reseed`, `shards/{id}/transfer`) sits behind the
+#     `members/remove`, `reseed`, `shards/{id}/transfer`) now require the separate
-#     SAME single bearer token as the data routes (tidal-server cluster/node.rs
+#     OPERATOR credential (`TIDAL_ADMIN_KEY`), so a data-plane key gets 403. They
-#     `protected`). There is no operator/data credential split, so publishing
+#     stay unpublished anyway: operator authority has no business being reachable
-#     these would let any client key destroy the cluster.
+#     from the internet, and this is the layer that makes that unconditional.
-#   * `/cluster/status` + `/cluster/status/local` - these are UNAUTHENTICATED
+#   * `/cluster/status` + `/cluster/status/local` - now authenticated (they moved
-#     (they live in the `public` router next to the health probes) and report
+#     out of the open probe group), but still not published: they report leader
-#     leader identity, membership, term, and per-shard applied/lag/commit seqnos.
+#     identity, membership, term, and per-shard applied/lag/commit seqnos, which
+#     no external client needs.
 #   * `/openapi.json` - unauthenticated, and enumerates the admin routes above.
 #   * `/metrics` - never reachable here: :9091 is published only on the headless
 #     `tidaldb-peers` Service, not on the client Service this Ingress targets.

k8s/cluster/kustomization.yaml

@@ -7,12 +7,16 @@
 # standalone set's replicas:1 is load-bearing).
 #
 # Create the credentials secret FIRST (deliberately excluded so no key is
-# committed — see secret.example.yaml). m11p7 shape carries BOTH the bearer and
+# committed - see secret.example.yaml). Three keys: the data-plane bearer, the
-# the cluster key:
+# OPERATOR key that gates the destructive /cluster/* verbs, and the cluster key:
 #   kubectl -n tidaldb-cluster create secret generic tidaldb-credentials \
 #     --from-literal=TIDAL_API_KEY="$(openssl rand -hex 32)" \
+#     --from-literal=TIDAL_ADMIN_KEY="$(openssl rand -hex 32)" \
 #     --from-literal=TIDAL_CLUSTER_KEY="$(openssl rand -hex 32)"
 #
+# Omitting TIDAL_ADMIN_KEY is supported (the mount is optional) but leaves every
+# admin verb reachable with the data bearer; the server WARNs at startup.
+#
 # certs.yaml (m11p7 inter-node TLS) requires cert-manager. If you do NOT run
 # cert-manager, comment certs.yaml out and provision the `tidaldb-cluster-tls`
 # Secret with scripts/gen-cluster-certs.sh instead.

k8s/cluster/secret.example.yaml

@@ -2,11 +2,19 @@
 # out-of-band and is deliberately excluded from kustomization.yaml so no key
 # lands in git.
 #
-# Secret shape: `tidaldb-credentials` with TWO keys (m11p5 §4 + m11p7):
+# Secret shape: `tidaldb-credentials` with up to THREE keys:
-#   - TIDAL_API_KEY    — the external/operator bearer (the stress/Ref-A lineage;
+#   - TIDAL_API_KEY    — the DATA-PLANE bearer. Every pod and every external
-#     the StatefulSet injects it as the `TIDAL_API_KEY` env var). EVERY pod and
+#     client uses this one (forwarded requests pass the caller's Authorization
-#     EVERY external client uses the same bearer (forwarded requests pass the
+#     verbatim). Hand this to applications.
-#     caller's Authorization verbatim).
+#   - TIDAL_ADMIN_KEY  — OPTIONAL but strongly recommended: the OPERATOR
+#     credential. Without it the destructive verbs (/cluster/promote,
+#     /cluster/partition, /cluster/heal, /cluster/members/remove,
+#     /cluster/reseed, /cluster/shards/*) accept the DATA bearer, so any
+#     application key can remove a member or move a shard. With it they require
+#     this key (or a verified sibling node token) and the data bearer gets 403.
+#     It is a SUPERSET credential - it also authenticates the data routes - so
+#     keep it off application hosts. Mounted as a FILE
+#     (TIDAL_ADMIN_KEY_FILE) so rotation needs no restart.
 #   - TIDAL_CLUSTER_KEY — m11p7: the SHARED CLUSTER KEY. Mints/verifies per-node
 #     signed internal tokens so inter-node HTTP carries verifiable node identity
 #     and the x-tidal-internal marker is honored ONLY from a verified sibling.
@@ -16,6 +24,7 @@
 # Create the real one (do not apply this file):
 #   kubectl -n tidaldb-cluster create secret generic tidaldb-credentials \
 #     --from-literal=TIDAL_API_KEY="$(openssl rand -hex 32)" \
+#     --from-literal=TIDAL_ADMIN_KEY="$(openssl rand -hex 32)" \
 #     --from-literal=TIDAL_CLUSTER_KEY="$(openssl rand -hex 32)"
 #
 # Inter-node TLS material is a SEPARATE Secret (`tidaldb-cluster-tls`), issued by
@@ -36,4 +45,5 @@ metadata:
 type: Opaque
 stringData:
   TIDAL_API_KEY: "replace-me-do-not-commit"
+  TIDAL_ADMIN_KEY: "replace-me-do-not-commit-distinct-from-api-key"
   TIDAL_CLUSTER_KEY: "replace-me-do-not-commit-distinct-from-api-key"

k8s/cluster/statefulset.yaml

@@ -185,6 +185,13 @@ spec:
             # credential poller. Distinct secret data key from the bearer.
             - name: TIDAL_CLUSTER_KEY_FILE
               value: /etc/tidaldb/cluster-key/cluster-key
+            # The cluster-admin credential. Pointing at a path that may not exist
+            # is safe and deliberate: the server treats an unreadable admin-key
+            # file as "not configured" and WARNs, so the gate is opt-in. Populate
+            # TIDAL_ADMIN_KEY in tidaldb-credentials to enable it - no restart
+            # needed, the credential poller picks the file up.
+            - name: TIDAL_ADMIN_KEY_FILE
+              value: /etc/tidaldb/admin-key/admin-key
             - name: TIDAL_SERVER_LOG
               value: info
             - name: TIDAL_ALLOW_EXPERIMENTAL_CLUSTER
@@ -316,6 +323,9 @@ spec:
             - name: cluster-key
               mountPath: /etc/tidaldb/cluster-key
               readOnly: true
+            - name: admin-key
+              mountPath: /etc/tidaldb/admin-key
+              readOnly: true
             - name: tmp
               mountPath: /tmp
       volumes:
@@ -336,6 +346,20 @@ spec:
             items:
               - key: TIDAL_CLUSTER_KEY
                 path: cluster-key
+        # The cluster-ADMIN key: the separate credential the destructive
+        # /cluster/* verbs require. `optional: true` is load-bearing - without it
+        # a deployment whose Secret has no TIDAL_ADMIN_KEY would fail to MOUNT and
+        # never start. Absent ⇒ the server logs a startup WARN and the admin verbs
+        # keep accepting the data bearer (previous behavior). Add the key to
+        # tidaldb-credentials to turn the gate on; the file form means a rotation
+        # is picked up without a restart.
+        - name: admin-key
+          secret:
+            secretName: tidaldb-credentials
+            optional: true
+            items:
+              - key: TIDAL_ADMIN_KEY
+                path: admin-key
         - name: tmp
           emptyDir: {}
   volumeClaimTemplates:

tidal-server/src/cluster/join_boot.rs

@@ -501,7 +501,7 @@ fn join_via_leader(
         "http_addr": advertise_http,
     });
     let mut req = client.post(&url).json(&body);
-    if let Ok(key) = std::env::var("TIDAL_API_KEY") {
+    if let Some(key) = crate::cluster::security::bearer_from_env() {
         req = req.bearer_auth(key);
     }
     let resp = req
@@ -559,7 +559,7 @@ fn current_leader_dial_target(
 ) -> Option<(String, u16)> {
     let url = super::forward::peer_url(gateway_http, "/cluster/status/local");
     let mut req = client.get(&url);
-    if let Ok(key) = std::env::var("TIDAL_API_KEY") {
+    if let Some(key) = crate::cluster::security::bearer_from_env() {
         req = req.bearer_auth(key);
     }
     let json: serde_json::Value = req.send().ok()?.json().ok()?;
@@ -574,7 +574,7 @@ fn current_leader_dial_target(
 fn fetch_roster(client: &reqwest::blocking::Client, http_addr: &str) -> Result<Vec<MemberEntry>> {
     let url = super::forward::peer_url(http_addr, "/cluster/members");
     let mut req = client.get(&url);
-    if let Ok(key) = std::env::var("TIDAL_API_KEY") {
+    if let Some(key) = crate::cluster::security::bearer_from_env() {
         req = req.bearer_auth(key);
     }
     let resp = req

tidal-server/src/cluster/node.rs

@@ -3413,7 +3413,7 @@ impl ShardReplica {
                 .blocking_client
                 .get(&url)
                 .timeout(forward::STATUS_PEER_TIMEOUT);
-            if let Ok(key) = std::env::var("TIDAL_API_KEY") {
+            if let Some(key) = crate::cluster::security::bearer_from_env() {
                 req = req.bearer_auth(key);
             }
             if let Ok(resp) = req.send()
@@ -4992,15 +4992,46 @@ pub fn build_region_router(
     node: Arc<ClusterNode>,
     creds: Arc<crate::cluster::security::ClusterCreds>,
 ) -> Router {
+    // ONLY the probe contract and the OpenAPI document are unauthenticated.
+    //
+    // `/cluster/status` and `/cluster/status/local` used to live here. They report
+    // leader identity, membership, term, and per-shard applied/lag/commit seqnos,
+    // which is reconnaissance rather than a probe, so they moved to `protected`.
+    // Every internal caller already authenticates: seed-join and reseed leader
+    // discovery and the status fan-out all send the bearer (via
+    // `security::bearer_from_env`, which honours the `*_FILE` form too).
     let public = Router::new()
         .route("/health", get(region_health))
         .route("/health/startup", get(crate::health::health_startup))
         .route("/health/live", get(crate::health::health_live))
-        .route("/cluster/status/local", get(status_local))
-        .route("/cluster/status", get(cluster_status))
         .route("/openapi.json", get(crate::openapi::serve_region))
         .with_state(Arc::clone(&node));
 
+    // The DESTRUCTIVE operator verbs. Separated from the data surface because
+    // they used to share its credential: one bearer let any client key remove a
+    // member, force a partition, or move a shard. `admin_gate` requires the
+    // admin key (or a verified sibling) once one is configured.
+    //
+    // Peer-callable verbs deliberately DO NOT live here - `/cluster/catchup`
+    // (self-heal nudge), `/cluster/join` + `/cluster/members` (seed-join), and
+    // the `/cluster/reconcile*` pair are all dialled node-to-node with the plain
+    // bearer, so gating them on the admin key would break replication and
+    // joining.
+    let admin_creds = Arc::clone(&creds);
+    let admin = Router::new()
+        .route("/cluster/promote", post(cluster_promote))
+        .route("/cluster/partition", post(cluster_partition))
+        .route("/cluster/heal", post(cluster_heal))
+        .route("/cluster/reseed", post(cluster_reseed))
+        .route("/cluster/members/remove", post(cluster_member_remove))
+        // m11p6 L3 rebalancing verbs (per-group, reusing the m11p5 machinery).
+        .route("/cluster/shards/{id}/replicas", post(shard_replicas))
+        .route("/cluster/shards/{id}/transfer", post(shard_transfer))
+        .layer(middleware::from_fn(move |req: Request, next: Next| {
+            admin_gate(Arc::clone(&admin_creds), req, next)
+        }))
+        .with_state(Arc::clone(&node));
+
     let protected = Router::new()
         .route("/items", post(create_item))
         .route("/embeddings", post(write_embedding))
@@ -5009,18 +5040,13 @@ pub fn build_region_router(
         .route("/feed", get(feed))
         .route("/search", get(search))
         .route("/vector_search", post(vector_search))
-        .route("/cluster/promote", post(cluster_promote))
-        .route("/cluster/partition", post(cluster_partition))
-        .route("/cluster/heal", post(cluster_heal))
         .route("/cluster/catchup", post(cluster_catchup))
-        .route("/cluster/reseed", post(cluster_reseed))
         .route("/cluster/reconcile", post(cluster_reconcile))
         .route("/cluster/members", get(cluster_members))
-        .route("/cluster/members/remove", post(cluster_member_remove))
         .route("/cluster/join", post(cluster_join))
-        // m11p6 L3 rebalancing verbs (per-group, reusing the m11p5 machinery).
+        // Moved out of `public`: topology reconnaissance, not a probe.
-        .route("/cluster/shards/{id}/replicas", post(shard_replicas))
+        .route("/cluster/status/local", get(status_local))
-        .route("/cluster/shards/{id}/transfer", post(shard_transfer))
+        .route("/cluster/status", get(cluster_status))
         // The snapshot body is corpus-sized, so it gets its own cap BEFORE the
         // group's data-surface limit applies (an inner layer wins). See
         // `RECONCILE_BODY_LIMIT_BYTES` for why the shared 2 MiB made
@@ -5036,10 +5062,11 @@ pub fn build_region_router(
         .route("/sharded/signals", post(sharded_write_signal))
         .route("/sharded/feed", get(sharded_feed))
         .route("/sharded/search", get(sharded_search))
+        .with_state(node)
+        .merge(admin)
         .layer(axum::extract::DefaultBodyLimit::max(
             crate::router::BODY_LIMIT_BYTES,
-        ))
+        ));
-        .with_state(node);
 
     // m11p7 cluster auth, in one layer so `next` runs at most once. Extracted to
     // `cluster_auth_middleware` so the assembled-router composition (bearer ->
@@ -5083,9 +5110,7 @@ async fn cluster_auth_middleware(
     req: Request,
     next: Next,
 ) -> Response {
-    if let Some(key) = creds.bearer()
+    if !creds.authenticated(req.headers()) {
-        && !crate::router::bearer_token_ok(req.headers(), &key)
-    {
         return crate::router::unauthorized_response();
     }
     let marked =
@@ -5108,6 +5133,25 @@ async fn cluster_auth_middleware(
     next.run(req).await
 }
 
+/// Authorization gate for the DESTRUCTIVE cluster verbs, layered INSIDE
+/// [`cluster_auth_middleware`] so the bearer/marker/rate gates run first and this
+/// only decides authority.
+///
+/// Rejects with 403 (authenticated but not permitted) rather than 401 — the
+/// caller's bearer was valid, it simply is not an operator credential. Passes
+/// through untouched when no admin key is configured, so an existing deployment
+/// behaves exactly as before (with a startup WARN from `ClusterCreds::from_env`).
+async fn admin_gate(
+    creds: Arc<crate::cluster::security::ClusterCreds>,
+    req: Request,
+    next: Next,
+) -> Response {
+    if creds.admin_ok(req.headers()) {
+        return next.run(req).await;
+    }
+    crate::router::admin_forbidden_response()
+}
+
 // ── Health ──────────────────────────────────────────────────────────────────
 
 #[allow(clippy::significant_drop_tightening)]
@@ -5354,8 +5398,10 @@ pub struct ShardStatusRow {
     tag = "cluster",
     responses(
         (status = 200, description = "This region's local status", body = LocalStatusResponse),
+        (status = 401, description = "Missing or invalid credential"),
         (status = 503, description = "Server shutting down"),
     ),
+    security(("bearerAuth" = [])),
 )]
 #[allow(clippy::significant_drop_tightening)]
 pub async fn status_local(
@@ -5444,7 +5490,9 @@ pub struct AggregatedRegionStatus {
     tag = "cluster",
     responses(
         (status = 200, description = "Cluster-wide replication status", body = AggregatedStatusResponse),
+        (status = 401, description = "Missing or invalid credential"),
     ),
+    security(("bearerAuth" = [])),
 )]
 #[allow(clippy::significant_drop_tightening)]
 pub async fn cluster_status(

tidal-server/src/cluster/reseed.rs

@@ -500,7 +500,7 @@ pub fn run_boot_install_with(
     // m11p7: a TLS cluster's reseed discovery dials `https://` and trusts the
     // cluster CA, matching the rest of the inter-node HTTP plane.
     super::forward::set_inter_node_https(tls.is_some());
-    let api_key = std::env::var("TIDAL_API_KEY").ok();
+    let api_key = crate::cluster::security::bearer_from_env();
     let deadline = Instant::now() + handshake_window;
     let mut backoff = BACKOFF_MIN;
 

tidal-server/src/cluster/routes.rs

@@ -58,42 +58,62 @@ pub fn build_cluster_router(
         // modes can never advertise a different probe contract.
         .route("/health/startup", get(crate::health::health_startup))
         .route("/health/live", get(crate::health::health_live))
-        .route("/cluster/status", get(cluster_status))
+        // `/cluster/status` deliberately NOT here - see the region router: it
+        // reports leader, membership and seqnos, so it is protected.
         // Cluster-superset OpenAPI document (adds the /cluster/* and /sharded/*
         // routes). Unauthenticated, like the probes — contract only, no data.
         .route("/openapi.json", get(crate::openapi::serve_cluster))
         .with_state(Arc::clone(&state));
 
+    // The destructive operator verbs, gated on the cluster-admin credential when
+    // one is configured. Same rationale as the multi-process region router: these
+    // used to share the data-plane bearer, so any client key could promote,
+    // partition, or heal.
+    let admin_creds = Arc::clone(&creds);
+    let admin = Router::new()
+        .route("/cluster/promote", post(cluster_promote))
+        .route("/cluster/partition", post(cluster_partition))
+        .route("/cluster/heal", post(cluster_heal))
+        .layer(middleware::from_fn(move |req: Request, next: Next| {
+            let creds = Arc::clone(&admin_creds);
+            async move {
+                if creds.admin_ok(req.headers()) {
+                    return next.run(req).await;
+                }
+                crate::router::admin_forbidden_response()
+            }
+        }))
+        .with_state(Arc::clone(&state));
+
     let protected = Router::new()
         .route("/items", post(create_item))
         .route("/embeddings", post(write_embedding))
         .route("/signals", post(write_signal))
         .route("/feed", get(feed))
         .route("/search", get(search))
-        .route("/cluster/promote", post(cluster_promote))
+        .route("/cluster/status", get(cluster_status))
-        .route("/cluster/partition", post(cluster_partition))
-        .route("/cluster/heal", post(cluster_heal))
         // Sharded (scatter-gather) routes.
         .route("/sharded/items", post(sharded_create_item))
         .route("/sharded/embeddings", post(sharded_write_embedding))
         .route("/sharded/signals", post(sharded_write_signal))
         .route("/sharded/feed", get(sharded_feed))
         .route("/sharded/search", get(sharded_search))
+        .with_state(state)
+        .merge(admin)
         // Shared with the standalone router so the body cap can never drift
         // (raise one, forget the other). See [`crate::router::BODY_LIMIT_BYTES`].
         .layer(axum::extract::DefaultBodyLimit::max(
             crate::router::BODY_LIMIT_BYTES,
-        ))
+        ));
-        .with_state(state);
 
-    // Bearer key read PER REQUEST from `creds` so a rotation takes effect with no
+    // Credentials read PER REQUEST from `creds` so a rotation takes effect with no
-    // restart (m11p7). No key configured ⇒ pass through (open).
+    // restart (m11p7). No bearer configured ⇒ pass through (open). The admin key
+    // also authenticates here (it is a superset credential) so an operator
+    // presenting it is not rejected before the admin gate runs.
     let protected = protected.layer(middleware::from_fn(move |req: Request, next: Next| {
         let creds = Arc::clone(&creds);
         async move {
-            if let Some(key) = creds.bearer()
+            if !creds.authenticated(req.headers()) {
-                && !crate::router::bearer_token_ok(req.headers(), &key)
-            {
                 return crate::router::unauthorized_response();
             }
             let principal = creds.principal(req.headers());
@@ -194,7 +214,9 @@ pub struct RegionStatus {
     tag = "cluster",
     responses(
         (status = 200, description = "Cluster replication status", body = ClusterStatusResponse),
+        (status = 401, description = "Missing or invalid credential"),
     ),
+    security(("bearerAuth" = [])),
 )]
 pub async fn cluster_status(
     State(state): State<Arc<ClusterState>>,

tidal-server/src/cluster/security.rs

@@ -66,15 +66,25 @@ fn derive_key(secret: &[u8]) -> [u8; 32] {
     *blake3::hash(secret).as_bytes()
 }
 
-/// The reloadable HTTP-plane credentials (bearer key + cluster key).
+/// The reloadable HTTP-plane credentials (bearer key + admin key + cluster key).
 ///
-/// Both values live behind [`ArcSwapOption`] so [`reload`](Self::reload) can swap
+/// Every value lives behind [`ArcSwapOption`] so [`reload`](Self::reload) can swap
 /// them under load with no lock; the live request path reads them with a cheap
 /// atomic load. File-backed sources are re-read on reload (the rotation path);
 /// env-only sources are static (no file to watch).
 pub struct ClusterCreds {
     bearer: ArcSwapOption<String>,
     bearer_file: Option<PathBuf>,
+    /// The cluster-ADMIN key: the separate credential required by the
+    /// destructive `/cluster/*` verbs when it is configured.
+    ///
+    /// WHY THIS EXISTS: every admin verb used to sit behind the SAME bearer as
+    /// `/items` and `/search`, so any client key could remove a member, force a
+    /// partition, or transfer a shard. Absent ⇒ the admin verbs keep accepting
+    /// the plain bearer (backward compatible) and [`from_env`](Self::from_env)
+    /// emits a WARN naming the exposure.
+    admin: ArcSwapOption<String>,
+    admin_file: Option<PathBuf>,
     cluster_key: ArcSwapOption<[u8; 32]>,
     cluster_key_file: Option<PathBuf>,
     /// m11p7 per-principal HTTP rate limiter (the engine's token bucket, reused).
@@ -90,8 +100,10 @@ impl std::fmt::Debug for ClusterCreds {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("ClusterCreds")
             .field("bearer_configured", &self.bearer.load().is_some())
+            .field("admin_configured", &self.admin.load().is_some())
             .field("cluster_key_configured", &self.cluster_key.load().is_some())
             .field("bearer_file", &self.bearer_file)
+            .field("admin_file", &self.admin_file)
             .field("cluster_key_file", &self.cluster_key_file)
             .finish_non_exhaustive()
     }
@@ -126,12 +138,16 @@ impl ClusterCreds {
     /// rotates without restart.
     ///
     /// - bearer: `TIDAL_API_KEY_FILE` else `TIDAL_API_KEY`
+    /// - admin: `TIDAL_ADMIN_KEY_FILE` else `TIDAL_ADMIN_KEY`
     /// - cluster key: `TIDAL_CLUSTER_KEY_FILE` else `TIDAL_CLUSTER_KEY`
     #[must_use]
     pub fn from_env() -> Self {
         let bearer_file = file_source("TIDAL_API_KEY_FILE");
         let bearer = read_bearer(bearer_file.as_deref(), "TIDAL_API_KEY");
 
+        let admin_file = file_source("TIDAL_ADMIN_KEY_FILE");
+        let admin = read_bearer(admin_file.as_deref(), "TIDAL_ADMIN_KEY");
+
         let cluster_key_file = file_source("TIDAL_CLUSTER_KEY_FILE");
         let cluster_key = read_cluster_key(cluster_key_file.as_deref(), "TIDAL_CLUSTER_KEY");
 
@@ -143,9 +159,25 @@ impl ClusterCreds {
             );
         }
 
+        // Only worth warning about when a bearer IS set: an entirely open server
+        // already emits the louder unauthenticated warning, and adding a second
+        // one there would just be noise.
+        if admin.is_none() && bearer.is_some() {
+            tracing::warn!(
+                "TIDAL_ADMIN_KEY is not set — the destructive cluster verbs \
+                 (/cluster/promote, /cluster/partition, /cluster/heal, \
+                 /cluster/members/remove, /cluster/reseed, /cluster/shards/*) accept the \
+                 SAME bearer as the data routes, so any client key can remove a member, \
+                 force a partition, or transfer a shard. Set an admin key to separate \
+                 operator authority from data-plane access."
+            );
+        }
+
         Self {
             bearer: ArcSwapOption::from(bearer.map(Arc::new)),
             bearer_file,
+            admin: ArcSwapOption::from(admin.map(Arc::new)),
+            admin_file,
             cluster_key: ArcSwapOption::from(cluster_key.map(Arc::new)),
             cluster_key_file,
             rate_limiter: rate_limiter_from_env(),
@@ -160,6 +192,8 @@ impl ClusterCreds {
         Self {
             bearer: ArcSwapOption::from(None),
             bearer_file: None,
+            admin: ArcSwapOption::from(None),
+            admin_file: None,
             cluster_key: ArcSwapOption::from(None),
             cluster_key_file: None,
             rate_limiter: tidaldb::RateLimiter::new(tidaldb::RateLimiterConfig::default()),
@@ -171,9 +205,23 @@ impl ClusterCreds {
     /// directly rather than via env/files.
     #[must_use]
     pub fn with_keys(bearer: Option<String>, cluster_key: Option<&str>) -> Self {
+        Self::with_keys_admin(bearer, None, cluster_key)
+    }
+
+    /// Build creds with an explicit bearer, ADMIN key, and optional cluster key,
+    /// no file sources. The admin key is what separates operator authority from
+    /// data-plane access on the destructive `/cluster/*` verbs.
+    #[must_use]
+    pub fn with_keys_admin(
+        bearer: Option<String>,
+        admin: Option<String>,
+        cluster_key: Option<&str>,
+    ) -> Self {
         Self {
             bearer: ArcSwapOption::from(bearer.map(Arc::new)),
             bearer_file: None,
+            admin: ArcSwapOption::from(admin.map(Arc::new)),
+            admin_file: None,
             cluster_key: ArcSwapOption::from(
                 cluster_key.map(|s| Arc::new(derive_key(s.as_bytes()))),
             ),
@@ -189,6 +237,8 @@ impl ClusterCreds {
         Self {
             bearer: ArcSwapOption::from(None),
             bearer_file: None,
+            admin: ArcSwapOption::from(None),
+            admin_file: None,
             cluster_key: ArcSwapOption::from(None),
             cluster_key_file: None,
             rate_limiter: tidaldb::RateLimiter::new(tidaldb::RateLimiterConfig::limited(
@@ -207,6 +257,8 @@ impl ClusterCreds {
         Self {
             bearer: ArcSwapOption::from(None),
             bearer_file: None,
+            admin: ArcSwapOption::from(None),
+            admin_file: None,
             cluster_key: ArcSwapOption::from(Some(Arc::new(derive_key(cluster_key.as_bytes())))),
             cluster_key_file: None,
             rate_limiter: tidaldb::RateLimiter::new(tidaldb::RateLimiterConfig::limited(
@@ -241,6 +293,67 @@ impl ClusterCreds {
         self.bearer.load_full()
     }
 
+    /// The current cluster-admin key, if one is configured.
+    #[must_use]
+    pub fn admin(&self) -> Option<Arc<String>> {
+        self.admin.load_full()
+    }
+
+    /// Whether a separate cluster-admin key is configured. When `false` the admin
+    /// verbs accept the plain bearer (pre-existing behavior).
+    #[must_use]
+    pub fn admin_configured(&self) -> bool {
+        self.admin.load().is_some()
+    }
+
+    /// Whether this request may invoke a DESTRUCTIVE cluster verb.
+    ///
+    /// * No admin key configured ⇒ `true`. The bearer gate has already run, so
+    ///   this preserves the pre-existing single-credential behavior exactly;
+    ///   [`from_env`](Self::from_env) warns about the exposure at startup.
+    /// * Admin key configured ⇒ requires EITHER the admin key presented as
+    ///   `Authorization: Bearer <admin>`, OR a verified sibling node token.
+    ///
+    /// The node-token arm is load-bearing, not a convenience: siblings relay
+    /// operator verbs to the leader/target (`/cluster/promote` and
+    /// `/cluster/shards/{id}/transfer` are forwarded with the caller's
+    /// `Authorization` verbatim, and the legacy fan-out promote carries the
+    /// internal marker). Accepting a verified node keeps every inter-node path
+    /// working no matter which credential it forwarded, so enabling an admin key
+    /// can never partition the control plane.
+    #[must_use]
+    pub fn admin_ok(&self, headers: &HeaderMap) -> bool {
+        let Some(admin) = self.admin() else {
+            return true;
+        };
+        if crate::router::bearer_token_ok(headers, &admin) {
+            return true;
+        }
+        self.principal(headers).is_node()
+    }
+
+    /// Whether the request satisfies the AUTHENTICATION gate (as opposed to the
+    /// admin authorization gate).
+    ///
+    /// True when no bearer is configured (open server), or the presented token
+    /// matches EITHER the data bearer or the admin key. The admin key is a
+    /// superset credential, and it has to be: one request carries a single
+    /// `Authorization` header, so if the admin key did not also authenticate then
+    /// an operator presenting it would be rejected 401 by the bearer gate before
+    /// [`admin_ok`](Self::admin_ok) ever ran, making the admin verbs
+    /// unreachable by anyone.
+    #[must_use]
+    pub fn authenticated(&self, headers: &HeaderMap) -> bool {
+        let Some(bearer) = self.bearer() else {
+            return true;
+        };
+        if crate::router::bearer_token_ok(headers, &bearer) {
+            return true;
+        }
+        self.admin()
+            .is_some_and(|admin| crate::router::bearer_token_ok(headers, &admin))
+    }
+
     /// Whether a cluster key is configured (per-node tokens are enabled).
     #[must_use]
     pub fn cluster_key_enabled(&self) -> bool {
@@ -263,6 +376,16 @@ impl ClusterCreds {
             }
         }
 
+        if let Some(path) = &self.admin_file
+            && let Some(fresh) = read_file_secret(path)
+        {
+            let differs = self.admin().is_none_or(|cur| *cur != fresh);
+            if differs {
+                self.admin.store(Some(Arc::new(fresh)));
+                changed = true;
+            }
+        }
+
         if let Some(path) = &self.cluster_key_file
             && let Some(fresh) = read_file_secret(path)
         {
@@ -413,6 +536,23 @@ fn read_bearer(file: Option<&std::path::Path>, env: &str) -> Option<String> {
     read_env_secret(env)
 }
 
+/// The bearer key as the SERVER would resolve it: `TIDAL_API_KEY_FILE` content
+/// first, else the inline `TIDAL_API_KEY`.
+///
+/// For the boot/self-heal paths that dial a peer BEFORE (or outside) a live
+/// [`ClusterCreds`] — seed-join leader discovery, reseed discovery, and the
+/// self-heal catch-up nudge. Those read `std::env::var("TIDAL_API_KEY")`
+/// directly, which silently yields nothing on a `*_FILE`-only deployment: the
+/// node then dials an authenticated peer with NO credential. Use this instead so
+/// both credential shapes behave identically.
+#[must_use]
+pub fn bearer_from_env() -> Option<String> {
+    read_bearer(
+        file_source("TIDAL_API_KEY_FILE").as_deref(),
+        "TIDAL_API_KEY",
+    )
+}
+
 /// Read + derive the 32-byte cluster key from its file source else the inline
 /// env var.
 fn read_cluster_key(file: Option<&std::path::Path>, env: &str) -> Option<[u8; 32]> {
@@ -479,6 +619,8 @@ mod tests {
         ClusterCreds {
             bearer: ArcSwapOption::from(None),
             bearer_file: None,
+            admin: ArcSwapOption::from(None),
+            admin_file: None,
             cluster_key: ArcSwapOption::from(Some(Arc::new(derive_key(secret.as_bytes())))),
             cluster_key_file: None,
             rate_limiter: tidaldb::RateLimiter::new(tidaldb::RateLimiterConfig::default()),
@@ -527,6 +669,8 @@ mod tests {
         let creds = ClusterCreds {
             bearer: ArcSwapOption::from(None),
             bearer_file: None,
+            admin: ArcSwapOption::from(None),
+            admin_file: None,
             cluster_key: ArcSwapOption::from(None),
             cluster_key_file: None,
             rate_limiter: tidaldb::RateLimiter::new(tidaldb::RateLimiterConfig::default()),
@@ -572,6 +716,8 @@ mod tests {
         let creds = ClusterCreds {
             bearer: ArcSwapOption::from(None),
             bearer_file: None,
+            admin: ArcSwapOption::from(None),
+            admin_file: None,
             cluster_key: ArcSwapOption::from(Some(Arc::new(derive_key(b"first-key")))),
             cluster_key_file: Some(key_path.clone()),
             rate_limiter: tidaldb::RateLimiter::new(tidaldb::RateLimiterConfig::default()),
@@ -588,4 +734,112 @@ mod tests {
         let t2 = creds.mint_node_token("n").unwrap();
         assert_eq!(creds.verify_node_token(&t2).as_deref(), Some("n"));
     }
+
+    fn bearer_headers(token: &str) -> HeaderMap {
+        let mut h = HeaderMap::new();
+        h.insert(
+            axum::http::header::AUTHORIZATION,
+            format!("Bearer {token}").parse().unwrap(),
+        );
+        h
+    }
+
+    /// With no admin key configured the admin verbs must behave EXACTLY as they
+    /// did before the split, so enabling this feature is opt-in and upgrading
+    /// cannot lock an existing operator out of its own cluster.
+    #[test]
+    fn admin_gate_is_open_when_no_admin_key_is_configured() {
+        let creds = ClusterCreds::with_keys(Some("data-key".to_string()), None);
+        assert!(!creds.admin_configured());
+        assert!(creds.admin_ok(&bearer_headers("data-key")));
+        assert!(creds.admin_ok(&HeaderMap::new()));
+    }
+
+    /// The whole point: a data-plane key must not be able to remove a member,
+    /// force a partition, or move a shard.
+    #[test]
+    fn data_bearer_cannot_reach_admin_verbs_once_an_admin_key_exists() {
+        let creds = ClusterCreds::with_keys_admin(
+            Some("data-key".to_string()),
+            Some("admin-key".to_string()),
+            None,
+        );
+        assert!(creds.admin_configured());
+        assert!(
+            !creds.admin_ok(&bearer_headers("data-key")),
+            "the data bearer must NOT confer operator authority"
+        );
+        assert!(!creds.admin_ok(&HeaderMap::new()));
+        assert!(creds.admin_ok(&bearer_headers("admin-key")));
+    }
+
+    /// The admin key must ALSO satisfy the authentication gate. A request carries
+    /// one `Authorization` header, so if it did not, the bearer check would 401 an
+    /// operator before the admin gate ran and the admin verbs would be reachable
+    /// by nobody. Regression guard for exactly that.
+    #[test]
+    fn admin_key_authenticates_as_well_as_authorizes() {
+        let creds = ClusterCreds::with_keys_admin(
+            Some("data-key".to_string()),
+            Some("admin-key".to_string()),
+            None,
+        );
+        assert!(creds.authenticated(&bearer_headers("admin-key")));
+        assert!(creds.authenticated(&bearer_headers("data-key")));
+        assert!(!creds.authenticated(&bearer_headers("neither")));
+        assert!(!creds.authenticated(&HeaderMap::new()));
+    }
+
+    /// A verified sibling keeps its authority without holding the admin key: nodes
+    /// relay operator verbs to the leader/target carrying whatever credential the
+    /// operator sent, so requiring the admin key on that hop would partition the
+    /// control plane.
+    #[test]
+    fn verified_sibling_retains_admin_authority() {
+        let creds = ClusterCreds::with_keys_admin(
+            Some("data-key".to_string()),
+            Some("admin-key".to_string()),
+            Some("cluster-secret"),
+        );
+        let token = creds.mint_node_token("us-east").unwrap();
+        let mut headers = bearer_headers("data-key");
+        headers.insert(NODE_TOKEN_HEADER, token.parse().unwrap());
+        assert!(
+            creds.admin_ok(&headers),
+            "a relayed verb from a verified sibling must still be allowed"
+        );
+
+        // A forged token grants nothing.
+        let mut forged = bearer_headers("data-key");
+        forged.insert(NODE_TOKEN_HEADER, "not.a.real.token".parse().unwrap());
+        assert!(!creds.admin_ok(&forged));
+    }
+
+    #[test]
+    fn reload_swaps_changed_admin_key() {
+        let dir = tempfile::tempdir().unwrap();
+        let admin_path = dir.path().join("admin.key");
+        std::fs::write(&admin_path, b"first-admin\n").unwrap();
+        let creds = ClusterCreds {
+            bearer: ArcSwapOption::from(Some(Arc::new("data-key".to_string()))),
+            bearer_file: None,
+            admin: ArcSwapOption::from(Some(Arc::new("first-admin".to_string()))),
+            admin_file: Some(admin_path.clone()),
+            cluster_key: ArcSwapOption::from(None),
+            cluster_key_file: None,
+            rate_limiter: tidaldb::RateLimiter::new(tidaldb::RateLimiterConfig::default()),
+        };
+        assert!(creds.admin_ok(&bearer_headers("first-admin")));
+
+        std::fs::write(&admin_path, b"second-admin\n").unwrap();
+        assert!(
+            creds.reload(),
+            "a changed admin key file must report a rotation"
+        );
+        assert!(
+            !creds.admin_ok(&bearer_headers("first-admin")),
+            "the retired admin key must stop working"
+        );
+        assert!(creds.admin_ok(&bearer_headers("second-admin")));
+    }
 }

tidal-server/src/router.rs

@@ -268,6 +268,25 @@ pub(crate) fn unauthorized_response() -> Response {
         .into_response()
 }
 
+/// The 403 returned when a valid data-plane bearer is presented to a DESTRUCTIVE
+/// cluster verb but no cluster-admin credential (or verified sibling node token)
+/// accompanies it.
+///
+/// 403, not 401: the caller authenticated successfully: it simply lacks operator
+/// authority. Shared by both cluster routers so the two surfaces cannot drift.
+#[must_use]
+pub(crate) fn admin_forbidden_response() -> Response {
+    (
+        StatusCode::FORBIDDEN,
+        Json(serde_json::json!({
+            "error": "this cluster verb requires the cluster-admin credential \
+                      (TIDAL_ADMIN_KEY) or a verified sibling node token; the \
+                      data-plane bearer does not grant operator authority"
+        })),
+    )
+        .into_response()
+}
+
 /// The 429 returned when a principal exceeds its per-principal rate limit
 /// (m11p7), carrying a `Retry-After` header (seconds, ceil) and the limit + the
 /// millisecond hint in the body so a client can back off precisely.

tidal-server/tests/cluster_runbook.rs

@@ -1082,5 +1082,120 @@ fn runbook_auth_protected_routes_401_probes_open() {
         "auth: a correct bearer is accepted (204): {}",
         with_key.status()
     );
+    // `/cluster/status` is NOT a probe: it reports leader identity, membership,
+    // term and per-shard applied/lag/commit seqnos. It used to sit in the
+    // unauthenticated group next to /health, which handed cluster topology to any
+    // caller that could reach the port.
+    let bare_status = cluster.get(LEADER, "/cluster/status");
+    assert_eq!(
+        bare_status.status().as_u16(),
+        401,
+        "auth: bare /cluster/status must be 401, got {}",
+        bare_status.status()
+    );
+    let status_url = format!("{}/cluster/status", cluster.node(LEADER));
+    let status_with_key = cluster
+        .client()
+        .get(&status_url)
+        .header("Authorization", format!("Bearer {KEY}"))
+        .send()
+        .unwrap();
+    assert_eq!(
+        status_with_key.status().as_u16(),
+        200,
+        "auth: /cluster/status with the bearer must still serve operators"
+    );
+
+    // That this cluster reached a serving state AT ALL is the load-bearing part of
+    // this test now: with a bearer configured on every process, seed-join and the
+    // status fan-out have to authenticate their own inter-node polls. If moving
+    // status behind auth had broken leader discovery, startup would never converge.
     println!("[auth] protected routes 401 bare, probes + /openapi.json open, valid bearer 204");
 }
+
+// ── Auth: the admin key separates operator authority from data-plane access ────
+
+/// With BOTH `TIDAL_API_KEY` and `TIDAL_ADMIN_KEY` set, a data-plane bearer is
+/// authenticated for the data routes but REFUSED (403, not 401) on the
+/// destructive cluster verbs, while the admin key is accepted for both.
+///
+/// The exposure this pins: before the split, every `/cluster/*` mutation sat
+/// behind the same bearer as `/items` and `/search`, so any client key could
+/// remove a member, force a partition, or transfer a shard.
+#[test]
+fn runbook_admin_key_gates_destructive_cluster_verbs() {
+    const DATA_KEY: &str = "runbook-data-key";
+    const ADMIN_KEY: &str = "runbook-admin-key";
+    let opts = (0..3).fold(ClusterOptions::new(3), |opts, i| {
+        opts.with_env(i, "TIDAL_API_KEY", DATA_KEY)
+            .with_env(i, "TIDAL_ADMIN_KEY", ADMIN_KEY)
+    });
+    let cluster = MultiProcCluster::start_with(opts);
+
+    let promote_url = format!("{}/cluster/promote", cluster.node(LEADER));
+    let body = serde_json::json!({ "region": "eu-west" });
+
+    // The data key authenticates, then FAILS authorization: 403, not 401.
+    let as_data = cluster
+        .client()
+        .post(&promote_url)
+        .header("Authorization", format!("Bearer {DATA_KEY}"))
+        .json(&body)
+        .send()
+        .unwrap();
+    assert_eq!(
+        as_data.status().as_u16(),
+        403,
+        "admin gate: the data bearer must be FORBIDDEN on /cluster/promote, got {}",
+        as_data.status()
+    );
+
+    // The data key still works on the data plane - the split must not break it.
+    let data_write = cluster
+        .client()
+        .post(format!("{}/signals", cluster.node(LEADER)))
+        .header("Authorization", format!("Bearer {DATA_KEY}"))
+        .json(&serde_json::json!({ "entity_id": 7, "signal": "view", "weight": 1.0 }))
+        .send()
+        .unwrap();
+    assert_eq!(
+        data_write.status().as_u16(),
+        204,
+        "admin gate: the data bearer must still serve /signals"
+    );
+
+    // The admin key is a SUPERSET credential: it authenticates too, so it reads
+    // the protected status surface as well.
+    let admin_status = cluster
+        .client()
+        .get(format!("{}/cluster/status", cluster.node(LEADER)))
+        .header("Authorization", format!("Bearer {ADMIN_KEY}"))
+        .send()
+        .unwrap();
+    assert_eq!(
+        admin_status.status().as_u16(),
+        200,
+        "admin gate: the admin key must authenticate, not just authorize"
+    );
+
+    // And it clears the admin gate. `/cluster/heal` is the benign admin verb, so
+    // this asserts acceptance without perturbing the cluster: any status EXCEPT
+    // 401/403 proves the gate let it through to the handler.
+    let admin_heal = cluster
+        .client()
+        .post(format!("{}/cluster/heal", cluster.node(LEADER)))
+        .header("Authorization", format!("Bearer {ADMIN_KEY}"))
+        .json(&serde_json::json!({ "region": "eu-west" }))
+        .send()
+        .unwrap();
+    let code = admin_heal.status().as_u16();
+    assert!(
+        code != 401 && code != 403,
+        "admin gate: the admin key must pass the gate on /cluster/heal, got {code}"
+    );
+
+    println!(
+        "[auth] admin key gates destructive verbs: data bearer 403 on promote, \
+         204 on signals; admin key 200 on status and {code} on heal"
+    );
+}