music-assistant-server

1"""Sync Group Player implementation."""
2
3from __future__ import annotations
4
5import asyncio
6import time
7from contextlib import asynccontextmanager
8from typing import TYPE_CHECKING, Any, cast
9
10from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
11from music_assistant_models.constants import PLAYER_CONTROL_FAKE
12from music_assistant_models.enums import ConfigEntryType, PlaybackState, PlayerFeature, PlayerType
13from music_assistant_models.errors import PlayerCommandFailed, UnsupportedFeaturedException
14from propcache import under_cached_property as cached_property
15
16from music_assistant.constants import (
17    APPLICATION_NAME,
18    CONF_DYNAMIC_GROUP_MEMBERS,
19    CONF_GROUP_MEMBERS,
20    CONF_POWER_CONTROL,
21)
22from music_assistant.controllers.players.constants import PlayerLockPurpose
23from music_assistant.models.player import DeviceInfo, Player, PlayerMedia
24
25from .constants import (
26    CONF_ALLOWED_MEMBERS,
27    CONF_ENTRY_SGP_NOTE,
28    EXTRA_FEATURES_FROM_MEMBERS,
29    IDLE_GRACE_SECONDS,
30    PLAYBACK_START_TIMEOUT,
31    PROVIDERS_WITH_DYNAMIC_LEADER_SWITCH,
32    REFORM_DEBOUNCE_SECONDS,
33)
34
35if TYPE_CHECKING:
36    from collections.abc import AsyncIterator
37
38    from music_assistant_models.player import PlayerSource
39
40    from .provider import SyncGroupProvider
41
42
43class SyncGroupPlayer(Player):
44    """Sync Group Player implementation."""
45
46    _attr_type: PlayerType = PlayerType.GROUP
47    sync_leader: Player | None = None
48    """The active sync leader player for this syncgroup."""
49
50    def __init__(
51        self,
52        provider: SyncGroupProvider,
53        player_id: str,
54    ) -> None:
55        """Initialize SyncGroupPlayer instance."""
56        super().__init__(provider, player_id)
57        self._attr_name = self.config.name or self.config.default_name or f"SyncGroup {player_id}"
58        self._attr_available = True
59        self._attr_device_info = DeviceInfo(model=provider.name, manufacturer=APPLICATION_NAME)
60        # Group players default to "no opinion" on power. The session lifecycle
61        # (form on play, dissolve on stop, debounced idle deform) governs whether
62        # the group is considered active. Users who want an explicit on/off button
63        # can assign 'Fake power control' which is then reflected via extra_data.
64        self._attr_powered = None
65        self._attr_needs_poll = True
66        # task that dissolves the group after the idle grace window expires
67        self._idle_grace_task: asyncio.Task[None] | None = None
68        # task that re-forms the group (debounced) after the sync leader was removed
69        self._reform_task: asyncio.Task[None] | None = None
70        # protocol hint for the debounced re-form, snapshotted before the old
71        # leader was cleared so the new leader keeps protocol continuity
72        self._reform_protocol_domain: str | None = None
73        # monotonic timestamp of the last playback start issued to the leader
74        # (-inf means never)
75        self._playback_start_at: float = float("-inf")
76        self._update_attributes()
77
78    @cached_property
79    def is_dynamic(self) -> bool:
80        """Return if the player is a dynamic group player."""
81        return bool(self.config.get_value(CONF_DYNAMIC_GROUP_MEMBERS, False))
82
83    @property
84    def synced_to(self) -> str | None:
85        """Return the id of the player this player is synced to (sync leader)."""
86        # groups can't be synced
87        return None
88
89    @property
90    def is_active_session(self) -> bool:
91        """
92        Return whether this sync group is currently holding its members.
93
94        The session is considered active while a sync leader is set (formed and
95        potentially playing/paused), while the idle grace timer is still
96        pending, or while a debounced re-form is pending. ``__final_active_group``
97        reads this to decide whether the configured members should be marked as
98        ``active_group`` for this group.
99        """
100        return (
101            self.sync_leader is not None
102            or self._idle_grace_task is not None
103            or self._reform_task is not None
104        )
105
106    async def on_config_updated(self) -> None:
107        """Handle logic when the PlayerConfig is first loaded or updated."""
108        # Config is only available after the player was registered
109        self._cache.clear()  # clear to prevent loading old is_dynamic
110        preset_members = cast("list[str]", self.config.get_value(CONF_GROUP_MEMBERS, []))
111        if self.is_dynamic:
112            # In dynamic mode the configured members act as a preset: they are
113            # pulled in when the group is powered on
114            self._attr_static_group_members = []
115            self._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
116        else:
117            self._attr_static_group_members = list(preset_members)
118            self._attr_supported_features.discard(PlayerFeature.SET_MEMBERS)
119        # Only realign the effective member list to the preset when we are
120        # dormant. Otherwise a config save (e.g. user toggling an unrelated
121        # field) would wipe any dynamic joins that happened during this session.
122        if not self.is_active_session:
123            self._attr_group_members = list(preset_members)
124
125    @property
126    def supported_features(self) -> set[PlayerFeature]:
127        """Return the supported features of the player."""
128        # PlayerFeature.POWER is intentionally NOT advertised by default: it forces
129        # users to remember to power the group off to release its members, which
130        # makes "play X on a single member from HA" silently redirect to the whole
131        # group long after playback ended. The new lifecycle forms the group on
132        # play and dissolves it on stop (with a short idle grace), so explicit
133        # power control is no longer required.
134        # Users who DO want an explicit on/off button can assign 'Fake power
135        # control' in the player config; we then advertise POWER so the UI shows
136        # the toggle. The raw config value is read here to avoid recursion via
137        # the power_control property (which itself may inspect supported features).
138        base_features: set[PlayerFeature] = {PlayerFeature.PLAY_MEDIA}
139        raw_power_conf = self.mass.config.get_raw_player_config_value(
140            self.player_id, CONF_POWER_CONTROL
141        )
142        if raw_power_conf == PLAYER_CONTROL_FAKE:
143            base_features.add(PlayerFeature.POWER)
144        if self.is_dynamic:
145            base_features.add(PlayerFeature.SET_MEMBERS)
146        if self.sync_leader:
147            # add features supported by the sync leader
148            for feature in EXTRA_FEATURES_FROM_MEMBERS:
149                if feature in self.sync_leader.state.supported_features:
150                    base_features.add(feature)
151        else:
152            # derive features from all (configured) group members
153            # so that features like volume control are always advertised
154            for member_id in self._attr_group_members:
155                member_player = self.mass.players.get_player(member_id)
156                if member_player and member_player.state.available:
157                    for feature in EXTRA_FEATURES_FROM_MEMBERS:
158                        if feature in member_player.state.supported_features:
159                            base_features.add(feature)
160        return base_features
161
162    @property
163    def requires_flow_mode(self) -> bool:
164        """Return if the player needs flow mode."""
165        if leader := self.sync_leader:
166            return leader.flow_mode
167        return False
168
169    @property
170    def supported_sample_rates(self) -> list[tuple[int, int]] | None:
171        """Return supported sample rates as defined by the sync leader."""
172        # not cached: sync_leader can change during dynamic group reforms,
173        # so we always re-resolve to stay in sync with the current leader
174        if leader := self.sync_leader:
175            return leader.get_supported_sample_rates()
176        return [(44100, 16), (48000, 16)]
177
178    @property
179    def active_source(self) -> str | None:
180        """Return the active source id of the current media (if any)."""
181        # NOTE: Not using 'state' here as we need the 'raw' value provided by the sync leader player
182        if not self.sync_leader:
183            return None
184        # deal with output protocols on the sync leader
185        output_protocol_domain: str | None = None
186        if (
187            self.sync_leader.active_output_protocol
188            and self.sync_leader.active_output_protocol != "native"
189        ):
190            if protocol_player := self.mass.players.get_player(
191                self.sync_leader.active_output_protocol
192            ):
193                output_protocol_domain = protocol_player.provider.domain
194        # active source as reported by the player itself
195        if (
196            self.sync_leader.active_source
197            # try to catch cases where player reports an active source
198            # that is actually from an active output protocol (e.g. AirPlay)
199            and self.sync_leader.active_source.lower() != output_protocol_domain
200            and not (
201                # try to handle sendspin bridge where the player itself
202                # is reporting the bridged protocol as active source
203                # we need to ignore that
204                output_protocol_domain == "sendspin"
205                and (
206                    self.sync_leader.active_source.lower()
207                    in ("airplay", "cast", "chromecast", "network")
208                )
209            )
210        ):
211            return self.sync_leader.active_source
212        return None
213
214    @property
215    def source_list(self) -> list[PlayerSource]:
216        """Return list of available (native) sources for this player."""
217        # NOTE: Not using 'state' here as we need the 'raw' value provided by the sync leader player
218        return self.sync_leader.source_list if self.sync_leader else []
219
220    @property
221    def can_group_with(self) -> set[str]:
222        """Return the id's of players this player can group with."""
223        if not self.is_dynamic:
224            # in case of static members,
225            # we can only group with the players defined in the config, so we return those directly
226            return set(self._attr_static_group_members)
227        # Aggregate can_group_with from ALL current group members (not just the leader).
228        # A sync group can accommodate protocol switches, so a player compatible with
229        # ANY current member is a valid candidate to join.
230        member_ids = self._attr_group_members if self._attr_group_members else []
231        # current members bypass the allow-list filter (filter constrains joiners only)
232        current_members = set(member_ids)
233        can_group_with: set[str] = set()
234        for member_id in member_ids:
235            member_player = self.mass.players.get_player(member_id)
236            if member_player and member_player.state.available:
237                can_group_with.add(member_player.player_id)
238                can_group_with.update(member_player.state.can_group_with)
239        if can_group_with:
240            return {
241                pid
242                for pid in can_group_with
243                if pid in current_members or self._is_member_allowed(pid)
244            }
245        # Without any available member to derive compatibility from (empty group or
246        # all members offline), offer any compatible player.
247        # Actual compatibility is validated when adding members
248        can_group_with = set()
249        for player in self.mass.players.iter_players(return_unavailable=False):
250            if not player.available or player.type == PlayerType.GROUP:
251                # let's avoid showing group players as options to group with
252                continue
253            if (
254                PlayerFeature.SET_MEMBERS in player.state.supported_features
255                and player.state.can_group_with
256                and not player.state.active_group
257            ):
258                can_group_with.add(player.player_id)
259        return {pid for pid in can_group_with if self._is_member_allowed(pid)}
260
261    @property
262    def group_members(self) -> list[str]:
263        """Return the list of parent player id's that are part of this sync group."""
264        if (sync_leader := self.sync_leader) and sync_leader.state.group_members:
265            # use state.group_members here so protocol specific id's get correctly translated
266            return sync_leader.state.group_members
267        return self._attr_group_members
268
269    async def get_config_entries(self) -> list[ConfigEntry]:
270        """Return all (provider/player specific) Config Entries for the given player (if any)."""
271        # keep saved player ids so the UI can render a user friendly name
272        # prevents the bug where only the player ids show up during playback
273        saved_ids = {
274            *cast("list[str]", self.config.get_value(CONF_GROUP_MEMBERS, []) or []),
275            *cast("list[str]", self.config.get_value(CONF_ALLOWED_MEMBERS, []) or []),
276        }
277        possible_players = sorted(
278            [
279                ConfigValueOption(x.player_id, title=x.display_name)
280                for x in self.mass.players.all_players(True, False)
281                if x.type != PlayerType.GROUP
282                and (
283                    x.player_id in saved_ids
284                    or (
285                        PlayerFeature.SET_MEMBERS in x.state.supported_features
286                        # also include synced followers: can_group_with returns empty
287                        # while slaved, but the player is still group-capable
288                        and (x.state.can_group_with or x.state.synced_to)
289                    )
290                )
291            ],
292            key=lambda x: x.title or "",
293        )
294        entries: list[ConfigEntry] = [
295            # syncgroup specific entries
296            CONF_ENTRY_SGP_NOTE,
297            ConfigEntry(
298                key=CONF_GROUP_MEMBERS,
299                type=ConfigEntryType.STRING,
300                multi_value=True,
301                default_value=[],
302                required=False,  # needed for dynamic members (which allows empty members list)
303                options=possible_players,
304            ),
305            ConfigEntry(
306                key=CONF_DYNAMIC_GROUP_MEMBERS,
307                type=ConfigEntryType.BOOLEAN,
308                default_value=False,
309                required=False,
310            ),
311            ConfigEntry(
312                key=CONF_ALLOWED_MEMBERS,
313                type=ConfigEntryType.STRING,
314                multi_value=True,
315                default_value=[],
316                required=False,
317                options=possible_players,
318                depends_on=CONF_DYNAMIC_GROUP_MEMBERS,
319                advanced=True,
320            ),
321        ]
322        return entries
323
324    async def power(self, powered: bool) -> None:
325        """
326        Handle POWER command to group player.
327
328        Only called when the user has assigned a power control (native or fake)
329        to the group. Powering ON pre-forms the group so its members are
330        captured immediately (matching the legacy behaviour for users who opt
331        in). Powering OFF stops any playback and dissolves the group.
332
333        :param powered: True to power on (form/capture), False to power off (dissolve).
334        """
335        # always cancel any pending idle-grace timer on explicit power transitions
336        self._cancel_idle_grace_timer()
337
338        if not powered and self.playback_state in (
339            PlaybackState.PLAYING,
340            PlaybackState.PAUSED,
341        ):
342            # stop directly via the leader to avoid re-entering our own stop()
343            # logic (which would dissolve before we re-dissolve below).
344            if sync_leader := self.sync_leader:
345                await self.mass.players._handle_cmd_stop(sync_leader.player_id)
346
347        if powered:
348            # apply the configured preset members on power-on so unjoins
349            # during a powered session stick until the next power cycle
350            preset_members = cast("list[str]", self.config.get_value(CONF_GROUP_MEMBERS, []))
351            self._attr_group_members = [
352                *preset_members,
353                *[x for x in self._attr_group_members if x not in preset_members],
354            ]
355            # form syncgroup when powering on
356            await self._form_syncgroup()
357        else:
358            # dissolve syncgroup when powering off
359            await self._dissolve_syncgroup()
360
361        if self._attr_powered != powered:
362            self._attr_powered = powered
363            self._update_attributes()
364            self.update_state()
365
366    async def stop(self) -> None:
367        """
368        Send STOP command to given player.
369
370        An explicit stop on the group dissolves the syncgroup immediately so the
371        members are released back to individual control. The idle grace timer is
372        intentionally only used when the queue ends naturally (playback_state
373        transitions to IDLE without a stop command). Users who want the group
374        to stay formed across stops can assign Fake power control and use that
375        to pin the group as 'active'.
376        """
377        self._cancel_idle_grace_timer()
378        # an explicit stop also voids any pending debounced re-form and the
379        # startup marker — the user asked for silence
380        self._cancel_reform_timer()
381        self._playback_start_at = float("-inf")
382        self._attr_current_media = None
383        if sync_leader := self.sync_leader:
384            # Use internal handler to target the sync leader directly,
385            # bypassing group/sync redirect that would loop back to this player.
386            await self.mass.players._handle_cmd_stop(sync_leader.player_id)
387        # Skip the dissolve when the user has explicitly powered the group on
388        # via Fake power control — they expect the group to stay 'active' until
389        # they power it off, even after a stop.
390        if self._attr_powered is True:
391            return
392        await self._dissolve_syncgroup()
393
394    async def play(self) -> None:
395        """Send PLAY (unpause) command to given player."""
396        # The controller has already powered us on, but the group may not be
397        # formed (e.g. after _dissolve_and_reform left us powered with no leader).
398        # _form_syncgroup is idempotent so calling it here is cheap when already formed.
399        await self._form_syncgroup()
400        # Hold the group's playback lock until the leader actually reports playing
401        # so a concurrent (un)group command can't race the in-flight start — which
402        # would otherwise leave a player streaming outside the group.
403        async with self._await_leader_playback():
404            await self.mass.players.cmd_resume(
405                self.player_id, self._attr_active_source, self._attr_current_media
406            )
407
408    async def poll(self) -> None:
409        """Poll player for state updates."""
410        self._update_attributes()
411        self.update_state()
412
413    async def play_media(self, media: PlayerMedia) -> None:
414        """Handle PLAY MEDIA on given player."""
415        self._attr_current_media = media
416        self._attr_active_source = media.source_id or None
417        # The controller has already powered us on, but the group may not be
418        # formed (e.g. after _dissolve_and_reform left us powered with no leader).
419        # _form_syncgroup is idempotent so calling it here is cheap when already formed.
420        await self._form_syncgroup()
421        if sync_leader := self.sync_leader:
422            # Use internal handler to target the sync leader directly,
423            # bypassing group/sync redirect that would loop back to this player.
424            # Hold the group's playback lock until the leader confirms playback
425            # (see play()) so a concurrent (un)group command can't race the start.
426            async with self._await_leader_playback():
427                await self.mass.players._handle_play_media(sync_leader.player_id, media)
428        else:
429            raise RuntimeError("An empty group cannot play media, consider adding members first")
430
431    async def enqueue_next_media(self, media: PlayerMedia) -> None:
432        """Handle enqueuing of a next media item on the player."""
433        if sync_leader := self.sync_leader:
434            if PlayerFeature.ENQUEUE not in sync_leader.state.supported_features:
435                # this may happen in race conditions where we just switched sync leaders
436                # and the new leader doesn't support enqueueing next media.
437                return
438            # Use internal handler to bypass group redirect logic and avoid infinite loop
439            await self.mass.players._handle_enqueue_next_media(sync_leader.player_id, media)
440
441    async def set_members(  # noqa: PLR0915
442        self,
443        player_ids_to_add: list[str] | None = None,
444        player_ids_to_remove: list[str] | None = None,
445    ) -> None:
446        """Handle SET_MEMBERS command on the player."""
447        if not self.is_dynamic:
448            raise UnsupportedFeaturedException(
449                f"Group {self.display_name} does not allow dynamically adding/removing members!",
450                translation_key="not_dynamic",
451                translation_owner=self.translation_owner,
452                translation_args=[self.display_name],
453            )
454        sync_leader = self.sync_leader or self._select_sync_leader(new_members=player_ids_to_add)
455        # A start that was just issued to the leader may not be reflected in the
456        # (transient) device state yet, so treat the startup window as playing —
457        # otherwise a (un)group command racing an in-flight start misreads the
458        # group as idle and skips the resume. An explicit pause always wins:
459        # PAUSED is deliberate user intent, never startup noise.
460        was_playing = self.playback_state == PlaybackState.PLAYING or (
461            self.playback_state != PlaybackState.PAUSED and self._playback_recently_started
462        )
463
464        # handle additions
465        final_players_to_add: list[str] = []
466        can_group_with = sync_leader.state.can_group_with.copy() if sync_leader else set()
467        for member_id in player_ids_to_add or []:
468            if member_id == self.player_id:
469                continue  # can not add self as member
470            if not self._is_member_allowed(member_id):
471                self.logger.warning(
472                    "Player %s is not allowed to join group %s by the configured player filters, "
473                    "skipping adding it as a member to the group",
474                    member_id,
475                    self.display_name,
476                )
477                continue
478            member = self.mass.players.get_player(member_id)
479            if member is None or not member.available:
480                continue
481            if not sync_leader:
482                # no leader yet (e.g. empty group) - just register the member
483                # the leader and protocol selection happen on the next form/play
484                if member_id not in self._attr_group_members:
485                    self._attr_group_members.append(member_id)
486                continue
487            if member_id != sync_leader.player_id and member_id not in can_group_with:
488                # incompatible with the current leader's protocols - do NOT register
489                # the member or it will linger in _attr_group_members forever without
490                # ever actually being synced.
491                self.logger.debug(
492                    f"Cannot add {member.display_name} to group {self.display_name} since it's "
493                    f"not compatible with the (current) sync leader"
494                )
495                continue
496            if member_id not in self._attr_group_members:
497                self._attr_group_members.append(member_id)
498            if member_id != sync_leader.player_id:
499                final_players_to_add.append(member_id)
500
501        # handle removals
502        final_players_to_remove: list[str] = []
503        leader_removed = False
504        for member_id in player_ids_to_remove or []:
505            if member_id not in self._attr_group_members:
506                # Fallback for a member that was grouped to the sync leader
507                # outside of MA (e.g. via the Sonos app): it isn't part of our
508                # tracked member list but does show up in group_members via the
509                # leader's live state. Forward its removal to the sync leader
510                # instead of silently skipping it.
511                if (
512                    self.sync_leader
513                    and member_id != self.sync_leader.player_id
514                    and member_id in self.sync_leader.state.group_members
515                ):
516                    final_players_to_remove.append(member_id)
517                continue
518            if member_id in self._attr_static_group_members:
519                # static members can not be removed from the group
520                raise PlayerCommandFailed(
521                    f"Cannot remove {member_id} from group {self.display_name} "
522                    "since it's a static member!"
523                )
524            if self.sync_leader and member_id == self.sync_leader.player_id:
525                leader_removed = True
526                continue
527            if member_id == self.player_id:
528                raise PlayerCommandFailed(
529                    f"Cannot remove {self.display_name} from itself as a member!",
530                    translation_key="remove_self",
531                    translation_owner=self.translation_owner,
532                    translation_args=[self.display_name],
533                )
534            self._attr_group_members.remove(member_id)
535            final_players_to_remove.append(member_id)
536
537        if self.sync_leader and leader_removed and self._attr_group_members:
538            # we removed the current sync leader, but we still have members in the group
539            old_leader_id = self.sync_leader.player_id
540            session_player = self._active_session_player()
541            supports_handoff = (
542                session_player is not None
543                and session_player.provider.domain in PROVIDERS_WITH_DYNAMIC_LEADER_SWITCH
544            )
545
546            if was_playing and supports_handoff:
547                # protocol supports dynamic leader switching: try to remove
548                # only the departing leader and keep remaining members playing.
549                # _dynamic_leader_switch will fall back to dissolve+reform
550                # automatically if the chosen new leader isn't already part of
551                # the live session (e.g. a freshly-added player).
552                await self._dynamic_leader_switch(old_leader_id)
553            else:
554                # protocol doesn't support dynamic leader switching or not playing
555                await self._dissolve_and_reform(old_leader_id, resume_playback=was_playing)
556        elif self.sync_leader and (leader_removed or not self._attr_group_members):
557            # we removed the current sync leader, and we have no members left in the group
558            # or we just removed the last member from the group, so we dissolve the syncgroup
559            # Use internal handler to stop the sync leader directly,
560            # bypassing group redirect that would loop back to this player.
561            async with self.mass.players.wait_for_player_update(
562                self.sync_leader.player_id, timeout=5
563            ):
564                await self.mass.players._handle_cmd_stop(self.sync_leader.player_id)
565            await self._dissolve_syncgroup()
566
567        elif self.sync_leader:
568            # just a regular member(s) added/removed action,
569            # we can simply update the syncgroup members on the sync leader.
570            # `active_protocol_domain` is derived from live state, so the
571            # group will naturally downshift once it re-forms if the last
572            # protocol-requiring member was removed.
573            # use _handle_set_members directly to avoid the redirect loop
574            # (cmd_set_members redirects sync-leader targets back to this syncgroup)
575            async with self.mass.players.get_player_lock(
576                self.sync_leader.player_id, PlayerLockPurpose.PLAYBACK
577            ):
578                await self.mass.players._handle_set_members(
579                    self.sync_leader,
580                    player_ids_to_add=final_players_to_add,
581                    player_ids_to_remove=final_players_to_remove,
582                )
583        elif self._reform_task is not None:
584            # leaderless with a debounced re-form pending: membership just changed,
585            # so re-arm the window — the re-form picks up the final member list.
586            self._schedule_reform_timer()
587        # NOTE: If we weren't playing before, we don't need to do anything else,
588        # since the syncing will be done once playback starts
589        self.mass.players.trigger_player_update(self.player_id)
590
591    def on_group_member_updated(
592        self, member_player: Player, changed_values: dict[str, tuple[Any, Any]]
593    ) -> None:
594        """Handle callback when a group member of the group player is updated."""
595        self._update_attributes()
596        super().on_group_member_updated(member_player, changed_values)
597
598    async def on_unload(self) -> None:
599        """Handle logic when the player is unloaded from the Player controller."""
600        self._cancel_idle_grace_timer()
601        self._cancel_reform_timer()
602        await super().on_unload()
603        # the player is going away; make sure we don't leave the protocol-level
604        # sync group standing with a now-nonexistent leader behind it.
605        if self.sync_leader is not None:
606            await self._dissolve_syncgroup()
607
608    @property
609    def active_protocol_domain(self) -> str | None:
610        """
611        Derive the active protocol domain for this sync group on the fly.
612
613        Returns the domain of the protocol currently carrying the live stream
614        session, EXCEPT when no remaining member actually requires that
615        non-native protocol — in which case the group should downshift and
616        this returns the leader's native provider domain. Always computed
617        from live state so it cannot drift from reality.
618
619        Because of that downshift this is a hint for the next leader selection,
620        not an address for the live session: use ``_active_session_player()``
621        to reach the players that are carrying the stream right now.
622        """
623        session_player = self._active_session_player()
624        if session_player is None or self.sync_leader is None:
625            return None
626        domain = session_player.provider.domain
627        native_domain = self.sync_leader.provider.domain
628        # If a non-native protocol is in use, only keep it as "active" for
629        # leader-selection purposes when some member still requires it.
630        if domain != native_domain and not self._any_member_requires_protocol_domain(domain):
631            return native_domain
632        return domain
633
634    def _is_member_allowed(self, player_id: str) -> bool:
635        """Return whether a player is allowed to join this group given the configured filter."""
636        # preset members should always be allowed to re-join
637        preset_members = cast("list[str]", self.config.get_value(CONF_GROUP_MEMBERS, []) or [])
638        if player_id in preset_members:
639            return True
640        allowed_members = cast("list[str]", self.config.get_value(CONF_ALLOWED_MEMBERS, []) or [])
641        return not allowed_members or player_id in allowed_members
642
643    async def _form_syncgroup(self) -> None:
644        """Form syncgroup by syncing all (possible) members."""
645        # any in-flight grace or debounced re-form timer is moot now —
646        # we're (re)forming the group
647        self._cancel_idle_grace_timer()
648        self._cancel_reform_timer()
649        self.logger.debug(
650            "Forming syncgroup %s, _attr_group_members=%s, sync_leader=%s",
651            self.display_name,
652            self._attr_group_members,
653            self.sync_leader.display_name if self.sync_leader else None,
654        )
655        # select new sync leader if needed
656        if not self.sync_leader:
657            self.sync_leader = self._select_sync_leader()
658
659        # pin the leader ref: a concurrent command (e.g. a dissolve) may clear
660        # or replace self.sync_leader while we await below
661        leader = self.sync_leader
662        if not leader:
663            # we have no members in the group, so we can't form a syncgroup
664            return
665
666        # ensure the sync leader is first in the list
667        self._attr_group_members = [
668            leader.player_id,
669            *[x for x in self._attr_group_members if x != leader.player_id],
670        ]
671        # If the leader still believes it's synced to a previous leader (e.g. we
672        # just picked a new leader after dissolving the old session and the
673        # protocol-level state hasn't propagated yet), wait for it to settle.
674        # Without this, the subsequent play_media call hits the provider's
675        # "I'm synced to another player" guard and gets rejected.
676        if leader.state.synced_to is not None:
677            self.logger.debug(
678                "Waiting for new leader %s to report synced_to=None before forming",
679                leader.display_name,
680            )
681            if not await self._wait_member_unsynced(leader.player_id):
682                # Leader is genuinely stuck — bail out before issuing play_media
683                # so we don't trigger the provider's "synced to another player"
684                # rejection. The caller (play / play_media) will surface this as
685                # a no-op form; the next user action can retry once the
686                # protocol layer has caught up.
687                self.logger.error(
688                    "Aborting syncgroup form for %s: leader %s is stuck synced",
689                    self.display_name,
690                    leader.display_name,
691                )
692                self.sync_leader = None
693                return
694            if self.sync_leader is not leader:
695                # the group was dissolved or re-led while we waited —
696                # this form attempt is stale, abort
697                return
698        # Translate the leader's group_members (may be protocol IDs) to parent IDs
699        # so we can compare against our _attr_group_members (always parent IDs)
700        already_synced = set(self._translate_to_parent_ids(leader.state.group_members))
701        members_to_sync = [
702            x for x in self._attr_group_members if x != leader.player_id and x not in already_synced
703        ]
704        if members_to_sync:
705            # If the sync leader is playing something independently, stop it first
706            # to prevent protocol switching from trying to resume the previous playback
707            # (we're about to start new playback on the syncgroup).
708            # Wait for the leader to actually reach IDLE before adding members,
709            # since some providers reject set_members while still playing.
710            if leader.state.playback_state == PlaybackState.PLAYING:
711                async with self.mass.players.wait_for_player_update(
712                    leader.player_id,
713                    attribute_name="playback_state",
714                    attribute_value=PlaybackState.IDLE,
715                    timeout=5,
716                ):
717                    await self.mass.players._handle_cmd_stop(leader.player_id)
718                if self.sync_leader is not leader:
719                    # the group was dissolved or re-led while we waited —
720                    # this form attempt is stale, abort
721                    return
722            # use _handle_set_members directly to avoid the redirect loop
723            # (cmd_set_members redirects sync-leader targets back to this syncgroup)
724            async with self.mass.players.get_player_lock(
725                leader.player_id, PlayerLockPurpose.PLAYBACK
726            ):
727                await self.mass.players._handle_set_members(
728                    leader, player_ids_to_add=members_to_sync
729                )
730
731    @asynccontextmanager
732    async def _await_leader_playback(self) -> AsyncIterator[None]:
733        """
734        Wait for the sync leader to confirm playback for the command run in the body.
735
736        Wrap the play/resume call that targets the leader in this context manager.
737        The group's playback lock (held by the caller) then stays acquired until the
738        leader actually reports playing, so a concurrent (un)group command cannot
739        race a start that has not yet taken effect at the device. A no-op when there
740        is no leader to wait on.
741        """
742        if (leader := self.sync_leader) is None:
743            yield
744            return
745        # stamp the start: device state is unreliable while a start settles, so
746        # group-command decisions treat this window as playing (see set_members)
747        self._playback_start_at = time.monotonic()
748        async with self.mass.players.wait_for_player_update(
749            leader.player_id,
750            attribute_name="playback_state",
751            attribute_value=PlaybackState.PLAYING,
752            timeout=PLAYBACK_START_TIMEOUT,
753        ):
754            yield
755
756    async def _dissolve_syncgroup(self) -> None:
757        """Dissolve the current syncgroup by ungrouping all members."""
758        # a dissolve is happening now — any pending grace or re-form timer is no
759        # longer needed (_dissolve_and_reform re-arms the re-form right after)
760        # and the session whose start the marker tracked is gone
761        self._cancel_idle_grace_timer()
762        self._cancel_reform_timer()
763        self._playback_start_at = float("-inf")
764        if sync_leader := self.sync_leader:
765            # dissolve the temporary syncgroup from the player that holds the members:
766            # ungrouping from a leader that no longer holds them is a no-op and would
767            # leave the members grouped and streaming with no way back
768            group_leader = self._protocol_group_leader(sync_leader)
769            sync_children = [
770                x for x in group_leader.state.group_members if x != group_leader.player_id
771            ]
772            if sync_children:
773                # wait for the leader's state to reflect the ungroup
774                # use _handle_set_members directly to avoid the redirect loop
775                # (cmd_set_members redirects sync-leader targets back to this syncgroup)
776                async with (
777                    self.mass.players.wait_for_player_update(group_leader.player_id, timeout=5),
778                    self.mass.players.get_player_lock(
779                        group_leader.player_id, PlayerLockPurpose.PLAYBACK
780                    ),
781                ):
782                    await self.mass.players._handle_set_members(
783                        group_leader, player_ids_to_remove=sync_children
784                    )
785            if group_leader is not sync_leader:
786                # our callers only ever stop the leader we track, so a provider-promoted
787                # one would keep streaming on its own once the members are released
788                await self.mass.players._handle_cmd_stop(group_leader.player_id)
789                self.mass.players.schedule_active_output_protocol_clear(group_leader)
790        # Clear the leader's active protocol once it stops playing; the controller's
791        # clearing in _handle_cmd_stop is skipped for a still-grouped protocol player.
792        if sync_leader:
793            self.mass.players.schedule_active_output_protocol_clear(sync_leader)
794        self.sync_leader = None
795        self._update_attributes()
796        self.update_state()
797
798    def _select_sync_leader(
799        self,
800        new_members: list[str] | None = None,
801        preferred_protocol_domain: str | None = None,
802    ) -> Player | None:
803        """
804        Select a (new) sync leader, preferring protocol continuity.
805
806        :param new_members: Optional list of newly added member ids to consider
807            when no current/static members are available.
808        :param preferred_protocol_domain: If provided, prefer members that
809            support this protocol domain so playback keeps using the same
810            protocol.
811        """
812        if self.group_members and self.sync_leader and self.sync_leader.state.available:
813            # current leader is still available, no need to select a new one
814            return self.sync_leader
815        # with selecting a new leader, we prioritize the static group members
816        group_members = self.static_group_members or self.group_members or new_members or []
817
818        # if a preferred protocol is given, prefer members that support it
819        if preferred_protocol_domain:
820            for member_id in group_members:
821                member_player = self.mass.players.get_player(member_id)
822                if (
823                    member_player
824                    and member_player.state.available
825                    and self._member_supports_protocol_domain(
826                        member_player, preferred_protocol_domain
827                    )
828                ):
829                    self.logger.debug(
830                        "Auto-selected %s as sync leader for group %s "
831                        "(supports active protocol %s)",
832                        member_player.display_name,
833                        self.display_name,
834                        preferred_protocol_domain,
835                    )
836                    return member_player
837
838        # fallback: pick any available member
839        for member_id in group_members:
840            member_player = self.mass.players.get_player(member_id)
841            if member_player and member_player.state.available:
842                self.logger.debug(
843                    f"Auto-selected {member_player.display_name} as sync leader for "
844                    f"group {self.display_name}"
845                )
846                return member_player
847        return None
848
849    # -----------------------------------------------------------------------
850    # Protocol awareness
851    # -----------------------------------------------------------------------
852    # A sync group can contain members from multiple protocol domains (e.g. a
853    # Sonos that can play via either its native protocol or via AirPlay). The
854    # group needs to:
855    #   - track which protocol the live session is using (active_protocol_domain)
856    #   - resolve which player actually owns the protocol-level session
857    #     (_active_session_player) so leader/handoff bookkeeping is done on
858    #     the right object
859    #   - choose new leaders that keep protocol continuity
860    #     (_member_supports_protocol_domain / _select_sync_leader)
861    #   - downshift to the native protocol when the last protocol-requiring
862    #     member is gone (_any_member_requires_protocol_domain)
863    #   - stay aligned with the protocol's own view of the group, both in member
864    #     order (_align_members_with_session) and in who leads it
865    #     (_protocol_group_leader)
866    # The helpers below cover those needs. Composition decisions (which protocol
867    # to use given the current member mix) intentionally live in the group:
868    # individual protocol providers don't have visibility into the rest of the
869    # group's members.
870
871    def _translate_to_parent_ids(self, player_ids: list[str]) -> list[str]:
872        """
873        Translate a list of (possibly protocol) player IDs to parent player IDs.
874
875        Protocol players (e.g. AirPlay `apc...`) are translated to their parent
876        (e.g. Sonos `RINCON_...`). Non-protocol IDs pass through unchanged.
877
878        :param player_ids: List of player IDs that may be protocol or parent IDs.
879        """
880        result: list[str] = []
881        for pid in player_ids:
882            if player := self.mass.players.get_player(pid):
883                parent_id = player.protocol_parent_id or pid
884                if parent_id not in result:
885                    result.append(parent_id)
886            elif pid not in result:
887                result.append(pid)
888        return result
889
890    def _member_supports_protocol_domain(self, player: Player, domain: str) -> bool:
891        """
892        Check if a player supports the given protocol domain.
893
894        :param player: The player to check.
895        :param domain: The protocol domain string (e.g. "airplay", "sonos").
896        """
897        if player.provider.domain == domain:
898            return True
899        for protocol in player.linked_output_protocols:
900            if protocol.protocol_domain == domain and protocol.available:
901                return True
902        return False
903
904    def _any_member_requires_protocol_domain(self, domain: str) -> bool:
905        """
906        Return True if any current member can only play via the given protocol domain.
907
908        A member "requires" the protocol when all of its available playback
909        paths are on that domain — i.e. it has no native playback path outside
910        this domain AND no linked output protocol outside this domain. This
911        covers plain protocol-domain players (e.g. AirPlay) as well as
912        UniversalPlayer wrappers whose native ``provider.domain`` is
913        ``universal_player`` but which can still only play via a single
914        linked protocol.
915
916        :param domain: The protocol domain string (e.g. "airplay", "sonos").
917        """
918        for member_id in self._attr_group_members:
919            member = self.mass.players.get_player(member_id)
920            if member is None or not member.state.available:
921                continue
922            # Collect the set of available playback path domains for this member.
923            paths: set[str] = set()
924            if member.is_native_player:
925                paths.add(member.provider.domain)
926            for protocol in member.linked_output_protocols:
927                if protocol.available:
928                    paths.add(protocol.protocol_domain)
929            if not paths:
930                # nothing available at all, skip rather than force a protocol
931                continue
932            if paths == {domain}:
933                return True
934        return False
935
936    def _active_session_player(self) -> Player | None:
937        """
938        Return the player that owns the live sync session.
939
940        If the current sync leader has a non-native active output protocol,
941        returns the protocol player that carries the stream; otherwise returns
942        the native sync leader itself. Returns ``None`` if there is no leader.
943        """
944        if not self.sync_leader:
945            return None
946        if (
947            self.sync_leader.active_output_protocol
948            and self.sync_leader.active_output_protocol != "native"
949            and (
950                protocol_player := self.mass.players.get_player(
951                    self.sync_leader.active_output_protocol
952                )
953            )
954        ):
955            return protocol_player
956        return self.sync_leader
957
958    def _align_members_with_session(self, session_player: Player | None) -> None:
959        """
960        Re-order the tracked members to match the live session's member order.
961
962        Members that are not part of the live session keep their relative order at
963        the end of the list.
964
965        :param session_player: The player that owns the live sync session.
966        """
967        if session_player is None:
968            return
969        # the provider's own group_members, not state.group_members: the latter is
970        # set-derived for non-protocol players and loses the member order
971        session_order = [
972            x
973            for x in self._translate_to_parent_ids(session_player.group_members)
974            if x in self._attr_group_members
975        ]
976        if not session_order:
977            return
978        self._attr_group_members = [
979            *session_order,
980            *[x for x in self._attr_group_members if x not in session_order],
981        ]
982
983    def _protocol_group_leader(self, sync_leader: Player) -> Player:
984        """
985        Return the member that currently holds the protocol-level group.
986
987        This is normally the sync leader itself, but a provider may have promoted a
988        different member at the protocol level. Falls back to the sync leader when no
989        member reports holding others.
990
991        :param sync_leader: The sync leader tracked by this group.
992        """
993        if sync_leader.state.group_members:
994            return sync_leader
995        for member_id in self._attr_group_members:
996            if member_id == sync_leader.player_id:
997                continue
998            member = self.mass.players.get_player(member_id)
999            if member is None or not member.state.available:
1000                continue
1001            # a leader always lists itself alongside its members; only adopt one that
1002            # holds members of this group, never a group formed outside of MA
1003            held = [x for x in member.state.group_members if x != member_id]
1004            if held and not set(held).isdisjoint(self._attr_group_members):
1005                self.logger.warning(
1006                    "Syncgroup %s tracks %s as leader but %s holds the group members",
1007                    self.display_name,
1008                    sync_leader.display_name,
1009                    member.display_name,
1010                )
1011                return member
1012        return sync_leader
1013
1014    def _update_attributes(self) -> None:
1015        """Update dynamic attributes."""
1016        # NOTE on what reads from `.state.*` vs the leader's raw attributes below:
1017        # `__final_current_media` and `__final_active_source` on a player that has
1018        # an ``active_group`` route through the active_group's state — so reading
1019        # ``sync_leader.state.current_media`` from inside the group would loop
1020        # back through our own state derivation (group → leader.state →
1021        # active_group=group → group). For those two we MUST use the leader's
1022        # raw attributes. ``playback_state`` / ``elapsed_time`` do not route via
1023        # active_group and are safe to read from ``.state.*``.
1024        if (sync_leader := self.sync_leader) is None:
1025            # no sync leader, reset playback-related attributes to default values
1026            self._attr_playback_state = PlaybackState.IDLE
1027            self._attr_elapsed_time = None
1028            self._attr_elapsed_time_last_updated = None
1029            self._attr_current_media = None
1030            self._attr_active_source = None
1031            self._attr_poll_interval = 30
1032            return
1033        prev_state = self._attr_playback_state
1034        new_state = sync_leader.state.playback_state
1035        self._attr_playback_state = new_state
1036        self._attr_elapsed_time = sync_leader.state.elapsed_time
1037        self._attr_elapsed_time_last_updated = sync_leader.state.elapsed_time_last_updated
1038        # don't use 'state' for current_media here since that points back to this group
1039        # player when we're active_group, we need the 'raw' value from the sync leader
1040        # itself to avoid circular dependency and ensure it reflects the actual media
1041        # on the leader rather than the group.
1042        self._attr_current_media = sync_leader.current_media
1043        self._attr_active_source = sync_leader.active_source
1044        self._attr_poll_interval = 1 if new_state == PlaybackState.PLAYING else 30
1045        # idle grace handling: schedule a debounced dissolve when the leader
1046        # naturally transitions from PLAYING/PAUSED to IDLE. The dissolve is
1047        # skipped if the user has pinned the group with Fake power control.
1048        if new_state == PlaybackState.IDLE and prev_state in (
1049            PlaybackState.PLAYING,
1050            PlaybackState.PAUSED,
1051        ):
1052            if self._attr_powered is not True:
1053                self._schedule_idle_grace_timer()
1054        elif new_state in (PlaybackState.PLAYING, PlaybackState.PAUSED):
1055            # leader resumed playing, cancel any pending grace
1056            self._cancel_idle_grace_timer()
1057
1058    def _is_player_in_session(self, player: Player, session_player: Player | None) -> bool:
1059        """
1060        Return True if ``player`` is already a sync_client of the live session.
1061
1062        A seamless leader handoff only works when the candidate's resolved
1063        protocol player is already in the active ``AirPlayStreamSession`` (or
1064        equivalent). If not (e.g. a freshly-added player that has never played
1065        anything), we must fall back to dissolve + reform.
1066
1067        :param player: The candidate new leader.
1068        :param session_player: The protocol player that owns the live session
1069            (snapshot taken before the old leader was cleared via
1070            ``_active_session_player()``).
1071        """
1072        if session_player is None:
1073            return False
1074        session = getattr(getattr(session_player, "stream", None), "session", None)
1075        if session is None:
1076            # No session object (e.g. Snapcast, Sendspin) — assume handoff is
1077            # safe if the provider declared support for it.
1078            return True
1079        sync_clients = getattr(session, "sync_clients", None)
1080        if sync_clients is None:
1081            # Session exists but doesn't expose sync_clients — same assumption.
1082            return True
1083        # Resolve player to the protocol player that would own the session
1084        target: Player = player
1085        if (
1086            player.active_output_protocol
1087            and player.active_output_protocol != "native"
1088            and (p := self.mass.players.get_player(player.active_output_protocol))
1089        ):
1090            target = p
1091        return target in sync_clients
1092
1093    async def _dissolve_and_reform(
1094        self,
1095        old_leader_id: str,
1096        leader_to_stop: Player | None = None,
1097        resume_playback: bool = True,
1098        preferred_protocol_domain: str | None = None,
1099    ) -> None:
1100        """
1101        Stop the current sync session, dissolve the syncgroup and schedule a re-form.
1102
1103        Used when a seamless handoff isn't possible (e.g. the new leader is not
1104        part of the live session). The stop/dissolve happens immediately; the
1105        re-form (with resume) is debounced so cascaded unjoins coalesce into a
1106        single restart with the final member list.
1107
1108        :param old_leader_id: The player_id of the departing leader.
1109        :param leader_to_stop: The player to stop before dissolving. Defaults
1110            to ``self.sync_leader`` but callers should pass the old leader
1111            explicitly when ``self.sync_leader`` has already been cleared.
1112        :param resume_playback: If True, schedule the debounced re-form which
1113            restarts playback on the new leader. Pass False when the group was
1114            not actively playing (e.g. paused or idle).
1115        :param preferred_protocol_domain: Optional snapshot of the active
1116            protocol domain taken before the old leader was cleared, passed to
1117            :meth:`_select_sync_leader` so the new leader is chosen to keep the
1118            protocol session continuous where possible.
1119        """
1120        leader_to_stop = leader_to_stop or self.sync_leader
1121        if leader_to_stop:
1122            self.logger.info(
1123                "Dissolving syncgroup %s (leader %s) and re-forming with a new leader",
1124                self.display_name,
1125                leader_to_stop.display_name,
1126            )
1127            async with self.mass.players.wait_for_player_update(
1128                leader_to_stop.player_id, timeout=5
1129            ):
1130                await self.mass.players._handle_cmd_stop(leader_to_stop.player_id)
1131        await self._dissolve_syncgroup()
1132        if old_leader_id in self._attr_group_members:
1133            self._attr_group_members.remove(old_leader_id)
1134        if resume_playback and self._attr_group_members:
1135            self._reform_protocol_domain = preferred_protocol_domain
1136            self._schedule_reform_timer()
1137
1138    async def _wait_member_unsynced(self, member_id: str, timeout: float = 5.0) -> bool:
1139        """
1140        Wait until the given member reports as unsynced (synced_to is None).
1141
1142        Returns ``True`` when the member is verified unsynced (downstream flows
1143        like leader selection / play_media on the leader are safe to proceed),
1144        or ``False`` when the player is genuinely stuck (the caller should
1145        abort rather than issue a play_media that will be rejected by the
1146        provider's "I'm synced to another player" guard).
1147
1148        :param member_id: The player to wait on.
1149        :param timeout: Seconds to wait for the first state propagation.
1150        """
1151        async with self.mass.players.wait_for_player_update(
1152            member_id,
1153            attribute_name="synced_to",
1154            attribute_value=None,
1155            timeout=timeout,
1156        ):
1157            pass
1158        member = self.mass.players.get_player(member_id)
1159        if member is None or member.synced_to is None:
1160            return True
1161        # The provider didn't propagate within the timeout. Kick the member
1162        # from its stale parent, then wait again with a tighter budget.
1163        # This rescues the common "Sonos UPnP event lag" case.
1164        # NOTE: not the public cmd_ungroup - it re-enters this syncgroup's set_members.
1165        # if the stale parent is gone, no kick is possible - fall through to the final check
1166        if stale_parent := self.mass.players.get_player(member.synced_to):
1167            self.logger.warning(
1168                "Player %s still reports synced_to=%s after %ss; "
1169                "removing it from its stale parent and re-waiting",
1170                member.display_name,
1171                member.synced_to,
1172                timeout,
1173            )
1174            try:
1175                async with (
1176                    self.mass.players.wait_for_player_update(
1177                        member_id,
1178                        attribute_name="synced_to",
1179                        attribute_value=None,
1180                        timeout=2.0,
1181                    ),
1182                    self.mass.players.get_player_lock(
1183                        stale_parent.player_id, PlayerLockPurpose.PLAYBACK
1184                    ),
1185                ):
1186                    await self.mass.players._handle_set_members(
1187                        stale_parent, player_ids_to_remove=[member_id]
1188                    )
1189            except asyncio.CancelledError:
1190                raise
1191            except Exception as err:
1192                self.logger.debug(
1193                    "stale-parent removal recovery for %s raised: %s", member.display_name, err
1194                )
1195        member = self.mass.players.get_player(member_id)
1196        if member is None or member.synced_to is None:
1197            return True
1198        self.logger.error(
1199            "Player %s is stuck synced_to=%s; aborting dissolve+reform path",
1200            member.display_name,
1201            member.synced_to,
1202        )
1203        return False
1204
1205    async def _dynamic_leader_switch(self, old_leader_id: str) -> None:
1206        """
1207        Switch the sync leader without tearing down the stream session.
1208
1209        Used when the provider supports dynamic leader selection (e.g. AirPlay,
1210        Snapcast). The old leader is removed from the live session and the
1211        remaining members keep playing uninterrupted on a newly selected leader.
1212
1213        If the selected new leader is not already part of the live session (e.g.
1214        a freshly-added player), a seamless handoff isn't possible. In that case
1215        we fall back to dissolve + reform, accepting a brief audio gap.
1216
1217        :param old_leader_id: The player_id of the leader being removed.
1218        """
1219        old_leader = self.sync_leader
1220        assert old_leader is not None
1221
1222        self.logger.info(
1223            "Dynamic leader switch: removing %s from group %s, remaining members keep playing",
1224            old_leader.display_name,
1225            self.display_name,
1226        )
1227
1228        # Snapshot the currently active protocol and session before clearing
1229        # the leader — we need both for new-leader selection and for the
1230        # handoff-eligibility check.
1231        preferred_domain = self.active_protocol_domain
1232        session_player = self._active_session_player()
1233        # The domain the live session actually runs on. It differs from
1234        # `preferred_domain` once the group is due to downshift to native, and
1235        # a seamless handoff must stay on the protocol carrying the stream.
1236        session_domain = session_player.provider.domain if session_player else None
1237
1238        # Remove the old leader from our group members list
1239        if old_leader_id in self._attr_group_members:
1240            self._attr_group_members.remove(old_leader_id)
1241
1242        # A provider may hand the live session to its own first remaining member, so our
1243        # member order has to match the session's before we pick from it — otherwise we
1244        # end up tracking a different leader than the one that inherits the session.
1245        self._align_members_with_session(session_player)
1246
1247        # Pick a new leader preferring one that supports the currently active
1248        # protocol so the session continuation is seamless.
1249        self.sync_leader = None
1250        new_leader = self._select_sync_leader(preferred_protocol_domain=session_domain)
1251
1252        if not new_leader:
1253            # No remaining members to take over — stop the old leader's
1254            # session and dissolve the group entirely. Restore sync_leader
1255            # so _dissolve_syncgroup can properly ungroup protocol members.
1256            self.sync_leader = old_leader
1257            self.logger.info(
1258                "No remaining members for group %s after removing %s, stopping",
1259                self.display_name,
1260                old_leader.display_name,
1261            )
1262            async with self.mass.players.wait_for_player_update(old_leader.player_id, timeout=5):
1263                await self.mass.players._handle_cmd_stop(old_leader.player_id)
1264            await self._dissolve_syncgroup()
1265            return
1266
1267        # A seamless handoff requires the new leader to already be a
1268        # sync_client of the live session. If it's a freshly-added player
1269        # with no existing stream, fall back to dissolve + reform.
1270        if not self._is_player_in_session(new_leader, session_player):
1271            self.logger.info(
1272                "New leader %s is not in the live session; dissolving and re-forming syncgroup %s",
1273                new_leader.display_name,
1274                self.display_name,
1275            )
1276            # Restore sync_leader so _dissolve_and_reform -> _dissolve_syncgroup
1277            # can properly ungroup protocol-level members. Forward the protocol
1278            # hint so the new form keeps protocol continuity when possible.
1279            self.sync_leader = old_leader
1280            await self._dissolve_and_reform(
1281                old_leader_id,
1282                leader_to_stop=old_leader,
1283                preferred_protocol_domain=preferred_domain,
1284            )
1285            return
1286
1287        self.sync_leader = new_leader
1288        # Ensure the new leader is first in the members list
1289        self._attr_group_members = [
1290            new_leader.player_id,
1291            *[x for x in self._attr_group_members if x != new_leader.player_id],
1292        ]
1293        self.logger.info(
1294            "Dynamic leader switch complete: %s is now leader of group %s",
1295            new_leader.display_name,
1296            self.display_name,
1297        )
1298
1299        # Hand off at the protocol level. We already know:
1300        # - the old session player (the protocol player that owns the live session)
1301        # - the domain that session runs on
1302        # - the new leader (a parent player whose protocol player is in the session)
1303        # So we can talk to the protocol players directly and skip the controller's
1304        # protocol-translation overhead in cmd_set_members.
1305        new_target = self._resolve_session_target(new_leader, session_domain)
1306        remaining_protocol_ids: list[str] = []
1307        for member_id in self._attr_group_members:
1308            if member_id == new_leader.player_id:
1309                continue
1310            if member := self.mass.players.get_player(member_id):
1311                if target := self._resolve_session_target(member, session_domain):
1312                    remaining_protocol_ids.append(target.player_id)
1313
1314        # 1. Old leader's session protocol player steps out of the session.
1315        # Direct call (the controller's cmd_set_members would interpret this
1316        # self-removal as "dissolve the entire group"). The provider's set_members
1317        # implementation handles "remove self while other clients remain" by
1318        # promoting another sync_client at the protocol level.
1319        if session_player is not None:
1320            await session_player.set_members(player_ids_to_remove=[session_player.player_id])
1321
1322        # 2. New leader's protocol player takes over ownership tracking of the
1323        # remaining members. The members are already in the live session at the
1324        # protocol level (sync_clients), this just transfers the bookkeeping so
1325        # the new leader's protocol player reports them as its group members.
1326        if remaining_protocol_ids and new_target is not None:
1327            await new_target.set_members(player_ids_to_add=remaining_protocol_ids)
1328
1329        self.update_state()
1330
1331    def _schedule_idle_grace_timer(self) -> None:
1332        """Schedule a debounced dissolve after the leader becomes idle."""
1333        # any previously scheduled task is replaced so we don't end up with
1334        # two dissolves racing each other when the leader oscillates quickly
1335        self._cancel_idle_grace_timer()
1336        self.logger.debug(
1337            "Scheduling idle-grace dissolve for syncgroup %s in %ss",
1338            self.display_name,
1339            IDLE_GRACE_SECONDS,
1340        )
1341        self._idle_grace_task = self.mass.create_task(self._idle_grace_runner())
1342
1343    def _cancel_idle_grace_timer(self) -> None:
1344        """Cancel any pending idle-grace dissolve task."""
1345        if self._idle_grace_task is not None:
1346            if not self._idle_grace_task.done():
1347                self._idle_grace_task.cancel()
1348            self._idle_grace_task = None
1349
1350    async def _idle_grace_runner(self) -> None:
1351        """Wait the grace window, then dissolve if the group is still idle."""
1352        try:
1353            await asyncio.sleep(IDLE_GRACE_SECONDS)
1354        except asyncio.CancelledError:
1355            return
1356        # re-check state at fire time — playback may have resumed, the user
1357        # may have powered the group on, or another path may have dissolved
1358        # us already. Any of these means we should not dissolve here.
1359        self._idle_grace_task = None
1360        if self.sync_leader is None:
1361            return
1362        if self._attr_powered is True:
1363            return
1364        if self.sync_leader.state.playback_state != PlaybackState.IDLE:
1365            return
1366        self.logger.info(
1367            "Idle-grace expired for syncgroup %s, dissolving",
1368            self.display_name,
1369        )
1370        await self._dissolve_syncgroup()
1371
1372    @property
1373    def _playback_recently_started(self) -> bool:
1374        """Return whether a playback start was issued within the settle window."""
1375        return (time.monotonic() - self._playback_start_at) < PLAYBACK_START_TIMEOUT
1376
1377    def _schedule_reform_timer(self) -> None:
1378        """(Re)schedule the debounced re-form after the sync leader was removed."""
1379        # any previously scheduled task is replaced so cascaded unjoins coalesce
1380        # into a single re-form with the final member list
1381        self._cancel_reform_timer()
1382        self.logger.debug(
1383            "Scheduling debounced re-form for syncgroup %s in %ss",
1384            self.display_name,
1385            REFORM_DEBOUNCE_SECONDS,
1386        )
1387        self._reform_task = self.mass.create_task(self._reform_runner())
1388
1389    def _cancel_reform_timer(self) -> None:
1390        """Cancel any pending debounced re-form task."""
1391        if (task := self._reform_task) is None:
1392            return
1393        self._reform_task = None
1394        # never cancel ourselves: the runner ends up here via play() -> _form_syncgroup
1395        if task is not asyncio.current_task() and not task.done():
1396            task.cancel()
1397
1398    async def _reform_runner(self) -> None:
1399        """Wait the debounce window, then re-form the group and resume playback."""
1400        try:
1401            await asyncio.sleep(REFORM_DEBOUNCE_SECONDS)
1402        except asyncio.CancelledError:
1403            return
1404        try:
1405            # serialize with (un)group and playback commands targeting this group.
1406            # A cancellation (another unjoin re-arming the window, an explicit
1407            # stop) may still land while we wait for the lock.
1408            async with self.mass.players.get_player_lock(
1409                self.player_id, PlayerLockPurpose.PLAYBACK
1410            ):
1411                # re-check state at execution time — an explicit play may have
1412                # re-formed the group already and all members may have been
1413                # removed meanwhile
1414                if self.sync_leader is not None or not self._attr_group_members:
1415                    return
1416                # Wait for the remaining members to report as unsynced before
1417                # re-forming. Providers like Sonos propagate group state
1418                # asynchronously — the children can still report synced_to for
1419                # a few seconds after the leader's ungroup command returns.
1420                members = list(self._attr_group_members)
1421                unsync_results = await asyncio.gather(
1422                    *(self._wait_member_unsynced(m) for m in members),
1423                    return_exceptions=True,
1424                )
1425                stuck_members = [
1426                    members[i] for i, result in enumerate(unsync_results) if result is False
1427                ]
1428                if stuck_members:
1429                    self.logger.error(
1430                        "Members of group %s still report synced_to after recovery attempts: "
1431                        "%s; aborting re-form (no playback will resume on this call)",
1432                        self.display_name,
1433                        stuck_members,
1434                    )
1435                    return
1436                self.logger.info(
1437                    "Re-forming syncgroup %s with %s member(s) and resuming playback",
1438                    self.display_name,
1439                    len(members),
1440                )
1441                # Preselect the new leader with the protocol hint so the form
1442                # picks a member compatible with the previous session's protocol
1443                # (e.g. keep AirPlay if the session was AirPlay).
1444                if self._reform_protocol_domain is not None:
1445                    self.sync_leader = self._select_sync_leader(
1446                        preferred_protocol_domain=self._reform_protocol_domain
1447                    )
1448                await self.play()
1449        finally:
1450            # normal completion detaches us via play() -> _form_syncgroup already;
1451            # this covers the early-return paths so is_active_session settles
1452            if self._reform_task is asyncio.current_task():
1453                self._reform_task = None
1454                self.update_state()
1455
1456    def _resolve_session_target(self, player: Player, domain: str | None) -> Player | None:
1457        """
1458        Resolve the player that participates in the live session for ``domain``.
1459
1460        For a player whose own provider domain matches, returns the player itself.
1461        For a parent player with a linked protocol on that domain, returns the
1462        corresponding protocol player. Returns ``None`` when nothing matches.
1463
1464        :param player: The player to resolve (parent or protocol player).
1465        :param domain: The protocol domain string of the active session
1466            (e.g. "airplay"). May be None, in which case ``player`` is returned.
1467        """
1468        if domain is None:
1469            return player
1470        if player.provider.domain == domain:
1471            return player
1472        for linked in player.linked_output_protocols:
1473            if linked.protocol_domain == domain and linked.available:
1474                return self.mass.players.get_player(linked.output_protocol_id)
1475        return None
1476