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 (
429                self.mass.players.get_player_lock(
430                    sync_leader.player_id, PlayerLockPurpose.PLAYBACK
431                ),
432                self._await_leader_playback(),
433            ):
434                await self.mass.players._handle_play_media(sync_leader.player_id, media)
435        else:
436            raise RuntimeError("An empty group cannot play media, consider adding members first")
437
438    async def enqueue_next_media(self, media: PlayerMedia) -> None:
439        """Handle enqueuing of a next media item on the player."""
440        if sync_leader := self.sync_leader:
441            if PlayerFeature.ENQUEUE not in sync_leader.state.supported_features:
442                # this may happen in race conditions where we just switched sync leaders
443                # and the new leader doesn't support enqueueing next media.
444                return
445            # Use internal handler to bypass group redirect logic and avoid infinite loop
446            await self.mass.players._handle_enqueue_next_media(sync_leader.player_id, media)
447
448    async def set_members(  # noqa: PLR0915
449        self,
450        player_ids_to_add: list[str] | None = None,
451        player_ids_to_remove: list[str] | None = None,
452    ) -> None:
453        """Handle SET_MEMBERS command on the player."""
454        if not self.is_dynamic:
455            raise UnsupportedFeaturedException(
456                f"Group {self.display_name} does not allow dynamically adding/removing members!",
457                translation_key="not_dynamic",
458                translation_owner=self.translation_owner,
459                translation_args=[self.display_name],
460            )
461        sync_leader = self.sync_leader or self._select_sync_leader(new_members=player_ids_to_add)
462        # A start that was just issued to the leader may not be reflected in the
463        # (transient) device state yet, so treat the startup window as playing —
464        # otherwise a (un)group command racing an in-flight start misreads the
465        # group as idle and skips the resume. An explicit pause always wins:
466        # PAUSED is deliberate user intent, never startup noise.
467        was_playing = self.playback_state == PlaybackState.PLAYING or (
468            self.playback_state != PlaybackState.PAUSED and self._playback_recently_started
469        )
470
471        # handle additions
472        final_players_to_add: list[str] = []
473        can_group_with = sync_leader.state.can_group_with.copy() if sync_leader else set()
474        for member_id in player_ids_to_add or []:
475            if member_id == self.player_id:
476                continue  # can not add self as member
477            if not self._is_member_allowed(member_id):
478                self.logger.warning(
479                    "Player %s is not allowed to join group %s by the configured player filters, "
480                    "skipping adding it as a member to the group",
481                    member_id,
482                    self.display_name,
483                )
484                continue
485            member = self.mass.players.get_player(member_id)
486            if member is None or not member.available:
487                continue
488            if not sync_leader:
489                # no leader yet (e.g. empty group) - just register the member
490                # the leader and protocol selection happen on the next form/play
491                if member_id not in self._attr_group_members:
492                    self._attr_group_members.append(member_id)
493                continue
494            if member_id != sync_leader.player_id and member_id not in can_group_with:
495                # incompatible with the current leader's protocols - do NOT register
496                # the member or it will linger in _attr_group_members forever without
497                # ever actually being synced.
498                self.logger.debug(
499                    f"Cannot add {member.display_name} to group {self.display_name} since it's "
500                    f"not compatible with the (current) sync leader"
501                )
502                continue
503            if member_id not in self._attr_group_members:
504                self._attr_group_members.append(member_id)
505            if member_id != sync_leader.player_id:
506                final_players_to_add.append(member_id)
507
508        # handle removals
509        final_players_to_remove: list[str] = []
510        leader_removed = False
511        for member_id in player_ids_to_remove or []:
512            if member_id not in self._attr_group_members:
513                # Fallback for a member that was grouped to the sync leader
514                # outside of MA (e.g. via the Sonos app): it isn't part of our
515                # tracked member list but does show up in group_members via the
516                # leader's live state. Forward its removal to the sync leader
517                # instead of silently skipping it.
518                if (
519                    self.sync_leader
520                    and member_id != self.sync_leader.player_id
521                    and member_id in self.sync_leader.state.group_members
522                ):
523                    final_players_to_remove.append(member_id)
524                continue
525            if member_id in self._attr_static_group_members:
526                # static members can not be removed from the group
527                raise PlayerCommandFailed(
528                    f"Cannot remove {member_id} from group {self.display_name} "
529                    "since it's a static member!"
530                )
531            if self.sync_leader and member_id == self.sync_leader.player_id:
532                leader_removed = True
533                continue
534            if member_id == self.player_id:
535                raise PlayerCommandFailed(
536                    f"Cannot remove {self.display_name} from itself as a member!",
537                    translation_key="remove_self",
538                    translation_owner=self.translation_owner,
539                    translation_args=[self.display_name],
540                )
541            self._attr_group_members.remove(member_id)
542            final_players_to_remove.append(member_id)
543
544        if self.sync_leader and leader_removed and self._attr_group_members:
545            # we removed the current sync leader, but we still have members in the group
546            old_leader_id = self.sync_leader.player_id
547            session_player = self._active_session_player()
548            supports_handoff = (
549                session_player is not None
550                and session_player.provider.domain in PROVIDERS_WITH_DYNAMIC_LEADER_SWITCH
551            )
552
553            if was_playing and supports_handoff:
554                # protocol supports dynamic leader switching: try to remove
555                # only the departing leader and keep remaining members playing.
556                # _dynamic_leader_switch will fall back to dissolve+reform
557                # automatically if no remaining member is part of the live
558                # session (e.g. only freshly-added players are left).
559                await self._dynamic_leader_switch(old_leader_id)
560            else:
561                # protocol doesn't support dynamic leader switching or not playing
562                await self._dissolve_and_reform(old_leader_id, resume_playback=was_playing)
563        elif self.sync_leader and (leader_removed or not self._attr_group_members):
564            # we removed the current sync leader, and we have no members left in the group
565            # or we just removed the last member from the group, so we dissolve the syncgroup
566            # Use internal handler to stop the sync leader directly,
567            # bypassing group redirect that would loop back to this player.
568            async with self.mass.players.wait_for_player_update(
569                self.sync_leader.player_id, timeout=5
570            ):
571                await self.mass.players._handle_cmd_stop(self.sync_leader.player_id)
572            await self._dissolve_syncgroup()
573
574        elif self.sync_leader:
575            # just a regular member(s) added/removed action,
576            # we can simply update the syncgroup members on the sync leader.
577            # `active_protocol_domain` is derived from live state, so the
578            # group will naturally downshift once it re-forms if every
579            # remaining member can play on the leader's native protocol.
580            # use _handle_set_members directly to avoid the redirect loop
581            # (cmd_set_members redirects sync-leader targets back to this syncgroup)
582            async with self.mass.players.get_player_lock(
583                self.sync_leader.player_id, PlayerLockPurpose.PLAYBACK
584            ):
585                await self.mass.players._handle_set_members(
586                    self.sync_leader,
587                    player_ids_to_add=final_players_to_add,
588                    player_ids_to_remove=final_players_to_remove,
589                )
590        elif self._reform_task is not None:
591            # leaderless with a debounced re-form pending: membership just changed,
592            # so re-arm the window — the re-form picks up the final member list.
593            self._schedule_reform_timer()
594        # NOTE: If we weren't playing before, we don't need to do anything else,
595        # since the syncing will be done once playback starts
596        self.mass.players.trigger_player_update(self.player_id)
597
598    def on_group_member_updated(
599        self, member_player: Player, changed_values: dict[str, tuple[Any, Any]]
600    ) -> None:
601        """Handle callback when a group member of the group player is updated."""
602        self._update_attributes()
603        super().on_group_member_updated(member_player, changed_values)
604
605    async def on_unload(self) -> None:
606        """Handle logic when the player is unloaded from the Player controller."""
607        self._cancel_idle_grace_timer()
608        self._cancel_reform_timer()
609        await super().on_unload()
610        # the player is going away; make sure we don't leave the protocol-level
611        # sync group standing with a now-nonexistent leader behind it.
612        if self.sync_leader is not None:
613            await self._dissolve_syncgroup()
614
615    @property
616    def active_protocol_domain(self) -> str | None:
617        """
618        Derive the active protocol domain for this sync group on the fly.
619
620        Returns the domain of the protocol currently carrying the live stream
621        session, EXCEPT when every remaining member can also play on the
622        leader's native domain — in which case the group should downshift and
623        this returns that native domain. Always computed from live state so it
624        cannot drift from reality.
625
626        Because of that downshift this is a hint for the next leader selection,
627        not an address for the live session: use ``_active_session_player()``
628        to reach the players that are carrying the stream right now.
629        """
630        session_player = self._active_session_player()
631        if session_player is None or self.sync_leader is None:
632            return None
633        domain = session_player.provider.domain
634        native_domain = self.sync_leader.provider.domain
635        # Keep a non-native protocol "active" for leader-selection purposes unless
636        # the whole group can be reached on the leader's native domain.
637        if domain != native_domain and self._all_members_can_play_on_domain(native_domain):
638            return native_domain
639        return domain
640
641    def _is_member_allowed(self, player_id: str) -> bool:
642        """Return whether a player is allowed to join this group given the configured filter."""
643        # preset members should always be allowed to re-join
644        preset_members = cast("list[str]", self.config.get_value(CONF_GROUP_MEMBERS, []) or [])
645        if player_id in preset_members:
646            return True
647        allowed_members = cast("list[str]", self.config.get_value(CONF_ALLOWED_MEMBERS, []) or [])
648        return not allowed_members or player_id in allowed_members
649
650    async def _form_syncgroup(self) -> None:
651        """Form syncgroup by syncing all (possible) members."""
652        # any in-flight grace or debounced re-form timer is moot now —
653        # we're (re)forming the group
654        self._cancel_idle_grace_timer()
655        self._cancel_reform_timer()
656        self.logger.debug(
657            "Forming syncgroup %s, _attr_group_members=%s, sync_leader=%s",
658            self.display_name,
659            self._attr_group_members,
660            self.sync_leader.display_name if self.sync_leader else None,
661        )
662        # select new sync leader if needed
663        if not self.sync_leader:
664            self.sync_leader = self._select_sync_leader()
665
666        # pin the leader ref: a concurrent command (e.g. a dissolve) may clear
667        # or replace self.sync_leader while we await below
668        leader = self.sync_leader
669        if not leader:
670            # we have no members in the group, so we can't form a syncgroup
671            return
672
673        # ensure the sync leader is first in the list
674        self._attr_group_members = [
675            leader.player_id,
676            *[x for x in self._attr_group_members if x != leader.player_id],
677        ]
678        # If the leader still believes it's synced to a previous leader (e.g. we
679        # just picked a new leader after dissolving the old session and the
680        # protocol-level state hasn't propagated yet), wait for it to settle.
681        # Without this, the subsequent play_media call hits the provider's
682        # "I'm synced to another player" guard and gets rejected.
683        if leader.state.synced_to is not None:
684            self.logger.debug(
685                "Waiting for new leader %s to report synced_to=None before forming",
686                leader.display_name,
687            )
688            if not await self._wait_member_unsynced(leader.player_id):
689                # Leader is genuinely stuck — bail out before issuing play_media
690                # so we don't trigger the provider's "synced to another player"
691                # rejection. The caller (play / play_media) will surface this as
692                # a no-op form; the next user action can retry once the
693                # protocol layer has caught up.
694                self.logger.error(
695                    "Aborting syncgroup form for %s: leader %s is stuck synced",
696                    self.display_name,
697                    leader.display_name,
698                )
699                self.sync_leader = None
700                return
701            if self.sync_leader is not leader:
702                # the group was dissolved or re-led while we waited —
703                # this form attempt is stale, abort
704                return
705        # Translate the leader's group_members (may be protocol IDs) to parent IDs
706        # so we can compare against our _attr_group_members (always parent IDs)
707        already_synced = set(self._translate_to_parent_ids(leader.state.group_members))
708        members_to_sync = [
709            x for x in self._attr_group_members if x != leader.player_id and x not in already_synced
710        ]
711        if members_to_sync:
712            # If the sync leader is playing something independently, stop it first
713            # to prevent protocol switching from trying to resume the previous playback
714            # (we're about to start new playback on the syncgroup).
715            # Wait for the leader to actually reach IDLE before adding members,
716            # since some providers reject set_members while still playing.
717            if leader.state.playback_state == PlaybackState.PLAYING:
718                async with self.mass.players.wait_for_player_update(
719                    leader.player_id,
720                    attribute_name="playback_state",
721                    attribute_value=PlaybackState.IDLE,
722                    timeout=5,
723                ):
724                    await self.mass.players._handle_cmd_stop(leader.player_id)
725                if self.sync_leader is not leader:
726                    # the group was dissolved or re-led while we waited —
727                    # this form attempt is stale, abort
728                    return
729            # use _handle_set_members directly to avoid the redirect loop
730            # (cmd_set_members redirects sync-leader targets back to this syncgroup)
731            async with self.mass.players.get_player_lock(
732                leader.player_id, PlayerLockPurpose.PLAYBACK
733            ):
734                await self.mass.players._handle_set_members(
735                    leader, player_ids_to_add=members_to_sync
736                )
737
738    @asynccontextmanager
739    async def _await_leader_playback(self) -> AsyncIterator[None]:
740        """
741        Wait for the sync leader to confirm playback for the command run in the body.
742
743        Wrap the play/resume call that targets the leader in this context manager.
744        The group's playback lock (held by the caller) then stays acquired until the
745        leader actually reports playing, so a concurrent (un)group command cannot
746        race a start that has not yet taken effect at the device. A no-op when there
747        is no leader to wait on.
748        """
749        if (leader := self.sync_leader) is None:
750            yield
751            return
752        # stamp the start: device state is unreliable while a start settles, so
753        # group-command decisions treat this window as playing (see set_members)
754        self._playback_start_at = time.monotonic()
755        async with self.mass.players.wait_for_player_update(
756            leader.player_id,
757            attribute_name="playback_state",
758            attribute_value=PlaybackState.PLAYING,
759            timeout=PLAYBACK_START_TIMEOUT,
760        ):
761            yield
762
763    async def _dissolve_syncgroup(self) -> None:
764        """Dissolve the current syncgroup by ungrouping all members."""
765        # a dissolve is happening now — any pending grace or re-form timer is no
766        # longer needed (_dissolve_and_reform re-arms the re-form right after)
767        # and the session whose start the marker tracked is gone
768        self._cancel_idle_grace_timer()
769        self._cancel_reform_timer()
770        self._playback_start_at = float("-inf")
771        if sync_leader := self.sync_leader:
772            # dissolve the temporary syncgroup from the player that holds the members:
773            # ungrouping from a leader that no longer holds them is a no-op and would
774            # leave the members grouped and streaming with no way back
775            group_leader = self._protocol_group_leader(sync_leader)
776            sync_children = [
777                x for x in group_leader.state.group_members if x != group_leader.player_id
778            ]
779            if sync_children:
780                # wait for the leader's state to reflect the ungroup
781                # use _handle_set_members directly to avoid the redirect loop
782                # (cmd_set_members redirects sync-leader targets back to this syncgroup)
783                async with (
784                    self.mass.players.wait_for_player_update(group_leader.player_id, timeout=5),
785                    self.mass.players.get_player_lock(
786                        group_leader.player_id, PlayerLockPurpose.PLAYBACK
787                    ),
788                ):
789                    await self.mass.players._handle_set_members(
790                        group_leader, player_ids_to_remove=sync_children
791                    )
792            if group_leader is not sync_leader:
793                # our callers only ever stop the leader we track, so a provider-promoted
794                # one would keep streaming on its own once the members are released
795                await self.mass.players._handle_cmd_stop(group_leader.player_id)
796                self.mass.players.schedule_active_output_protocol_clear(group_leader)
797        # Clear the leader's active protocol once it stops playing; the controller's
798        # clearing in _handle_cmd_stop is skipped for a still-grouped protocol player.
799        if sync_leader:
800            self.mass.players.schedule_active_output_protocol_clear(sync_leader)
801        self.sync_leader = None
802        self._update_attributes()
803        self.update_state()
804
805    def _select_sync_leader(
806        self,
807        new_members: list[str] | None = None,
808        preferred_protocol_domain: str | None = None,
809        preferred_member_ids: Collection[str] | None = None,
810    ) -> Player | None:
811        """
812        Select a (new) sync leader, preferring session and protocol continuity.
813
814        :param new_members: Optional list of newly added member ids to consider
815            when no current/static members are available.
816        :param preferred_protocol_domain: If provided, prefer members that
817            support this protocol domain so playback keeps using the same
818            protocol.
819        :param preferred_member_ids: If provided, prefer members from this
820            collection (e.g. the ones a live session already feeds). Outranks
821            ``preferred_protocol_domain``.
822        """
823        if self.group_members and self.sync_leader and self.sync_leader.state.available:
824            # current leader is still available, no need to select a new one
825            return self.sync_leader
826        # with selecting a new leader, we prioritize the static group members
827        group_members = self.static_group_members or self.group_members or new_members or []
828        candidates = [
829            member_player
830            for member_id in group_members
831            if (member_player := self.mass.players.get_player(member_id))
832            and member_player.state.available
833        ]
834        preferred_ids = set(preferred_member_ids or ())
835        # preference tiers, most specific first: a member that is already fed by the
836        # live session can take it over without restarting playback, and one that
837        # supports the active protocol at least keeps the session on that protocol
838        for reason, matches in (
839            (
840                "takes part in the live session",
841                [x for x in candidates if x.player_id in preferred_ids],
842            ),
843            (
844                f"supports active protocol {preferred_protocol_domain}",
845                [
846                    x
847                    for x in candidates
848                    if preferred_protocol_domain
849                    and self._member_supports_protocol_domain(x, preferred_protocol_domain)
850                ],
851            ),
852            ("first available member", candidates),
853        ):
854            if not matches:
855                continue
856            self.logger.debug(
857                "Auto-selected %s as sync leader for group %s (%s)",
858                matches[0].display_name,
859                self.display_name,
860                reason,
861            )
862            return matches[0]
863        return None
864
865    # -----------------------------------------------------------------------
866    # Protocol awareness
867    # -----------------------------------------------------------------------
868    # A sync group can contain members from multiple protocol domains (e.g. a
869    # Sonos that can play via either its native protocol or via AirPlay). The
870    # group needs to:
871    #   - track which protocol the live session is using (active_protocol_domain)
872    #   - resolve which player actually owns the protocol-level session
873    #     (_active_session_player) so leader/handoff bookkeeping is done on
874    #     the right object
875    #   - choose new leaders that keep protocol continuity
876    #     (_member_supports_protocol_domain / _select_sync_leader)
877    #   - downshift to the native protocol once every remaining member can play
878    #     on it (_all_members_can_play_on_domain)
879    #   - stay aligned with the protocol's own view of the group, both in member
880    #     order (_align_members_with_session) and in who leads it
881    #     (_protocol_group_leader)
882    # The helpers below cover those needs. Composition decisions (which protocol
883    # to use given the current member mix) intentionally live in the group:
884    # individual protocol providers don't have visibility into the rest of the
885    # group's members.
886
887    def _translate_to_parent_ids(self, player_ids: list[str]) -> list[str]:
888        """
889        Translate a list of (possibly protocol) player IDs to parent player IDs.
890
891        Protocol players (e.g. AirPlay `apc...`) are translated to their parent
892        (e.g. Sonos `RINCON_...`). Non-protocol IDs pass through unchanged.
893
894        :param player_ids: List of player IDs that may be protocol or parent IDs.
895        """
896        result: list[str] = []
897        for pid in player_ids:
898            if player := self.mass.players.get_player(pid):
899                parent_id = player.protocol_parent_id or pid
900                if parent_id not in result:
901                    result.append(parent_id)
902            elif pid not in result:
903                result.append(pid)
904        return result
905
906    def _member_supports_protocol_domain(self, player: Player, domain: str) -> bool:
907        """
908        Check if a player can be reached on the given protocol domain right now.
909
910        :param player: The player to check.
911        :param domain: The protocol domain string (e.g. "airplay", "sonos").
912        """
913        return domain in player.playback_domains
914
915    def _all_members_can_play_on_domain(self, domain: str) -> bool:
916        """
917        Return True if every current member has a playback path on the given domain.
918
919        Members that are unavailable or expose no playback path at all are
920        ignored, so they never hold the group on a protocol.
921
922        :param domain: The playback path domain to check (e.g. "airplay", "sonos").
923        """
924        for member_id in self._attr_group_members:
925            member = self.mass.players.get_player(member_id)
926            if member is None or not member.state.available:
927                continue
928            paths = member.playback_domains
929            if paths and domain not in paths:
930                return False
931        return True
932
933    def _active_session_player(self) -> Player | None:
934        """
935        Return the player that owns the live sync session.
936
937        If the current sync leader has a non-native active output protocol,
938        returns the protocol player that carries the stream; otherwise returns
939        the native sync leader itself. Returns ``None`` if there is no leader.
940        """
941        if not self.sync_leader:
942            return None
943        if (
944            self.sync_leader.active_output_protocol
945            and self.sync_leader.active_output_protocol != "native"
946            and (
947                protocol_player := self.mass.players.get_player(
948                    self.sync_leader.active_output_protocol
949                )
950            )
951        ):
952            return protocol_player
953        return self.sync_leader
954
955    def _align_members_with_session(self, session_player: Player | None) -> None:
956        """
957        Re-order the tracked members to match the live session's member order.
958
959        Members that are not part of the live session keep their relative order at
960        the end of the list.
961
962        :param session_player: The player that owns the live sync session.
963        """
964        if session_player is None:
965            return
966        # the provider's own group_members, not state.group_members: the latter is
967        # set-derived for non-protocol players and loses the member order. Not
968        # live_session_members either: that answers who is in the session, not in
969        # which order, and a provider may derive it without preserving any.
970        session_order = [
971            x
972            for x in self._translate_to_parent_ids(session_player.group_members)
973            if x in self._attr_group_members
974        ]
975        if not session_order:
976            return
977        self._attr_group_members = [
978            *session_order,
979            *[x for x in self._attr_group_members if x not in session_order],
980        ]
981
982    def _protocol_group_leader(self, sync_leader: Player) -> Player:
983        """
984        Return the member that currently holds the protocol-level group.
985
986        This is normally the sync leader itself, but a provider may have promoted a
987        different member at the protocol level. Falls back to the sync leader when no
988        member reports holding others.
989
990        :param sync_leader: The sync leader tracked by this group.
991        """
992        if sync_leader.state.group_members:
993            return sync_leader
994        for member_id in self._attr_group_members:
995            if member_id == sync_leader.player_id:
996                continue
997            member = self.mass.players.get_player(member_id)
998            if member is None or not member.state.available:
999                continue
1000            # a leader always lists itself alongside its members; only adopt one that
1001            # holds members of this group, never a group formed outside of MA
1002            held = [x for x in member.state.group_members if x != member_id]
1003            if held and not set(held).isdisjoint(self._attr_group_members):
1004                self.logger.warning(
1005                    "Syncgroup %s tracks %s as leader but %s holds the group members",
1006                    self.display_name,
1007                    sync_leader.display_name,
1008                    member.display_name,
1009                )
1010                return member
1011        return sync_leader
1012
1013    def _update_attributes(self) -> None:
1014        """Update dynamic attributes."""
1015        # NOTE on what reads from `.state.*` vs the leader's raw attributes below:
1016        # `__final_current_media` and `__final_active_source` on a player that has
1017        # an ``active_group`` route through the active_group's state — so reading
1018        # ``sync_leader.state.current_media`` from inside the group would loop
1019        # back through our own state derivation (group → leader.state →
1020        # active_group=group → group). For those two we MUST use the leader's
1021        # raw attributes. ``playback_state`` / ``elapsed_time`` do not route via
1022        # active_group and are safe to read from ``.state.*``.
1023        if (sync_leader := self.sync_leader) is None:
1024            # no sync leader, reset playback-related attributes to default values
1025            self._attr_playback_state = PlaybackState.IDLE
1026            self._attr_elapsed_time = None
1027            self._attr_elapsed_time_last_updated = None
1028            self._attr_current_media = None
1029            self._attr_active_source = None
1030            self._attr_poll_interval = 30
1031            return
1032        prev_state = self._attr_playback_state
1033        new_state = sync_leader.state.playback_state
1034        self._attr_playback_state = new_state
1035        self._attr_elapsed_time = sync_leader.state.elapsed_time
1036        self._attr_elapsed_time_last_updated = sync_leader.state.elapsed_time_last_updated
1037        # don't use 'state' for current_media here since that points back to this group
1038        # player when we're active_group, we need the 'raw' value from the sync leader
1039        # itself to avoid circular dependency and ensure it reflects the actual media
1040        # on the leader rather than the group.
1041        self._attr_current_media = sync_leader.current_media
1042        self._attr_active_source = sync_leader.active_source
1043        self._attr_poll_interval = 1 if new_state == PlaybackState.PLAYING else 30
1044        # idle grace handling: schedule a debounced dissolve when the leader
1045        # naturally transitions from PLAYING/PAUSED to IDLE. The dissolve is
1046        # skipped if the user has pinned the group with Fake power control.
1047        if new_state == PlaybackState.IDLE and prev_state in (
1048            PlaybackState.PLAYING,
1049            PlaybackState.PAUSED,
1050        ):
1051            if self._attr_powered is not True:
1052                self._schedule_idle_grace_timer()
1053        elif new_state in (PlaybackState.PLAYING, PlaybackState.PAUSED):
1054            # leader resumed playing, cancel any pending grace
1055            self._cancel_idle_grace_timer()
1056
1057    async def _dissolve_and_reform(
1058        self,
1059        old_leader_id: str,
1060        leader_to_stop: Player | None = None,
1061        resume_playback: bool = True,
1062        preferred_protocol_domain: str | None = None,
1063    ) -> None:
1064        """
1065        Stop the current sync session, dissolve the syncgroup and schedule a re-form.
1066
1067        Used when a seamless handoff isn't possible (e.g. the new leader is not
1068        part of the live session). The stop/dissolve happens immediately; the
1069        re-form (with resume) is debounced so cascaded unjoins coalesce into a
1070        single restart with the final member list.
1071
1072        :param old_leader_id: The player_id of the departing leader.
1073        :param leader_to_stop: The player to stop before dissolving. Defaults
1074            to ``self.sync_leader`` but callers should pass the old leader
1075            explicitly when ``self.sync_leader`` has already been cleared.
1076        :param resume_playback: If True, schedule the debounced re-form which
1077            restarts playback on the new leader. Pass False when the group was
1078            not actively playing (e.g. paused or idle).
1079        :param preferred_protocol_domain: Optional snapshot of the active
1080            protocol domain taken before the old leader was cleared, passed to
1081            :meth:`_select_sync_leader` so the new leader is chosen to keep the
1082            protocol session continuous where possible.
1083        """
1084        leader_to_stop = leader_to_stop or self.sync_leader
1085        if leader_to_stop:
1086            self.logger.info(
1087                "Dissolving syncgroup %s (leader %s) and re-forming with a new leader",
1088                self.display_name,
1089                leader_to_stop.display_name,
1090            )
1091            async with self.mass.players.wait_for_player_update(
1092                leader_to_stop.player_id, timeout=5
1093            ):
1094                await self.mass.players._handle_cmd_stop(leader_to_stop.player_id)
1095        await self._dissolve_syncgroup()
1096        if old_leader_id in self._attr_group_members:
1097            self._attr_group_members.remove(old_leader_id)
1098        if resume_playback and self._attr_group_members:
1099            self._reform_protocol_domain = preferred_protocol_domain
1100            self._schedule_reform_timer()
1101
1102    async def _wait_member_unsynced(self, member_id: str, timeout: float = 5.0) -> bool:
1103        """
1104        Wait until the given member reports as unsynced (synced_to is None).
1105
1106        Returns ``True`` when the member is verified unsynced (downstream flows
1107        like leader selection / play_media on the leader are safe to proceed),
1108        or ``False`` when the player is genuinely stuck (the caller should
1109        abort rather than issue a play_media that will be rejected by the
1110        provider's "I'm synced to another player" guard).
1111
1112        :param member_id: The player to wait on.
1113        :param timeout: Seconds to wait for the first state propagation.
1114        """
1115        async with self.mass.players.wait_for_player_update(
1116            member_id,
1117            attribute_name="synced_to",
1118            attribute_value=None,
1119            timeout=timeout,
1120        ):
1121            pass
1122        member = self.mass.players.get_player(member_id)
1123        if member is None or member.synced_to is None:
1124            return True
1125        # The provider didn't propagate within the timeout. Kick the member
1126        # from its stale parent, then wait again with a tighter budget.
1127        # This rescues the common "Sonos UPnP event lag" case.
1128        # NOTE: not the public cmd_ungroup - it re-enters this syncgroup's set_members.
1129        # if the stale parent is gone, no kick is possible - fall through to the final check
1130        if stale_parent := self.mass.players.get_player(member.synced_to):
1131            self.logger.warning(
1132                "Player %s still reports synced_to=%s after %ss; "
1133                "removing it from its stale parent and re-waiting",
1134                member.display_name,
1135                member.synced_to,
1136                timeout,
1137            )
1138            try:
1139                async with (
1140                    self.mass.players.wait_for_player_update(
1141                        member_id,
1142                        attribute_name="synced_to",
1143                        attribute_value=None,
1144                        timeout=2.0,
1145                    ),
1146                    self.mass.players.get_player_lock(
1147                        stale_parent.player_id, PlayerLockPurpose.PLAYBACK
1148                    ),
1149                ):
1150                    await self.mass.players._handle_set_members(
1151                        stale_parent, player_ids_to_remove=[member_id]
1152                    )
1153            except asyncio.CancelledError:
1154                raise
1155            except Exception as err:
1156                self.logger.debug(
1157                    "stale-parent removal recovery for %s raised: %s", member.display_name, err
1158                )
1159        member = self.mass.players.get_player(member_id)
1160        if member is None or member.synced_to is None:
1161            return True
1162        self.logger.error(
1163            "Player %s is stuck synced_to=%s; aborting dissolve+reform path",
1164            member.display_name,
1165            member.synced_to,
1166        )
1167        return False
1168
1169    async def _dynamic_leader_switch(self, old_leader_id: str) -> None:
1170        """
1171        Switch the sync leader without tearing down the stream session.
1172
1173        Used when the provider supports dynamic leader selection (e.g. AirPlay,
1174        Snapcast). The old leader is removed from the live session and the
1175        remaining members keep playing uninterrupted on a newly selected leader.
1176
1177        If no remaining member takes part in the live session (e.g. only
1178        freshly-added players are left), a seamless handoff isn't possible. In
1179        that case we fall back to dissolve + reform, accepting a brief audio gap.
1180
1181        :param old_leader_id: The player_id of the leader being removed.
1182        """
1183        old_leader = self.sync_leader
1184        assert old_leader is not None
1185
1186        self.logger.info(
1187            "Dynamic leader switch: removing %s from group %s, remaining members keep playing",
1188            old_leader.display_name,
1189            self.display_name,
1190        )
1191
1192        # Snapshot the currently active protocol and session before clearing
1193        # the leader — we need both for new-leader selection and for the
1194        # handoff-eligibility check.
1195        preferred_domain = self.active_protocol_domain
1196        session_player = self._active_session_player()
1197        # The domain the live session actually runs on. It differs from
1198        # `preferred_domain` once the group is due to downshift to native, and
1199        # a seamless handoff must stay on the protocol carrying the stream.
1200        session_domain = session_player.provider.domain if session_player else None
1201        # The members the session feeds right now: only a member from this set can take
1202        # it over without a restart. Tracked membership is not enough — a member can be
1203        # dropped from the session (or never make it in) while still being listed.
1204        live_member_ids = (
1205            self._translate_to_parent_ids(session_player.live_session_members)
1206            if session_player
1207            else []
1208        )
1209
1210        # Remove the old leader from our group members list
1211        if old_leader_id in self._attr_group_members:
1212            self._attr_group_members.remove(old_leader_id)
1213
1214        # A provider may hand the live session to its own first remaining member, so our
1215        # member order has to match the session's before we pick from it — otherwise we
1216        # end up tracking a different leader than the one that inherits the session.
1217        self._align_members_with_session(session_player)
1218
1219        # Pick a new leader, preferring one that is already fed by the live session
1220        # so the session continuation is seamless.
1221        self.sync_leader = None
1222        new_leader = self._select_sync_leader(
1223            preferred_protocol_domain=session_domain,
1224            preferred_member_ids=live_member_ids,
1225        )
1226
1227        if not new_leader:
1228            # No remaining members to take over — stop the old leader's
1229            # session and dissolve the group entirely. Restore sync_leader
1230            # so _dissolve_syncgroup can properly ungroup protocol members.
1231            self.sync_leader = old_leader
1232            self.logger.info(
1233                "No remaining members for group %s after removing %s, stopping",
1234                self.display_name,
1235                old_leader.display_name,
1236            )
1237            async with self.mass.players.wait_for_player_update(old_leader.player_id, timeout=5):
1238                await self.mass.players._handle_cmd_stop(old_leader.player_id)
1239            await self._dissolve_syncgroup()
1240            return
1241
1242        # A seamless handoff requires the new leader to already be a sync_client of
1243        # the live session. Selection prefers such a member, so reaching this means
1244        # no remaining member has a stream to inherit: fall back to dissolve + reform.
1245        if new_leader.player_id not in live_member_ids:
1246            self.logger.info(
1247                "New leader %s is not in the live session; dissolving and re-forming syncgroup %s",
1248                new_leader.display_name,
1249                self.display_name,
1250            )
1251            # Restore sync_leader so _dissolve_and_reform -> _dissolve_syncgroup
1252            # can properly ungroup protocol-level members. Forward the protocol
1253            # hint so the new form keeps protocol continuity when possible.
1254            self.sync_leader = old_leader
1255            await self._dissolve_and_reform(
1256                old_leader_id,
1257                leader_to_stop=old_leader,
1258                preferred_protocol_domain=preferred_domain,
1259            )
1260            return
1261
1262        self.sync_leader = new_leader
1263        # Ensure the new leader is first in the members list
1264        self._attr_group_members = [
1265            new_leader.player_id,
1266            *[x for x in self._attr_group_members if x != new_leader.player_id],
1267        ]
1268        self.logger.info(
1269            "Dynamic leader switch complete: %s is now leader of group %s",
1270            new_leader.display_name,
1271            self.display_name,
1272        )
1273
1274        # Hand off at the protocol level. We already know:
1275        # - the old session player (the protocol player that owns the live session)
1276        # - the domain that session runs on
1277        # - the new leader (a parent player whose protocol player is in the session)
1278        # So we can talk to the protocol players directly and skip the controller's
1279        # protocol-translation overhead in cmd_set_members.
1280        new_target = self._resolve_session_target(new_leader, session_domain)
1281        remaining_protocol_ids: list[str] = []
1282        for member_id in self._attr_group_members:
1283            if member_id == new_leader.player_id:
1284                continue
1285            if member := self.mass.players.get_player(member_id):
1286                if target := self._resolve_session_target(member, session_domain):
1287                    remaining_protocol_ids.append(target.player_id)
1288
1289        # 1. Old leader's session protocol player steps out of the session.
1290        # Direct call (the controller's cmd_set_members would interpret this
1291        # self-removal as "dissolve the entire group"). The provider's set_members
1292        # keeps the live session running for the members that stay behind and releases
1293        # them, so they are briefly without a leader until step 2 picks them back up.
1294        if session_player is not None:
1295            await session_player.set_members(player_ids_to_remove=[session_player.player_id])
1296
1297        # 2. New leader's protocol player takes over ownership tracking of the
1298        # remaining members. The members are already in the live session at the
1299        # protocol level (sync_clients), this just transfers the bookkeeping so
1300        # the new leader's protocol player reports them as its group members.
1301        if remaining_protocol_ids and new_target is not None:
1302            await new_target.set_members(player_ids_to_add=remaining_protocol_ids)
1303
1304        self.update_state()
1305
1306    def _schedule_idle_grace_timer(self) -> None:
1307        """Schedule a debounced dissolve after the leader becomes idle."""
1308        # any previously scheduled task is replaced so we don't end up with
1309        # two dissolves racing each other when the leader oscillates quickly
1310        self._cancel_idle_grace_timer()
1311        self.logger.debug(
1312            "Scheduling idle-grace dissolve for syncgroup %s in %ss",
1313            self.display_name,
1314            IDLE_GRACE_SECONDS,
1315        )
1316        self._idle_grace_task = self.mass.create_task(self._idle_grace_runner())
1317
1318    def _cancel_idle_grace_timer(self) -> None:
1319        """Cancel any pending idle-grace dissolve task."""
1320        if self._idle_grace_task is not None:
1321            if not self._idle_grace_task.done():
1322                self._idle_grace_task.cancel()
1323            self._idle_grace_task = None
1324
1325    async def _idle_grace_runner(self) -> None:
1326        """Wait the grace window, then dissolve if the group is still idle."""
1327        try:
1328            await asyncio.sleep(IDLE_GRACE_SECONDS)
1329        except asyncio.CancelledError:
1330            return
1331        # re-check state at fire time — playback may have resumed, the user
1332        # may have powered the group on, or another path may have dissolved
1333        # us already. Any of these means we should not dissolve here.
1334        self._idle_grace_task = None
1335        if self.sync_leader is None:
1336            return
1337        if self._attr_powered is True:
1338            return
1339        if self.sync_leader.state.playback_state != PlaybackState.IDLE:
1340            return
1341        self.logger.info(
1342            "Idle-grace expired for syncgroup %s, dissolving",
1343            self.display_name,
1344        )
1345        await self._dissolve_syncgroup()
1346
1347    @property
1348    def _playback_recently_started(self) -> bool:
1349        """Return whether a playback start was issued within the settle window."""
1350        return (time.monotonic() - self._playback_start_at) < PLAYBACK_START_TIMEOUT
1351
1352    def _schedule_reform_timer(self) -> None:
1353        """(Re)schedule the debounced re-form after the sync leader was removed."""
1354        # any previously scheduled task is replaced so cascaded unjoins coalesce
1355        # into a single re-form with the final member list
1356        self._cancel_reform_timer()
1357        self.logger.debug(
1358            "Scheduling debounced re-form for syncgroup %s in %ss",
1359            self.display_name,
1360            REFORM_DEBOUNCE_SECONDS,
1361        )
1362        self._reform_task = self.mass.create_task(self._reform_runner())
1363
1364    def _cancel_reform_timer(self) -> None:
1365        """Cancel any pending debounced re-form task."""
1366        if (task := self._reform_task) is None:
1367            return
1368        self._reform_task = None
1369        # never cancel ourselves: the runner ends up here via play() -> _form_syncgroup
1370        if task is not asyncio.current_task() and not task.done():
1371            task.cancel()
1372
1373    async def _reform_runner(self) -> None:
1374        """Wait the debounce window, then re-form the group and resume playback."""
1375        try:
1376            await asyncio.sleep(REFORM_DEBOUNCE_SECONDS)
1377        except asyncio.CancelledError:
1378            return
1379        try:
1380            # serialize with (un)group and playback commands targeting this group.
1381            # A cancellation (another unjoin re-arming the window, an explicit
1382            # stop) may still land while we wait for the lock.
1383            async with self.mass.players.get_player_lock(
1384                self.player_id, PlayerLockPurpose.PLAYBACK
1385            ):
1386                # re-check state at execution time — an explicit play may have
1387                # re-formed the group already and all members may have been
1388                # removed meanwhile
1389                if self.sync_leader is not None or not self._attr_group_members:
1390                    return
1391                # Wait for the remaining members to report as unsynced before
1392                # re-forming. Providers like Sonos propagate group state
1393                # asynchronously — the children can still report synced_to for
1394                # a few seconds after the leader's ungroup command returns.
1395                members = list(self._attr_group_members)
1396                unsync_results = await asyncio.gather(
1397                    *(self._wait_member_unsynced(m) for m in members),
1398                    return_exceptions=True,
1399                )
1400                stuck_members = [
1401                    members[i] for i, result in enumerate(unsync_results) if result is False
1402                ]
1403                if stuck_members:
1404                    self.logger.error(
1405                        "Members of group %s still report synced_to after recovery attempts: "
1406                        "%s; aborting re-form (no playback will resume on this call)",
1407                        self.display_name,
1408                        stuck_members,
1409                    )
1410                    return
1411                self.logger.info(
1412                    "Re-forming syncgroup %s with %s member(s) and resuming playback",
1413                    self.display_name,
1414                    len(members),
1415                )
1416                # Preselect the new leader with the protocol hint so the form
1417                # picks a member compatible with the previous session's protocol
1418                # (e.g. keep AirPlay if the session was AirPlay).
1419                if self._reform_protocol_domain is not None:
1420                    self.sync_leader = self._select_sync_leader(
1421                        preferred_protocol_domain=self._reform_protocol_domain
1422                    )
1423                await self.play()
1424        finally:
1425            # normal completion detaches us via play() -> _form_syncgroup already;
1426            # this covers the early-return paths so is_active_session settles
1427            if self._reform_task is asyncio.current_task():
1428                self._reform_task = None
1429                self.update_state()
1430
1431    def _resolve_session_target(self, player: Player, domain: str | None) -> Player | None:
1432        """
1433        Resolve the player that participates in the live session for ``domain``.
1434
1435        For a player whose own provider domain matches, returns the player itself.
1436        For a parent player with a linked protocol on that domain, returns the
1437        corresponding protocol player. Returns ``None`` when nothing matches.
1438
1439        :param player: The player to resolve (parent or protocol player).
1440        :param domain: The protocol domain string of the active session
1441            (e.g. "airplay"). May be None, in which case ``player`` is returned.
1442        """
1443        if domain is None:
1444            return player
1445        if player.provider.domain == domain:
1446            return player
1447        for linked in player.linked_output_protocols:
1448            if linked.protocol_domain == domain:
1449                return self.mass.players.get_player(linked.output_protocol_id)
1450        return None
1451