diff --git a/apps/server/src/__tests__/metrics-orchestrator.test.ts b/apps/server/src/__tests__/metrics-orchestrator.test.ts
index a2b615ad..3cc6e4a0 100644
--- a/apps/server/src/__tests__/metrics-orchestrator.test.ts
+++ b/apps/server/src/__tests__/metrics-orchestrator.test.ts
@@ -6,6 +6,8 @@ import {
   metricsServerMap,
   stopAllMetricsServers,
   reconcileClusterMetricsServers,
+  updateClusterNodeRegistry,
+  resolveClusterRefreshTarget,
   clients,
   clusterNodesRegistry,
   __test__,
@@ -229,12 +231,6 @@ describe("metrics-orchestrator", () => {
 
       // Mock all side-effectful internal functions
       mock.method(__test__, "createClient", async () => ({}))
-      mock.method(__test__, "getClusterTopology", async () => ({
-        clusterNodes: {
-          node1: { host: "127.0.0.1", port: "6379", tls: false, verifyTlsCertificate: false },
-        },
-        clusterId: "cluster-1",
-      }))
       mock.method(__test__, "updateClusterNodeRegistry", async () => mockClusterNodesRegistry)
       mock.method(__test__, "updateMetricsServers", async () => {})
       mock.method(__test__, "findDiff", async () => ({ nodesToAdd: {}, nodesToRemove: [] }))
@@ -271,4 +267,103 @@ describe("metrics-orchestrator", () => {
       assert.strictEqual(updateMetricsServers.mock.callCount(), 0)
     })
   })
+
+  describe("topology refresh", () => {
+    beforeEach(() => {
+      mock.restoreAll()
+      clients.clear()
+      metricsServerMap.clear()
+      clusterNodesRegistry.clear()
+    })
+    afterEach(() => {
+      mock.restoreAll()
+      clients.clear()
+      metricsServerMap.clear()
+      clusterNodesRegistry.clear()
+    })
+
+    it("updateClusterNodeRegistry replaces stale topology with the freshly discovered one", async () => {
+      // Seed a stale registry: cluster "node-1" knows only one node.
+      clusterNodesRegistry.set("node-1", {
+        "192-168-1-1-6379": { host: "192.168.1.1", port: 6379, tls: false, verifyTlsCertificate: false },
+      })
+
+      // A live client for that cluster whose CLUSTER SLOTS now reports three
+      // primaries. (SLOTS parsing itself is covered in connection.test.ts.)
+      const client = {
+        customCommand: async (args: string[]) =>
+          args[0] === "CLUSTER" && args[1] === "SLOTS"
+            ? [
+              [0, 5460, ["192.168.1.1", 6379, "node-1"]],
+              [5461, 10922, ["192.168.1.3", 6379, "node-2"]],
+              [10923, 16383, ["192.168.1.4", 6379, "node-3"]],
+            ]
+            : [],
+      } as never
+
+      const sampleNode = Object.values(clusterNodesRegistry.get("node-1") ?? {})[0]
+      await updateClusterNodeRegistry(client, sampleNode)
+
+      assert.deepStrictEqual(
+        Object.keys(clusterNodesRegistry.get("node-1") ?? {}).sort(),
+        ["192-168-1-1-6379", "192-168-1-3-6379", "192-168-1-4-6379"],
+        "registry should reflect the newly discovered topology, not the stale snapshot",
+      )
+    })
+
+    it("overwrites the existing entry when a known clusterId is passed, even if the derived id changed", async () => {
+      // Existing cluster tracked under "orig-id".
+      clusterNodesRegistry.set("orig-id", {
+        "192-168-1-1-6379": { host: "192.168.1.1", port: 6379, tls: false, verifyTlsCertificate: false },
+      })
+
+      // After a failover, CLUSTER SLOTS now lists a different first primary, so the
+      // derived clusterId would be "new-first-primary" — which would orphan "orig-id".
+      const client = {
+        customCommand: async (args: string[]) =>
+          args[0] === "CLUSTER" && args[1] === "SLOTS"
+            ? [[0, 16383, ["192.168.1.9", 6379, "new-first-primary"]]]
+            : [],
+      } as never
+
+      const sampleNode = Object.values(clusterNodesRegistry.get("orig-id") ?? {})[0]
+      await updateClusterNodeRegistry(client, sampleNode, "orig-id")
+
+      assert.deepStrictEqual(
+        [...clusterNodesRegistry.keys()],
+        ["orig-id"],
+        "the known clusterId should be overwritten in place, leaving no orphaned entry",
+      )
+      assert.deepStrictEqual(
+        Object.keys(clusterNodesRegistry.get("orig-id") ?? {}),
+        ["192-168-1-9-6379"],
+        "the entry should hold the freshly discovered node",
+      )
+    })
+
+    it("resolveClusterRefreshTarget uses the user client and carries node metadata forward", async () => {
+      const userClient = { id: "user-client" } as never
+      const clusterNodes = {
+        node1: { host: "10.0.0.1", port: 6379, tls: true, verifyTlsCertificate: false, username: "admin", authType: "iam" as const },
+      }
+
+      const target = await resolveClusterRefreshTarget("cluster-1", clusterNodes, userClient)
+
+      assert.strictEqual(target?.client, userClient, "should refresh with the cluster's own live client")
+      // nodeInfo must be a node from THIS cluster (preserving its tls/username/auth),
+      // not initialConnectionDetails.
+      assert.deepStrictEqual(target?.nodeInfo, clusterNodes.node1, "should carry the cluster's own node metadata forward")
+    })
+
+    it("resolveClusterRefreshTarget skips a cluster with no live client when not preconfigured", async () => {
+      // DEPLOYMENT_MODE is unset in this file, so preConfiguredConnection is falsy
+      // and there is no initial client to fall back to.
+      const target = await resolveClusterRefreshTarget(
+        "cluster-1",
+        { node1: { host: "10.0.0.1", port: 6379, tls: false, verifyTlsCertificate: false } },
+        undefined,
+      )
+      assert.strictEqual(target, undefined, "a cluster with no live client and no preconfigured fallback is skipped")
+    })
+  })
 })
