fix(cluster): 晋升 leader 前必须问环，防止滚动重启后分裂成两个环

用户报「只要 .60 和 .106 重启，.30 不重启就会脱环」。线上取证确认是分裂脑，
且违反设计文稿（docs/cluster-api.md §容错行为：非leader宕机应"环跳过死亡节点，
任务重挂"；leader宕机应"上邻居晋升 + StartRing"——环都应继续转）。

实测状态（cycle 两边同为 458198，共享同一段历史）：

    .60 视角: leader=.30   节点: [.60 alive, .30 alive=False, .106 alive]  topology=3
    .30 视角: leader=.30   节点: [.60 alive, .30 alive=False, .106 alive]  topology=0  pending=3

两个节点同时自认 leader、互判对方已死；.30 的 topology 被清空成 0、三条转发
全部退回 pending 且无人处理，环停止推进。

## 根因：领导权是"自封"的

becomeLeader() 无条件把自己设为 leader，不问环上是否已有存活 leader。
于是分区两侧各自"心跳失败 → 晋升"，两个环同时运行。更糟的是 WatchLeader
里 MarkOffline + OfflineReassign 是**无条件**的：被分区的那一侧会把仍然活着的
leader 标死，并把它的全部转发搬进 pending —— 这正是两边 topology=0、
pending=3 的直接来源。

## 修法

1. becomeLeader() 先问环：若存在**其它**存活且自称 leader 的节点，就拒绝晋升
   （我们才是被分区的一侧），保留存活 leader 的环。真正失去 leader（全员离线）
   时无此节点，晋升照常发生。

2. 抽出 takeOverFromDeadLeader()：MarkOffline → 条件晋升 → **仅在真正接管后**
   才 OfflineReassign 并 StartRing。原先无条件重挂是破坏活环的直接原因。
   同时让这段逻辑可测（WatchLeader 本身是无限循环 goroutine，不能直接测）。

## 关于"leader 要证明自己活着"

我一开始判断错方向，曾加过一个 leader 反向心跳自检 —— 那纯属多余且会引入新的
自封风险：OnToken 每次都会 UpsertNode(self, Alive:true) 自我保活，重启节点按设计
本就该主动用缓存上家/leader 重入（retryRejoinCached，线上三台缓存均完好）。已撤销。

## 测试

新增 4 条并全部双向验证（回退修复行 → 变红 → 还原变绿）：

- TestBecomeLeaderDeclinesToLiveLeader —— 环上已有存活 leader 时拒绝晋升
- TestBecomeLeaderPromotesWhenSoleSurvivor —— 守卫不能破坏真正的 failover
- TestTakeoverLeavesLiveLeadersForwardsAlone —— 拒绝接管时**不得**动对方转发
- TestTakeoverReassignsWhenSoleSurvivor —— 真正接管时转发必须重挂

★ 写测试时踩到一个陷阱：`AdoptState` 是 `e.state = s` 赋值，State.Nodes 是切片，
同一 seed 分给三个引擎会**共享底层数组**，互相改写——看起来像生产 bug，其实只是
测试构造问题（真实进程各自从 token 反序列化）。测试改为经 JSON 往返，各自独立持有。

★ 另一个自省：`git add -A` 把工作区里两个与本次无关的既有文件（docs/how-it-works.md、
scripts/probe-with-virtual-frps.sh）卷进了这个提交，已 reset 分离，它们另立提交。
本仓库再次印证：**只用 `git add <具体路径>`**。

go build / go vet / go test ./... 全绿，gofmt 干净。

internal/cluster/ring_leader.go

