Merge: robot orchestrator lease 感知调度，消除跨节点 CONFLICT 与补偿风暴

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

services/room-service/internal/room/service/admin_robot_room.go

@@ -278,9 +278,37 @@ func (s *Service) startActiveRobotRooms(ctx context.Context) {
 		return
 	}
 	logx.Info(ctx, "robot_room_runtime_scan_completed", slog.Int("active_room_count", len(configs)))
+	// 在启动/停止决策前先记录已运行集合；扫描期间由管理端并发新建的房间不在本轮 active 列表里，
+	// 不能被误判为已停用。
+	runningRoomIDs := s.runningRobotRoomIDs()
+	activeRoomIDs := make(map[string]bool, len(configs))
 	for _, config := range configs {
+		activeRoomIDs[config.RoomID] = true
+		ownerNodeID, ownedElsewhere, err := s.roomLeaseOwnedElsewhere(ctx, config.AppCode, config.RoomID)
+		if err != nil {
+			// 目录抖动时保持现状：不认领也不停止，等下一轮扫描重新判定。
+			logx.Warn(ctx, "robot_room_runtime_lease_lookup_failed", slog.String("room_id", config.RoomID), slog.String("error", err.Error()))
+			continue
+		}
+		if ownedElsewhere {
+			// 其他节点持有有效 lease 时，本节点对该房间的任何命令都只会得到 CONFLICT；
+			// 不启动新循环，并停掉执行权已迁走后仍残留的本地循环。
+			if s.stopRobotRoomRuntime(config.RoomID) {
+				logx.Warn(ctx, "robot_room_runtime_ownership_lost", slog.String("room_id", config.RoomID), slog.String("owner_node_id", ownerNodeID))
+			}
+			continue
+		}
 		s.startRobotRoomRuntime(ctx, config)
 	}