diff --git a/apps/server/src/__tests__/orchestrator-k8s-register.test.ts b/apps/server/src/__tests__/orchestrator-k8s-register.test.ts
new file mode 100644
index 00000000..3512fabf
--- /dev/null
+++ b/apps/server/src/__tests__/orchestrator-k8s-register.test.ts
@@ -0,0 +1,114 @@
+import { describe, it, afterEach } from "node:test"
+import assert from "node:assert"
+import {
+  ORCHESTRATOR_AUTH_DOMAIN,
+  ORCHESTRATOR_AUTH_HEADER,
+  createOrchestratorAuthCredential
+} from "valkey-common"
+import type { Request, Response } from "express"
+
+// These tests exercise the Kubernetes-only branches of the orchestrator:
+//   - resolveCollectorKey falls back to the shared ORCHESTRATOR_KEY, and
+//   - handleRegister admits a sidecar whose nodeId is part of the discovered
+//     cluster topology even though the orchestrator never spawned it.
+// isKubernetes is a module-load constant, so DEPLOYMENT_MODE and the shared key
+// must be set BEFORE importing the module.
+process.env.DEPLOYMENT_MODE = "K8"
+process.env.ORCHESTRATOR_KEY = "k8s-shared-secret"
+
+const {
+  metricsServerMap,
+  clusterNodesRegistry,
+  resolveCollectorKey,
+  __test__,
+} = await import("../metrics-orchestrator")
+
+const NODE_ID = "valkey-6-valkey-headless-valkey-svc-cluster-local-6379"
+const URI = "http://10.42.0.32:3000"
+const SHARED_KEY = "k8s-shared-secret"
+
+const makeRes = () => {
+  const captured: { statusCode: number; body?: unknown } = { statusCode: 200 }
+  const res = {
+    status(code: number) { captured.statusCode = code; return res },
+    send(body?: unknown) { captured.body = body; return res },
+    sendStatus(code: number) { captured.statusCode = code; return res },
+  }
+  return { res: res as unknown as Response, captured }
+}
+
+const makeReq = (body: unknown, credential?: string) => ({
+  body,
+  headers: credential === undefined ? {} : { [ORCHESTRATOR_AUTH_HEADER]: credential },
+}) as unknown as Request
+
+const signRegister = (fields: { nodeId?: string; metricsServerUri?: string; timestamp?: number } = {}, key = SHARED_KEY) =>
+  createOrchestratorAuthCredential(key, ORCHESTRATOR_AUTH_DOMAIN.REGISTER, {
+    nodeId: NODE_ID,
+    metricsServerUri: URI,
+    timestamp: Date.now(),
+    ...fields,
+  }) as string
+
+// Register a node into the discovered topology so it is a known cluster member.
+const seedTopology = (nodeId = NODE_ID) => {
+  clusterNodesRegistry.set("cluster-1", {
+    [nodeId]: { host: "10.42.0.32", port: 6379, tls: false, verifyTlsCertificate: false },
+  })
+}
+
+describe("K8s shared-key registration", () => {
+  afterEach(() => {
+    metricsServerMap.clear()
+    clusterNodesRegistry.clear()
+    __test__.collectorKeys.clear()
+  })
+
+  describe("resolveCollectorKey", () => {
+    it("falls back to the shared ORCHESTRATOR_KEY for a node the orchestrator never spawned", () => {
+      assert.strictEqual(resolveCollectorKey("never-spawned-node"), SHARED_KEY)
+    })
+
+    it("still prefers a per-node minted key when one exists", () => {
+      __test__.collectorKeys.set("spawned-node", "per-node-key")
+      assert.strictEqual(resolveCollectorKey("spawned-node"), "per-node-key")
+    })
+  })
+
+  describe("handleRegister topology-membership gate", () => {
+    it("admits a correctly signed sidecar for a known cluster node with no pre-existing entry", () => {
+      seedTopology()
+      assert.strictEqual(metricsServerMap.has(NODE_ID), false, "precondition: no orchestrator-spawned entry")
+
+      const { res, captured } = makeRes()
+      __test__.handleRegister(makeReq({ nodeId: NODE_ID, metricsServerUri: URI, timestamp: Date.now() }, signRegister()), res)
+
+      assert.strictEqual(captured.statusCode, 200, "a known cluster node should be allowed to register")
+      assert.strictEqual(metricsServerMap.get(NODE_ID)?.metricsURI, URI, "the sidecar's entry should be created on first register")
+    })
+
+    it("rejects a correctly signed sidecar whose nodeId is not in the discovered topology", () => {
+      // Topology known, but for a different node.
+      seedTopology("some-other-node-6379")
+
+      const { res, captured } = makeRes()
+      __test__.handleRegister(makeReq({ nodeId: NODE_ID, metricsServerUri: URI, timestamp: Date.now() }, signRegister()), res)
+
+      assert.strictEqual(captured.statusCode, 401, "an unknown node must be rejected even with a valid signature")
+      assert.strictEqual(metricsServerMap.has(NODE_ID), false)
+    })
+
+    it("rejects a known cluster node when the signature uses the wrong key", () => {
+      seedTopology()
+
+      const { res, captured } = makeRes()
+      __test__.handleRegister(
+        makeReq({ nodeId: NODE_ID, metricsServerUri: URI, timestamp: Date.now() }, signRegister({}, "wrong-key")),
+        res,
+      )
+
+      assert.strictEqual(captured.statusCode, 401, "a bad signature must be rejected before the membership check")
+      assert.strictEqual(metricsServerMap.has(NODE_ID), false)
+    })
+  })
+})
diff --git a/apps/server/src/__tests__/topology-refresh-preconfigured.test.ts b/apps/server/src/__tests__/topology-refresh-preconfigured.test.ts
new file mode 100644
index 00000000..83ee5fa0
--- /dev/null
+++ b/apps/server/src/__tests__/topology-refresh-preconfigured.test.ts
@@ -0,0 +1,70 @@
+import { describe, it, afterEach, mock } from "node:test"
+import assert from "node:assert"
+
+// preConfiguredConnection is a module-load constant derived from VALKEY_HOST /
+// VALKEY_PORT, so it must be set BEFORE importing the module to exercise the
+// headless-refresh fallback (no live user client → initial client).
+process.env.VALKEY_HOST = "valkey-0.example"
+process.env.VALKEY_PORT = "6379"
+
+const {
+  resolveClusterRefreshTarget,
+  setPreconfiguredClusterId,
+  initialConnectionDetails,
+  __test__,
+} = await import("../metrics-orchestrator")
+
+const PRECONFIGURED_ID = "preconfigured-cluster"
+
+describe("resolveClusterRefreshTarget (preconfigured / headless)", () => {
+  afterEach(() => {
+    mock.restoreAll()
+    setPreconfiguredClusterId(undefined)
+  })
+
+  it("refreshes the preconfigured cluster via the initial client + initialConnectionDetails", async () => {
+    setPreconfiguredClusterId(PRECONFIGURED_ID)
+    const initialClient = { id: "initial-client" } as never
+    mock.method(__test__, "getInitialClient", async () => initialClient)
+
+    const target = await resolveClusterRefreshTarget(
+      PRECONFIGURED_ID,
+      { node1: { host: "10.0.0.1", port: 6379, tls: false, verifyTlsCertificate: false } },
+      undefined,
+    )
+
+    assert.strictEqual(target?.client, initialClient, "the preconfigured cluster should refresh via the initial client")
+    assert.strictEqual(
+      target?.nodeInfo,
+      initialConnectionDetails,
+      "the preconfigured cluster decorates with initialConnectionDetails",
+    )
+  })
+
+  it("does NOT use the initial client for a different client-less cluster", async () => {
+    // Preconfigured cluster is A; refreshing a stale, client-less cluster B must
+    // not rediscover A's topology through the initial client.
+    setPreconfiguredClusterId(PRECONFIGURED_ID)
+    const getInitial = mock.method(__test__, "getInitialClient", async () => ({ id: "initial-client" } as never))
+
+    const target = await resolveClusterRefreshTarget(
+      "some-other-cluster",
+      { node1: { host: "10.9.9.9", port: 6379, tls: false, verifyTlsCertificate: false } },
+      undefined,
+    )
+
+    assert.strictEqual(target, undefined, "an unrelated client-less cluster must be left unchanged")
+    assert.strictEqual(getInitial.mock.callCount(), 0, "the initial client must not be consulted for other clusters")
+  })
+
+  it("still prefers a user client over the initial client when one is present", async () => {
+    const userClient = { id: "user-client" } as never
+    mock.method(__test__, "getInitialClient", async () => ({ id: "initial-client" } as never))
+
+    const clusterNodes = { node1: { host: "10.0.0.1", port: 6379, tls: true, verifyTlsCertificate: false } }
+    const target = await resolveClusterRefreshTarget("any-cluster", clusterNodes, userClient)
+
+    assert.strictEqual(target?.client, userClient)
+    assert.deepStrictEqual(target?.nodeInfo, clusterNodes.node1)
+  })
+})
diff --git a/apps/server/src/connection.ts b/apps/server/src/connection.ts
index 191853e9..7587490d 100644
--- a/apps/server/src/connection.ts
+++ b/apps/server/src/connection.ts
@@ -22,7 +22,8 @@ import {
   reconcileClusterMetricsServers, 
   isKubernetes, 
   forgetCollectorKey,
-  ClusterNodeMap } from "./metrics-orchestrator"
+  ClusterNodeMap,
+  type NodeInfo } from "./metrics-orchestrator"
 import { subscribe } from "./node-watchers"
 import { clearCpuSamples } from "./node-utilization"
 import { createClusterValkeyClient, createStandaloneValkeyClient } from "./valkey-client"
