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