+	for _, roomID := range runningRoomIDs {
+		if activeRoomIDs[roomID] {
+			continue
+		}
+		// 停用/删除请求可能落在其他节点；配置不再 active 的房间不能在任何节点继续跑循环。
+		if s.stopRobotRoomRuntime(roomID) {
+			logx.Info(ctx, "robot_room_runtime_stopped_inactive", slog.String("room_id", roomID))
+		}
+	}
 }
 
 func (s *Service) startRobotRoomRuntime(parent context.Context, config RobotRoomConfig) {
@@ -314,6 +342,23 @@ func (s *Service) startRobotRoomRuntime(parent context.Context, config RobotRoom
 	}
 	s.robotRuntimeMu.Unlock()
 
+	owned, ownerNodeID, err := s.claimRobotRoomOwnership(parent, config)
+	if err != nil {
+		logx.Warn(parent, "robot_room_runtime_lease_claim_failed", slog.String("room_id", config.RoomID), slog.String("error", err.Error()))
+		return
+	}
+	if !owned {
+		// 机器人房循环只能在 lease owner 节点运行；非 owner 节点启动只会对 owner 的房间刷 CONFLICT。
+		logx.Info(parent, "robot_room_runtime_start_skipped",
+			slog.String("room_id", config.RoomID),
+			slog.String("status", config.Status),
+			slog.Int("robot_count", len(config.RobotUserIDs)),
+			slog.String("reason", "not_owner"),
+			slog.String("owner_node_id", ownerNodeID),
+		)
+		return
+	}
+
 	if err := s.ensureRobotRoomParticipants(parent, config); err != nil {
 		logx.Warn(parent, "robot_room_runtime_prepare_failed", slog.String("room_id", config.RoomID), slog.String("error", err.Error()))
 		return
@@ -339,16 +384,97 @@ func (s *Service) startRobotRoomRuntime(parent context.Context, config RobotRoom
 		defer s.clearRobotRoomRuntime(config.RoomID, token)
 		s.runRobotRoomRuntime(ctx, config)
 	}()
+	go s.runRobotRoomLeaseKeeper(ctx, config, token)
 }
 
-func (s *Service) stopRobotRoomRuntime(roomID string) {
+// claimRobotRoomOwnership 在无有效 owner 或本节点已是 owner 时认领/续租机器人房 lease；
+// 其他节点持有有效 lease 时返回现任 owner，调用方必须放弃启动循环。
+func (s *Service) claimRobotRoomOwnership(ctx context.Context, config RobotRoomConfig) (bool, string, error) {
+	if s.directory == nil {
+		// 测试中的极简 Service 可能没有装配 directory；生产 HealthCheck 会拒绝这种配置。
+		return true, "", nil
+	}
+	lease, err := s.directory.EnsureOwner(ctx, runtimeRoomKey(config.AppCode, config.RoomID), s.nodeID, s.clock.Now(), s.leaseTTL)
+	if err != nil {
+		return false, "", err
+	}
+	if lease.NodeID != s.nodeID {
+		return false, lease.NodeID, nil
+	}
+	return true, "", nil
+}
+
+// runRobotRoomLeaseKeeper 以短于 leaseTTL 的周期续租机器人房 lease。
+// 机器人房的写命令间隔可能远大于 leaseTTL，只靠命令续租会让 lease 频繁过期、被其他节点的
+// 扫描抢走；续租发现执行权已被接管时立即停掉本地 runtime，不再对新 owner 的房间空转 CONFLICT。
+func (s *Service) runRobotRoomLeaseKeeper(ctx context.Context, config RobotRoomConfig, token string) {
+	if s.directory == nil {
+		return
+	}
+	interval := s.leaseTTL / 3
+	if interval <= 0 {
+		interval = time.Second
+	}
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+		}
+		lease, err := s.directory.EnsureOwner(ctx, runtimeRoomKey(config.AppCode, config.RoomID), s.nodeID, s.clock.Now(), s.leaseTTL)
+		if err != nil {
+			// 目录抖动时保留 runtime 等下一次续租；期间的房间命令也会经 ensureCell 续租。
+			logx.Warn(ctx, "robot_room_lease_renew_failed", slog.String("room_id", config.RoomID), slog.String("error", err.Error()))
+			continue
+		}
+		if lease.NodeID != s.nodeID {
+			logx.Warn(ctx, "robot_room_runtime_ownership_lost",
+				slog.String("room_id", config.RoomID),
+				slog.String("owner_node_id", lease.NodeID),
+			)
+			s.stopRobotRoomRuntimeToken(config.RoomID, token)
+			return
+		}
+	}
+}
+
+// stopRobotRoomRuntime 停止指定房间的本地循环；返回是否确实存在运行中的 runtime。
+func (s *Service) stopRobotRoomRuntime(roomID string) bool {
 	s.robotRuntimeMu.Lock()
-	runtime := s.robotRuntimes[roomID]
+	runtime, exists := s.robotRuntimes[roomID]
 	delete(s.robotRuntimes, roomID)
 	s.robotRuntimeMu.Unlock()
 	if runtime.cancel != nil {
 		runtime.cancel()
 	}
+	return exists
+}
+
+// stopRobotRoomRuntimeToken 只停止仍由该 token 代表的 runtime，避免误停同房间的新一轮启动。
+func (s *Service) stopRobotRoomRuntimeToken(roomID string, token string) {
+	s.robotRuntimeMu.Lock()
+	runtime, exists := s.robotRuntimes[roomID]
+	if !exists || runtime.token != token {
+		s.robotRuntimeMu.Unlock()
+		return
+	}
+	delete(s.robotRuntimes, roomID)
+	s.robotRuntimeMu.Unlock()
+	if runtime.cancel != nil {
+		runtime.cancel()
+	}
+}
+
+func (s *Service) runningRobotRoomIDs() []string {
+	s.robotRuntimeMu.Lock()
+	defer s.robotRuntimeMu.Unlock()
+	roomIDs := make([]string, 0, len(s.robotRuntimes))
+	for roomID := range s.robotRuntimes {
+		roomIDs = append(roomIDs, roomID)
+	}
+	return roomIDs
 }
 
 func (s *Service) stopAllRobotRoomRuntimes() {

services/room-service/internal/room/service/admin_robot_room_lease_test.go

@@ -0,0 +1,184 @@
+package service
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"hyapp/pkg/appcode"
+	"hyapp/services/room-service/internal/router"
+)
+
+type robotRoomScanRepository struct {
+	Repository
+	configs []RobotRoomConfig
+}
+
+func (r *robotRoomScanRepository) ListActiveRobotRooms(context.Context) ([]RobotRoomConfig, error) {
+	return r.configs, nil
+}
+
+func activeRobotRoomConfig(roomID string) RobotRoomConfig {
+	return RobotRoomConfig{
+		AppCode:      appcode.Default,
+		RoomID:       roomID,
+		Status:       robotRoomStatusActive,
+		RobotUserIDs: []int64{1001, 1002},
+	}
+}
+
+func TestClaimRobotRoomOwnershipRespectsOtherValidOwner(t *testing.T) {
+	ctx := context.Background()
+	now := time.Date(2026, 7, 12, 9, 0, 0, 0, time.UTC)
+	directory := router.NewMemoryDirectory()
+	svc := &Service{
+		nodeID:    "room-node-a",
+		leaseTTL:  10 * time.Second,
+		clock:     fixedRoomTestClock{now: now},
+		directory: directory,
+	}
+	config := activeRobotRoomConfig("robot-claim-test")
+
+	owned, ownerNodeID, err := svc.claimRobotRoomOwnership(ctx, config)
+	if err != nil || !owned || ownerNodeID != "" {
+		t.Fatalf("unowned robot room should be claimed by this node: owned=%v owner=%q err=%v", owned, ownerNodeID, err)
+	}
+	if owned, _, err := svc.claimRobotRoomOwnership(ctx, config); err != nil || !owned {
+		t.Fatalf("self-owned robot room should renew: owned=%v err=%v", owned, err)
+	}
+
+	directory.ForceExpire(runtimeRoomKey(config.AppCode, config.RoomID))
+	if _, err := directory.EnsureOwner(ctx, runtimeRoomKey(config.AppCode, config.RoomID), "room-node-b", now, 10*time.Second); err != nil {
+		t.Fatalf("other node takeover: %v", err)
+	}
+	owned, ownerNodeID, err = svc.claimRobotRoomOwnership(ctx, config)
+	if err != nil || owned || ownerNodeID != "room-node-b" {
+		t.Fatalf("valid lease on another node must not be claimed: owned=%v owner=%q err=%v", owned, ownerNodeID, err)
+	}
+}
+
+func TestStartRobotRoomRuntimeSkipsWhenOwnedElsewhere(t *testing.T) {
+	ctx := context.Background()
+	now := time.Date(2026, 7, 12, 9, 0, 0, 0, time.UTC)
+	directory := router.NewMemoryDirectory()
+	svc := &Service{
+		nodeID:    "room-node-a",
+		leaseTTL:  10 * time.Second,
+		clock:     fixedRoomTestClock{now: now},
+		directory: directory,
+	}
+	config := activeRobotRoomConfig("robot-owned-elsewhere")
+	if _, err := directory.EnsureOwner(ctx, runtimeRoomKey(config.AppCode, config.RoomID), "room-node-b", now, 10*time.Second); err != nil {
+		t.Fatalf("seed other-node lease: %v", err)
+	}
+
+	// repository 故意留空：非 owner 分支必须在触达任何房间命令/仓储调用前返回。
+	svc.startRobotRoomRuntime(ctx, config)
+	if len(svc.robotRuntimes) != 0 {
+		t.Fatalf("non-owner node must not start robot room runtime, got %d", len(svc.robotRuntimes))
+	}
+}
+
+func TestStartActiveRobotRoomsReconcilesOwnershipAndInactiveRooms(t *testing.T) {
+	ctx := context.Background()
+	now := time.Date(2026, 7, 12, 9, 0, 0, 0, time.UTC)
+	directory := router.NewMemoryDirectory()
+	ownedConfig := activeRobotRoomConfig("robot-owned")
+	lostConfig := activeRobotRoomConfig("robot-lease-lost")
+
+	lostCancelled := false
+	staleCancelled := false
+	ownedCancelled := false
+	svc := &Service{
+		nodeID:     "room-node-a",
+		leaseTTL:   10 * time.Second,
+		clock:      fixedRoomTestClock{now: now},
+		directory:  directory,
+		repository: &robotRoomScanRepository{configs: []RobotRoomConfig{ownedConfig, lostConfig}},
+		robotRuntimes: map[string]robotRoomRuntime{
+			ownedConfig.RoomID: {cancel: func() { ownedCancelled = true }, token: "owned-token"},
+			lostConfig.RoomID:  {cancel: func() { lostCancelled = true }, token: "lost-token"},
+			"robot-stale":      {cancel: func() { staleCancelled = true }, token: "stale-token"},
+		},
+	}
+	if _, err := directory.EnsureOwner(ctx, runtimeRoomKey(ownedConfig.AppCode, ownedConfig.RoomID), svc.nodeID, now, svc.leaseTTL); err != nil {
+		t.Fatalf("seed self lease: %v", err)
+	}
+	if _, err := directory.EnsureOwner(ctx, runtimeRoomKey(lostConfig.AppCode, lostConfig.RoomID), "room-node-b", now, svc.leaseTTL); err != nil {
+		t.Fatalf("seed other-node lease: %v", err)
+	}
+
+	svc.startActiveRobotRooms(ctx)
+
+	if _, exists := svc.robotRuntimes[ownedConfig.RoomID]; !exists || ownedCancelled {
+		t.Fatalf("runtime owned by this node must keep running: exists=%v cancelled=%v", exists, ownedCancelled)
+	}
+	if _, exists := svc.robotRuntimes[lostConfig.RoomID]; exists || !lostCancelled {
+		t.Fatalf("runtime whose lease moved to another node must stop: exists=%v cancelled=%v", exists, lostCancelled)
+	}
+	if _, exists := svc.robotRuntimes["robot-stale"]; exists || !staleCancelled {
+		t.Fatalf("runtime without active config must stop: exists=%v cancelled=%v", exists, staleCancelled)
+	}
+}
+
+func TestRobotRoomLeaseKeeperStopsRuntimeWhenOwnershipMoves(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	now := time.Date(2026, 7, 12, 9, 0, 0, 0, time.UTC)
+	directory := router.NewMemoryDirectory()
+	svc := &Service{
+		nodeID:    "room-node-a",
+		leaseTTL:  30 * time.Millisecond,
+		clock:     fixedRoomTestClock{now: now},
+		directory: directory,
+	}
+	config := activeRobotRoomConfig("robot-keeper")
+	key := runtimeRoomKey(config.AppCode, config.RoomID)
+	if _, err := directory.EnsureOwner(ctx, key, svc.nodeID, now, svc.leaseTTL); err != nil {
+		t.Fatalf("seed self lease: %v", err)
+	}
+
+	runtimeCancelled := make(chan struct{})
+	svc.robotRuntimes = map[string]robotRoomRuntime{
+		config.RoomID: {cancel: func() { close(runtimeCancelled) }, token: "keeper-token"},
+	}
+	go svc.runRobotRoomLeaseKeeper(ctx, config, "keeper-token")
+
+	// 续租周期内 keeper 必须保持 owner 且不打断 runtime。
+	time.Sleep(50 * time.Millisecond)
+	select {
+	case <-runtimeCancelled:
+		t.Fatal("keeper must not stop runtime while this node still owns the lease")
+	default:
+	}
+	if lease, exists, err := directory.Lookup(ctx, key); err != nil || !exists || lease.NodeID != svc.nodeID || !lease.ValidAt(now) {
+		t.Fatalf("keeper should keep renewing the lease: lease=%+v exists=%v err=%v", lease, exists, err)
+	}
+
+	// keeper 会在 ForceExpire 和另一节点抢占之间继续续租，重试直到 node-b 真正拿到有效 lease。
+	takeoverDeadline := time.Now().Add(2 * time.Second)
+	for {
+		directory.ForceExpire(key)
+		if _, err := directory.EnsureOwner(ctx, key, "room-node-b", now, time.Minute); err != nil {
+			t.Fatalf("other node takeover: %v", err)
+		}
+		if lease, exists, err := directory.Lookup(ctx, key); err == nil && exists && lease.NodeID == "room-node-b" {
+			break
+		}
+		if time.Now().After(takeoverDeadline) {
+			t.Fatal("other node could not take over the lease")
+		}
+	}
+
+	select {
+	case <-runtimeCancelled:
+	case <-time.After(2 * time.Second):
+		t.Fatal("keeper must stop the local runtime after ownership moved to another node")
+	}
+	svc.robotRuntimeMu.Lock()
+	_, exists := svc.robotRuntimes[config.RoomID]
+	svc.robotRuntimeMu.Unlock()
+	if exists {
+		t.Fatal("keeper must remove the runtime entry after ownership loss")
+	}
+}

