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