music-assistant-server

1"""
2MusicAssistant Player Queues Controller.
3
4Handles all logic to PLAY Media Items, provided by Music Providers to supported players.
5
6It is loosely coupled to the MusicAssistant Music Controller and Player Controller.
7A Music Assistant Player always has a PlayerQueue associated with it
8which holds the queue items and state.
9
10The PlayerQueue is in that case the active source of the player,
11but it can also be something else, hence the loose coupling.
12"""
13
14from __future__ import annotations
15
16import asyncio
17import random
18import time
19from typing import TYPE_CHECKING, Any, Final, cast
20
21import shortuuid
22from music_assistant_models.auth import Scope
23from music_assistant_models.enums import (
24    EventType,
25    MediaType,
26    PlaybackState,
27    PlayerType,
28    QueueOption,
29    RepeatMode,
30)
31from music_assistant_models.errors import (
32    AudioError,
33    InsufficientPermissions,
34    InvalidCommand,
35    InvalidDataError,
36    MediaNotFoundError,
37    MusicAssistantError,
38    PlayerUnavailableError,
39    QueueEmpty,
40)
41from music_assistant_models.media_items import (
42    Audiobook,
43    ItemMapping,
44    MediaItemType,
45    PlayableMediaItemType,
46    Playlist,
47    PodcastEpisode,
48    SoundEffect,
49    Track,
50)
51from music_assistant_models.player_queue import PlayerQueue
52
53from music_assistant.constants import (
54    ATTR_ANNOUNCEMENT_IN_PROGRESS,
55    MASS_LOGO_ONLINE,
56    PLAYLIST_MEDIA_TYPES,
57)
58from music_assistant.controllers.player_queues.autoplay import Autoplay
59from music_assistant.controllers.player_queues.config import (
60    core_config_entries,
61    queue_config_entries,
62)
63from music_assistant.controllers.player_queues.constants import (
64    CACHE_CATEGORY_PLAYER_QUEUE_ITEMS,
65    CACHE_CATEGORY_PLAYER_QUEUE_STATE,
66    PLAYBACK_START_TIMEOUT,
67    QUEUE_CACHE_SAVE_DELAY,
68)
69from music_assistant.controllers.player_queues.helpers import (
70    get_current_playback_speed,
71    handle_play_action,
72    is_dynamic_source,
73)
74from music_assistant.controllers.player_queues.managed_pool import ManagedPool
75from music_assistant.controllers.player_queues.media_resolver import MediaResolver
76from music_assistant.controllers.player_queues.playback_tracker import PlaybackTrackerMixin
77from music_assistant.controllers.player_queues.queue_loader import QueueLoaderMixin
78from music_assistant.controllers.player_queues.smart_shuffle import SmartShuffle
79from music_assistant.controllers.player_queues.state import PlayerQueueData
80from music_assistant.controllers.player_queues.stream_feeder import StreamFeederMixin
81from music_assistant.controllers.webserver.helpers.auth_middleware import get_current_user
82from music_assistant.helpers.api import api_command
83from music_assistant.helpers.uri import parse_uri
84from music_assistant.models.music_provider import ProviderStreamLimitError
85from music_assistant.models.player import Player, PlayerMedia
86
87if TYPE_CHECKING:
88    from collections.abc import Iterator
89
90    from music_assistant_models import BackgroundTask
91    from music_assistant_models.config_entries import (
92        ConfigEntry,
93        ConfigValueOption,
94        CoreConfig,
95    )
96    from music_assistant_models.queue_item import QueueItem
97
98    from music_assistant import MusicAssistant
99    from music_assistant.constants import PlaylistPlayableItem
100    from music_assistant.controllers.music.recency import RecencyWindows
101    from music_assistant.helpers.json import SerializableType
102    from music_assistant.models.player import Player
103
104
105# the container media types worth surfacing as a queue "source" for clients to display. Individual
106# items (single tracks, radio streams, podcast episodes, live audio sources, ...) carry no grouping
107# and only clutter the "playing from" representation, so they are omitted from the wire `sources`.
108_WIRE_SOURCE_MEDIA_TYPES: Final = frozenset(
109    {
110        MediaType.ARTIST,
111        MediaType.ALBUM,
112        MediaType.PLAYLIST,
113        MediaType.PODCAST,
114        MediaType.AUDIOBOOK,
115    }
116)
117
118
119async def _is_audio_source(item: MediaItemType | ItemMapping | str) -> bool:
120    """
121    Return whether the given media names a live audio source.
122
123    :param item: One entry of a play request.
124    """
125    if not isinstance(item, str):
126        return item.media_type == MediaType.AUDIO_SOURCE
127    try:
128        media_type, _, _ = await parse_uri(item)
129    except MusicAssistantError:
130        return False
131    return media_type == MediaType.AUDIO_SOURCE
132
133
134async def _resolve_audio_source_request(
135    media: MediaItemType | ItemMapping | str | list[MediaItemType | ItemMapping | str],
136) -> str | None:
137    """
138    Return the uri of the live audio source a play request names, if it names one.
139
140    A live source is selected on a player rather than queued, so it cannot be lined
141    up behind or alongside other media: naming one among others is a caller error
142    rather than a request to interpret.
143
144    :param media: The media a play request was given.
145    :raises InvalidCommand: When a live source is named among other media.
146    """
147    items = media if isinstance(media, list) else [media]
148    sources = [item for item in items if await _is_audio_source(item)]
149    if not sources:
150        return None
151    if len(items) > 1:
152        raise InvalidCommand(
153            "A live audio source plays on its own: it can not be combined with other media"
154        )
155    item = sources[0]
156    return item if isinstance(item, str) else item.uri
157
158
159class PlayerQueuesController(QueueLoaderMixin, PlaybackTrackerMixin, StreamFeederMixin):
160    """
161    Controller holding all logic to enqueue music for players.
162
163    The loading, playback-tracking and stream-feeding logic lives in mixins (over the shared base);
164    this class owns the public API surface, the per-queue records and the stateful helper services.
165    """
166
167    def __init__(self, mass: MusicAssistant) -> None:
168        """Initialize core controller."""
169        super().__init__(mass)
170        # server-side per-queue records, keyed by queue_id; each bundles the wire PlayerQueue with
171        # its items, dynamic-source items and runtime-only state (see PlayerQueueData)
172        self._queue_data: dict[str, PlayerQueueData] = {}
173        # stateful helper services (own per-queue state + lifecycle), constructed with self
174        self._autoplay = Autoplay(self)
175        self._smart_shuffle = SmartShuffle(self)
176        self._managed_pool = ManagedPool(self)
177        self._media_resolver = MediaResolver(self)
178        self.manifest.name = "Player Queues controller"
179        self.manifest.description = (
180            "Music Assistant's core controller which manages the queues for all players."
181        )
182        self.manifest.icon = "playlist-music"
183
184    async def close(self) -> None:
185        """Cleanup on exit."""
186        # stop all playback
187        for queue in self.all():
188            if queue.state in (PlaybackState.PLAYING, PlaybackState.PAUSED):
189                await self.stop(queue.queue_id)
190        # flush any pending (debounced) state writes so the latest queue survives shutdown/update
191        for queue in self.all():
192            self.mass.cancel_timer(f"save_queue_cache_{queue.queue_id}")
193            await self._save_queue_to_cache(queue.queue_id)
194
195    async def get_diagnostics(self) -> dict[str, SerializableType]:
196        """Return diagnostics info for this controller to include in diagnostics reports."""
197        queues = [queue_data.queue for queue_data in self._queue_data.values()]
198        by_state: dict[str, int] = {}
199        for queue in queues:
200            by_state[queue.state.value] = by_state.get(queue.state.value, 0) + 1
201        return {
202            "total": len(queues),
203            "active": sum(queue.active for queue in queues),
204            "by_state": by_state,
205            "flow_mode_active": sum(queue.flow_mode for queue in queues),
206            "dynamic_mode_active": sum(queue.is_dynamic for queue in queues),
207            "total_items": sum(queue.items for queue in queues),
208        }
209
210    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
211        """Return the core-module (global) config entries: the queue-controller defaults."""
212        # kept cheap (no library lookup): the config controller populates the global autoplay
213        # playlist dropdown for the UI, so this stays fast on the config value/parse path
214        return core_config_entries(self.mass)
215
216    async def update_config(self, config: CoreConfig, changed_keys: set[str]) -> None:
217        """Apply a global queue-settings change: refresh derived per-queue state and notify clients."""
218        await super().update_config(config, changed_keys)
219        if not any(key.startswith("values/") for key in changed_keys):
220            return
221        # queues that follow a changed global value may flip their derived indicators, so refresh
222        # and signal them (mirrors what save_player_queue_config does for a single queue)
223        for queue in self.all():
224            queue.smart_fades_active = self.mass.streams.is_smart_fades_active(queue)
225            queue.smart_shuffle_active = self.is_smart_shuffle_active(queue)
226            self.signal_update(queue.queue_id)
227
228    def get_queue_config_entries(
229        self, playlist_options: list[ConfigValueOption] | None = None
230    ) -> list[ConfigEntry]:
231        """
232        Return the per-queue config entries.
233
234        The autoplay_mode select disables the 'similar' option when no provider can supply
235        similar tracks. The crossfade_mode select's options and default depend on whether smart
236        fades are available: the smart option is disabled (shown but not selectable) and the
237        default falls back to standard crossfade when smart fades can't be used on this server.
238
239        :param playlist_options: Library playlists to offer for the 'playlist' autoplay mode.
240            Only populated when serving the entries to the UI; the parse path can omit it.
241        """
242        return queue_config_entries(self.mass, playlist_options)
243
244    def __iter__(self) -> Iterator[PlayerQueue]:
245        """Iterate over (available) players."""
246        return iter(queue_data.queue for queue_data in self._queue_data.values())
247
248    @api_command("player_queues/all", required_scope=Scope.QUEUES_READ)
249    def all(self) -> tuple[PlayerQueue, ...]:
250        """Return all registered PlayerQueues."""
251        return tuple(queue_data.queue for queue_data in self._queue_data.values())
252
253    @api_command("player_queues/get", required_scope=Scope.QUEUES_READ)
254    def get(self, queue_id: str) -> PlayerQueue | None:
255        """Return PlayerQueue by queue_id or None if not found."""
256        queue_data = self._queue_data.get(queue_id)
257        return queue_data.queue if queue_data else None
258
259    def queue_data(self, queue_id: str) -> PlayerQueueData:
260        """
261        Return the server-side record for a queue (raises if the queue is unknown).
262
263        Internal accessor for the stateful helper services so they reach per-queue state through
264        the controller rather than its private store.
265        """
266        return self._queue_data[queue_id]
267
268    def queue_data_or_none(self, queue_id: str) -> PlayerQueueData | None:
269        """Return the server-side record for a queue, or None if it is not registered."""
270        return self._queue_data.get(queue_id)
271
272    @api_command("player_queues/items", required_scope=Scope.QUEUES_READ)
273    def items(self, queue_id: str, limit: int = 500, offset: int = 0) -> list[QueueItem]:
274        """Return all QueueItems for given PlayerQueue."""
275        if (queue_data := self._queue_data.get(queue_id)) is None:
276            return []
277        return queue_data.items[offset : offset + limit]
278
279    @api_command("player_queues/get_active_queue", required_scope=Scope.QUEUES_READ)
280    def get_active_queue(self, player_id: str) -> PlayerQueue | None:
281        """Return the current active/synced queue for a player."""
282        if player := self.mass.players.get_player(player_id):
283            return self.mass.players.get_active_queue(player)
284        return None
285
286    # Queue commands
287
288    @api_command("player_queues/shuffle", required_scope=Scope.QUEUES_CONTROL)
289    async def set_shuffle(self, queue_id: str, shuffle_enabled: bool) -> None:
290        """Configure shuffle setting on the the queue."""
291        queue = self._queue_data[queue_id].queue
292        if queue.is_dynamic:
293            # a dynamic queue is an always-on, recency-orchestrated smart mix; manual shuffle
294            # (and plain linear order) have no meaning here so the toggle is locked
295            raise InvalidCommand("Cannot change shuffle while the queue is in dynamic mode")
296        if queue.shuffle_enabled == shuffle_enabled:
297            return  # no change
298        await self._apply_local_shuffle(queue_id, shuffle_enabled)
299
300    def is_smart_shuffle_active(self, queue: PlayerQueue) -> bool:
301        """
302        Return whether smart shuffle is currently in effect for the queue.
303
304        A dynamic queue is always an orchestrated smart mix (the managed pool), so it always counts
305        as active; otherwise smart shuffle is active when shuffle is on and the per-queue
306        smart-shuffle setting is enabled.
307
308        :param queue: The queue to evaluate.
309        """
310        if queue.is_dynamic:
311            return True
312        return queue.shuffle_enabled and self._smart_shuffle.is_enabled(queue.queue_id)
313
314    @api_command("player_queues/autoplay", required_scope=Scope.QUEUES_CONTROL)
315    def set_autoplay(self, queue_id: str, autoplay_enabled: bool) -> None:
316        """Configure Autoplay setting on the queue."""
317        queue_data = self._queue_data[queue_id]
318        queue = queue_data.queue
319        queue.autoplay_enabled = autoplay_enabled
320        # if we're already at/near the end of the queue, kick off a refill right away
321        # (an active dynamic source manages its own refills, so leave it be)
322        if (
323            queue.autoplay_enabled
324            and not queue.is_dynamic
325            and queue.current_index is not None
326            and (queue.items - queue.current_index) < 5
327        ):
328            task_id = f"fill_autoplay_tracks_{queue_id}"
329            self.mass.call_later(5, self._fill_autoplay_tracks, queue_id, task_id=task_id)
330        self.signal_update(queue_id=queue_id)
331
332    @api_command(
333        "player_queues/dont_stop_the_music", required_scope=Scope.QUEUES_CONTROL, alias=True
334    )
335    def set_dont_stop_the_music(self, queue_id: str, dont_stop_the_music_enabled: bool) -> None:
336        """Backwards-compatible alias for the autoplay command, used by older clients."""
337        self.set_autoplay(queue_id, dont_stop_the_music_enabled)
338
339    @api_command("player_queues/repeat", required_scope=Scope.QUEUES_CONTROL)
340    async def set_repeat(self, queue_id: str, repeat_mode: RepeatMode) -> None:
341        """Configure repeat setting on the the queue."""
342        queue = self._queue_data[queue_id].queue
343        if queue.is_dynamic:
344            # a dynamic queue is an always-on flowing mix of its sources; repeat has no meaning here
345            raise InvalidCommand("Cannot change repeat while the queue is in dynamic mode")
346        if queue.repeat_mode == repeat_mode:
347            return  # no change
348        queue.repeat_mode = repeat_mode
349        self.signal_update(queue_id)
350        if (
351            queue.state == PlaybackState.PLAYING
352            and queue.index_in_buffer is not None
353            and queue.index_in_buffer == queue.current_index
354        ):
355            # if the queue is playing,
356            # ensure to (re)queue the next track because it might have changed
357            # note that we only do this if the player has loaded the current track
358            # if not, we wait until it has loaded to prevent conflicts
359            if next_item := self.get_next_item(queue_id, queue.index_in_buffer):
360                self._enqueue_next_item(queue_id, next_item)
361
362    @api_command("player_queues/crossfade", required_scope=Scope.QUEUES_CONTROL)
363    def set_crossfade(self, queue_id: str, crossfade_enabled: bool) -> None:
364        """Enable or disable crossfade on the queue."""
365        queue = self._queue_data[queue_id].queue
366        if queue.crossfade_enabled == crossfade_enabled:
367            return  # no change
368        queue.crossfade_enabled = crossfade_enabled
369        # refresh the derived smart-fades indicator so the update we signal reflects the new state
370        queue.smart_fades_active = self.mass.streams.is_smart_fades_active(queue)
371        self.signal_update(queue_id)
372        if (
373            queue.state == PlaybackState.PLAYING
374            and queue.index_in_buffer is not None
375            and queue.index_in_buffer == queue.current_index
376        ):
377            # re-enqueue the next track so the new crossfade behaviour applies to the
378            # upcoming transition (only when the player has already loaded the current track)
379            if next_item := self.get_next_item(queue_id, queue.index_in_buffer):
380                self._enqueue_next_item(queue_id, next_item)
381
382    @api_command("player_queues/overlay", required_scope=Scope.QUEUES_CONTROL)
383    async def set_overlay(
384        self,
385        queue_id: str,
386        enabled: bool | None = None,
387        source: str | None = None,
388        volume: int | None = None,
389    ) -> None:
390        """
391        Configure the audio overlay for the given queue.
392
393        The audio overlay mixes a looping sound effect (e.g. rain or white noise)
394        into the queue's audio stream. Changes take effect immediately: if the
395        queue is playing, playback is restarted from the current position.
396
397        :param queue_id: queue_id of the queue to configure.
398        :param enabled: Enable or disable the audio overlay. Omit to leave unchanged.
399        :param source: URI of the sound effect item to mix in. Omit to leave unchanged.
400        :param volume: Overlay loudness relative to the music in percent
401            (0-200, 100 = equally loud). Omit to leave unchanged.
402        """
403        queue = self._queue_data[queue_id].queue
404        changed = audible_change = False
405        if source is not None:
406            item = await self.mass.music.get_item_by_uri(source)
407            if item.media_type != MediaType.SOUND_EFFECT:
408                raise InvalidDataError("Audio overlay source must be a sound effect item")
409            mapping = ItemMapping.from_item(cast("SoundEffect", item))
410            if queue.overlay_source != mapping:
411                queue.overlay_source = mapping
412                changed = True
413                audible_change = queue.overlay_enabled
414        if volume is not None:
415            if not (0 <= volume <= 200):
416                raise InvalidDataError(f"Overlay volume must be between 0 and 200, got {volume}")
417            if queue.overlay_volume != volume:
418                queue.overlay_volume = volume
419                changed = True
420                audible_change |= queue.overlay_enabled
421        if enabled is not None and queue.overlay_enabled != enabled:
422            if enabled and queue.overlay_source is None:
423                raise InvalidCommand("Can not enable audio overlay: no overlay source selected")
424            queue.overlay_enabled = enabled
425            changed = audible_change = True
426        if not changed:
427            return
428        self.signal_update(queue_id)
429        if audible_change and queue.state == PlaybackState.PLAYING:
430            # restart playback from the current position so the change is heard
431            # immediately instead of after the player's audio buffer drains
432            await self.resume(queue_id)
433
434    # Two timebases are used in this controller when variable playback speed is in
435    # effect (atempo applied server-side):
436    #   "stream-time"  — seconds of audio the player has played (post-atempo).
437    #   "media-time"   — seconds of the original content the listener has heard.
438    #                    What the user expects to see on the progress bar and what
439    #                    we use for resume positions.
440    # Conversion: media-time = stream-time x playback_speed.
441    @api_command("player_queues/set_playback_speed", required_scope=Scope.QUEUES_CONTROL)
442    async def set_playback_speed(
443        self, queue_id: str, speed: float, queue_item_id: str | None = None
444    ) -> None:
445        """
446        Set the playback speed for the given queue item.
447
448        Variable playback speed is supported only for audiobooks and podcast episodes.
449
450        If queue_item_id is not provided,
451        the speed will be set for the current item in the queue.
452
453        :param queue_id: queue_id of the queue to configure.
454        :param speed: playback speed multiplier (0.5 to 3.0). 1.0 = normal speed.
455        """
456        if not (0.5 <= speed <= 3.0):
457            raise InvalidDataError(f"Playback speed must be between 0.5 and 3.0, got {speed}")
458        queue = self._queue_data[queue_id].queue
459        if not queue.current_item:
460            raise QueueEmpty("Cannot set playback speed: queue is empty")
461        queue_item_id = queue_item_id or queue.current_item.queue_item_id
462        queue_item = self.get_item(queue_id, queue_item_id)
463        if not queue_item:
464            raise InvalidDataError(f"Queue item {queue_item_id} not found in queue")
465        if queue_item.media_type not in (MediaType.AUDIOBOOK, MediaType.PODCAST_EPISODE):
466            raise InvalidCommand(
467                "Variable playback speed is only supported for audiobooks and podcast episodes"
468            )
469        if not queue_item.duration:
470            raise InvalidCommand("Cannot set playback speed for items with unknown duration")
471        current_speed = float(queue_item.extra_attributes.get("playback_speed") or 1.0)
472        if abs(current_speed - speed) < 0.001:
473            return  # no change
474        # use extra_attributes of the queue item to store the playback speed
475        queue_item.extra_attributes["playback_speed"] = speed
476        # mirror onto the queue so corrected_elapsed_time advances in media-time
477        # immediately, before the next on_player_elapsed_time_corrected snapshot.
478        if queue.current_item and queue.current_item.queue_item_id == queue_item_id:
479            # close off the wallclock seconds that already ticked by at the old speed
480            # before switching, so corrected_elapsed_time doesn't multiply them by the new speed
481            if queue.state == PlaybackState.PLAYING:
482                queue.elapsed_time = queue.corrected_elapsed_time
483                queue.elapsed_time_last_updated = time.time()
484            queue.playback_speed = speed
485        self.signal_update(queue_id)
486        if queue.state == PlaybackState.PLAYING:
487            await self.resume(queue_id)
488
489    @api_command(
490        "player_queues/play_media", required_scope=Scope.QUEUES_CONTROL, allow_impersonation=True
491    )
492    async def play_media(
493        self,
494        queue_id: str,
495        media: MediaItemType | ItemMapping | str | list[MediaItemType | ItemMapping | str],
496        option: QueueOption | None = None,
497        radio_mode: bool = False,
498        start_item: PlayableMediaItemType | str | None = None,
499        sort_by: str | None = None,
500        start_from_beginning: bool = False,
501        shuffle: bool | None = None,
502    ) -> None:
503        """
504        Play media item(s) on the given queue.
505
506        :param queue_id: The queue_id of the queue to play media on.
507        :param media: Media that should be played (MediaItem(s) and/or uri's).
508        :param option: Which enqueue mode to use.
509        :param radio_mode: Deprecated — translated to a radio_playlist:// dynamic playlist;
510            prefer enqueuing that URI directly.
511        :param start_item: Optional item to start the playlist or album from.
512        :param sort_by: Optional sort key to order tracks before applying start_item.
513        :param start_from_beginning: Start a podcast episode at position 0, ignoring any
514            saved resume position. The stored progress itself is left untouched.
515        :param shuffle: Play the media shuffled (or explicitly in order). Only applies to the
516            options that start playing right away (play/replace), and never to a dynamic source
517            (an always-on smart mix). Omit to follow the queue's own shuffle setting, which media
518            with an order of its own (album, podcast, episode, audiobook, audio source) switches
519            off; the first item of a batch decides for the whole batch.
520        """
521        self._check_player_permission(queue_id)
522        if not self.get(queue_id):
523            raise PlayerUnavailableError(f"Queue {queue_id} is not available")
524        # A live source is not queue content: it plays on the player while the queue
525        # keeps its own items. Selecting it is the real operation, so a play request
526        # naming one is forwarded there rather than enqueued.
527        if (source_uri := await _resolve_audio_source_request(media)) is not None:
528            await self.mass.players.select_source(queue_id, source_uri)
529            return
530        # Lock is acquired by the @handle_play_action decorator on the internal handler
531        await self._handle_play_media(
532            queue_id,
533            media,
534            option,
535            radio_mode,
536            start_item,
537            sort_by,
538            start_from_beginning,
539            shuffle,
540        )
541
542    @api_command("player_queues/move_item", required_scope=Scope.QUEUES_CONTROL)
543    def move_item(self, queue_id: str, queue_item_id: str, pos_shift: int = 1) -> None:
544        """
545        Move queue item x up/down the queue.
546
547        - queue_id: id of the queue to process this request.
548        - queue_item_id: the item_id of the queueitem that needs to be moved.
549        - pos_shift: move item x positions down if positive value
550        - pos_shift: move item x positions up if negative value
551        - pos_shift:  move item to top of queue as next item if 0.
552        """
553        queue = self._queue_data[queue_id].queue
554        item_index = self.index_by_id(queue_id, queue_item_id)
555        if item_index is None:
556            raise InvalidDataError(f"Item {queue_item_id} not found in queue")
557        if queue.index_in_buffer is not None and item_index <= queue.index_in_buffer:
558            msg = f"{item_index} is already played/buffered"
559            raise IndexError(msg)
560
561        queue_items = self._queue_data[queue_id].items
562        queue_items = queue_items.copy()
563
564        if pos_shift == 0 and queue.state == PlaybackState.PLAYING:
565            new_index = (queue.current_index or 0) + 1
566        elif pos_shift == 0:
567            new_index = queue.current_index or 0
568        else:
569            new_index = item_index + pos_shift
570        if (new_index < (queue.current_index or 0)) or (new_index > len(queue_items)):
571            return
572        # move the item in the list
573        queue_items.insert(new_index, queue_items.pop(item_index))
574        self.update_items(queue_id, queue_items)
575
576    @api_command("player_queues/move_item_end", required_scope=Scope.QUEUES_CONTROL)
577    def move_item_end(self, queue_id: str, queue_item_id: str) -> None:
578        """
579        Move queue item to the end the queue.
580
581        - queue_id: id of the queue to process this request.
582        - queue_item_id: the item_id of the queueitem that needs to be moved.
583        """
584        queue = self._queue_data[queue_id].queue
585        item_index = self.index_by_id(queue_id, queue_item_id)
586        if item_index is None:
587            raise InvalidDataError(f"Item {queue_item_id} not found in queue")
588        if queue.index_in_buffer is not None and item_index <= queue.index_in_buffer:
589            msg = f"{item_index} is already played/buffered"
590            raise IndexError(msg)
591
592        queue_items = self._queue_data[queue_id].items
593        if item_index == (len(queue_items) - 1):
594            return
595        queue_items = queue_items.copy()
596
597        new_index = len(self._queue_data[queue_id].items) - 1
598
599        # move the item in the list
600        queue_items.insert(new_index, queue_items.pop(item_index))
601        self.update_items(queue_id, queue_items)
602
603    @api_command("player_queues/delete_item", required_scope=Scope.QUEUES_CONTROL)
604    def delete_item(self, queue_id: str, item_id_or_index: int | str) -> None:
605        """Delete item (by id or index) from the queue."""
606        if isinstance(item_id_or_index, str):
607            item_index = self.index_by_id(queue_id, item_id_or_index)
608            if item_index is None:
609                raise InvalidDataError(f"Item {item_id_or_index} not found in queue")
610        else:
611            item_index = item_id_or_index
612        queue = self._queue_data[queue_id].queue
613        if queue.index_in_buffer is not None and item_index <= queue.index_in_buffer:
614            # ignore request if track already loaded in the buffer
615            # the frontend should guard so this is just in case
616            self.logger.warning("delete requested for item already loaded in buffer")
617            return
618        queue_items = self._queue_data[queue_id].items.copy()
619        queue_items.pop(item_index)
620        self.update_items(queue_id, queue_items)
621
622    @api_command("player_queues/clear", required_scope=Scope.QUEUES_CONTROL)
623    def clear(self, queue_id: str, skip_stop: bool = False) -> None:
624        """Clear all items in the queue, switching shuffle off with them."""
625        self._clear(queue_id, skip_stop)
626        # clearing is an explicit "start over" gesture by the user, so a shuffle that belonged to
627        # the discarded content must not carry over into whatever is played next
628        self._reset_shuffle(queue_id)
629
630    def mark_ended(self, queue_id: str) -> None:
631        """
632        Mark a queue as played to its end, keeping its items so it can be replayed.
633
634        The playback position is parked on the last item rather than cleared: a null index is
635        indistinguishable from a queue that was loaded but never started, and an index past the
636        end is silently misread by everything that does arithmetic on it. `ended` is what tells
637        clients the queue finished, and pressing play starts it over from the first item.
638
639        :param queue_id: The queue_id of the queue that reached its end.
640        """
641        queue_data = self._queue_data[queue_id]
642        queue = queue_data.queue
643        if not queue_data.items:
644            # nothing to replay, so there is nothing to advertise as finished either
645            self._clear(queue_id)
646            return
647        self.mass.streams.audio_processing.clear(queue_id)
648        queue.ended = True
649        queue.current_index = len(queue_data.items) - 1
650        queue.current_item = queue_data.items[-1]
651        queue.next_item = None
652        queue.elapsed_time = 0
653        queue.elapsed_time_last_updated = time.time()
654        queue.index_in_buffer = None
655        queue.resume_pos = 0
656        self.mass.create_task(self._cleanup_queue_audio_data(queue_id))
657        self.signal_update(queue_id)
658
659    @api_command("player_queues/save_as_playlist", required_scope=Scope.LIBRARY_WRITE)
660    async def save_as_playlist(self, queue_id: str, name: str) -> BackgroundTask:
661        """
662        Save the current queue items as a new playlist.
663
664        :param queue_id: The queue_id of the queue to save.
665        :param name: The name for the new playlist.
666        """
667        if not self.get(queue_id):
668            raise PlayerUnavailableError(f"Queue {queue_id} is not available")
669        queue_items = queue_data.items if (queue_data := self._queue_data.get(queue_id)) else []
670        if not queue_items:
671            raise QueueEmpty("Cannot save an empty queue as a playlist.")
672        # collect URIs from queue items that are playlist-compatible
673        uris: list[str] = []
674        for item in queue_items:
675            if item.uri and item.media_type in PLAYLIST_MEDIA_TYPES:
676                uris.append(item.uri)
677        if not uris:
678            raise InvalidDataError("No valid items in queue to save as playlist.")
679        playlist = await self.mass.music.playlists.create_playlist(name)
680        return await self.mass.music.playlists.add_playlist_tracks(playlist.item_id, uris)
681
682    @api_command("player_queues/stop", required_scope=Scope.QUEUES_CONTROL)
683    @handle_play_action
684    async def stop(self, queue_id: str) -> None:
685        """
686        Handle STOP command for given queue.
687
688        - queue_id: queue_id of the playerqueue to handle the command.
689        """
690        self._check_player_permission(queue_id)
691        # cancel any pending play_index calls for this queue to prevent conflicts
692        self.mass.cancel_timer(f"queue_play_index_{queue_id}")
693        # cancel in-flight preload/enqueue-next so it can't enqueue after stop
694        self.mass.cancel_task(f"preload_next_item_{queue_id}")
695        self.mass.cancel_timer(f"enqueue_next_item_{queue_id}")
696        self.mass.cancel_task(f"enqueue_next_item_{queue_id}")
697        self._set_transitioning(queue_id, False)
698        queue_data = self._queue_data[queue_id]
699        session_id = queue_data.session_id
700        queue_player = self.mass.players.get_player(queue_id, True)
701        if queue_player is None:
702            raise PlayerUnavailableError(f"Player {queue_id} is not available")
703        if (queue := self.get(queue_id)) and queue.active:
704            if queue.state == PlaybackState.PLAYING:
705                queue.resume_pos = int(queue.corrected_elapsed_time)
706        # Use internal handler to avoid circular redirect:
707        # public cmd_stop redirects to queue.stop when a queue is active,
708        # which would loop back here indefinitely.
709        await self.mass.players._handle_cmd_stop(queue_id)
710        if queue_data.session_id == session_id:
711            queue_data.session_id = None
712        self.mass.streams.audio_processing.clear(queue_id, session_id)
713        self.mass.create_task(self._cleanup_queue_audio_data(queue_id))
714
715    @api_command("player_queues/play", required_scope=Scope.QUEUES_CONTROL)
716    async def play(self, queue_id: str) -> None:
717        """
718        Handle PLAY command for given queue.
719
720        :param queue_id: queue_id of the playerqueue to handle the command.
721        """
722        self._check_player_permission(queue_id)
723        if not self.get(queue_id):
724            raise PlayerUnavailableError(f"Queue {queue_id} is not available")
725        await self._handle_play(queue_id)
726
727    @api_command("player_queues/pause", required_scope=Scope.QUEUES_CONTROL)
728    async def pause(self, queue_id: str) -> None:
729        """
730        Handle PAUSE command for given queue.
731
732        - queue_id: queue_id of the playerqueue to handle the command.
733        """
734        self._check_player_permission(queue_id)
735        # cancel any pending play_index calls for this queue to prevent conflicts
736        self.mass.cancel_timer(f"queue_play_index_{queue_id}")
737        self._set_transitioning(queue_id, False)
738        if not (queue := self.get(queue_id)):
739            return
740        queue_active = queue.active
741        if queue.active and queue.state == PlaybackState.PLAYING:
742            queue.resume_pos = int(queue.corrected_elapsed_time)
743        # Use internal handler to avoid circular redirect
744        # (cmd_pause redirects to queue.pause, which calls cmd_pause again)
745        await self.mass.players._handle_cmd_pause(queue_id)
746
747        async def _watch_pause(player: Player) -> None:
748            count = 0
749            # wait for pause
750            while count < 5 and player.state.playback_state == PlaybackState.PLAYING:
751                count += 1
752                await asyncio.sleep(1)
753            # wait for unpause
754            if player.state.playback_state != PlaybackState.PAUSED:
755                return
756            count = 0
757            while count < 30 and player.state.playback_state == PlaybackState.PAUSED:
758                count += 1
759                await asyncio.sleep(1)
760            # if player is still paused when the limit is reached, send stop
761            if player.state.playback_state == PlaybackState.PAUSED:
762                await self.stop(queue_id)
763
764        # we auto stop a player from paused when its paused for 30 seconds
765        if (
766            queue_active
767            and (queue_player := self.mass.players.get_player(queue_id))
768            and not queue_player.extra_data.get(ATTR_ANNOUNCEMENT_IN_PROGRESS)
769        ):
770            self.mass.create_task(_watch_pause(queue_player))
771
772    @api_command("player_queues/play_pause", required_scope=Scope.QUEUES_CONTROL)
773    async def play_pause(self, queue_id: str) -> None:
774        """
775        Toggle play/pause on given playerqueue.
776
777        - queue_id: queue_id of the queue to handle the command.
778        """
779        if (queue := self.get(queue_id)) and queue.state == PlaybackState.PLAYING:
780            await self.pause(queue_id)
781            return
782        await self.play(queue_id)
783
784    @api_command("player_queues/next", required_scope=Scope.QUEUES_CONTROL)
785    @handle_play_action
786    async def next(self, queue_id: str) -> None:
787        """
788        Handle NEXT TRACK command for given queue.
789
790        :param queue_id: queue_id of the queue to handle the command.
791        """
792        self._check_player_permission(queue_id)
793        if (queue := self.get(queue_id)) is None or not queue.active:
794            raise InvalidCommand(f"Queue {queue_id} is not active")
795        self._set_transitioning(queue_id, True)
796        idx = self._queue_data[queue_id].queue.current_index
797        if idx is None:
798            self.logger.warning("Queue %s has no current index", queue.display_name)
799            self._set_transitioning(queue_id, False)
800            return
801        next_index = self._get_next_index(queue_id, idx, True)
802        if next_index is None:
803            self._set_transitioning(queue_id, False)
804            return
805
806        # immediately update current item so UI shows the new track right away
807        queue.current_index = next_index
808        queue.current_item = self.get_item(queue_id, next_index)
809        queue.elapsed_time = 0
810        queue.elapsed_time_last_updated = time.time()
811        self.signal_update(queue_id)
812        if queue_player := self.mass.players.get_player(queue_id, True):
813            queue_player.update_state()
814
815        # debounce rapid next button presses using call_later
816        self.mass.call_later(
817            1,
818            self.play_index,
819            queue_id,
820            next_index,
821            task_id=f"queue_play_index_{queue_id}",
822        )
823
824    @api_command("player_queues/previous", required_scope=Scope.QUEUES_CONTROL)
825    @handle_play_action
826    async def previous(self, queue_id: str) -> None:
827        """
828        Handle PREVIOUS TRACK command for given queue.
829
830        :param queue_id: queue_id of the queue to handle the command.
831        """
832        self._check_player_permission(queue_id)
833        if (queue := self.get(queue_id)) is None or not queue.active:
834            raise InvalidCommand(f"Queue {queue_id} is not active")
835        self._set_transitioning(queue_id, True)
836        current_index = self._queue_data[queue_id].queue.current_index
837        if current_index is None:
838            self._set_transitioning(queue_id, False)
839            return
840        prev_index = int(current_index)
841        # restart current track if elapsed > 5s, otherwise go to previous
842        if self._queue_data[queue_id].queue.elapsed_time < 5:
843            prev_index = max(current_index - 1, 0)
844
845        # immediately update current item so UI shows the new track right away
846        queue.current_index = prev_index
847        queue.current_item = self.get_item(queue_id, prev_index)
848        queue.elapsed_time = 0
849        queue.elapsed_time_last_updated = time.time()
850        self.signal_update(queue_id)
851        if queue_player := self.mass.players.get_player(queue_id, True):
852            queue_player.update_state()
853
854        # debounce rapid previous button presses using call_later
855        self.mass.call_later(
856            1,
857            self.play_index,
858            queue_id,
859            prev_index,
860            task_id=f"queue_play_index_{queue_id}",
861        )
862
863    @api_command("player_queues/skip", required_scope=Scope.QUEUES_CONTROL)
864    async def skip(self, queue_id: str, seconds: int = 10) -> None:
865        """
866        Handle SKIP command for given queue.
867
868        - queue_id: queue_id of the queue to handle the command.
869        - seconds: number of seconds to skip in track. Use negative value to skip back.
870        """
871        if (queue := self.get(queue_id)) is None or not queue.active:
872            raise InvalidCommand(f"Queue {queue_id} is not active")
873        await self.seek(queue_id, int(self._queue_data[queue_id].queue.elapsed_time + seconds))
874
875    @api_command("player_queues/seek", required_scope=Scope.QUEUES_CONTROL)
876    async def seek(self, queue_id: str, position: int = 10) -> None:
877        """
878        Handle SEEK command for given queue.
879
880        - queue_id: queue_id of the queue to handle the command.
881        - position: position in seconds to seek to in the current playing item.
882        """
883        if (queue := self.get(queue_id)) is None or not queue.active:
884            raise InvalidCommand(f"Queue {queue_id} is not active")
885        queue_player = self.mass.players.get_player(queue_id, True)
886        if queue_player is None:
887            raise PlayerUnavailableError(f"Player {queue_id} is not available")
888        if not queue.current_item:
889            raise InvalidCommand(f"Queue {queue_player.state.name} has no item(s) loaded.")
890        if not queue.current_item.duration:
891            raise InvalidCommand("Can not seek items without duration.")
892        position = max(0, int(position))
893        if position > queue.current_item.duration:
894            raise InvalidCommand("Can not seek outside of duration range.")
895        if queue.current_index is None:
896            raise InvalidCommand(f"Queue {queue_player.state.name} has no current index.")
897        # Publish the seek target before rebuilding the stream to prevent progress snapback.
898        queue.elapsed_time = position
899        queue.elapsed_time_last_updated = time.time()
900        self.signal_update(queue_id)
901        await self.play_index(queue_id, queue.current_index, seek_position=position)
902
903    @api_command("player_queues/resume", required_scope=Scope.QUEUES_CONTROL)
904    @handle_play_action
905    async def resume(self, queue_id: str, fade_in: bool | None = None) -> None:
906        """
907        Handle RESUME command for given queue.
908
909        - queue_id: queue_id of the queue to handle the command.
910        """
911        self._check_player_permission(queue_id)
912        queue = self._queue_data[queue_id].queue
913        queue_items = self._queue_data[queue_id].items
914        resume_item = queue.current_item
915        if queue.state == PlaybackState.PLAYING:
916            # resume requested while already playing,
917            # use current position as resume position
918            resume_pos = queue.corrected_elapsed_time
919            fade_in = False
920        else:
921            resume_pos = queue.resume_pos or queue.elapsed_time
922
923        if queue.ended and len(queue_items) > 0:
924            # the queue played to its end and is parked on its last item,
925            # so pressing play starts it over from the beginning
926            resume_item = queue_items[0]
927            resume_pos = 0
928        elif not resume_item and queue.current_index is not None and len(queue_items) > 0:
929            resume_item = self.get_item(queue_id, queue.current_index)
930            resume_pos = 0
931        elif not resume_item and queue.current_index is None and len(queue_items) > 0:
932            # items available in queue but no previous track, start at 0
933            resume_item = self.get_item(queue_id, 0)
934            resume_pos = 0
935
936        if resume_item is not None:
937            queue_player = self.mass.players.get_player(queue_id)
938            if queue_player is None:
939                raise PlayerUnavailableError(f"Player {queue_id} is not available")
940            if (
941                fade_in is None
942                and queue_player.state.playback_state == PlaybackState.IDLE
943                and (time.time() - queue.elapsed_time_last_updated) > 60
944            ):
945                # enable fade in effect if the player is idle for a while
946                fade_in = resume_pos > 0
947            if resume_item.media_type == MediaType.RADIO:
948                # we're not able to skip in online radio so this is pointless
949                resume_pos = 0
950            await self.play_index(
951                queue_id, resume_item.queue_item_id, int(resume_pos), fade_in or False
952            )
953        else:
954            msg = f"Resume queue requested but queue {queue.display_name} is empty"
955            raise QueueEmpty(msg)
956
957    @api_command("player_queues/play_index", required_scope=Scope.QUEUES_CONTROL)
958    @handle_play_action
959    async def play_index(  # noqa: PLR0915
960        self,
961        queue_id: str,
962        index: int | str,
963        seek_position: int = 0,
964        fade_in: bool = False,
965    ) -> None:
966        """Play item at index (or item_id) X in queue."""
967        self._check_player_permission(queue_id)
968        # cancel any pending play_index calls for this queue to prevent conflicts
969        self.mass.cancel_timer(f"queue_play_index_{queue_id}")
970        # we set a flag to notify the update logic that we're transitioning to a new track
971        self._set_transitioning(queue_id, True)
972        try:
973            queue_data = self._queue_data[queue_id]
974            queue = queue_data.queue
975            queue.resume_pos = 0
976            # A queue picked up from its end plays its items over from the start, so a resume point
977            # left on an audiobook/episode must not pull it back to where it was left off. The flag
978            # itself is only cleared once an item actually loaded below, so a start that never got
979            # off the ground leaves the queue finished instead of stranding it without a position.
980            restarting_ended_queue = queue.ended
981            if isinstance(index, str):
982                temp_index = self.index_by_id(queue_id, index)
983                if temp_index is None:
984                    raise InvalidDataError(f"Item {index} not found in queue")
985                index = temp_index
986            # At this point index is guaranteed to be int
987            queue.index_in_buffer = index
988            queue_data.flow_mode_stream_log = []
989            queue_data.flow_buffer_completed = None
990            queue_data.flow_queue_exhausted = None
991            target_player = self.mass.players.get_player(queue_id)
992            if target_player is None:
993                raise PlayerUnavailableError(f"Player {queue_id} is not available")
994            queue_data.next_item_id_enqueued = None
995            # always update session id when we start a new playback session
996            queue_data.session_id = shortuuid.random(length=8)
997            self.mass.streams.audio_processing.start_session(
998                queue_id,
999                queue_data.session_id,
1000            )
1001            # handle resume point of audiobook(chapter) or podcast(episode)
1002            if (
1003                not seek_position
1004                and not restarting_ended_queue
1005                and (queue_item := self.get_item(queue_id, index))
1006                and (resume_position_ms := getattr(queue_item.media_item, "resume_position_ms", 0))
1007            ):
1008                # the client may have fetched the item before its duration was known
1009                await self._restore_probed_duration(queue_item)
1010                if queue_item.duration or getattr(queue_item.media_item, "duration", 0):
1011                    seek_position = max(0, int((resume_position_ms - 500) / 1000))
1012                else:
1013                    # seeking needs a duration, which is determined while streaming
1014                    self.logger.debug(
1015                        "Can not resume %s at %ss: its duration is not known (yet)",
1016                        queue_item.name,
1017                        int(resume_position_ms / 1000),
1018                    )
1019
1020            # restore the persisted playback speed for a freshly queued audiobook/episode
1021            # (an in-session item already carries its speed in extra_attributes)
1022            if (
1023                (queue_item := self.get_item(queue_id, index))
1024                and queue_item.media_item is not None
1025                and queue_item.media_type in (MediaType.AUDIOBOOK, MediaType.PODCAST_EPISODE)
1026                and "playback_speed" not in queue_item.extra_attributes
1027            ):
1028                stored_speed = await self.mass.music.get_playback_speed(
1029                    cast("Audiobook | PodcastEpisode", queue_item.media_item),
1030                    userid=queue_data.userid,
1031                )
1032                if stored_speed != 1.0:
1033                    queue_item.extra_attributes["playback_speed"] = stored_speed
1034
1035            # try to load the item, retry with next item if it fails
1036            for attempt in range(5):
1037                try:
1038                    queue_item = self.get_item(queue_id, index)
1039                    if not queue_item:
1040                        continue  # guard
1041                    await self._load_item(
1042                        queue_item,
1043                        self._get_next_index(queue_id, index),
1044                        is_start=True,
1045                        seek_position=seek_position if attempt == 0 else 0,
1046                        fade_in=fade_in if attempt == 0 else False,
1047                    )
1048                    # if we reach this point, loading the item succeeded, break the loop
1049                    queue.current_index = index
1050                    queue.current_item = queue_item
1051                    # playback is under way, so the queue is no longer sitting at its end
1052                    queue.ended = False
1053                    # reset the elapsed clock together with the item switch (like
1054                    # next/previous do), so queue updates signaled before the player
1055                    # reports position don't carry the previous item's elapsed_time
1056                    queue.elapsed_time = seek_position if attempt == 0 else 0
1057                    queue.elapsed_time_last_updated = time.time()
1058                    break
1059                except (MediaNotFoundError, AudioError) as err:
1060                    item_name = queue_item.name if queue_item else "unknown"
1061                    if isinstance(err, ProviderStreamLimitError):
1062                        # the requested item is playable, its provider is just at capacity:
1063                        # report that instead of silently advancing to another item
1064                        self.logger.error("%s", err)
1065                        await self.stop(queue_id)
1066                        raise
1067                    # Only MediaNotFoundError (item unreachable) is persistent;
1068                    # keep AudioError items available so a retry can resurface
1069                    # the same actionable error.
1070                    if queue_item and isinstance(err, MediaNotFoundError):
1071                        queue_item.available = False
1072                    next_index = self._get_next_index(queue_id, index, allow_repeat=False)
1073                    if next_index is None:
1074                        # Surface an AudioError's own (actionable) message;
1075                        # MediaNotFoundError gets the generic wording.
1076                        if isinstance(err, AudioError) and str(err):
1077                            msg = str(err)
1078                        else:
1079                            msg = f"Playback failed for {item_name} - no more tracks available"
1080                        self.logger.error(msg)
1081                        await self.stop(queue_id)
1082                        raise MediaNotFoundError(msg) from err
1083                    self.logger.warning(
1084                        "Skipping unplayable item %s",
1085                        item_name,
1086                    )
1087                    index = next_index
1088            else:
1089                # all attempts to find a playable item failed
1090                await self.stop(queue_id)
1091                raise MediaNotFoundError("No playable item found to start playback")
1092
1093            # Reset flow_mode - the streams controller will set it if flow mode is used.
1094            queue.flow_mode = False
1095            player_media = await self.player_media_from_queue_item(queue_item)
1096            # Hold the play action until the player confirms playback so the UI keeps
1097            # showing the command as in progress instead of falling back to a play button
1098            # for the time the player still needs to connect and start. The queue update
1099            # for the new item goes out first, so the item shows while it is starting.
1100            async with self.mass.players.wait_for_player_update(
1101                queue_id,
1102                attribute_name="playback_state",
1103                attribute_value=PlaybackState.PLAYING,
1104                timeout=PLAYBACK_START_TIMEOUT,
1105            ):
1106                await self.mass.players.play_media(queue_id, player_media)
1107                queue.current_index = index
1108                queue.current_item = queue_item
1109                self.signal_update(queue_id)
1110        finally:
1111            self._set_transitioning(queue_id, False)
1112
1113    @api_command("player_queues/transfer", required_scope=Scope.QUEUES_CONTROL)
1114    async def transfer_queue(
1115        self,
1116        source_queue_id: str,
1117        target_queue_id: str,
1118        auto_play: bool | None = None,
1119    ) -> None:
1120        """Transfer queue to another queue."""
1121        if not (source_queue := self.get(source_queue_id)):
1122            raise PlayerUnavailableError(f"Queue {source_queue_id} is not available")
1123        if not (target_queue := self.get(target_queue_id)):
1124            raise PlayerUnavailableError(f"Queue {target_queue_id} is not available")
1125        if auto_play is None:
1126            auto_play = source_queue.state == PlaybackState.PLAYING
1127
1128        target_player = self.mass.players.get_player(target_queue_id)
1129        if target_player is None:
1130            raise PlayerUnavailableError(f"Player {target_queue_id} is not available")
1131        if target_player.state.active_group or target_player.state.synced_to:
1132            # edge case: the user wants to move playback from the group as a whole, to a single
1133            # player in the group or it is grouped and the command targeted at the single player.
1134            # We need to dissolve the group/sync first, and wait for the state to actually
1135            # propagate before we hand the queue over to the target player.
1136            group_id = target_player.state.active_group or target_player.state.synced_to
1137            assert group_id is not None  # checked in if condition above
1138            # For an ad-hoc sync group (target is a sync member of a regular leader),
1139            # ungroup the target itself so only it is freed - ungrouping the leader would
1140            # transfer leadership to a remaining member and recurse back into this method.
1141            # For a virtual group player (active_group), release the group so its static
1142            # members are handled correctly.
1143            ungroup_target = (
1144                target_queue_id
1145                if target_player.state.synced_to and not target_player.state.active_group
1146                else group_id
1147            )
1148            async with self.mass.players.wait_for_player_update(
1149                target_queue_id,
1150                attribute_name=(
1151                    "active_group" if target_player.state.active_group else "synced_to"
1152                ),
1153                attribute_value=None,
1154                timeout=5,
1155            ):
1156                await self.mass.players.cmd_ungroup(ungroup_target)
1157
1158        # capture source state before stopping (stop resets these)
1159        source_items = self._queue_data[source_queue_id].items
1160        if source_queue.state == PlaybackState.PLAYING:
1161            # use the live playback clock while actively playing
1162            source_resume_pos = int(source_queue.corrected_elapsed_time)
1163        else:
1164            # when not playing the live clock is stale, so use the stored resume position
1165            source_resume_pos = int(source_queue.resume_pos or source_queue.elapsed_time or 0)
1166        source_current_index = source_queue.current_index
1167        source_current_item = source_queue.current_item
1168
1169        # stop the source player synchronously to prevent the async stop from
1170        # clear() racing with the target's sync group formation/protocol switching
1171        if source_queue.state != PlaybackState.IDLE:
1172            await self.stop(source_queue_id)
1173
1174        target_queue.repeat_mode = source_queue.repeat_mode
1175        target_queue.shuffle_enabled = source_queue.shuffle_enabled
1176        target_queue.crossfade_enabled = source_queue.crossfade_enabled
1177        # refresh the derived smart-fades indicator for the target's own config/availability
1178        target_queue.smart_fades_active = self.mass.streams.is_smart_fades_active(target_queue)
1179        target_queue.autoplay_enabled = source_queue.autoplay_enabled
1180        self._queue_data[target_queue_id].source_items = list(
1181            self._queue_data[source_queue_id].source_items
1182        )
1183        target_queue.sources = list(source_queue.sources)
1184        target_queue.is_dynamic = source_queue.is_dynamic
1185        target_queue.smart_shuffle_active = self.is_smart_shuffle_active(target_queue)
1186        self._queue_data[target_queue_id].enqueued_media_items = list(
1187            self._queue_data[source_queue_id].enqueued_media_items
1188        )
1189        self._queue_data[target_queue_id].credited_albums = set(
1190            self._queue_data[source_queue_id].credited_albums
1191        )
1192        target_queue.resume_pos = source_resume_pos
1193        target_queue.current_index = source_current_index
1194        if source_current_item:
1195            target_queue.current_item = source_current_item
1196            target_queue.current_item.queue_id = target_queue_id
1197        self._clear(source_queue_id, skip_stop=True)
1198
1199        await self.load(target_queue_id, source_items, keep_remaining=False, keep_played=False)
1200        for item in source_items:
1201            item.queue_id = target_queue_id
1202        self.update_items(target_queue_id, source_items)
1203        if auto_play:
1204            await self.resume(target_queue_id)
1205
1206    # Interaction with player
1207
1208    async def on_player_register(self, player: Player) -> None:
1209        """Register PlayerQueue for given player/queue id."""
1210        queue_id = player.player_id
1211        queue_data: PlayerQueueData | None = None
1212        # try to restore previous state
1213        try:
1214            if prev_state := await self.mass.cache.get(
1215                key=queue_id,
1216                provider=self.domain,
1217                category=CACHE_CATEGORY_PLAYER_QUEUE_STATE,
1218            ):
1219                prev_items = await self.mass.cache.get(
1220                    key=queue_id,
1221                    provider=self.domain,
1222                    category=CACHE_CATEGORY_PLAYER_QUEUE_ITEMS,
1223                    default=[],
1224                )
1225                queue_data = PlayerQueueData.from_cache(prev_state, prev_items)
1226        except Exception as err:
1227            self.logger.warning(
1228                "Failed to restore the queue(items) for %s - %s",
1229                player.state.name,
1230                str(err),
1231            )
1232            # Reset to clean state on failure
1233            queue_data = None
1234        if queue_data is None:
1235            queue_data = PlayerQueueData(
1236                queue=PlayerQueue(
1237                    queue_id=queue_id,
1238                    active=False,
1239                    display_name=player.state.name,
1240                    available=player.state.available,
1241                    # Autoplay starts out on for a brand new queue; the player's own Autoplay
1242                    # switch owns it from here on (and is restored above for a queue we know)
1243                    autoplay_enabled=True,
1244                    items=0,
1245                )
1246            )
1247
1248        self._queue_data[queue_id] = queue_data
1249        # always call update to calculate state etc
1250        self.on_player_update(player, {})
1251        self.mass.signal_event(EventType.QUEUE_ADDED, object_id=queue_id, data=queue_data.queue)
1252
1253    def on_player_update(
1254        self,
1255        player: Player,
1256        changed_values: dict[str, tuple[Any, Any]],
1257    ) -> None:
1258        """
1259        Call when a PlayerQueue needs to be updated (e.g. when player updates).
1260
1261        NOTE: This is called every second if the player is playing.
1262        """
1263        if player.type == PlayerType.PROTOCOL:
1264            # protocol players do not have a queue on their own
1265            return
1266        queue_id = player.player_id
1267        if (queue := self.get(queue_id)) is None:
1268            # race condition
1269            return
1270        if player.extra_data.get(ATTR_ANNOUNCEMENT_IN_PROGRESS):
1271            # do nothing while the announcement is in progress
1272            return
1273        # determine if this queue is currently active for this player
1274        queue.active = player.state.active_source in (queue.queue_id, None)
1275        if not queue.active and self._queue_data[queue_id].prev_state is None:
1276            queue.state = PlaybackState.IDLE
1277            # return early if the queue is not active and we have no previous state
1278            return
1279        if self._queue_data[queue_id].transitioning:
1280            # we're currently transitioning to a new track,
1281            # ignore updates from the player during this time
1282            return
1283        # queue is active and preflight checks passed, update the queue details
1284        self._update_queue_from_player(player)
1285
1286    def on_player_elapsed_time_corrected(self, player: Player) -> None:
1287        """Correct the queue's timing base if the player's real elapsed_time diverged."""
1288        if player.type == PlayerType.PROTOCOL:
1289            return
1290        queue_id = player.player_id
1291        if (queue := self.get(queue_id)) is None:
1292            return
1293        if not queue.active:
1294            return
1295        player_elapsed = player.state.corrected_elapsed_time
1296        if player_elapsed is None:
1297            return
1298        now = time.time()
1299        # queue.elapsed_time is stored in media-time so it can be displayed and
1300        # used as a resume position directly. The player reports stream-time
1301        # (post-atempo), so we scale by the current item's playback_speed.
1302        speed = get_current_playback_speed(queue)
1303        if queue.flow_mode:
1304            # _get_flow_queue_stream_index returns media-time in the current item
1305            # using each playlog entry's recorded speed.
1306            _, elapsed_time = self._get_flow_queue_stream_index(queue, player)
1307        else:
1308            elapsed_time = player_elapsed * speed
1309            if queue.current_item and queue.current_item.streamdetails:
1310                if seek_pos := queue.current_item.streamdetails.seek_position:
1311                    elapsed_time += seek_pos
1312        queue.elapsed_time = elapsed_time
1313        queue.elapsed_time_last_updated = now
1314        queue.playback_speed = speed
1315        self.mass.signal_event(
1316            EventType.QUEUE_TIME_UPDATED,
1317            object_id=queue_id,
1318            data=queue.elapsed_time,
1319        )
1320
1321    def on_player_remove(self, player_id: str, permanent: bool) -> None:
1322        """Call when a player is removed from the registry."""
1323        self.mass.streams.audio_processing.clear(player_id)
1324        # cancel any pending play_index calls for this queue to prevent conflicts
1325        self.mass.cancel_timer(f"queue_play_index_{player_id}")
1326        # cancel a pending debounced cache write AND an already-started one, so neither can
1327        # recreate a deleted entry after the player is gone (the timer becomes a task once it fires)
1328        self.mass.cancel_timer(f"save_queue_cache_{player_id}")
1329        self.mass.cancel_task(f"save_queue_cache_{player_id}")
1330        self._set_transitioning(player_id, False)
1331        if permanent:
1332            self.purge_saved_queue(player_id)
1333        self._queue_data.pop(player_id, None)
1334        self._managed_pool.forget(player_id)
1335
1336    def purge_saved_queue(self, queue_id: str) -> None:
1337        """Delete the persisted state and items of the given queue."""
1338        for category in (CACHE_CATEGORY_PLAYER_QUEUE_STATE, CACHE_CATEGORY_PLAYER_QUEUE_ITEMS):
1339            # a removal runs both the player teardown and the config cleanup, so keep the
1340            # delete to one task per category instead of one per caller
1341            self.mass.create_task(
1342                self.mass.cache.delete(
1343                    key=queue_id,
1344                    provider=self.domain,
1345                    category=category,
1346                ),
1347                task_id=f"purge_saved_queue_{queue_id}_{category}",
1348            )
1349
1350    async def load_next_queue_item(
1351        self,
1352        queue_id: str,
1353        current_item_id: str,
1354    ) -> QueueItem:
1355        """
1356        Call when a player wants the next queue item to play.
1357
1358        Raises QueueEmpty if there are no more tracks left.
1359        """
1360        queue = self.get(queue_id)
1361        if not queue:
1362            msg = f"PlayerQueue {queue_id} is not available"
1363            raise PlayerUnavailableError(msg)
1364        cur_index = self.index_by_id(queue_id, current_item_id)
1365        if cur_index is None:
1366            # this is just a guard for bad data
1367            raise QueueEmpty("Invalid item id for queue given.")
1368        next_item: QueueItem | None = None
1369        idx = 0
1370        while True:
1371            next_index = self._get_next_index(queue_id, cur_index + idx)
1372            if next_index is None:
1373                raise QueueEmpty("No more tracks left in the queue.")
1374            queue_item = self.get_item(queue_id, next_index)
1375            if queue_item is None:
1376                raise QueueEmpty("No more tracks left in the queue.")
1377            if idx >= 10:
1378                # we only allow 10 retries to prevent infinite loops
1379                raise QueueEmpty("No more (playable) tracks left in the queue.")
1380            try:
1381                await self._load_item(queue_item, self._get_next_index(queue_id, next_index))
1382                # we're all set, this is our next item
1383                next_item = queue_item
1384                break
1385            except ProviderStreamLimitError:
1386                # transient source capacity, do not burn a playable item over it
1387                raise
1388            except MediaNotFoundError, AudioError:
1389                # No stream details found, skip this QueueItem
1390                self.logger.warning(
1391                    "Skipping unplayable item %s (%s)", queue_item.name, queue_item.uri
1392                )
1393                queue_item.available = False
1394                idx += 1
1395        if idx != 0:
1396            # we skipped some items, signal a queue items update
1397            self.update_items(queue_id, self._queue_data[queue_id].items)
1398        if next_item is None:
1399            raise QueueEmpty("No more (playable) tracks left in the queue.")
1400
1401        # carry playback_speed forward across consecutive audiobook/podcast items
1402        current_item = self.get_item(queue_id, current_item_id)
1403        if (
1404            current_item
1405            and current_item.media_type in (MediaType.AUDIOBOOK, MediaType.PODCAST_EPISODE)
1406            and next_item.media_type in (MediaType.AUDIOBOOK, MediaType.PODCAST_EPISODE)
1407        ):
1408            next_item.extra_attributes["playback_speed"] = current_item.extra_attributes.get(
1409                "playback_speed", 1.0
1410            )
1411
1412        return next_item
1413
1414    def track_loaded_in_buffer(self, queue_id: str, item_id: str) -> None:
1415        """Call when a player has (started) loading a track in the buffer."""
1416        queue = self.get(queue_id)
1417        if not queue:
1418            msg = f"PlayerQueue {queue_id} is not available"
1419            raise PlayerUnavailableError(msg)
1420        # store the index of the item that is currently (being) loaded in the buffer
1421        # which helps us a bit to determine how far the player has buffered ahead
1422        current_index = self.index_by_id(queue_id, item_id)
1423        queue.index_in_buffer = current_index
1424        self.logger.debug("PlayerQueue %s loaded item %s in buffer", queue.display_name, item_id)
1425        self.signal_update(queue_id)
1426        # preload next streamdetails
1427        self._preload_next_item(queue_id, item_id)
1428        # clean up stale audio buffers for old queue items to prevent memory leaks
1429        if current_index is not None:
1430            self.mass.create_task(self._cleanup_stale_queue_buffers(queue_id, current_index))
1431
1432    def queue_buffer_completed(self, queue_id: str, queue_exhausted: bool) -> None:
1433        """
1434        Call when the flow stream has finished generating all audio data for a queue.
1435
1436        At this point all audio data for the queue has been passed to the encoding pipeline.
1437        The player will go idle once it finishes playing the remaining buffered audio.
1438
1439        We start a background task that waits for the player to go idle and checks if new
1440        items have been added to the queue in the meantime, resuming playback if so.
1441
1442        :param queue_id: The queue ID.
1443        :param queue_exhausted: Whether the flow ended because the queue ran out of items,
1444            as opposed to ending early to restart on a format change or a live item.
1445        """
1446        queue = self.get(queue_id)
1447        if not queue:
1448            return
1449        self.logger.debug("Queue flow buffer completed for %s", queue.display_name)
1450
1451        # capture session_id so we can bail out if playback restarts
1452        queue_data = self._queue_data[queue_id]
1453        original_session_id = queue_data.session_id
1454        # record so player providers can detect flow EOF without an idle report
1455        if original_session_id is not None:
1456            queue_data.flow_buffer_completed = original_session_id
1457            if queue_exhausted:
1458                queue_data.flow_queue_exhausted = original_session_id
1459
1460        async def _resume_on_idle() -> None:
1461            # wait for the player to finish playing the buffered audio and go idle
1462            idle_detected = False
1463            for _ in range(60):
1464                await asyncio.sleep(1)
1465                if not queue.active or queue_data.session_id != original_session_id:
1466                    return
1467                if queue.state == PlaybackState.IDLE:
1468                    idle_detected = True
1469                    break
1470            if not idle_detected:
1471                return
1472            # player went idle, give it a brief moment to settle
1473            await asyncio.sleep(1)
1474            if queue.state != PlaybackState.IDLE or queue_data.session_id != original_session_id:
1475                return
1476            # check if new items were added to the queue after the flow stream ended
1477            if queue.current_index is not None and (
1478                next_item := self.get_next_item(queue_id, queue.current_index)
1479            ):
1480                next_index = self.index_by_id(queue_id, next_item.queue_item_id)
1481                if next_index is not None:
1482                    self.logger.info(
1483                        "Resuming playback after flow stream completed for %s",
1484                        queue.display_name,
1485                    )
1486                    await self.play_index(queue_id, next_index)
1487
1488        task_id = f"queue_buffer_completed_{queue_id}"
1489        self.mass.create_task(_resume_on_idle(), task_id=task_id)
1490
1491    def flow_stream_finished(self, queue_id: str) -> bool:
1492        """
1493        Return whether the flow stream for the current playback session is fully generated.
1494
1495        Lets player providers detect flow EOF when the device does not report idle
1496        (e.g. a Cast group that underruns the LIVE flow stream and keeps reporting playing).
1497
1498        :param queue_id: The queue ID.
1499        """
1500        queue_data = self.queue_data_or_none(queue_id)
1501        if queue_data is None or queue_data.session_id is None:
1502            return False
1503        return queue_data.flow_buffer_completed == queue_data.session_id
1504
1505    def flow_queue_exhausted(self, queue_id: str, session_id: str) -> bool:
1506        """
1507        Return whether the given flow stream session played the queue to its end.
1508
1509        False while a session is still streaming, and for a flow stream that ended early
1510        to be restarted (a format change or a live item), where the player is expected to
1511        pick up the next stream right away.
1512
1513        :param queue_id: The queue ID.
1514        :param session_id: The stream session to check.
1515        """
1516        queue_data = self.queue_data_or_none(queue_id)
1517        if queue_data is None or queue_data.session_id != session_id:
1518            return False
1519        return queue_data.flow_queue_exhausted == session_id
1520
1521    # Main queue manipulation methods
1522
1523    async def load(
1524        self,
1525        queue_id: str,
1526        queue_items: list[QueueItem],
1527        insert_at_index: int = 0,
1528        keep_remaining: bool = True,
1529        keep_played: bool = True,
1530        shuffle: bool = False,
1531        pin_first: bool = False,
1532    ) -> None:
1533        """
1534        Load new items at index.
1535
1536        - queue_id: id of the queue to process this request.
1537        - queue_items: a list of QueueItems
1538        - insert_at_index: insert the item(s) at this index
1539        - keep_remaining: keep the remaining items after the insert
1540        - shuffle: (re)shuffle the items after insert index
1541        - pin_first: keep the first item at the insert index instead of letting the shuffle
1542          move it; only meaningful together with shuffle
1543        """
1544        prev_items = self._queue_data[queue_id].items[:insert_at_index] if keep_played else []
1545        next_items = queue_items
1546
1547        # if keep_remaining, append the old 'next' items
1548        if keep_remaining:
1549            next_items += self._queue_data[queue_id].items[insert_at_index:]
1550
1551        # we set the original insert order as attribute so we can un-shuffle
1552        for index, item in enumerate(next_items):
1553            item.sort_index += insert_at_index + index
1554        # (re)shuffle the final batch if needed: smart shuffle when enabled, else pure random
1555        if shuffle:
1556            queue = self._queue_data[queue_id].queue
1557            # a user-picked item must stay the one that plays, so hold it out of the shuffle
1558            pinned = next_items[:1] if pin_first else []
1559            shuffled = next_items[1:] if pin_first else next_items
1560            if self._smart_shuffle.is_enabled(queue_id):
1561                shuffled = await self._smart_shuffle.arrange(queue, shuffled)
1562            else:
1563                shuffled = random.sample(shuffled, len(shuffled))
1564            next_items = pinned + shuffled
1565        self.update_items(queue_id, prev_items + next_items)
1566
1567    def update_items(self, queue_id: str, queue_items: list[QueueItem]) -> None:
1568        """Update the existing queue items, mostly caused by reordering."""
1569        self._queue_data[queue_id].items = queue_items
1570        queue = self._queue_data[queue_id].queue
1571        queue.items = len(self._queue_data[queue_id].items)
1572        self.signal_update(queue_id, True)
1573        if (
1574            queue.state == PlaybackState.PLAYING
1575            and queue.index_in_buffer is not None
1576            and queue.index_in_buffer == queue.current_index
1577        ):
1578            # if the queue is playing,
1579            # ensure to (re)queue the next track because it might have changed
1580            # note that we only do this if the player has loaded the current track
1581            # if not, we wait until it has loaded to prevent conflicts
1582            if next_item := self.get_next_item(queue_id, queue.index_in_buffer):
1583                self._enqueue_next_item(queue_id, next_item)
1584
1585    # Helper methods
1586
1587    def get_item(self, queue_id: str, item_id_or_index: int | str | None) -> QueueItem | None:
1588        """Get queue item by index or item_id."""
1589        if item_id_or_index is None:
1590            return None
1591        if (queue_data := self._queue_data.get(queue_id)) is None:
1592            return None
1593        queue_items = queue_data.items
1594        if isinstance(item_id_or_index, int) and len(queue_items) > item_id_or_index:
1595            return queue_items[item_id_or_index]
1596        if isinstance(item_id_or_index, str):
1597            return next((x for x in queue_items if x.queue_item_id == item_id_or_index), None)
1598        return None
1599
1600    def signal_update(self, queue_id: str, items_changed: bool = False) -> None:
1601        """Signal state changed of given queue."""
1602        if (queue_data := self._queue_data.get(queue_id)) is None:
1603            return
1604        queue = queue_data.queue
1605        # a mirrored shuffle write (streams controller) changes what smart shuffle
1606        # resolves to, so refresh the derived flag with every signaled update
1607        queue.smart_shuffle_active = self.is_smart_shuffle_active(queue)
1608        if items_changed:
1609            queue_data.items_cache_dirty = True
1610            self.mass.signal_event(EventType.QUEUE_ITEMS_UPDATED, object_id=queue_id, data=queue)
1611        self.mass.streams.audio_processing.prune(queue_id)
1612        # always send the base event
1613        self.mass.signal_event(EventType.QUEUE_UPDATED, object_id=queue_id, data=queue)
1614        # also signal update to the player itself so it can update its current_media
1615        self.mass.players.trigger_player_update(queue_id)
1616        # persist the (settings-bearing) queue state, debounced so a burst of updates or the
1617        # per-track updates during playback collapse into a single cache write
1618        self.mass.call_later(
1619            QUEUE_CACHE_SAVE_DELAY,
1620            self._save_queue_to_cache,
1621            queue_id,
1622            task_id=f"save_queue_cache_{queue_id}",
1623        )
1624
1625    def index_by_id(self, queue_id: str, queue_item_id: str) -> int | None:
1626        """Get index by queue_item_id."""
1627        if (queue_data := self._queue_data.get(queue_id)) is None:
1628            return None
1629        for index, item in enumerate(queue_data.items):
1630            if item.queue_item_id == queue_item_id:
1631                return index
1632        return None
1633
1634    async def get_tracks_for_playback(self, media_item: MediaItemType) -> list[Track]:
1635        """
1636        Return the playable tracks a media item resolves to, honoring the user's selection prefs.
1637
1638        :param media_item: The media item to resolve to playable tracks.
1639        """
1640        return await self._media_resolver.get_tracks_for_playback(media_item)
1641
1642    async def get_playlist_tracks(
1643        self, playlist: Playlist, start_item: str | None = None, sort_by: str | None = None
1644    ) -> list[PlaylistPlayableItem]:
1645        """
1646        Return the playable tracks for a playlist, honoring the user's selection prefs.
1647
1648        :param playlist: The playlist to resolve.
1649        :param start_item: Optional item URI to start the playlist from.
1650        :param sort_by: Optional sort key for the returned tracks.
1651        """
1652        return await self._media_resolver.get_playlist_tracks(playlist, start_item, sort_by)
1653
1654    async def get_dynamic_source_tracks(self, item: MediaItemType) -> list[Track]:
1655        """
1656        Return a fresh batch of tracks for a dynamic source (a dynamic playlist or radio station).
1657
1658        :param item: The dynamic playlist or radio station to fetch the next batch for.
1659        """
1660        return await self._media_resolver.get_dynamic_source_tracks(item)
1661
1662    def recency_windows(self) -> RecencyWindows:
1663        """Return the configured recency windows (a global setting; used for recency-aware gating)."""
1664        return self._smart_shuffle.windows()
1665
1666    async def player_media_from_queue_item(self, queue_item: QueueItem) -> PlayerMedia:
1667        """
1668        Parse PlayerMedia from QueueItem.
1669
1670        :param queue_item: The queue item to create media from.
1671        """
1672        queue_data = self._queue_data[queue_item.queue_id]
1673        stream_duration: int | None = None
1674        if queue_item.streamdetails:
1675            # prefer netto duration
1676            duration = queue_item.streamdetails.duration or queue_item.duration
1677            if duration and queue_item.streamdetails.seek_position:
1678                # the audio handed to the player starts at the seek position, so it is
1679                # shorter than the media item itself. seeking to (or past) the end
1680                # leaves no stream to describe, so the full length is kept instead.
1681                remaining = int(duration - queue_item.streamdetails.seek_position)
1682                stream_duration = remaining if remaining > 0 else None
1683        else:
1684            duration = queue_item.duration
1685        if queue_data.session_id is None:
1686            raise InvalidDataError("Queue session_id is None")
1687        media = PlayerMedia(
1688            uri=queue_item.uri,
1689            media_type=queue_item.media_type,
1690            title=queue_item.name,
1691            image_url=MASS_LOGO_ONLINE,
1692            duration=duration,
1693            stream_duration=stream_duration,
1694            source_id=queue_item.queue_id,
1695            queue_item_id=queue_item.queue_item_id,
1696            queue_session_id=queue_data.session_id,
1697            custom_data={
1698                "original_uri": queue_item.uri,
1699            },
1700        )
1701        if queue_item.media_item:
1702            media.title = queue_item.media_item.name
1703            media.artist = getattr(queue_item.media_item, "artist_str", "")
1704            media.album = (
1705                album.name if (album := getattr(queue_item.media_item, "album", None)) else ""
1706            )
1707            if queue_item.image:
1708                # the image format needs to be 512x512 jpeg for maximum compatibility with players
1709                # we prefer the imageproxy on the streamserver here because this request is sent
1710                # to the player itself which may not be able to reach the regular webserver
1711                media.image_url = self.mass.metadata.get_image_url(
1712                    queue_item.image, size=512, image_format="jpeg", prefer_stream_server=True
1713                )
1714        return media
1715
1716    def get_next_item(self, queue_id: str, cur_index: int | str) -> QueueItem | None:
1717        """Return next QueueItem for given queue."""
1718        index: int
1719        if isinstance(cur_index, str):
1720            resolved_index = self.index_by_id(queue_id, cur_index)
1721            if resolved_index is None:
1722                return None  # guard
1723            index = resolved_index
1724        else:
1725            index = cur_index
1726        # At this point index is guaranteed to be int
1727        for skip in range(5):
1728            if (next_index := self._get_next_index(queue_id, index + skip)) is None:
1729                break
1730            next_item = self.get_item(queue_id, next_index)
1731            if next_item is None:
1732                continue
1733            if not next_item.available:
1734                # ensure that we skip unavailable items (set by load_next track logic)
1735                continue
1736            return next_item
1737        return None
1738
1739    def store_sources(self, queue: PlayerQueue, items: list[MediaItemType]) -> None:
1740        """
1741        Hold the queue's full dynamic-source items server-side and project them onto `sources`.
1742
1743        :param queue: The queue whose sources are being set.
1744        :param items: The full source media items; an empty list clears the queue's sources.
1745        """
1746        self._queue_data[queue.queue_id].source_items = items
1747        # keep every occurrence server-side (a source added more than once weights it up in the
1748        # managed pool), but expose only the distinct container sources on the wire for clients to
1749        # show. Individual items (tracks, live radio streams, podcast episodes, ...) are omitted; see
1750        # `_WIRE_SOURCE_MEDIA_TYPES`. Autoplay/pool refill reads the full `source_items` above, not
1751        # this projected list, so it is unaffected.
1752        seen: set[str] = set()
1753        sources: list[ItemMapping] = []
1754        for item in items:
1755            if item.media_type not in _WIRE_SOURCE_MEDIA_TYPES and not is_dynamic_source(item):
1756                continue
1757            mapping = ItemMapping.from_item(item)
1758            if mapping.uri and mapping.uri in seen:
1759                continue
1760            if mapping.uri:
1761                seen.add(mapping.uri)
1762            sources.append(mapping)
1763        queue.sources = sources
1764        # release any materialized finite-source state whose source is no longer present
1765        self._managed_pool.retain(
1766            queue.queue_id, {item.uri for item in items if item.uri is not None}
1767        )
1768
1769    async def _save_queue_to_cache(self, queue_id: str) -> None:
1770        """Persist the queue's state (and its items when changed) to the cache."""
1771        if (queue_data := self._queue_data.get(queue_id)) is None:
1772            return
1773        try:
1774            # persistent so a cache clear/reset does not wipe the user's queues; the default
1775            # expiration still applies but is refreshed on every write. Skip the state write when its
1776            # persist-worthy content is unchanged (i.e. only playback progress advanced).
1777            state = queue_data.to_cache()
1778            significant = queue_data.cache_significant(state)
1779            if significant != queue_data.last_saved_state:
1780                await self.mass.cache.set(
1781                    key=queue_id,
1782                    data=state,
1783                    provider=self.domain,
1784                    category=CACHE_CATEGORY_PLAYER_QUEUE_STATE,
1785                    persistent=True,
1786                )
1787                queue_data.last_saved_state = significant
1788            if queue_data.items_cache_dirty:
1789                # only cache items with a valid media_item
1790                await self.mass.cache.set(
1791                    key=queue_id,
1792                    data=queue_data.items_to_cache(),
1793                    provider=self.domain,
1794                    category=CACHE_CATEGORY_PLAYER_QUEUE_ITEMS,
1795                    persistent=True,
1796                )
1797                queue_data.items_cache_dirty = False
1798        except Exception as err:
1799            self.logger.warning("Failed to persist the queue for %s - %s", queue_id, err)
1800
1801    def _check_player_permission(self, queue_id: str) -> None:
1802        """
1803        Check if the current user has permission to control this player/queue.
1804
1805        :param queue_id: The queue/player ID to check access for.
1806        :raises InsufficientPermissions: If the user lacks access.
1807        """
1808        current_user = get_current_user()
1809        if (
1810            current_user
1811            and current_user.player_filter
1812            and queue_id not in current_user.player_filter
1813        ):
1814            msg = f"{current_user.username} does not have access to player {queue_id}"
1815            raise InsufficientPermissions(msg)
1816
1817    @handle_play_action
1818    async def _handle_play(self, queue_id: str) -> None:
1819        """Handle play without acquiring the queue lock."""
1820        queue_player = self.mass.players.get_player(queue_id, True)
1821        if queue_player is None:
1822            raise PlayerUnavailableError(f"Player {queue_id} is not available")
1823        if (queue := self.get(queue_id)) and queue.active and queue.state == PlaybackState.PAUSED:
1824            # forward the actual play/unpause command to the player,
1825            # holding the action until the player confirms it resumed playback
1826            async with self.mass.players.wait_for_player_update(
1827                queue_id,
1828                attribute_name="playback_state",
1829                attribute_value=PlaybackState.PLAYING,
1830                timeout=PLAYBACK_START_TIMEOUT,
1831            ):
1832                await queue_player.play()
1833            return
1834        # player is not paused, perform resume instead
1835        await self.resume(queue_id)
1836
1837    def _set_transitioning(self, queue_id: str, value: bool) -> None:
1838        """Mark (or clear) whether a queue is mid-transition (no-op if it is not registered)."""
1839        if (queue_data := self._queue_data.get(queue_id)) is not None:
1840            queue_data.transitioning = value
1841
1842    def _clear(self, queue_id: str, skip_stop: bool = False) -> None:
1843        """Drop the queue's items and playback position, leaving user settings untouched."""
1844        queue = self._queue_data[queue_id].queue
1845        self.mass.streams.audio_processing.clear(queue_id)
1846        self.store_sources(queue, [])
1847        if queue.is_dynamic:
1848            # Dynamic sources impose shuffle, so clearing the source clears that shuffle too.
1849            queue.shuffle_enabled = False
1850        queue.is_dynamic = False
1851        # dropping the dynamic source changes what smart shuffle resolves to, so the derived
1852        # flag has to follow or clients keep showing a smart mix on a plain queue
1853        queue.smart_shuffle_active = self.is_smart_shuffle_active(queue)
1854        queue.ended = False
1855        if queue.state != PlaybackState.IDLE and not skip_stop:
1856            self.mass.create_task(self.stop(queue_id))
1857        queue.current_index = None
1858        queue.current_item = None
1859        queue.elapsed_time = 0
1860        queue.elapsed_time_last_updated = time.time()
1861        queue.index_in_buffer = None
1862        self.mass.create_task(self._cleanup_queue_audio_data(queue_id))
1863        self.update_items(queue_id, [])
1864
1865    def _reset_shuffle(self, queue_id: str) -> None:
1866        """Switch shuffle off."""
1867        queue = self._queue_data[queue_id].queue
1868        if not queue.shuffle_enabled:
1869            return
1870        queue.shuffle_enabled = False
1871        queue.smart_shuffle_active = self.is_smart_shuffle_active(queue)
1872        self.signal_update(queue_id)
1873
1874    async def _apply_shuffle(
1875        self, queue_id: str, option: QueueOption, shuffle: bool | None
1876    ) -> None:
1877        """
1878        Settle the queue's shuffle state for a play command before its items are resolved.
1879
1880        :param queue_id: The queue the media is played on.
1881        :param option: The enqueue option this command resolved to.
1882        :param shuffle: The state to put the queue's shuffle in; None to leave it as it is.
1883        """
1884        queue = self._queue_data[queue_id].queue
1885        if queue.is_dynamic and option in (
1886            QueueOption.PLAY,
1887            QueueOption.REPLACE,
1888            QueueOption.REPLACE_NEXT,
1889        ):
1890            # These are the options that replace the queue's sources, so the smart mix may be on
1891            # its way out - and its shuffle is never the user's own (a dynamic queue's toggle is
1892            # locked), so it must not outlive the source that imposed it. Recorded directly
1893            # because set_shuffle refuses a queue that is still a smart mix, and the items are
1894            # resolved against this flag. The state is provisional until the sources are known:
1895            # `_enter_dynamic_mode` forces shuffle back on if the queue stays dynamic.
1896            if option == QueueOption.REPLACE_NEXT:
1897                # staging leaves the shuffle the user chose alone, so it never carries a request
1898                # of its own to honour here
1899                queue.shuffle_enabled = False
1900            else:
1901                queue.shuffle_enabled = bool(shuffle)
1902            return
1903        if shuffle is None or option not in (QueueOption.PLAY, QueueOption.REPLACE):
1904            # nothing to settle: the media brings no order of its own to protect, or the option
1905            # only stages items for later and leaves the queue's shuffle state alone
1906            return
1907        if queue.shuffle_enabled == shuffle:
1908            return
1909        # routed through set_shuffle so switching shuffle off also restores the order of
1910        # the items that stay in the queue: a play keeps them, and a tail left in shuffled
1911        # order behind a queue that now reads unshuffled would contradict its own flag
1912        await self.set_shuffle(queue_id, shuffle)
1913
1914    async def _apply_local_shuffle(self, queue_id: str, shuffle_enabled: bool) -> None:
1915        """
1916        Record the queue's shuffle state and re-order the un-played tail accordingly.
1917
1918        :param queue_id: The queue to apply the shuffle state to.
1919        :param shuffle_enabled: The shuffle state to record and apply to the tail.
1920        """
1921        queue = self._queue_data[queue_id].queue
1922        queue.shuffle_enabled = shuffle_enabled
1923        queue.smart_shuffle_active = self.is_smart_shuffle_active(queue)
1924        queue_items = self._queue_data[queue_id].items
1925        cur_index = (
1926            queue.index_in_buffer if queue.index_in_buffer is not None else queue.current_index
1927        )
1928        if cur_index is not None:
1929            next_index = cur_index + 1
1930            next_items = queue_items[next_index:]
1931        else:
1932            next_items = []
1933            next_index = 0
1934        if not shuffle_enabled:
1935            # shuffle disabled, try to restore original sort order of the remaining items
1936            next_items.sort(key=lambda x: x.sort_index, reverse=False)
1937        await self.load(
1938            queue_id=queue_id,
1939            queue_items=next_items,
1940            insert_at_index=next_index,
1941            keep_remaining=False,
1942            shuffle=shuffle_enabled,
1943        )
1944