services/room-service/internal/room/service/create_room.go

@@ -145,7 +145,7 @@ func (s *Service) CreateRoom(ctx context.Context, req *roomv1.CreateRoomRequest)
 
 	if lease.NodeID != s.nodeID {
 		// 有其他有效 owner 时当前节点不能创建或接管该房间。
-		return nil, xerr.New(xerr.Conflict, fmt.Sprintf("room is owned by %s", lease.NodeID))
+		return nil, roomOwnedByOtherNodeError(lease.NodeID)
 	}
 
 	if _, exists, err := s.repository.GetRoomMeta(ctx, cmd.RoomID()); err != nil {

services/room-service/internal/room/service/gift_operation.go

@@ -203,8 +203,9 @@ func (f *giftFlow) handleGiftOperationFailure(err error) {
 	ctx, cancel := context.WithTimeout(context.WithoutCancel(f.ctx), giftOperationPersistTimeout)
 	defer cancel()
 	now := f.s.clock.Now()
-	if !f.debitCompleted && !giftOperationErrorRetryable(err) {
+	if !f.debitCompleted && !giftOperationErrorRetryable(err) && !isRoomOwnedByOtherNodeError(err) {
 		// wallet 明确拒绝（余额不足、配置非法等）保证没有扣款，pending 可以安全关闭；同载荷的用户主动重试会重新激活。
+		// 跨节点执行权冲突不在此列：它只说明本节点不是 owner，首次尝试可能已经扣款，必须留给 owner 节点重试发现回执。
 		if persistErr := f.s.repository.MarkGiftOperationCompensated(ctx, f.cmd.RoomID(), f.cmd.ID(), err.Error(), now.UnixMilli()); persistErr != nil {
 			slog.WarnContext(ctx, "mark gift operation compensated failed", "room_id", f.cmd.RoomID(), "command_id", f.cmd.ID(), "error", persistErr)
 		}
@@ -334,6 +335,15 @@ func (s *Service) recoverGiftOperation(workerCtx context.Context, operation Gift
 	defer cancel()
 
 	now := s.clock.Now()
+	if ownerNodeID, ownedElsewhere, err := s.roomLeaseOwnedElsewhere(operationCtx, operation.AppCode, operation.RoomID); err == nil && ownedElsewhere {
+		// 非 owner 节点执行恢复只会在 ensureCell 得到跨节点 CONFLICT；释放 claim 让 owner 节点的
+		// worker 在下一轮完成结算，避免恢复任务在错误节点空转。目录查询失败时按本地可执行处理，
+		// ensureCell 会再做权威判定。
+		if persistErr := s.repository.MarkGiftOperationRetry(operationCtx, operation.RoomID, operation.CommandID, roomOwnedByOtherNodeMsgPrefix+ownerNodeID, now.Add(giftOperationRetryDelay).UnixMilli(), now.UnixMilli(), false); persistErr != nil {
+			slog.WarnContext(operationCtx, "defer gift operation to owner failed", "room_id", operation.RoomID, "command_id", operation.CommandID, "error", persistErr)
+		}
+		return
+	}
 	var req roomv1.SendGiftRequest
 	if err := proto.Unmarshal(operation.RequestPayload, &req); err != nil {
 		_ = s.repository.MarkGiftOperationRetry(operationCtx, operation.RoomID, operation.CommandID, "stored request decode failed", now.Add(time.Minute).UnixMilli(), now.UnixMilli(), false)

services/room-service/internal/room/service/gift_operation_internal_test.go

@@ -13,8 +13,32 @@ import (
 	"hyapp/pkg/appcode"
 	"hyapp/pkg/xerr"
 	"hyapp/services/room-service/internal/room/command"
+	"hyapp/services/room-service/internal/router"
 )
 
+type giftOperationOwnershipRepository struct {
+	Repository
+	retries     []giftOperationRetryCall
+	compensated []giftOperationRetryCall
+}
+
+type giftOperationRetryCall struct {
+	roomID        string
+	commandID     string
+	lastError     string
+	preserveLease bool
+}
+
+func (r *giftOperationOwnershipRepository) MarkGiftOperationRetry(_ context.Context, roomID string, commandID string, lastError string, _ int64, _ int64, preserveLease bool) error {
+	r.retries = append(r.retries, giftOperationRetryCall{roomID: roomID, commandID: commandID, lastError: lastError, preserveLease: preserveLease})
+	return nil
+}
+
+func (r *giftOperationOwnershipRepository) MarkGiftOperationCompensated(_ context.Context, roomID string, commandID string, lastError string, _ int64) error {
+	r.compensated = append(r.compensated, giftOperationRetryCall{roomID: roomID, commandID: commandID, lastError: lastError})
+	return nil
+}
+
 type compensatedGiftOperationRepository struct {
 	Repository
 	operation       GiftOperation
@@ -64,6 +88,88 @@ func TestGiftOperationRecoveryDeadlineKeepsClaimUntilCellCommitBudgetEnds(t *tes
 	}
 }
 
+func TestGiftOperationFailureOwnershipConflictRetriesInsteadOfCompensating(t *testing.T) {
+	ctx := appcode.WithContext(context.Background(), appcode.Default)
+	now := time.Date(2026, 7, 12, 9, 0, 0, 0, time.UTC)
+	req := &roomv1.SendGiftRequest{
+		Meta:         &roomv1.RequestMeta{AppCode: appcode.Default, RoomId: "room-owner-conflict", CommandId: "cmd-owner-conflict", ActorUserId: 101},
+		TargetType:   "user",
+		TargetUserId: 202,
+		GiftId:       "rose",
+		GiftCount:    1,
+	}
+
+	repository := &giftOperationOwnershipRepository{}
+	flow := newGiftFlow(&Service{repository: repository, clock: fixedRoomTestClock{now: now}}, ctx, req, giftSendOptions{})
+	flow.operationStarted = true
+	flow.recovering = true
+	flow.handleGiftOperationFailure(roomOwnedByOtherNodeError("room-node-b"))
+	if len(repository.compensated) != 0 {
+		t.Fatalf("cross-node ownership conflict must not compensate a pending saga: %+v", repository.compensated)
+	}
+	if len(repository.retries) != 1 || repository.retries[0].preserveLease {
+		t.Fatalf("ownership conflict should release the claim for the owner node to retry: %+v", repository.retries)
+	}
+
+	repository = &giftOperationOwnershipRepository{}
+	flow = newGiftFlow(&Service{repository: repository, clock: fixedRoomTestClock{now: now}}, ctx, req, giftSendOptions{})
+	flow.operationStarted = true
+	flow.handleGiftOperationFailure(xerr.New(xerr.Conflict, "command_id payload conflict"))
+	if len(repository.compensated) != 1 || len(repository.retries) != 0 {
+		t.Fatalf("business conflict before debit must stay compensated: compensated=%+v retries=%+v", repository.compensated, repository.retries)
+	}
+}
+
+func TestRecoverGiftOperationDefersToOwnerNode(t *testing.T) {
+	ctx := context.Background()
+	now := time.Date(2026, 7, 12, 9, 0, 0, 0, time.UTC)
+	directory := router.NewMemoryDirectory()
+	repository := &giftOperationOwnershipRepository{}
+	svc := &Service{
+		nodeID:     "room-node-a",
+		leaseTTL:   10 * time.Second,
+		clock:      fixedRoomTestClock{now: now},
+		directory:  directory,
+		repository: repository,
+	}
+	operation := GiftOperation{
+		AppCode:   appcode.Default,
+		RoomID:    "room-recovery-owner",
+		CommandID: "cmd-recovery-owner",
+		Status:    GiftOperationStatusPending,
+	}
+	if _, err := directory.EnsureOwner(ctx, runtimeRoomKey(operation.AppCode, operation.RoomID), "room-node-b", now, 10*time.Second); err != nil {
+		t.Fatalf("seed other-node lease: %v", err)
+	}
+
+	// RequestPayload 故意留空：非 owner 分支必须在解码存量请求、进入 Room Cell 之前把任务让给 owner。
+	svc.recoverGiftOperation(ctx, operation)
+	if len(repository.retries) != 1 {
+		t.Fatalf("non-owner recovery must defer exactly once, got %+v", repository.retries)
+	}
+	if repository.retries[0].preserveLease || repository.retries[0].lastError != roomOwnedByOtherNodeMsgPrefix+"room-node-b" {
+		t.Fatalf("defer must release the claim and record the current owner: %+v", repository.retries[0])
+	}
+	if len(repository.compensated) != 0 {
+		t.Fatalf("defer must not compensate the operation: %+v", repository.compensated)
+	}
+}
+
+func TestIsRoomOwnedByOtherNodeError(t *testing.T) {
+	if !isRoomOwnedByOtherNodeError(roomOwnedByOtherNodeError("room-node-b")) {
+		t.Fatal("ownership conflict error must be recognized")
+	}
+	for _, err := range []error{
+		xerr.New(xerr.Conflict, "command_id payload conflict"),
+		xerr.New(xerr.Unavailable, roomOwnedByOtherNodeMsgPrefix+"room-node-b"),
+		errors.New(roomOwnedByOtherNodeMsgPrefix + "room-node-b"),
+	} {
+		if isRoomOwnedByOtherNodeError(err) {
+			t.Fatalf("must not treat %v as cross-node ownership conflict", err)
+		}
+	}
+}
+
 func TestCompensatedGiftOperationReactivatesOnlyAfterPayloadMatch(t *testing.T) {
 	ctx := appcode.WithContext(context.Background(), appcode.Default)
 	originalRequest := &roomv1.SendGiftRequest{

services/room-service/internal/room/service/helpers.go

@@ -130,6 +130,22 @@ func runtimeRoomKeyFromContext(ctx context.Context, roomID string) string {
 	return runtimeRoomKey(appcode.FromContext(ctx), roomID)
 }
 
+// roomLeaseOwnedElsewhere 判断房间执行权当前是否由其他节点持有有效 lease；只读判定，不接管也不续租。
+func (s *Service) roomLeaseOwnedElsewhere(ctx context.Context, appCode string, roomID string) (string, bool, error) {
+	if s.directory == nil {
+		// 测试中的极简 Service 可能没有装配 directory；生产 HealthCheck 会拒绝这种配置。
+		return "", false, nil
+	}
+	lease, exists, err := s.directory.Lookup(ctx, runtimeRoomKey(appCode, roomID))
+	if err != nil {
+		return "", false, err
+	}
+	if exists && lease.ValidAt(s.clock.Now()) && lease.NodeID != s.nodeID {
+		return lease.NodeID, true, nil
+	}
+	return "", false, nil
+}
+
 func (s *Service) loadedRoomLease(ctx context.Context, roomRef loadedRoomRef, now time.Time) (router.Lease, bool, error) {
 	// 后台 worker 通过 EnsureOwner 续租当前 Cell，不能在没有 lease 的情况下读取旧内存态。
 	lease, err := s.directory.EnsureOwner(ctx, runtimeRoomKey(roomRef.AppCode, roomRef.RoomID), s.nodeID, now, s.leaseTTL)

services/room-service/internal/room/service/recovery.go

@@ -3,6 +3,7 @@ package service
 import (
 	"context"
 	"fmt"
+	"strings"
 
 	"google.golang.org/protobuf/proto"
 	roomv1 "hyapp.local/api/proto/room/v1"
@@ -14,6 +15,21 @@ import (
 	"hyapp/services/room-service/internal/router"
 )
 
+// roomOwnedByOtherNodeMsgPrefix 是跨节点执行权冲突的稳定错误文案前缀；
+// isRoomOwnedByOtherNodeError 依赖它把这类短暂冲突和业务终态 Conflict 区分开。
+const roomOwnedByOtherNodeMsgPrefix = "room is owned by "
+
+// roomOwnedByOtherNodeError 表示房间执行权由其他节点的有效 lease 持有。
+func roomOwnedByOtherNodeError(nodeID string) error {
+	return xerr.New(xerr.Conflict, roomOwnedByOtherNodeMsgPrefix+nodeID)
+}
+
+// isRoomOwnedByOtherNodeError 判断错误是否为跨节点执行权冲突。
+// 它只在 lease TTL 窗口内成立，不代表业务不可重试；补偿判定不能把它当作终态。
+func isRoomOwnedByOtherNodeError(err error) bool {
+	return xerr.CodeOf(err) == xerr.Conflict && strings.HasPrefix(xerr.MessageOf(err), roomOwnedByOtherNodeMsgPrefix)
+}
+
 // ensureCell 确保当前节点拥有房间执行权，并在本地缺少 Cell 时从 MySQL 恢复。
 func (s *Service) ensureCell(ctx context.Context, roomID string) (*cell.RoomCell, router.Lease, error) {
 	// 每次写命令先刷新或接管 lease，确保同一房间同一时刻只有一个执行节点。
@@ -24,7 +40,7 @@ func (s *Service) ensureCell(ctx context.Context, roomID string) (*cell.RoomCell
 
 	if lease.NodeID != s.nodeID {
 		// 仍有其他有效 owner 时当前节点不能执行该房间命令。
-		return nil, router.Lease{}, xerr.New(xerr.Conflict, fmt.Sprintf("room is owned by %s", lease.NodeID))
+		return nil, router.Lease{}, roomOwnedByOtherNodeError(lease.NodeID)
 	}
 
 	if roomCell := s.loadCellForLease(ctx, roomID, lease); roomCell != nil {