@@ -146,7 +146,14 @@ func (e *Engine) forwardToNext(ctx context.Context, tk *Token) error {
 		// the token is lost — start fresh as the new leader (plan §leader
 		// 补充/监控: "上家邻居探测到 leader 崩溃 → 自身成为新 leader").
 		if next == e.state.LeaderID {
-			e.becomeLeader()
+			if !e.becomeLeader() {
+				// Another live leader exists (we are the partitioned side):
+				// keep its ring intact and let it re-drive the token. Re-raise
+				// the node we just marked dead only if the send really was the
+				// problem — otherwise we would paper over a real outage.
+				log.Printf("ring[%s] not promoting: deferring to live leader", e.ID)
+				return nil
+			}
 			if e.Log != nil {
 				_, _ = e.Log.Append(e.ID, LogLeaderChange, map[string]string{"leader": e.ID})
 			}
@@ -161,13 +168,42 @@ func (e *Engine) forwardToNext(ctx context.Context, tk *Token) error {
 	return nil
 }
 
-// becomeLeader promotes this node (used when the monitored leader dies).
-func (e *Engine) becomeLeader() {
+// becomeLeader promotes this node, but ONLY when the ring agrees there is no
+// live leader to defer to.
+//
+// The guard matters because leadership was previously self-declared: any node
+// that failed to reach the leader it believed in would immediately promote
+// itself. Two nodes partitioned from each other both hit that path and both
+// ended up believing they were the leader — a split brain with two token
+// rings, duplicate claims, and each side reassigning the other's forwards. That
+// was observed live: after a rolling restart, two nodes reported the same
+// leader while marking each other dead, and the topology was emptied into
+// pending on both sides.
+//
+// Before promoting, ask the ring: is any OTHER node currently alive and claiming
+// leadership? If so, we are the partitioned one — demote ourselves and let the
+// live leader keep the ring. A genuine single-leader loss (the real leader is
+// dead or unreachable) leaves no such node, so promotion still happens.
+//
+// CreateCluster does not come through here: it assigns LeaderID directly, so
+// the "no other leader exists" case never needs an override.
+func (e *Engine) becomeLeader() bool {
+	for i := range e.state.Nodes {
+		n := &e.state.Nodes[i]
+		if n.ID == e.ID {
+			continue
+		}
+		if n.Alive && n.IsLeader {
+			log.Printf("ring[%s] decline promotion: %s is alive and is the leader", e.ID, n.ID)
+			return false
+		}
+	}
 	e.state.LeaderID = e.ID
 	for i := range e.state.Nodes {
 		e.state.Nodes[i].IsLeader = e.state.Nodes[i].ID == e.ID
 	}
 	log.Printf("ring[%s] promoted to leader", e.ID)
+	return true
 }
 
 // WatchLeader runs the FALLBACK leader liveness monitor. The PRIMARY path is
@@ -209,20 +245,43 @@ func (e *Engine) WatchLeader(ctx context.Context) {
 				cancel()
 				if err != nil {
 					log.Printf("ring[%s] heartbeat to leader %s failed: %v", e.ID, e.state.LeaderID, err)
-					e.state.MarkOffline(e.state.LeaderID)
-					e.state.OfflineReassign(e.state.LeaderID)
-					e.becomeLeader()
-					// Kick off a fresh cycle: the ring died with the old
-					// leader (no token inflight → WatchTokenLoss won't
-					// fire). Without this the newly promoted leader would
-					// sit idle and the ring would stay dead.
-					e.StartRing(ctx)
+					e.takeOverFromDeadLeader(ctx)
 				}
 			}
 		}
 	}
 }
 
+// takeOverFromDeadLeader handles "the leader I could not reach is gone": mark it
+// dead and, if the ring agrees there is no live leader left, promote this node
+// and re-seed the cycle.
+//
+// The promotion is deliberately conditional, and so is the reassignment. Doing
+// the reassignment unconditionally is what destroyed the live ring: a node on
+// the partitioned side would mark the (still very much alive) leader dead and
+// move ALL of its forwards into pending, so both sides ended up with an empty
+// topology and a queue nobody was serving. Sequencing matters — the forward
+// re-hang is only correct once we have actually taken leadership.
+//
+// Extracted from WatchLeader's loop so the decision is testable (the loop itself
+// is an endless goroutine).
+func (e *Engine) takeOverFromDeadLeader(ctx context.Context) {
+	old := e.state.LeaderID
+	e.state.MarkOffline(old)
+	if !e.becomeLeader() {
+		log.Printf("ring[%s] leader %s unreachable but a live leader remains; "+
+			"declining takeover and leaving its forwards untouched", e.ID, old)
+		return
+	}
+	e.state.OfflineReassign(old)
+	if e.Log != nil {
+		_, _ = e.Log.Append(e.ID, LogLeaderChange, map[string]string{"leader": e.ID})
+	}
+	// The ring died with the old leader (no token inflight → WatchTokenLoss
+	// won't fire). Without this the newly promoted leader would sit idle.
+	e.StartRing(ctx)
+}
+
 // WatchTokenLoss runs the leader's token-loss judge: if a token was sent and
 // does not return within LossTimeout, the leader issues a fresh token (all
 // nodes drop older stamps via the SentAt guard, so at most one circulates).