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