@@ -595,7 +596,7 @@ function sendStandaloneConnectFulfilled(ws: WebSocket, payload: StandaloneConnec
 
 export async function discoverCluster(
   client: GlideClient | GlideClusterClient, 
-  payload: { connectionDetails: ConnectionDetails, connectionId?: string;},
+  payload: { connectionDetails: NodeInfo, connectionId?: string;},
 )  {
   try {
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts
index b12006ae..149cdb9f 100644
--- a/apps/server/src/index.ts
+++ b/apps/server/src/index.ts
@@ -47,7 +47,9 @@ import {
   isKubernetes,
   preConfiguredConnection,
   getInitialClient,
-  updateClusterNodeRegistry
+  updateClusterNodeRegistry,
+  resolveClusterRefreshTarget,
+  setPreconfiguredClusterId
 } from "./metrics-orchestrator"
 import { isAllowedWebSocketOrigin } from "./websocket-origin"
 import { ensureSession, hasAuthorizedSession, isConnectionAuthorized, setSessionExpiryListener } from "./session"
@@ -143,7 +145,14 @@ const wss = new WebSocketServer({ noServer: true })
 
 const delay = (ms: number) => new Promise((res) => setTimeout(res, ms))
 
-function refreshAllClusterRegistries() {
+// Upper bound on a single cluster's topology re-discovery. Without it, one hung
+// CLUSTER SLOTS would leave the awaited refresh unsettled and stall the broadcast
+// for every cluster, not just the unresponsive one.
+const TOPOLOGY_REDISCOVERY_TIMEOUT_MS = 10_000
+
+async function refreshAllClusterRegistries() {
+  // Group live connections by cluster once; used both to pick a client to
+  // re-discover each cluster's topology and to target the broadcast below.
   const connectionIdsByCluster = new Map<string, string[]>()
   for (const [connectionId, entry] of clients) {
     if (!entry.clusterId) continue
@@ -152,6 +161,28 @@ function refreshAllClusterRegistries() {
     connectionIdsByCluster.set(entry.clusterId, ids)
   }
 
+  // Re-discover each tracked cluster before broadcasting. User-connected clusters
+  // use their own client + node metadata; a preconfigured cluster (K8s / headless
+  // Web) has no entry in `clients`, so it falls back to the initial client so
+  // scaled-in nodes are picked up without a UI session. Time-bounded so one hung
+  // node can't stall the whole loop.
+  await Promise.all(
+    [...clusterNodesRegistry.entries()].map(async ([clusterId, clusterNodes]) => {
+      const connectionId = connectionIdsByCluster.get(clusterId)?.[0]
+      const userClient = connectionId ? clients.get(connectionId)?.client : undefined
+
+      const target = await resolveClusterRefreshTarget(clusterId, clusterNodes, userClient)
+      if (!target) return
+
+      await Promise.race([
+        updateClusterNodeRegistry(target.client, target.nodeInfo, clusterId),
+        delay(TOPOLOGY_REDISCOVERY_TIMEOUT_MS).then(() =>
+          console.warn(`Topology re-discovery for cluster ${clusterId} timed out; broadcasting last known nodes.`),
+        ),
+      ])
+    }),
+  )
+
   for (const [clusterId, clusterNodes] of clusterNodesRegistry) {
     const connectionIds = connectionIdsByCluster.get(clusterId)
     if (!connectionIds) continue
@@ -172,7 +203,7 @@ function refreshAllClusterRegistries() {
 async function refreshAllClusterRegistriesLoop() {
   while (true) {
     try {
-      refreshAllClusterRegistries()
+      await refreshAllClusterRegistries()
     } catch (err) {
       console.warn("Unable to refresh cluster topologies. ", err)
     } finally {
@@ -184,7 +215,8 @@ async function refreshAllClusterRegistriesLoop() {
 
 async function updateRegistryforK8() {
   const client = await getInitialClient()
-  updateClusterNodeRegistry(client, initialConnectionDetails)
+  const clusterId = await updateClusterNodeRegistry(client, initialConnectionDetails)
+  setPreconfiguredClusterId(clusterId)
 }
 
 // Electron: bind to localhost only — Origin headers are forgeable by non-browser clients,
diff --git a/apps/server/src/metrics-orchestrator.ts b/apps/server/src/metrics-orchestrator.ts
index 4d88a414..b71792cb 100644
--- a/apps/server/src/metrics-orchestrator.ts
+++ b/apps/server/src/metrics-orchestrator.ts
@@ -43,7 +43,7 @@ export type MetricsServerMap = Map<string,
   }
 >
 
-type NodeInfo = {
+export type NodeInfo = {
   host: string;
   port: number | string;
   username?: string;
@@ -64,6 +64,11 @@ export const clusterNodesRegistry: Map<string, ClusterNodeMap> = new Map()
 
 export const clusterCredentials: Map<string, string | undefined> = new Map()
 
+let preconfiguredClusterId: string | undefined
+export function setPreconfiguredClusterId(clusterId: string | undefined) {
+  preconfiguredClusterId = clusterId
+}
+
 export const metricsServerMap: MetricsServerMap = new Map()
 
 /**
@@ -79,7 +84,14 @@ const collectorKeys: Map<string, string> = new Map()
  * replayed against another.
  */
 export function resolveCollectorKey(nodeId: string): string | undefined {
-  return collectorKeys.get(nodeId)
+  const spawnedKey = collectorKeys.get(nodeId)
+  if (spawnedKey) return spawnedKey
+  // K8s collectors are external sidecars the orchestrator never spawns, so there
+  // is no per-node minted key. They authenticate with a shared key provisioned to
+  // both the orchestrator and every sidecar via a Kubernetes Secret. Scoped to K8s
+  // so the per-node property is preserved for spawned (Web/Electron) collectors.
+  if (isKubernetes) return process.env[ORCHESTRATOR_AUTH_KEY_ENV]
+  return undefined
 }
 
 /**
@@ -225,6 +237,18 @@ function isPinnedMetricsHost(uri: string): boolean {
   }
 }
 
+/**
+ * True when `nodeId` is part of a discovered cluster topology. In K8s the
+ * orchestrator only tracks topology (it never spawns collectors), so this is
+ * how a sidecar's registration is authorized as belonging to the cluster.
+ */
+function isKnownClusterNode(nodeId: string): boolean {
+  for (const clusterNodes of clusterNodesRegistry.values()) {
+    if (flattenClusterNodeMap(clusterNodes)[nodeId]) return true
+  }
+  return false
+}
+
 /**
  * `POST /orchestrator/register` — a collector advertising where it can be
  * reached.
@@ -269,14 +293,22 @@ function handleRegister(req: Request, res: Response): void {
     return
   }
 
-  const entry = metricsServerMap.get(nodeId)
-  if (!entry) {
-    // Key material without an entry means the spawn did not complete.
+  // Allowed if we already have an entry (orchestrator-spawned) or, in K8s,
+  // the node is part of the discovered cluster (sidecars are external, so
+  // they create their entry on first register).
+  const allowed = metricsServerMap.has(nodeId) || (isKubernetes && isKnownClusterNode(nodeId))
+  if (!allowed) {
     console.warn(`Rejected metrics registration for ${nodeId}: no metrics server entry`)
     res.status(401).send("Unauthorized")
     return
   }
 
+  let entry = metricsServerMap.get(nodeId)
+  if (!entry) {
+    entry = { metricsURI: "", pid: undefined, lastSeen: Date.now() }
+    metricsServerMap.set(nodeId, entry)
+  }
+
   entry.metricsURI = metricsServerUri
   entry.lastSeen = Date.now()
   console.log(`Metrics server registered for ${nodeId} at ${metricsServerUri}`)
@@ -359,18 +391,21 @@ async function createClient(connectionDetails: ConnectionDetails) {
   return await createOrchestratorValkeyClient({ addresses, credentials, useTLS: tls, verifyTlsCertificate, databaseId: db })
 }
 
-async function getClusterTopology(client: GlideClusterClient | GlideClient | null, node: ConnectionDetails) {
-  if (!client) client = await createClient(node)
-
-  const { discoveredClusterNodes, clusterId } = await discoverCluster(client, { connectionDetails: node })
-
-  return { discoveredClusterNodes, clusterId }
-}
-
-export async function updateClusterNodeRegistry(client: GlideClusterClient | GlideClient | null, connectionDetails = initialConnectionDetails) {
+export async function updateClusterNodeRegistry(
+  client: GlideClusterClient | GlideClient,
+  nodeInfo: NodeInfo,
+  clusterId?: string,
+) {
   try {
-    const { discoveredClusterNodes, clusterId } = await getClusterTopology(client, connectionDetails)
-    if (clusterId && discoveredClusterNodes) clusterNodesRegistry.set(clusterId, discoveredClusterNodes)
+    const { discoveredClusterNodes, clusterId: discoveredClusterId } = await discoverCluster(client, { connectionDetails: nodeInfo })
+    // Prefer the caller's known clusterId when refreshing an existing cluster: the
+    // discovered id is derived from the first primary in CLUSTER SLOTS, which can
+    // change on failover/resharding and would otherwise orphan the old entry.
+    const key = clusterId ?? discoveredClusterId
+    if (key && discoveredClusterNodes) {
+      clusterNodesRegistry.set(key, discoveredClusterNodes)
+      return key
+    }
   }
   catch (err) {
     if (err instanceof ConnectionError) {
@@ -378,7 +413,29 @@ export async function updateClusterNodeRegistry(client: GlideClusterClient | Gli
     }
     console.error(err)
   }
-  return clusterNodesRegistry
+  return undefined
+}
+
+/**
+ * Decide how to refresh one tracked cluster. A user-connected cluster is refreshed
+ * with its own client and metadata carried forward from an existing node (so a
+ * refresh doesn't revert TLS/auth to server defaults). A preconfigured cluster
+ * (K8s / headless Web) has no live user client, so it falls back to the initial
+ * client with initialConnectionDetails. Returns undefined when the cluster can't
+ * be refreshed (no client available).
+ */
+export async function resolveClusterRefreshTarget(
+  clusterId: string,
+  clusterNodes: ClusterNodeMap,
+  userClient: GlideClusterClient | GlideClient | undefined,
+): Promise<{ client: GlideClusterClient | GlideClient; nodeInfo: NodeInfo } | undefined> {
+  if (userClient) return { client: userClient, nodeInfo: Object.values(clusterNodes)[0] }
+
+  // Only the preconfigured cluster may be refreshed via the initial client.
+  if (preConfiguredConnection && clusterId === preconfiguredClusterId) {
+    return { client: await internals.getInitialClient(), nodeInfo: initialConnectionDetails }
+  }
+  return undefined
 }
 
 async function findDiff(metricsServerMap: MetricsServerMap, clusterNodeMap: ClusterNodeMap) {
@@ -582,9 +639,10 @@ export async function startPreconfiguredMetricsServers() {
   const client = await getInitialClient()
   if (await belongsToCluster(client)) {
     if (isWebMode) {
-      const { discoveredClusterNodes, clusterId } = await internals.getClusterTopology(client, initialConnectionDetails)
+      const { discoveredClusterNodes, clusterId } = await discoverCluster(client, { connectionDetails: initialConnectionDetails })
       if (clusterId && discoveredClusterNodes) {
         clusterNodesRegistry.set(clusterId, discoveredClusterNodes)
+        setPreconfiguredClusterId(clusterId)
         if (!clusterCredentials.has(clusterId)) clusterCredentials.set(clusterId, initialConnectionDetails.password)
       }
       runReconcileLoop()
@@ -616,7 +674,7 @@ export function cleanupOrchestratorResources() {
 const internals =  {
   startMetricsServers,
   createClient,
-  getClusterTopology,
+  getInitialClient,
   updateClusterNodeRegistry,
   findDiff,
   flattenClusterNodeMap,
diff --git a/docs-site/src/content/docs/deployment/kubernetes.md b/docs-site/src/content/docs/deployment/kubernetes.md
index 74444ca5..4e503e90 100644
--- a/docs-site/src/content/docs/deployment/kubernetes.md
+++ b/docs-site/src/content/docs/deployment/kubernetes.md
@@ -55,6 +55,21 @@ For a non-local deployment, use published images from a container registry and u
 
 ### 3. Deploy the app server
 
+The metrics sidecars authenticate to the orchestrator's `/orchestrator/register`
+and `/orchestrator/ping` endpoints with a shared key. Because sidecars are external
+(the orchestrator does not spawn them in Kubernetes), the same key must be
+provisioned to both the app server and every sidecar via a Secret. Create it first
+with your own random value:
+
+```bash
+kubectl create secret generic valkey-admin-orchestrator-key -n valkey \
+  --from-literal=ORCHESTRATOR_KEY="$(openssl rand -hex 32)"
+```
+
+`app.yaml` intentionally does **not** define this Secret — it only references it —
+so `kubectl apply` can never overwrite your generated key with a committed
+placeholder. Create the Secret first (above), then deploy the app server:
+
 ```bash
 kubectl apply -f examples/k8s/app.yaml
 kubectl rollout status deployment/valkey-admin-app -n valkey
@@ -181,15 +196,30 @@ kubectl logs -n valkey valkey-0 -c metrics
 
 You want to see `Register success` in the metrics sidecar log.
 
-:::caution[Sidecar registration is not yet functional]
-Sidecar registration does not currently succeed in `DEPLOYMENT_MODE=K8`. The orchestrator only tracks collectors it spawned itself, and it does not spawn sidecars, so a sidecar's registration is rejected and the sidecar exits after 30 attempts:
+:::caution[Sidecar registration requires the shared key]
+Registration and ping are authenticated. Each sidecar signs its requests with the
+shared `ORCHESTRATOR_KEY`, and the orchestrator verifies against the same key, so
+the Secret from step 3 must be present on **both** the app Deployment and the
+sidecar. If the key is missing or does not match, the sidecar exits after 30
+attempts:
 
 ```text
 Register failed: 401 Unauthorized
 Failed to register with server after 30 attempts. Shutting down.
 ```
 
-Support  is in progress. Until then the Kubernetes sidecar topology on this page will deploy but will not populate metrics panels. Use the [Docker deployment](/deployment/docker/) for a working metrics path.
+If you see this, confirm the `valkey-admin-orchestrator-key` Secret exists in the
+`valkey` namespace and that both pods reference it (`kubectl get pod ... -o yaml |
+grep -A3 ORCHESTRATOR_KEY`).
+
+The key is shared cluster-wide rather than per-node, so it authenticates a sidecar
+as *a* member of the cluster, not as one specific node. Treat it as a
+cluster-scoped credential: restrict read access on the Secret, and rotate it by
+recreating the Secret and restarting the app Deployment and the sidecars. Note that
+in `DEPLOYMENT_MODE=K8` the registered metrics host is not pinned to loopback, so
+any holder of the key can point a node's metrics URI at an arbitrary host that the
+server will then fetch from — keep the key tightly scoped and the namespace's
+network egress constrained accordingly.
 :::
 
 ### Charts Empty in the UI
diff --git a/examples/k8s/app.yaml b/examples/k8s/app.yaml
index f06dc692..a3718145 100644
--- a/examples/k8s/app.yaml
+++ b/examples/k8s/app.yaml
@@ -1,3 +1,15 @@
+# NOTE: the metrics control plane (/orchestrator/register, /orchestrator/ping)
+# authenticates sidecars with a shared key read from the ORCHESTRATOR_KEY env
+# var below, sourced from the `valkey-admin-orchestrator-key` Secret. Create that
+# Secret with your OWN random value before applying this manifest — it is
+# intentionally NOT defined here so `kubectl apply` can never overwrite a real
+# key with a committed placeholder:
+#
+#   kubectl create secret generic valkey-admin-orchestrator-key -n valkey \
+#     --from-literal=ORCHESTRATOR_KEY="$(openssl rand -hex 32)"
+#
+# The same Secret is referenced by every metrics sidecar
+# (see valkey-statefulset-sidecar-patch.yaml).
 apiVersion: apps/v1
 kind: Deployment
 metadata:
@@ -28,6 +40,11 @@ spec:
               value: valkey-0.valkey-headless.valkey.svc.cluster.local
             - name: VALKEY_PORT
               value: "6379"
+            - name: ORCHESTRATOR_KEY
+              valueFrom:
+                secretKeyRef:
+                  name: valkey-admin-orchestrator-key
+                  key: ORCHESTRATOR_KEY
           ports:
             - containerPort: 8080
               name: http
diff --git a/examples/k8s/valkey-statefulset-sidecar-patch.yaml b/examples/k8s/valkey-statefulset-sidecar-patch.yaml
index ef574190..22104c95 100644
--- a/examples/k8s/valkey-statefulset-sidecar-patch.yaml
+++ b/examples/k8s/valkey-statefulset-sidecar-patch.yaml
@@ -34,6 +34,11 @@ spec:
                   fieldPath: status.podIP
             - name: METRICS_ADVERTISE_PORT
               value: "3000"
+            - name: ORCHESTRATOR_KEY
+              valueFrom:
+                secretKeyRef:
+                  name: valkey-admin-orchestrator-key
+                  key: ORCHESTRATOR_KEY
           ports:
             - containerPort: 3000
               name: metrics
diff --git a/examples/k8s/valkey-statefulset.yaml b/examples/k8s/valkey-statefulset.yaml
index 5aaf2e66..cc5442fb 100644
--- a/examples/k8s/valkey-statefulset.yaml
+++ b/examples/k8s/valkey-statefulset.yaml
@@ -1,3 +1,18 @@
+# Example Valkey cluster StatefulSet with an inline metrics sidecar.
+#
+# PREREQUISITES — the metrics sidecars will crash-loop (and, because StatefulSet
+# rollout is ordered, block the whole rollout at valkey-0) unless both are met:
+#
+#   1. Create the shared orchestrator key Secret first (the sidecar exits with
+#      "Missing ORCHESTRATOR_KEY ... Shutting down" without it):
+#        kubectl create secret generic valkey-admin-orchestrator-key -n valkey \
+#          --from-literal=ORCHESTRATOR_KEY="$(openssl rand -hex 32)"
+#
+#   2. Form the cluster after the pods are up. These nodes start with
+#      --cluster-enabled yes but do NOT self-form, so CLUSTER SLOTS is empty and
+#      the sidecar dies on "Cluster(No topology views found)". Assign slots once:
+#        valkey-cli --cluster create <pod IPs>:6379 --cluster-replicas <n>
+#
 apiVersion: v1
 kind: Service
 metadata:
@@ -105,6 +120,11 @@ spec:
                   fieldPath: status.podIP
             - name: METRICS_ADVERTISE_PORT
               value: "3000"
+            - name: ORCHESTRATOR_KEY
+              valueFrom:
+                secretKeyRef:
+                  name: valkey-admin-orchestrator-key
+                  key: ORCHESTRATOR_KEY
           ports:
             - containerPort: 3000
               name: metrics