fix(relay): publish admission authority under the room lock

A guest joining while the host handed the room to someone else could end up
following the host everyone else had just left behind.

join installs the client into the room and snapshots HostPeerID under
room.mu, then released the lock before enqueueing joined. The client is a
hostChanged recipient and a legal transfer target from the moment it is
installed, so a transfer committing in that gap enqueued the newer authority
first and the stale joined frame overwrote it on the client. transferHost
already enqueues while holding the lock for exactly this reason; joined
carries the same authority field and did not.

The admission frame is now built and enqueued while the room is still locked,
and enqueueFrame never blocks, so nothing waits on network I/O under the
lock. A beforeJoinRoomAck barrier alongside the two existing test hooks pins
the ordering: a transfer cannot commit while an admission is unpublished.

server/main.go

@@ -1094,6 +1094,7 @@ type Server struct {
 	oauth                  *oauthProxy // nil when OAUTH_BASE_URL is unset
 	removalErrors          removalErrorThrottle
 	beforeJoinRoomLock     func() // test-only deterministic admission barrier
+	beforeJoinRoomAck      func() // test-only authority-publication ordering barrier
 	beforeLeaveRoomLock    func() // test-only mutation/capture ordering barrier
 	beforeTerminalDelivery func() // test-only post-persistence, pre-delivery barrier
 	mu                     sync.RWMutex
@@ -2075,9 +2076,33 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
 			}
 			room.LastActivityAt = time.Now()
 			peers := room.peerIDs()
-			hostPeerID := room.HostPeerID
 			roomProtocolVersion := room.ProtocolVersion
+			existingPeers := make([]string, 0, len(peers)-1)
+			for _, peerID := range peers {
+				if peerID != msg.PeerID {
+					existingPeers = append(existingPeers, peerID)
+				}
+			}
 			s.snap.recordMutation()
+			// The admission carries host authority, so like hostChanged it is
+			// enqueued while the room is still locked. This client became a
+			// hostChanged recipient at room.Peers[msg.PeerID] above, and its
+			// reservation makes it a legal transfer target immediately; a
+			// transfer committing in the gap would otherwise enqueue the newer
+			// authority first and leave this stale one to overwrite it.
+			// enqueueFrame never blocks (a full queue closes the client), so
+			// holding room.mu across it is safe.
+			if s.beforeJoinRoomAck != nil {
+				s.beforeJoinRoomAck()
+			}
+			client.sendJSON(serverMsg{
+				Type:            relayTypeJoined,
+				SessionID:       msg.SessionID,
+				HostPeerID:      room.HostPeerID,
+				ReconnectToken:  responseToken,
+				ProtocolVersion: roomProtocolVersion,
+				Peers:           existingPeers,
+			})
 			room.mu.Unlock()
 
 			currentRoom = room
@@ -2086,21 +2111,6 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
 				existingClient.close()
 			}
 			log.Printf("peer %s joined room %s", msg.PeerID, msg.SessionID)
-
-			existingPeers := make([]string, 0, len(peers)-1)
-			for _, peerID := range peers {
-				if peerID != msg.PeerID {
-					existingPeers = append(existingPeers, peerID)
-				}
-			}
-			client.sendJSON(serverMsg{
-				Type:            relayTypeJoined,
-				SessionID:       msg.SessionID,
-				HostPeerID:      hostPeerID,
-				ReconnectToken:  responseToken,
-				ProtocolVersion: roomProtocolVersion,
-				Peers:           existingPeers,
-			})
 			room.broadcastExcept(msg.PeerID, serverMsg{Type: relayTypePeerJoined, PeerID: msg.PeerID})
 
 		case relayTypeLeave:

server/main_test.go

@@ -5385,6 +5385,69 @@ func TestTransferHostSwapsAuthorityAndBroadcastsHostChanged(t *testing.T) {
 	}
 }
 
+// An admission and a host transfer both publish room authority, and a joining
+// client is a broadcast recipient — and a legal transfer target — from the
+// moment it is installed. So the admission must publish its authority while it
+// still holds room.mu: a transfer committing in the gap would enqueue the newer
+// authority first and let the stale joined frame re-pin the client to the
+// demoted host.
+func TestJoinPublishesHostAuthorityUnderRoomLock(t *testing.T) {
+	h := newRelayHarness(t)
+	host, _, _, _ := createModernRoomWithGuest(t, h, "XFER_ORDER", "6.3.4.1", "6.3.4.2")
+
+	atAck := make(chan struct{})
+	releaseAck := make(chan struct{})
+	var once sync.Once
+	h.srv.beforeJoinRoomAck = func() {
+		once.Do(func() {
+			close(atAck)
+			<-releaseAck
+		})
+	}
+
+	newcomerToken, _ := mustReconnectToken(t)
+	newcomer := h.dial(t, "6.3.4.3")
+	newcomer.send(clientMsg{
+		Type:            relayTypeJoin,
+		SessionID:       "XFER_ORDER",
+		PeerID:          "N",
+		ReconnectToken:  newcomerToken,
+		ProtocolVersion: relayProtocolVersion,
+	})
+
+	select {
+	case <-atAck:
+	case <-time.After(2 * time.Second):
+		t.Fatal("admission never reached the authority-publication barrier")
+	}
+
+	h.srv.mu.RLock()
+	room := h.srv.rooms["XFER_ORDER"]
+	h.srv.mu.RUnlock()
+	if room == nil {
+		close(releaseAck)
+		t.Fatal("room XFER_ORDER disappeared")
+	}
+	unlocked := room.mu.TryLock()
+	if unlocked {
+		room.mu.Unlock()
+	}
+	close(releaseAck)
+	if unlocked {
+		t.Fatal("join published host authority without holding room.mu")
+	}
+
+	// End to end: the transfer lands strictly after the admission, so the
+	// newcomer's last authority frame is the room's real host.
+	host.send(clientMsg{Type: relayTypeTransferHost, To: "G", ProtocolVersion: relayProtocolVersion})
+	if first := newcomer.recv(); first.Type != relayTypeJoined || first.HostPeerID != "H" {
+		t.Fatalf("first newcomer frame=%+v, want joined{H}", first)
+	}
+	if second := newcomer.recv(); second.Type != relayTypeHostChanged || second.HostPeerID != "G" {
+		t.Fatalf("second newcomer frame=%+v, want hostChanged{G}", second)
+	}
+}
+
 func TestTransferHostRejectsNonHostSender(t *testing.T) {
 	h := newRelayHarness(t)
 	host, guest, hostToken, _ := createModernRoomWithGuest(t, h, "XFER_NOT_HOST", "6.3.1.1", "6.3.1.2")