music-assistant-server

1"""Yandex Ynison plugin provider for Music Assistant."""
2
3from __future__ import annotations
4
5import asyncio
6import hashlib
7import random
8import time
9from collections.abc import AsyncGenerator, Callable
10from contextlib import aclosing, suppress
11from dataclasses import dataclass
12from typing import TYPE_CHECKING, Any, ClassVar, Literal, cast
13
14from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
15from music_assistant_models.enums import (
16    ConfigEntryType,
17    ContentType,
18    EventType,
19    MediaType,
20    PlaybackState,
21    ProviderFeature,
22    ProviderType,
23    SourceControl,
24    StreamType,
25)
26from music_assistant_models.errors import (
27    InvalidDataError,
28    LoginFailed,
29    MediaNotFoundError,
30    PlayerCommandFailed,
31    UnsupportedFeaturedException,
32)
33from music_assistant_models.media_items import AudioSource, ProviderMapping
34from music_assistant_models.streamdetails import StreamDetails, StreamMetadata
35from ya_passport_auth import SecretStr
36from ya_passport_auth.ma import BorrowedCredentialSource
37
38from music_assistant.controllers.streams.constants import STREAM_SLOT_PLAYBACK_WAIT_TIMEOUT
39from music_assistant.helpers.ffmpeg import get_ffmpeg_stream
40from music_assistant.helpers.throttle_retry import BYPASS_THROTTLER, ThrottlerManager
41from music_assistant.models.plugin import PluginProvider, SourceControlValue
42
43from .auth import refresh_music_token
44from .constants import (
45    CONF_ALLOW_PLAYER_SWITCH,
46    CONF_DEVICE_ID,
47    CONF_MASS_PLAYER_ID,
48    CONF_OUTPUT_BIT_DEPTH,
49    CONF_OUTPUT_SAMPLE_RATE,
50    CONF_PUBLISH_NAME,
51    CONF_TOKEN,
52    CONF_X_TOKEN,
53    CONF_YM_INSTANCE,
54    DEFAULT_DISPLAY_NAME,
55    OUTPUT_AUTO,
56    PLAYER_ID_AUTO,
57    YANDEX_MUSIC_CONF_QUALITY,
58    YANDEX_MUSIC_LOSSLESS_QUALITIES,
59    YM_INSTANCE_OWN,
60)
61from .protocols import YandexMusicProviderLike
62from .streaming import (
63    PCM_LOSSLESS_PARAMS,
64    PCM_LOSSY_PARAMS,
65    PROBE_ARGS,
66    make_pcm_format,
67)
68from .ynison_client import (
69    YnisonClient,
70    YnisonDeviceInfo,
71    YnisonSendError,
72    YnisonState,
73    generate_device_id,
74    make_version_block,
75)
76
77if TYPE_CHECKING:
78    from music_assistant_models.config_entries import ProviderConfig
79    from music_assistant_models.event import MassEvent
80    from music_assistant_models.media_items import AudioFormat
81    from music_assistant_models.provider import ProviderManifest
82
83    from music_assistant.mass import MusicAssistant
84
85# How often (seconds) to sync progress to MA UI and Ynison.
86_PROGRESS_SYNC_INTERVAL = 5.0
87
88# Grace window after our own REPLACE/seek during which incoming Ynison
89# progress updates are treated as our own echo (not a user seek).
90_ECHO_GRACE_PERIOD = 3.0
91
92# Bound on the synchronous pre-fetch in _prefetch_format_for_track. A slow
93# pre-fetch is treated like a failed one — fall back to the current format
94# and let the in-stream `_get_stream_details_with_retry` handle retries.
95_PREFETCH_FORMAT_TIMEOUT = 2.5
96
97# Idempotency cache TTL for outbound peer-commands.
98_COMMAND_IDEMPOTENCY_TTL = 1.0
99
100# stable id for the single AudioSource this provider exposes;
101# combined with the provider instance_id this forms the persistent uri
102AUDIO_SOURCE_ID = "main"
103
104# Retry settings for transient Yandex API failures
105_API_MAX_RETRIES = 3
106_API_INITIAL_BACKOFF = 2.0
107_API_MAX_BACKOFF = 30.0
108
109# Cache TTL for stream details (seconds)
110_STREAM_DETAILS_CACHE_TTL = 300  # 5 minutes
111
112# In-memory music-token cache TTL (seconds). Yandex music tokens live ~60 min;
113# 50 min leaves 10 min headroom before the server would reject them. Tied to
114# the borrow-mode-with-only-x_token + 401-storm path described in spec 0004.
115_MUSIC_TOKEN_TTL_S = 50 * 60
116
117# Maximum number of distinct x_token entries kept in the own-mode music-token
118# cache (borrow mode caches inside BorrowedCredentialSource). 4 keeps headroom
119# for an x_token rotation with one refresh in flight.
120_MUSIC_TOKEN_CACHE_MAX = 4
121
122# Accepted non-auto values for output format overrides; mirrors the options
123# offered in CONF_OUTPUT_SAMPLE_RATE / CONF_OUTPUT_BIT_DEPTH config entries.
124# Used defensively to reject stale/tampered values without raising.
125_VALID_SAMPLE_RATES: frozenset[str] = frozenset({"44100", "48000", "96000"})
126_VALID_BIT_DEPTHS: frozenset[str] = frozenset({"16", "24"})
127
128
129class _StreamOwnerMismatchError(InvalidDataError):
130    """Raised when linked-provider stream details belong to another instance."""
131
132
133@dataclass(frozen=True)
134class _CachedToken:
135    """
136    Music token entry in the in-memory cache.
137
138    `expires_monotonic` is compared against the provider's `_now()` seam.
139    """
140
141    token: SecretStr
142    expires_monotonic: float
143
144
145def _hash_x_token(x_token: str) -> str:
146    """
147    Return the SHA-256 hex digest of an x_token, used as cache key.
148
149    The raw x_token is never stored in dict keys (defence-in-depth against
150    accidental log / dump leakage of the cache structure).
151    """
152    return hashlib.sha256(x_token.encode("utf-8")).hexdigest()
153
154
155class YandexYnisonProvider(PluginProvider):
156    """Implementation of the Yandex Music Connect (Ynison) Plugin."""
157
158    # PluginProvider base does not declare `is_streaming_provider`; MA's
159    # audio-analysis path raises AttributeError for live sources without
160    # an explicit opt-out. Analysing transient external-source tracks
161    # buys nothing.
162    is_streaming_provider: bool = False
163
164    @property
165    def instance_name_postfix(self) -> str | None:
166        """Return display name as instance postfix for multi-instance setups."""
167        name = self._display_name
168        return name if name != DEFAULT_DISPLAY_NAME else None
169
170    def __init__(
171        self,
172        mass: MusicAssistant,
173        manifest: ProviderManifest,
174        config: ProviderConfig,
175        supported_features: set[ProviderFeature],
176    ) -> None:
177        """Initialize the Ynison plugin provider."""
178        super().__init__(mass, manifest, config, supported_features)
179
180        # Setup identity and playback options
181        self._default_player_id: str = (
182            cast("str", self.get_setup_value(CONF_MASS_PLAYER_ID)) or PLAYER_ID_AUTO
183        )
184        allow_switch_value = self.config.get_value(CONF_ALLOW_PLAYER_SWITCH)
185        self._allow_player_switch: bool = (
186            cast("bool", allow_switch_value) if allow_switch_value is not None else True
187        )
188        self._cfg_sample_rate: str = (
189            cast("str", self.config.get_value(CONF_OUTPUT_SAMPLE_RATE)) or OUTPUT_AUTO
190        )
191        self._cfg_bit_depth: str = (
192            cast("str", self.config.get_value(CONF_OUTPUT_BIT_DEPTH)) or OUTPUT_AUTO
193        )
194        self._display_name: str = (
195            cast("str", self.get_setup_value(CONF_PUBLISH_NAME)) or DEFAULT_DISPLAY_NAME
196        )
197
198        # Token source — None = own (manually entered CONF_TOKEN);
199        # otherwise the instance_id of a linked yandex_music provider to borrow from.
200        ym_instance_value = cast("str | None", self.get_setup_value(CONF_YM_INSTANCE))
201        self._ym_instance_id: str | None = (
202            ym_instance_value
203            if ym_instance_value and ym_instance_value != YM_INSTANCE_OWN
204            else None
205        )
206        # Borrow mode: read-only credential source over the linked
207        # yandex_music instance (shared auth layer). The owner stays the
208        # single writer/rotator of persisted credentials; minted music
209        # tokens are cached in-memory inside the source (TTL + LRU +
210        # coalesced refreshes per its spec).
211        self._borrow_source: BorrowedCredentialSource | None = (
212            BorrowedCredentialSource(self.mass, self._ym_instance_id)
213            if self._ym_instance_id is not None
214            else None
215        )
216
217        # Device ID — persist in config so re-registration uses the same ID
218        device_id = cast("str | None", self.config.get_value(CONF_DEVICE_ID))
219        if not device_id:
220            device_id = generate_device_id()
221            self._update_config_value(CONF_DEVICE_ID, device_id)
222        self._device_id: str = device_id
223
224        # Runtime state
225        self._active_player_id: str | None = None
226        self._ynison: YnisonClient | None = None
227        self._runner_task: asyncio.Task[None] | None = None
228        self._on_unload_callbacks: list[Callable[..., None]] = []
229        self._yandex_provider: YandexMusicProviderLike | None = None
230        self._current_streaming_track_id: str | None = None
231        self._track_changed_event = asyncio.Event()
232        self._stream_stop_event = asyncio.Event()
233        self._seek_position_ms: int = 0
234        self._seek_grace_until: float = 0.0
235        self._last_player_update_time: float = 0.0
236        self._actual_duration_ms: int = 0
237        self._prefetched_list: list[dict[str, Any]] | None = None
238        self._prefetch_task: asyncio.Task[Any] | None = None
239        self._normalized_params: dict[str, Any] = PCM_LOSSY_PARAMS
240        self._normalized_format: AudioFormat = make_pcm_format(PCM_LOSSY_PARAMS)
241
242        # Rate limiter for Yandex API calls (max 2 req/s)
243        self._api_throttler = ThrottlerManager(rate_limit=2, period=1.0)
244
245        # Progress tracking — byte counter is the single source of truth
246        # during active streaming; Ynison echoes are detected via
247        # YnisonState.last_update_is_echo and ignored.
248        self._streaming_progress_ms: int = 0
249
250        # AudioSource MediaItem + per-stream state
251        self._stream_metadata = StreamMetadata(
252            title=f"Yandex Music Connect | {self._display_name}",
253        )
254        self._audio_source = AudioSource(
255            item_id=AUDIO_SOURCE_ID,
256            provider=self.instance_id,
257            name=self.name,
258            provider_mappings={
259                ProviderMapping(
260                    item_id=AUDIO_SOURCE_ID,
261                    provider_domain=self.domain,
262                    provider_instance=self.instance_id,
263                    # Fresh AudioFormat copy: AudioFormat is mutable and MA's
264                    # FFMpeg._log_reader_task sets `input_format.codec_type`
265                    # in-place. Sharing `self._normalized_format` here would
266                    # let that mutation leak into the ProviderMapping and into
267                    # later StreamDetails snapshots.
268                    audio_format=make_pcm_format(self._normalized_params),
269                )
270            },
271            can_play_pause=False,
272            can_seek=False,
273            can_next_previous=False,
274            exclusive=True,
275            allow_external_trigger=True,
276        )
277        # _in_use_by_player tracks the queue currently consuming our stream
278        self._in_use_by_player: str | None = None
279        # _active_session_id is the controller-provided token for the current
280        # stream request — used to reject stale on_source_unselected callbacks
281        # after a same-queue reconnect supersedes the previous request.
282        self._active_session_id: str | None = None
283
284        # Idempotency cache for outbound peer-commands. Suppresses duplicate
285        # (action, key) pairs inside `_COMMAND_IDEMPOTENCY_TTL` — protects
286        # against echo-storms where the same Ynison broadcast lands on our
287        # state-handler twice in quick succession.
288        self._command_idempotency: dict[tuple[str, str | None], float] = {}
289
290        # "Ynison paused us externally — expect a resume that needs
291        # `play_media` re-issuance." Set in `_pause_playback`, read in
292        # `_activate_playback`. Survives a stray `_stream_stop_event` clear
293        # independent of the stop signal (which covers non-pause stop reasons).
294        self._externally_paused: bool = False
295
296        # In-memory music-token cache keyed by SHA-256(x_token). 50-min TTL,
297        # 4-entry LRU. Coalesces concurrent refresh attempts via a single
298        # asyncio.Lock so a reconnect storm makes at most one Passport call.
299        # `_now` is a seam for tests to advance the clock.
300        self._token_cache: dict[str, _CachedToken] = {}
301        self._token_refresh_lock = asyncio.Lock()
302        self._now: Callable[[], float] = time.monotonic
303
304    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
305        """
306        Return Config entries to configure this provider.
307
308        Account, player and device identity are collected by the interactive setup flow;
309        only runtime playback options live here.
310        """
311        return (
312            ConfigEntry(
313                key=CONF_ALLOW_PLAYER_SWITCH,
314                type=ConfigEntryType.BOOLEAN,
315                default_value=True,
316            ),
317            ConfigEntry(
318                key=CONF_OUTPUT_SAMPLE_RATE,
319                type=ConfigEntryType.STRING,
320                default_value=OUTPUT_AUTO,
321                options=[
322                    ConfigValueOption(OUTPUT_AUTO),
323                    ConfigValueOption("44100"),
324                    ConfigValueOption("48000"),
325                    ConfigValueOption("96000"),
326                ],
327                advanced=True,
328            ),
329            ConfigEntry(
330                key=CONF_OUTPUT_BIT_DEPTH,
331                type=ConfigEntryType.STRING,
332                default_value=OUTPUT_AUTO,
333                options=[
334                    ConfigValueOption(OUTPUT_AUTO),
335                    ConfigValueOption("16"),
336                    ConfigValueOption("24"),
337                ],
338                advanced=True,
339            ),
340            ConfigEntry(
341                key=CONF_DEVICE_ID,
342                type=ConfigEntryType.STRING,
343                hidden=True,
344                required=False,
345            ),
346        )
347
348    # ------------------------------------------------------------------
349    # Provider lifecycle
350    # ------------------------------------------------------------------
351
352    async def handle_async_init(self) -> None:
353        """Handle async initialization of the provider."""
354        if self._ym_instance_id is not None:
355            self.logger.info(
356                "Borrowing credentials from yandex_music instance '%s'",
357                self._ym_instance_id,
358            )
359        else:
360            self.logger.info("Using manually configured Yandex Music token (no auto-refresh)")
361        token = await self._resolve_token()
362
363        device_info = YnisonDeviceInfo(
364            device_id=self._device_id,
365            title=self._display_name,
366        )
367
368        self._ynison = YnisonClient(
369            token=token,
370            device_info=device_info,
371            on_state_update=self._handle_ynison_state,
372            logger=self.logger,
373            on_auth_failure=self._refresh_ynison_token,
374        )
375
376        self._runner_task = self.mass.create_task(self._ynison.connect())
377
378        # Subscribe to provider events to detect linked yandex_music provider
379        self._on_unload_callbacks.append(
380            self.mass.subscribe(
381                self._on_provider_event,
382                EventType.PROVIDERS_UPDATED,
383            )
384        )
385        # Initial check for matching provider
386        self.mass.create_task(self._check_yandex_provider_match())
387
388    async def unload(self, is_removed: bool = False) -> None:
389        """Handle close/cleanup of the provider."""
390        if self._prefetch_task and not self._prefetch_task.done():
391            self._prefetch_task.cancel()
392            with suppress(asyncio.CancelledError):
393                await self._prefetch_task
394
395        if self._ynison:
396            await self._ynison.disconnect()
397
398        if self._runner_task and not self._runner_task.done():
399            self._runner_task.cancel()
400            with suppress(asyncio.CancelledError):
401                await self._runner_task
402
403        for callback in self._on_unload_callbacks:
404            with suppress(KeyError):
405                callback()
406
407    async def get_audio_sources(self) -> list[AudioSource]:
408        """Return the AudioSources this plugin currently exposes."""
409        return [self._audio_source]
410
411    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
412        """
413        Return StreamDetails for streaming the Yandex Music Connect audio.
414
415        Side-effect-free: ownership is claimed in on_source_selected (which the
416        streams controller fires before this method on the actual stream
417        request). Keeping this idempotent means preload paths like
418        player_queues._load_item can fetch streamdetails without claiming the
419        source and blocking a subsequent cross-queue handoff.
420        """
421        if item_id != AUDIO_SOURCE_ID:
422            raise MediaNotFoundError(f"Unknown AudioSource: {item_id}")
423        return StreamDetails(
424            provider=self.instance_id,
425            item_id=item_id,
426            # Fresh AudioFormat copy per call: MA's ffmpeg mutates
427            # input_format.codec_type in place, so a shared instance would
428            # propagate that mutation into future stream-details snapshots.
429            audio_format=make_pcm_format(self._normalized_params),
430            media_type=MediaType.AUDIO_SOURCE,
431            stream_type=StreamType.CUSTOM,
432            stream_metadata=self._stream_metadata,
433        )
434
435    async def on_source_control(
436        self,
437        source_id: str,
438        action: SourceControl,
439        value: SourceControlValue = None,
440    ) -> None:
441        """Proxy playback control commands to Yandex via the linked Yandex Music provider."""
442        if source_id != AUDIO_SOURCE_ID:
443            return
444        if action == SourceControl.PLAY:
445            await self._on_play()
446        elif action == SourceControl.PAUSE:
447            await self._on_pause()
448        elif action == SourceControl.NEXT:
449            await self._on_next()
450        elif action == SourceControl.PREVIOUS:
451            await self._on_previous()
452        elif (
453            action == SourceControl.SEEK
454            # tolerate float positions from internal callers; bool is an int
455            # subclass, so a misrouted toggle must not become a 1-second seek
456            and isinstance(value, (int, float))
457            and not isinstance(value, bool)
458        ):
459            await self._on_seek(int(value))
460
461    async def get_audio_stream(  # noqa: PLR0915
462        self, streamdetails: StreamDetails, seek_position: int = 0
463    ) -> AsyncGenerator[bytes]:
464        """
465        Return continuous audio stream following Ynison track changes.
466
467        Streams the current track, then waits for track changes and streams
468        the next track automatically. Runs until the source is deselected.
469
470        The PCM format is frozen at session start to match what the outer
471        ffmpeg captured from ``self._normalized_format``.  If
472        ``_update_normalized_format()`` fires mid-session (e.g. a provider
473        reload), the new format takes effect only on the *next* session —
474        preventing bit-depth/sample-rate mismatches that cause noise.
475        """
476        self._stream_stop_event.clear()
477        # snapshot the consumer at session start; the rest of this generator
478        # treats the queue_id as the player_id (they are the same by convention).
479        # The lock may legitimately be empty here — MA's `_load_item` preload
480        # path drives the generator to fill an initial audio buffer BEFORE
481        # `on_source_selected` has been dispatched, so `_in_use_by_player` is
482        # still None on that call. `had_claim` records whether a lock was
483        # already in force at entry; only in that case do we enforce
484        # cross-session invariants on the loop and the `finally` cleanup.
485        player_id = self._in_use_by_player or ""
486        had_claim = self._in_use_by_player is not None
487        # Snapshot the active session id too so a same-queue reconnect (which
488        # updates _active_session_id but not _in_use_by_player) is treated as a
489        # superseding session: the loop exits early, and the finally clear
490        # below skips the release so it doesn't clobber the new claim.
491        captured_session_id = self._active_session_id
492
493        # MA's streams controller may pass a non-zero seek_position (e.g. a
494        # resume initiated through a path that does NOT go through Ynison and
495        # therefore did not set `_seek_position_ms`). Honor it as the seed for
496        # the upcoming track. The Ynison-driven seek path (`_activate_playback`
497        # / `_on_seek`) keeps writing `_seek_position_ms` directly, which
498        # subsequent track iterations consume — only the seed differs.
499        if seek_position > 0 and self._seek_position_ms == 0:
500            self._seek_position_ms = seek_position * 1000
501
502        # Freeze format for this streaming session so every inner ffmpeg
503        # produces data matching the outer ffmpeg's captured input_format.
504        session_params: dict[str, Any] = dict(self._normalized_params)
505        session_fmt: AudioFormat = make_pcm_format(session_params)
506
507        try:
508            while not self._stream_stop_event.is_set() and (
509                # Preload path: no claim was active at entry — drive the
510                # loop purely off Ynison state and the stop event.
511                not had_claim or not self._session_lost(player_id, captured_session_id)
512            ):
513                if not self._ynison or not self._ynison.state.current_track_id:
514                    # Wait for a track to appear
515                    self._track_changed_event.clear()
516                    try:
517                        await asyncio.wait_for(self._track_changed_event.wait(), timeout=30.0)
518                    except TimeoutError:
519                        continue
520                    continue
521
522                # Clear event before reading state so any subsequent update
523                # re-sets the event instead of being silently cleared.
524                self._track_changed_event.clear()
525                track_id = self._ynison.state.current_track_id
526                self._current_streaming_track_id = track_id
527
528                # `_pause_playback` set the stop event; finalize.
529                if self._ynison.state.is_paused:
530                    return
531
532                if not self._yandex_provider:
533                    self.logger.warning(
534                        "No linked Yandex Music provider — cannot stream track %s", track_id
535                    )
536                    self._stream_stop_event.set()
537                    if self._in_use_by_player == player_id:
538                        await self.mass.players.cmd_stop(player_id)
539                    return
540
541                # Stream the current track
542                seek_ms = self._seek_position_ms
543                self._seek_position_ms = 0
544                bytes_yielded = 0
545                self._streaming_progress_ms = seek_ms
546                last_progress_sync = time.monotonic()
547
548                track_fmt = make_pcm_format(session_params)
549                track_stream = self._stream_track(
550                    track_id, seek_ms=seek_ms, session_params=session_params
551                )
552                # aclosing: breaking out below must finalize the generator right away,
553                # otherwise the linked provider's stream slot stays charged until GC.
554                async with aclosing(track_stream):
555                    async for chunk in track_stream:
556                        yield chunk
557                        bytes_yielded += len(chunk)
558                        now_mono = time.monotonic()
559                        if now_mono - last_progress_sync >= _PROGRESS_SYNC_INTERVAL:
560                            last_progress_sync = now_mono
561                            await self._sync_progress(
562                                seek_ms, bytes_yielded, player_id, session_fmt
563                            )
564                        if (
565                            self._track_changed_event.is_set()
566                            or self._stream_stop_event.is_set()
567                            or (had_claim and self._session_lost(player_id, captured_session_id))
568                        ):
569                            break
570
571                # Align to PCM frame boundary — prevents misalignment in MA's
572                # downstream ffmpeg when a track stream is interrupted mid-chunk.
573                # We pad with zeros (can't un-yield bytes already sent downstream).
574                frame_size = (track_fmt.bit_depth // 8) * track_fmt.channels
575                if frame_size > 0:
576                    excess = bytes_yielded % frame_size
577                    if excess:
578                        yield b"\x00" * (frame_size - excess)
579
580                # Don't clear _current_streaming_track_id yet — keep it set
581                # during advance/wait so Ynison echo of the same track doesn't
582                # trigger a false track-change detection in _activate_playback.
583
584                if self._stream_stop_event.is_set():
585                    break
586
587                # Differentiate "track finished naturally" from "inner loop
588                # broke out early". Signalling completion on an
589                # interrupted track makes Yandex auto-advance the queue —
590                # surfaces as an unwanted skip on pause / handoff.
591                broke_for_pause = self._ynison is not None and self._ynison.state.is_paused
592                broke_for_session_change = had_claim and self._session_lost(
593                    player_id, captured_session_id
594                )
595                natural_end = (
596                    not self._track_changed_event.is_set()
597                    and not broke_for_pause
598                    and not broke_for_session_change
599                    and self._ynison is not None
600                )
601                if natural_end:
602                    self.logger.info("Track %s finished, advancing to next", track_id)
603                    await self._signal_track_completion()
604                    if not await self._wait_for_track_change(track_id):
605                        self._stream_stop_event.set()
606                        break
607
608                # Clear before next iteration — the new track ID will be set at
609                # the top of the loop from the latest Ynison state.
610                self._current_streaming_track_id = None
611        finally:
612            # Release ownership only if THIS generator owned the claim at
613            # entry AND no one else has superseded it since. The double-guard
614            # protects against a same-queue reconnect refreshing the session
615            # id without changing the queue id; clearing the lock on the old
616            # generator's teardown would otherwise clobber the new session's
617            # claim. `had_claim` keeps the preload path from touching the lock
618            # at all (no claim ever existed to release).
619            if had_claim and not self._session_lost(player_id, captured_session_id):
620                self._in_use_by_player = None
621            self._current_streaming_track_id = None
622
623    async def on_source_selected(
624        self,
625        source_id: str,
626        player_id: str,
627        owner_player_id: str,
628        stream_session_id: str,
629    ) -> None:
630        """Handle callback when this AudioSource has been selected/started on a player."""
631        if source_id != AUDIO_SOURCE_ID or not player_id:
632            return
633
634        # Check if manual player switching is allowed
635        if not self._allow_player_switch:
636            current_target = self._get_target_player_id()
637            if player_id != current_target and current_target:
638                # Redirect to the configured target, but only once per
639                # idempotency window. The target may be a sendspin bridge /
640                # sync-group whose stream is consumed under a player id that
641                # never equals `current_target`, so each redirect re-triggers
642                # selection here. Re-issuing `play_media` on every rejection
643                # turns that into an unbounded AudioError storm; the raise
644                # below still aborts every wrong-player stream regardless.
645                if self._idempotent("source_redirect", current_target):
646                    self.logger.debug(
647                        "Player switching disabled, redirecting selection from %s to %s",
648                        player_id,
649                        current_target,
650                    )
651                    await self.mass.player_queues.play_media(
652                        current_target, str(self._audio_source.uri)
653                    )
654                msg = f"Player switching is disabled; source must remain on {current_target}"
655                raise RuntimeError(msg)
656
657        # Stop previous player if switching. The lock claim a few lines below
658        # replaces the previous queue's claim; the previous stream loop notices
659        # the queue change and exits cleanly.
660        if self._active_player_id and self._active_player_id != player_id:
661            prev_player_id = self._active_player_id
662            self.logger.info(
663                "Source selected on %s, stopping %s",
664                player_id,
665                prev_player_id,
666            )
667            try:
668                await self.mass.players.cmd_stop(prev_player_id)
669            except Exception as err:
670                self.logger.debug(
671                    "Failed to stop previous player %s: %s",
672                    prev_player_id,
673                    err,
674                )
675
676        # Claim ownership for this queue. The lock lives here (not in
677        # get_stream_details) so preload paths can fetch streamdetails without
678        # accidentally blocking a subsequent cross-queue handoff at the actual
679        # stream request.
680        self._in_use_by_player = owner_player_id
681        # Record this request's session id so a later on_source_unselected can
682        # tell whether it is the live teardown or a stale callback from a
683        # superseded same-queue request.
684        self._active_session_id = stream_session_id
685        self._active_player_id = player_id
686        self.logger.debug("Active player set to: %s", player_id)
687
688    async def on_source_unselected(
689        self, source_id: str, owner_player_id: str, stream_session_id: str
690    ) -> None:
691        """Release the queue-scoped exclusive claim when MA tears down the stream."""
692        if source_id != AUDIO_SOURCE_ID:
693            return
694        # Reject stale callbacks: only release if this is still the active
695        # session. A owner_player_id check alone is not sufficient — same-queue
696        # reconnects (player drops + reopens the same stream URL before the
697        # original request's finally fires) would otherwise let the old
698        # request's late callback clear the live claim of the new stream.
699        if self._active_session_id != stream_session_id:
700            return
701        self._active_session_id = None
702        if self._in_use_by_player == owner_player_id:
703            self._in_use_by_player = None
704
705    async def _wait_for_track_change(self, old_track_id: str, timeout: float = 30.0) -> bool:
706        """
707        Wait for Ynison to report a different track, ignoring echoes.
708
709        After _signal_track_completion sends update_playing_status, Ynison
710        echoes back the same track with updated progress.  Only return True
711        once current_track_id actually differs from old_track_id.
712        """
713        deadline = time.monotonic() + timeout
714        while not self._stream_stop_event.is_set():
715            # Check state BEFORE clearing the event.  Ynison may have already
716            # advanced between _signal_track_completion() returning and this
717            # method running; clearing first would drop the set() that went
718            # with the state update, leaving us to wait until timeout.
719            # Check is race-free: no await between the read and clear() below.
720            # None means empty/unreadable queue — treat as "not advanced."
721            if self._ynison:
722                current = self._ynison.state.current_track_id
723                if current is not None and current != old_track_id:
724                    return True
725            self._track_changed_event.clear()
726            remaining = deadline - time.monotonic()
727            if remaining <= 0:
728                break
729            try:
730                await asyncio.wait_for(self._track_changed_event.wait(), timeout=remaining)
731            except TimeoutError:
732                break
733        self.logger.info("No new track from Ynison after completion, stopping stream")
734        return False
735
736    async def _stream_track(
737        self,
738        track_id: str,
739        seek_ms: int = 0,
740        session_params: dict[str, Any] | None = None,
741    ) -> AsyncGenerator[bytes]:
742        """
743        Stream a single track, normalizing to fixed PCM via per-track ffmpeg.
744
745        Every track is decoded through its own ffmpeg process to produce a
746        fixed PCM output (s16le or s24le based on YM quality setting). This
747        ensures MA's single ffmpeg process never encounters mid-stream format
748        changes (codec, bit depth, sample rate).
749
750        *session_params* — frozen format dict from the enclosing
751        ``get_audio_stream()`` session.  Falls back to the current
752        ``_normalized_params`` when called outside a session.
753        """
754        provider = self._yandex_provider
755        if provider is None:
756            self.logger.warning(
757                "Linked Yandex Music provider unavailable — stopping track %s",
758                track_id,
759            )
760            self._stream_stop_event.set()
761            return
762        # In-flight stream fetch outranks unrelated 429 cooldowns:
763        # dropping a stream the user is actively trying to play is
764        # worse than risking another captcha. Prefetch deliberately
765        # stays throttled (see `_prefetch_format_for_track`).
766        bypass_token = BYPASS_THROTTLER.set(True)
767        try:
768            stream_details = await self._get_stream_details_with_retry(track_id, provider=provider)
769        except Exception:
770            self.logger.exception("Failed to get stream details for track %s", track_id)
771            self._stream_stop_event.set()
772            return
773        finally:
774            BYPASS_THROTTLER.reset(bypass_token)
775
776        if not self._linked_provider_is_current(provider):
777            self.logger.warning(
778                "Linked Yandex Music provider changed mid-stream — stopping track %s",
779                track_id,
780            )
781            self._stream_stop_event.set()
782            return
783
784        await self._update_metadata_from_stream(stream_details, seek_ms)
785        if not self._linked_provider_is_current(provider):
786            self.logger.warning(
787                "Linked Yandex Music provider changed while preparing track %s",
788                track_id,
789            )
790            self._stream_stop_event.set()
791            return
792
793        # No -re here: MA's realtime pacer is the single pacing authority for
794        # AudioSources. Pacing the decode a second time would pin it to realtime
795        # and forbid the small read-ahead that absorbs CDN jitter; back-pressure
796        # through the generator chain still bounds memory.
797        extra_input_args = list(PROBE_ARGS)
798        if seek_ms > 0:
799            extra_input_args += ["-ss", f"{seek_ms / 1000.0:.3f}"]
800
801        # Use session format when available, otherwise current normalized params
802        params = session_params if session_params is not None else self._normalized_params
803        out_fmt = make_pcm_format(params)
804        # Log the output rate + bit depth alongside the source format: with the
805        # passthrough fast path this PCM IS the delivered audio, so the line must
806        # let an operator read rate passthrough vs a resample, not just codec.
807        self.logger.info(
808            "Streaming track %s → %s/%dHz/%dbit: input=%s seek=%dms",
809            track_id,
810            out_fmt.content_type.value,
811            out_fmt.sample_rate,
812            out_fmt.bit_depth,
813            stream_details.audio_format,
814            seek_ms,
815        )
816        async with provider.acquire_stream_slot(STREAM_SLOT_PLAYBACK_WAIT_TIMEOUT):
817            if not self._linked_provider_is_current(provider):
818                self.logger.warning(
819                    "Linked Yandex Music provider changed before starting track %s",
820                    track_id,
821                )
822                self._stream_stop_event.set()
823                return
824            raw_stream = provider.get_audio_stream(stream_details)
825            ffmpeg_stream = get_ffmpeg_stream(
826                audio_input=raw_stream,
827                input_format=stream_details.audio_format,
828                output_format=out_fmt,
829                extra_input_args=extra_input_args,
830            )
831            async with aclosing(raw_stream), aclosing(ffmpeg_stream):
832                async for chunk in ffmpeg_stream:
833                    if not self._linked_provider_is_current(provider):
834                        self.logger.warning(
835                            "Linked Yandex Music provider changed while streaming track %s",
836                            track_id,
837                        )
838                        self._stream_stop_event.set()
839                        break
840                    yield chunk
841
842    async def _get_stream_details_with_retry(
843        self,
844        track_id: str,
845        media_type: MediaType = MediaType.TRACK,
846        *,
847        provider: YandexMusicProviderLike | None = None,
848    ) -> StreamDetails:
849        """Fetch stream details with caching, throttling, and retry."""
850        # Capture the linked yandex_music provider into a local ref at entry.
851        # self._yandex_provider can flip to None mid-await when the linked
852        # MusicProvider is unloaded (see _check_yandex_provider_match, which
853        # runs as a background task on provider-loaded/unloaded events).
854        # Dereferencing the attribute after an await would raise
855        # AttributeError and hard-stop the audio generator.
856        provider = provider or self._yandex_provider
857        if provider is None:
858            raise LoginFailed(
859                "Linked Yandex Music provider is not loaded — cannot fetch stream details"
860            )
861
862        cache_key = self._stream_details_cache_key(provider.instance_id, track_id)
863        cached = await self.mass.cache.get(
864            cache_key,
865            provider=self.instance_id,
866            base_class=StreamDetails,
867        )
868        if cached is not None:
869            cached_streamdetails = cast("StreamDetails", cached)
870            if cached_streamdetails.provider == provider.instance_id:
871                self.logger.debug("Stream details cache hit for %s", track_id)
872                return cached_streamdetails
873            await self.mass.cache.delete(cache_key, provider=self.instance_id)
874            self.logger.warning(
875                "Discarded stream details for %s owned by %s instead of %s",
876                track_id,
877                cached_streamdetails.provider,
878                provider.instance_id,
879            )
880
881        backoff = _API_INITIAL_BACKOFF
882        last_err: Exception | None = None
883        for attempt in range(_API_MAX_RETRIES):
884            async with self._api_throttler.acquire() as delay:
885                if delay > 0:
886                    self.logger.debug("get_stream_details throttled %.1fs", delay)
887            try:
888                sd = await provider.get_stream_details(track_id, media_type)
889                if sd.provider != provider.instance_id:
890                    raise _StreamOwnerMismatchError(
891                        f"Stream details for {track_id} belong to {sd.provider}, "
892                        f"expected {provider.instance_id}"
893                    )
894                # StreamDetails.data has serialize="omit", so to_dict()
895                # strips it. Manually include it so cached entries keep
896                # the URL / decryption key needed by get_audio_stream().
897                cache_value = sd.to_dict()
898                cache_value["data"] = sd.data
899                # Respect the provider's expiration (e.g. yandex_music sets
900                # 50 s because CDN URLs expire after ~60 s).  Fall back to
901                # our default TTL when the provider does not override.
902                cache_ttl = min(_STREAM_DETAILS_CACHE_TTL, sd.expiration)
903                if cache_ttl > 0:
904                    await self.mass.cache.set(
905                        cache_key,
906                        cache_value,
907                        expiration=cache_ttl,
908                        provider=self.instance_id,
909                    )
910                return sd
911            except asyncio.CancelledError:
912                raise
913            except _StreamOwnerMismatchError:
914                raise
915            except Exception as err:
916                last_err = err
917                if attempt < _API_MAX_RETRIES - 1:
918                    jitter = backoff * random.uniform(0.75, 1.25)
919                    self.logger.warning(
920                        "get_stream_details attempt %d/%d failed: %s, retrying in %.1fs",
921                        attempt + 1,
922                        _API_MAX_RETRIES,
923                        err,
924                        jitter,
925                    )
926                    await asyncio.sleep(jitter)
927                    backoff = min(backoff * 2, _API_MAX_BACKOFF)
928        msg = f"get_stream_details failed after {_API_MAX_RETRIES} attempts for {track_id}"
929        raise RuntimeError(msg) from last_err
930
931    async def _invalidate_stream_cache(
932        self, track_id: str, provider_instance_id: str | None = None
933    ) -> None:
934        """
935        Evict cached stream details for a track so the next fetch is fresh.
936
937        :param track_id: Track whose cached stream details should be dropped.
938        :param provider_instance_id: Linked provider instance that owns the entry,
939            defaulting to the currently linked one.
940        """
941        if provider_instance_id is None:
942            if self._yandex_provider is None:
943                return
944            provider_instance_id = self._yandex_provider.instance_id
945        cache_key = self._stream_details_cache_key(provider_instance_id, track_id)
946        await self.mass.cache.delete(cache_key, provider=self.instance_id)
947        self.logger.debug("Invalidated stream cache for %s", track_id)
948
949    @staticmethod
950    def _stream_details_cache_key(provider_instance_id: str, track_id: str) -> str:
951        """Return the cache key for one linked provider instance and track."""
952        return f"ynison_sd_{provider_instance_id}_{track_id}"
953
954    def _linked_provider_is_current(self, provider: YandexMusicProviderLike) -> bool:
955        """Return whether the captured linked provider still owns streaming."""
956        return self._yandex_provider is provider and provider.available
957
958    # ------------------------------------------------------------------
959    # Token handling
960    # ------------------------------------------------------------------
961
962    async def _refresh_via_x_token(self, x_token: str) -> SecretStr:
963        """
964        Refresh the music token from an x_token, caching the result.
965
966        Within :data:`_MUSIC_TOKEN_TTL_S` of a successful refresh, subsequent
967        calls for the same x_token return the cached :class:`SecretStr`
968        without hitting Yandex Passport. Concurrent callers coalesce via
969        :attr:`_token_refresh_lock`.
970
971        :param x_token: Long-lived session token to exchange for a music
972            token. Hashed before use as a cache key; the raw value is
973            never stored in dict keys or logs.
974        :returns: Fresh or cached music-scoped :class:`SecretStr`.
975        :raises LoginFailed: When Yandex explicitly rejects the x_token
976            (propagated from :func:`provider.auth.refresh_music_token`).
977        :raises ResourceTemporarilyUnavailable: On transient Passport
978            failures (network, rate limit) — retry later, credentials
979            are still good.
980        """
981        cache_key = _hash_x_token(x_token)
982        cached = self._token_cache.get(cache_key)
983        now = self._now()
984        if cached is not None and cached.expires_monotonic > now:
985            return cached.token
986
987        async with self._token_refresh_lock:
988            # Double-check inside the lock — a peer caller may have refreshed
989            # while we were waiting for the lock, in which case we reuse
990            # their fresh entry instead of issuing a duplicate Passport call.
991            cached = self._token_cache.get(cache_key)
992            now = self._now()
993            if cached is not None and cached.expires_monotonic > now:
994                return cached.token
995
996            token = await refresh_music_token(SecretStr(x_token))
997            self._store_cached_token(cache_key, token)
998            return token
999
1000    def _store_cached_token(self, cache_key: str, token: SecretStr) -> None:
1001        """
1002        Insert a cache entry, enforcing the LRU bound.
1003
1004        Refreshing an existing key bumps its position to most-recent. When
1005        a new key would push the cache over :data:`_MUSIC_TOKEN_CACHE_MAX`,
1006        the oldest entry is evicted first.
1007        """
1008        # Reordering: pop-then-set positions the (possibly-new) key as
1009        # most-recent in Python's insertion-ordered dict.
1010        self._token_cache.pop(cache_key, None)
1011        while len(self._token_cache) >= _MUSIC_TOKEN_CACHE_MAX:
1012            oldest = next(iter(self._token_cache))
1013            self._token_cache.pop(oldest)
1014        self._token_cache[cache_key] = _CachedToken(
1015            token=token,
1016            expires_monotonic=self._now() + _MUSIC_TOKEN_TTL_S,
1017        )
1018
1019    def _invalidate_cached_token(self, x_token: str) -> None:
1020        """Drop the cache entry for an x_token (e.g. after a 401)."""
1021        self._token_cache.pop(_hash_x_token(x_token), None)
1022
1023    async def _resolve_token(self) -> SecretStr:
1024        """
1025        Resolve the Yandex Music OAuth token for the Ynison connection.
1026
1027        In borrow mode: read from the linked yandex_music provider's config.
1028        If only x_token is present (YM hasn't refreshed yet), do a cached
1029        in-memory refresh without writing back — YM owns token persistence.
1030
1031        In own mode: return CONF_TOKEN if set; otherwise, when CONF_X_TOKEN
1032        is present (QR-with-Remember-session path), cached in-memory refresh.
1033        """
1034        if self._borrow_source is not None:
1035            return await self._borrow_source.resolve_music_token()
1036
1037        token = cast("str | None", self.get_setup_value(CONF_TOKEN))
1038        if token:
1039            return SecretStr(token)
1040        x_token = cast("str | None", self.get_setup_value(CONF_X_TOKEN))
1041        if x_token:
1042            self.logger.debug("Own-mode token not present — refreshing from stored x_token")
1043            return await self._refresh_via_x_token(x_token)
1044        raise LoginFailed("No Yandex Music token configured")
1045
1046    async def _refresh_ynison_token(self) -> SecretStr:
1047        """
1048        Refresh the OAuth token for Ynison reconnection.
1049
1050        Called by YnisonClient on auth failure (401/403) during reconnect.
1051
1052        In borrow mode: re-read the linked YM instance's x_token and refresh
1053        in-memory only (no config writes — YM owns token persistence).
1054
1055        In own mode: refresh from stored CONF_X_TOKEN when present (QR with
1056        "Remember session" enabled). When absent (manual token paste only),
1057        surface LoginFailed so the user knows to paste a new token.
1058
1059        The cached token entry for the current x_token is invalidated up
1060        front — this method is reached only on a server-rejected token, so
1061        the cached value is provably stale.
1062        """
1063        if self._borrow_source is not None:
1064            ym_music_token, ym_x_token = self._borrow_source.read_tokens()
1065            if ym_x_token is None:
1066                raise LoginFailed("Cannot refresh: linked Yandex Music instance has no x_token")
1067            # Both the minted entry AND the owner's persisted token may be the
1068            # value the server just rejected — invalidate both so the source
1069            # can't re-serve either; it will mint fresh from x_token.
1070            if ym_music_token is not None:
1071                self._borrow_source.invalidate(ym_music_token)
1072            self._borrow_source.invalidate(ym_x_token)
1073            self.logger.info("Refreshing Yandex Music token for Ynison reconnect (borrow mode)")
1074            return await self._borrow_source.resolve_music_token()
1075
1076        x_token = cast("str | None", self.get_setup_value(CONF_X_TOKEN))
1077        if x_token:
1078            self._invalidate_cached_token(x_token)
1079            self.logger.info("Refreshing Yandex Music token for Ynison reconnect (own mode)")
1080            return await self._refresh_via_x_token(x_token)
1081
1082        raise LoginFailed(
1083            "Token expired and no stored x_token to refresh from. Re-authenticate "
1084            "via QR or paste a fresh Yandex Music token."
1085        )
1086
1087    # ------------------------------------------------------------------
1088    # Ynison state handling
1089    # ------------------------------------------------------------------
1090
1091    async def _handle_ynison_state(self, state: YnisonState) -> None:
1092        """Handle state update from Ynison."""
1093        is_our_device = state.active_device_id == self._device_id
1094
1095        # Detailed queue logging for diagnostics
1096        queue = state.player_state.get("player_queue", {})
1097        playable_list = queue.get("playable_list", [])
1098        current_index = queue.get("current_playable_index", -1)
1099        entity_type = queue.get("entity_type", "")
1100        entity_id = queue.get("entity_id", "")
1101        track_id = state.current_track_id
1102        self.logger.debug(
1103            "Ynison state: active_device=%s (ours=%s) track=%s "
1104            "index=%d/%d entity=%s type=%s paused=%s progress=%dms",
1105            state.active_device_id,
1106            is_our_device,
1107            track_id,
1108            current_index,
1109            len(playable_list),
1110            entity_id[:40] if entity_id else "<none>",
1111            entity_type,
1112            state.is_paused,
1113            state.progress_ms,
1114        )
1115
1116        # Post-reconnect settle window: the first inbound state after a WS
1117        # reconnect may reflect pre-reconnect peer state (active device etc).
1118        # Acting on it would re-issue play_media, mirror a stale paused flag
1119        # to MA, or worst case clobber a fresh local claim. The 2 s window in
1120        # YnisonClient._connect_state gives the server time to emit a state
1121        # broadcast that reflects our re-registered presence; until then we
1122        # only log.
1123        if self._ynison and self._ynison.in_post_reconnect_settle:
1124            self.logger.debug(
1125                "Skipping state inside post-reconnect settle window (track=%s paused=%s)",
1126                track_id,
1127                state.is_paused,
1128            )
1129            return
1130
1131        if is_our_device and not state.is_paused:
1132            self.logger.info(
1133                "Ynison → playing (track=%s progress=%dms)", track_id, state.progress_ms
1134            )
1135            # Pre-fetch next batch when playing second-to-last track
1136            self._maybe_prefetch(current_index, playable_list, entity_id, entity_type)
1137            await self._activate_playback(state)
1138        elif is_our_device and state.is_paused:
1139            self.logger.info(
1140                "Ynison → paused (track=%s progress=%dms)", track_id, state.progress_ms
1141            )
1142            await self._pause_playback()
1143        elif self._in_use_by_player:
1144            self.logger.info(
1145                "Ynison → other device active (was=%s), clearing",
1146                state.active_device_id,
1147            )
1148            self._clear_active_player()
1149
1150    async def _activate_playback(self, state: YnisonState) -> None:  # noqa: PLR0915
1151        """Activate playback on the target MA player."""
1152        target_player_id = self._get_target_player_id()
1153        if not target_player_id:
1154            self.logger.warning("Ynison active on our device but no MA player available")
1155            return
1156
1157        # Resume after pause / fresh start: either signal triggers
1158        # play_media below. `_externally_paused` survives a stray stop-event
1159        # clear; the stop event covers non-pause stop reasons
1160        # (`_stream_track` warning branch, `_clear_active_player`).
1161        needs_reselect = self._stream_stop_event.is_set() or self._externally_paused
1162        self._stream_stop_event.clear()
1163        self._externally_paused = False
1164
1165        # Start playback via the standard play_media flow if not already active.
1166        # Guard on _active_player_id (set immediately) rather than in_use_by_queue
1167        # (set by get_stream_details when the streams controller picks up the request)
1168        # to prevent queuing redundant play_media calls during the ~5s gap.
1169        if self._active_player_id != target_player_id or needs_reselect:
1170            # Pre-fetch the upcoming track's real format BEFORE submitting
1171            # play_media so the AudioSource's provider_mapping carries the
1172            # right audio_format when the streams controller calls
1173            # get_stream_details(). Skip on same-track same-player resume —
1174            # the cached format is still correct for that case.
1175            upcoming = state.current_track_id
1176            switching_player = self._active_player_id != target_player_id
1177            self._active_player_id = target_player_id
1178            if upcoming and (switching_player or upcoming != self._current_streaming_track_id):
1179                await self._prefetch_format_for_track(upcoming)
1180            self.mass.create_task(
1181                self.mass.player_queues.play_media(target_player_id, str(self._audio_source.uri))
1182            )
1183
1184        # Signal track change if track_id changed
1185        significant_change = False
1186        new_track = state.current_track_id
1187        if new_track and new_track != self._current_streaming_track_id:
1188            self.logger.info("Track changed: %s -> %s", self._current_streaming_track_id, new_track)
1189            self._current_streaming_track_id = new_track
1190            self._seek_position_ms = state.progress_ms
1191            self._track_changed_event.set()
1192            significant_change = True
1193            # Grace period: ignore seek detection for a few seconds after
1194            # track change — Ynison echoes can report stale progress that
1195            # looks like a large drift.
1196            self._seek_grace_until = time.monotonic() + _ECHO_GRACE_PERIOD
1197        elif new_track and new_track == self._current_streaming_track_id:
1198            # Same-track resume after pause: explicitly seek to the Ynison position
1199            # so the new stream starts at the right offset.
1200            if needs_reselect:
1201                self._seek_position_ms = state.progress_ms
1202                self._track_changed_event.set()
1203                self._seek_grace_until = time.monotonic() + _ECHO_GRACE_PERIOD
1204                significant_change = True
1205            else:
1206                # Detect seek: compare Ynison progress against our stream position.
1207                # Ignore Ynison echoes (updates authored by our own device_id) to
1208                # prevent feedback loops where our own progress triggers false seeks.
1209                now = time.monotonic()
1210                if now < self._seek_grace_until:
1211                    pass  # Skip during grace period after track change or seek
1212                elif state.last_update_is_echo:
1213                    pass  # Echo of our own update — ignore
1214                else:
1215                    our_ms = self._streaming_progress_ms
1216                    if our_ms >= 0:
1217                        verdict = self._classify_drift(state.progress_ms, our_ms)
1218                        if verdict == "seek":
1219                            drift_ms = abs(state.progress_ms - our_ms)
1220                            self.logger.info(
1221                                "Seek detected on track %s: "
1222                                "expected ~%dms, Ynison at %dms (drift %dms)",
1223                                new_track,
1224                                our_ms,
1225                                state.progress_ms,
1226                                int(drift_ms),
1227                            )
1228                            self._seek_position_ms = state.progress_ms
1229                            self._track_changed_event.set()
1230                            self._seek_grace_until = now + _ECHO_GRACE_PERIOD
1231                            significant_change = True
1232                        elif verdict == "queue_rebuild":
1233                            self.logger.debug(
1234                                "Drift on track %s classified as queue-rebuild "
1235                                "echo (Ynison=%dms, ours=%dms) — not seeking",
1236                                new_track,
1237                                state.progress_ms,
1238                                our_ms,
1239                            )
1240
1241        # Update metadata from state
1242        self._update_metadata(state)
1243
1244        # Always trigger player update on significant changes;
1245        # throttle regular updates to avoid UI churn (every 5 seconds).
1246        # Use force_update on seek/track change so the server broadcasts a full
1247        # PLAYER_UPDATED event instead of a lightweight elapsed-time-only one
1248        # that the frontend may not handle for AudioSource players.
1249        now_mono = time.monotonic()
1250        if significant_change or needs_reselect or now_mono - self._last_player_update_time >= 5.0:
1251            self.mass.players.trigger_player_update(
1252                target_player_id, force_update=significant_change
1253            )
1254            self._last_player_update_time = now_mono
1255
1256    def _update_metadata(self, state: YnisonState) -> None:
1257        """Update AudioSource metadata from Ynison state."""
1258        meta = self._stream_metadata
1259
1260        # Update duration (prefer actual from stream_details) and elapsed time
1261        best_duration = self._best_duration_ms()
1262        if best_duration:
1263            meta.duration = best_duration // 1000
1264        # Only update elapsed from Ynison when NOT actively streaming —
1265        # during streaming, _sync_progress provides byte-accurate progress.
1266        if state.progress_ms is not None and not self._in_use_by_player:
1267            meta.elapsed_time = state.progress_ms // 1000
1268            meta.elapsed_time_last_updated = time.time()
1269
1270        # Extract track info from player state if available
1271        queue = state.player_state.get("player_queue", {})
1272        playable_list = queue.get("playable_list", [])
1273        index = queue.get("current_playable_index", 0)
1274        if playable_list and 0 <= index < len(playable_list):
1275            playable = playable_list[index]
1276            title = playable.get("title")
1277            if title:
1278                meta.title = title
1279            cover = playable.get("cover_url_optional")
1280            if cover and not cover.startswith("http"):
1281                cover = f"https://{cover}"
1282            if cover:
1283                # Replace %% placeholder with size
1284                cover = cover.replace("%%", "400x400")
1285            meta.image_url = cover
1286
1287    async def _update_metadata_from_stream(
1288        self, stream_details: StreamDetails, seek_ms: int = 0
1289    ) -> None:
1290        """Update AudioSource metadata from stream details (authoritative for duration)."""
1291        meta = self._stream_metadata
1292        if stream_details.duration:
1293            meta.duration = stream_details.duration
1294            self._actual_duration_ms = stream_details.duration * 1000
1295            # Push the real duration to Ynison so the YM app shows
1296            # the correct value (we send duration_ms=0 on advance to
1297            # prevent stale propagation, so this corrects it).
1298            if self._ynison:
1299                await self._send_progress_to_ynison(
1300                    progress_ms=seek_ms,
1301                    duration_ms=self._actual_duration_ms,
1302                    paused=self._ynison.state.is_paused,
1303                )
1304        meta.elapsed_time = seek_ms // 1000 if seek_ms else 0
1305        meta.elapsed_time_last_updated = time.time()
1306        # `trigger_player_update` expects a player_id; `_in_use_by_player` is
1307        # a queue identifier which only happens to coincide with player_id
1308        # when there is no protocol bridge. Use `_active_player_id` — the
1309        # real player wrapping our stream (bridge if any).
1310        if self._active_player_id:
1311            self.mass.players.trigger_player_update(self._active_player_id, force_update=True)
1312
1313    async def _send_progress_to_ynison(
1314        self,
1315        progress_ms: int,
1316        duration_ms: int,
1317        paused: bool,
1318        *,
1319        strict: bool = False,
1320    ) -> None:
1321        """
1322        Send progress to Ynison.
1323
1324        Progress is clamped to duration because Ynison rejects updates where
1325        progress > duration (error 400030001) and disconnects the WebSocket.
1326        The byte counter can slightly overshoot duration at end-of-stream.
1327
1328        Echo detection is done upstream via YnisonState.last_update_is_echo,
1329        which is set when Ynison rebroadcasts an update we authored.
1330
1331        :param progress_ms: Current playback position in milliseconds.
1332        :param duration_ms: Current track duration in milliseconds.
1333        :param paused: Whether playback is paused.
1334        :param strict: When ``True``, propagate transport failures as
1335            :class:`provider.ynison_client.YnisonSendError`. Used by user-command
1336            and end-of-track callers. Heartbeat callers leave the default.
1337        """
1338        if duration_ms <= 0:
1339            # Ynison rejects progress > duration; skip until duration is known.
1340            return
1341        if not self._ynison or not self._ynison.connected:
1342            if strict:
1343                raise YnisonSendError("Ynison not connected")
1344            return
1345        progress_ms = min(progress_ms, duration_ms)
1346        await self._ynison.update_playing_status(
1347            progress_ms=progress_ms,
1348            duration_ms=duration_ms,
1349            paused=paused,
1350            strict=strict,
1351        )
1352
1353    def _bytes_to_ms(self, byte_count: int, fmt: AudioFormat | None = None) -> int:
1354        """Convert PCM byte count to milliseconds using the given format."""
1355        bps = (fmt or self._normalized_format).pcm_sample_size
1356        if bps == 0:
1357            return 0
1358        return (byte_count * 1000) // bps
1359
1360    async def _sync_progress(
1361        self,
1362        seek_ms: int,
1363        bytes_yielded: int,
1364        player_id: str | None,
1365        fmt: AudioFormat | None = None,
1366    ) -> None:
1367        """Push real playback progress to MA metadata and Ynison."""
1368        elapsed_ms = seek_ms + self._bytes_to_ms(bytes_yielded, fmt)
1369        self._streaming_progress_ms = elapsed_ms
1370        # Update MA metadata
1371        meta = self._stream_metadata
1372        if meta:
1373            meta.elapsed_time = elapsed_ms // 1000
1374            meta.elapsed_time_last_updated = time.time()
1375        if player_id:
1376            self.mass.players.trigger_player_update(player_id)
1377        # Update Ynison so the Yandex app shows correct position
1378        await self._send_progress_to_ynison(
1379            progress_ms=elapsed_ms,
1380            duration_ms=self._best_duration_ms(),
1381            paused=False,
1382        )
1383
1384    async def _pause_playback(self) -> None:
1385        """
1386        Release the active player on external pause.
1387
1388        ``cmd_stop`` is the only mechanism that flips ``PlaybackState``
1389        to IDLE for an AudioSource queue item; ``cmd_pause`` and
1390        ``queue.pause`` both short-circuit back to ``on_source_control``
1391        and leave MA's state untouched. Pattern matches upstream
1392        ``AriaCastReceiver._handle_playback_state_update``. Resume
1393        re-runs ``play_media`` (preload + ffmpeg startup) so it costs
1394        a few seconds — the alternative kept resume instant but left
1395        MA's UI stuck on PLAYING.
1396        """
1397        target = self._in_use_by_player
1398        if not target:
1399            self.logger.info("Pause requested but no active queue (_in_use_by_player is None)")
1400            return
1401        self.logger.info("Pause: cmd_stop(%s)", target)
1402        # stop event ends the audio generator; finally clears the lock.
1403        self._stream_stop_event.set()
1404        try:
1405            await self.mass.players.cmd_stop(target)
1406        except Exception:
1407            # cmd_stop is the only mechanism that flips MA's PlaybackState
1408            # to IDLE for an AudioSource. A silent failure here resurrects
1409            # the very UX bug this code path exists to fix.
1410            self.logger.warning(
1411                "cmd_stop(%s) failed during external pause — MA UI may stay PLAYING",
1412                target,
1413                exc_info=True,
1414            )
1415            return
1416        # Demote `_active_player_id` from the bridge MA streams to
1417        # (e.g. `spb_*`) back to the queue id; queues live on the bare
1418        # UUID. Without this, resume's `play_media(_active_player_id,
1419        # …)` would target the bridge and raise
1420        # `PlayerUnavailableError`. Post-success only so a failure
1421        # path keeps the bridge id intact for the next attempt.
1422        self._active_player_id = target
1423        self._externally_paused = True
1424
1425    # ------------------------------------------------------------------
1426    # Player selection
1427    # ------------------------------------------------------------------
1428
1429    def _get_target_player_id(self) -> str | None:
1430        """Determine the target player ID for playback."""
1431        # If there's an active player, validate it still exists
1432        if self._active_player_id:
1433            if self.mass.players.get_player(self._active_player_id):
1434                return self._active_player_id
1435            self._active_player_id = None
1436
1437        # Auto selection
1438        if self._default_player_id == PLAYER_ID_AUTO:
1439            all_players = list(self.mass.players.all_players(False, False))
1440            # Prefer currently playing player
1441            for player in all_players:
1442                if player.state.playback_state == PlaybackState.PLAYING:
1443                    self.logger.debug("Auto-selecting playing player: %s", player.display_name)
1444                    return str(player.player_id)
1445            # Fallback to first available
1446            if all_players:
1447                return str(all_players[0].player_id)
1448            return None
1449
1450        # Specific configured player
1451        if self.mass.players.get_player(self._default_player_id):
1452            return self._default_player_id
1453
1454        self.logger.warning(
1455            "Configured default player '%s' no longer exists",
1456            self._default_player_id,
1457        )
1458        return None
1459
1460    def _session_lost(self, player_id: str, session_id: str | None) -> bool:
1461        """
1462        Return ``True`` when our claim no longer matches the live session.
1463
1464        :param player_id: Queue id captured at generator entry.
1465        :param session_id: ``_active_session_id`` captured at generator entry.
1466        """
1467        return self._in_use_by_player != player_id or self._active_session_id != session_id
1468
1469    def _idempotent(self, action: str, key: str | None) -> bool:
1470        """
1471        Return ``True`` if ``(action, key)`` was not seen within the TTL window.
1472
1473        :param action: A short string identifying the command kind.
1474        :param key: Sub-key inside the action namespace, or ``None``.
1475        """
1476        now = time.monotonic()
1477        for stale_key in [
1478            k for k, ts in self._command_idempotency.items() if now - ts > _COMMAND_IDEMPOTENCY_TTL
1479        ]:
1480            self._command_idempotency.pop(stale_key, None)
1481        composite = (action, key)
1482        last = self._command_idempotency.get(composite)
1483        if last is not None and now - last < _COMMAND_IDEMPOTENCY_TTL:
1484            return False
1485        self._command_idempotency[composite] = now
1486        return True
1487
1488    @staticmethod
1489    def _classify_drift(
1490        ynison_ms: int,
1491        our_ms: int,
1492        threshold_ms: int = 3000,
1493    ) -> Literal["ignore", "queue_rebuild", "seek"]:
1494        """
1495        Classify drift between Ynison-reported and our local position.
1496
1497        Returns one of:
1498
1499        - ``"ignore"`` — drift at or below ``threshold_ms``; no seek needed.
1500        - ``"queue_rebuild"`` — Ynison reports near-zero progress while we
1501          are past 5s into the track; treat as a RADIO queue-rebuild echo,
1502          not a user seek (otherwise we'd yank playback to the start every
1503          time the rotor station refills the queue).
1504        - ``"seek"`` — genuine drift; honor it.
1505
1506        :param ynison_ms: Position reported by Ynison in milliseconds.
1507        :param our_ms: Position tracked locally in milliseconds.
1508        :param threshold_ms: Minimum drift to consider non-ignorable.
1509        """
1510        drift = abs(ynison_ms - our_ms)
1511        if drift <= threshold_ms:
1512            return "ignore"
1513        if ynison_ms < 1000 and our_ms > 5000:
1514            return "queue_rebuild"
1515        return "seek"
1516
1517    async def _prefetch_format_for_track(self, track_id: str) -> None:
1518        """
1519        Pre-fetch stream details for *track_id* and adapt PCM format.
1520
1521        Best-effort: bounded by ``_PREFETCH_FORMAT_TIMEOUT`` so a slow Yandex
1522        API does not stall ``_activate_playback``. On timeout / error the
1523        current format stays in place and the in-stream
1524        ``_get_stream_details_with_retry`` handles retries.
1525
1526        :param track_id: Yandex Music track id to query.
1527        """
1528        if not self._yandex_provider:
1529            return
1530        try:
1531            stream_details = await asyncio.wait_for(
1532                self._get_stream_details_with_retry(track_id),
1533                timeout=_PREFETCH_FORMAT_TIMEOUT,
1534            )
1535        except TimeoutError:
1536            self.logger.info(
1537                "Pre-fetch of stream details for %s exceeded %.1fs — "
1538                "keeping current format; in-stream fetch will retry",
1539                track_id,
1540                _PREFETCH_FORMAT_TIMEOUT,
1541            )
1542            return
1543        except Exception:
1544            self.logger.warning(
1545                "Pre-fetch of stream details failed for %s — keeping current format",
1546                track_id,
1547                exc_info=True,
1548            )
1549            return
1550        old_sr = self._normalized_params.get("sample_rate")
1551        old_bd = self._normalized_params.get("bit_depth")
1552        self._update_normalized_format(hint=stream_details.audio_format)
1553        new_sr = self._normalized_params.get("sample_rate")
1554        new_bd = self._normalized_params.get("bit_depth")
1555        if (old_sr, old_bd) != (new_sr, new_bd):
1556            self.logger.info(
1557                "Pre-fetch adapted format for %s: %dHz/%dbit -> %dHz/%dbit (source=%s)",
1558                track_id,
1559                old_sr or 0,
1560                old_bd or 0,
1561                new_sr or 0,
1562                new_bd or 0,
1563                stream_details.audio_format,
1564            )
1565
1566    def _clear_active_player(self) -> None:
1567        """Clear the active player and reset plugin state."""
1568        prev_player_id = self._active_player_id
1569        # the owner is the user-facing MA player; _active_player_id can be the protocol
1570        # player that consumed the stream, which is not what holds the source session
1571        owner_player_id = self._in_use_by_player
1572        self._active_player_id = None
1573        self._in_use_by_player = None
1574        self._active_session_id = None
1575        self._stream_stop_event.set()
1576        self._streaming_progress_ms = 0
1577        self._prefetched_list = None
1578        self._command_idempotency.clear()
1579        self._externally_paused = False
1580        if self._prefetch_task and not self._prefetch_task.done():
1581            self._prefetch_task.cancel()
1582
1583        if prev_player_id:
1584            self.logger.debug(
1585                "Playback ended on player %s, clearing active player",
1586                prev_player_id,
1587            )
1588            if owner_player_id:
1589                # give the source back as well as stopping: a session left on the player
1590                # keeps it publishing this source, so its own queue stays unreachable
1591                self.mass.create_task(self.mass.players.deselect_source(owner_player_id))
1592            self.mass.players.trigger_player_update(prev_player_id)
1593
1594    # ------------------------------------------------------------------
1595    # Yandex Music provider matching
1596    # ------------------------------------------------------------------
1597
1598    def _on_provider_event(self, event: MassEvent) -> None:
1599        """Handle provider added/removed events."""
1600        self.mass.create_task(self._check_yandex_provider_match())
1601
1602    async def _check_yandex_provider_match(self) -> None:
1603        """
1604        Check if a Yandex Music provider is available for audio streaming.
1605
1606        In borrow mode (self._ym_instance_id set), match strictly by instance_id
1607        so that audio and credentials come from the same account. In own mode,
1608        accept any yandex_music music-provider (prior behavior).
1609        """
1610        for provider in self.mass.get_providers():
1611            if provider.domain != "yandex_music" or provider.type != ProviderType.MUSIC:
1612                continue
1613            if self._ym_instance_id is not None and provider.instance_id != self._ym_instance_id:
1614                continue
1615            self.logger.debug("Found Yandex Music provider — enabling playback control")
1616            self._yandex_provider = cast("YandexMusicProviderLike", provider)
1617            self._update_normalized_format()
1618            self._update_source_capabilities()
1619            return
1620
1621        if self._yandex_provider is not None:
1622            self.logger.debug(
1623                "Yandex Music provider no longer available — disabling playback control"
1624            )
1625            self._yandex_provider = None
1626            self._update_source_capabilities()
1627
1628    def _snap_rate_to_player(self, rate: int) -> int:
1629        """
1630        Snap *rate* down to the nearest sample rate the target player accepts.
1631
1632        Best-effort: returns *rate* unchanged when no target player or
1633        supported-rate set can be resolved, and never raises.
1634
1635        :param rate: The sample rate the hint / floor logic chose.
1636        :return: A rate the target player can play (``rate`` itself when it is
1637            already supported or no player is resolvable).
1638        """
1639        # Mirror MA's _select_audio_source_pcm_format so the declared format
1640        # equals what the AudioSource passthrough picks — keeping MA off its
1641        # second resampling ffmpeg.
1642        try:
1643            player_id = self._get_target_player_id()
1644            if not player_id:
1645                return rate
1646            player = self.mass.players.get_player(player_id)
1647            if player is None:
1648                return rate
1649            supported = [sr for sr, _ in player.get_supported_sample_rates()]
1650            if not supported or rate in supported:
1651                return rate
1652            return max((r for r in supported if r <= rate), default=min(supported))
1653        except Exception:
1654            self.logger.debug(
1655                "Could not snap sample rate to player capabilities; keeping %d Hz",
1656                rate,
1657                exc_info=True,
1658            )
1659            return rate
1660
1661    def _update_normalized_format(self, hint: AudioFormat | None = None) -> None:
1662        """
1663        Set PCM normalization profile based on config and YM quality.
1664
1665        Priority: explicit config values > hint from real stream_details >
1666        auto-detection from YM quality. The hint is fed by
1667        ``_prefetch_format_for_track`` when ``CONF_OUTPUT_SAMPLE_RATE`` is
1668        ``auto`` so the AudioSource ``provider_mapping.audio_format`` matches
1669        the actual source rate of the upcoming track. Without a hint, falls
1670        back to YM-quality-based detection (superb/lossless → 24bit/44.1kHz,
1671        else → 16bit/44.1kHz). The resulting auto/hint rate is then snapped
1672        down to the nearest rate the target player supports; a valid explicit
1673        override is delivered verbatim and never snapped.
1674
1675        Creates fresh AudioFormat instances each time to prevent mutation by
1676        MA's FFMpeg._log_reader_task (which sets input_format.codec_type
1677        in-place on the object passed as input_format to the outer ffmpeg).
1678
1679        :param hint: Optional real source AudioFormat (from a stream-details
1680            pre-fetch). Lifts auto mode from the quality-based default to the
1681            track's actual sample rate and bit depth.
1682        """
1683        # Start with auto-detected base from YM quality config
1684        # (yandex_music does not expose get_quality(); read from its ProviderConfig instead)
1685        quality = ""
1686        if self._yandex_provider is not None:
1687            provider_config = getattr(self._yandex_provider, "config", None)
1688            if provider_config is not None and hasattr(provider_config, "get_value"):
1689                config_quality = provider_config.get_value(YANDEX_MUSIC_CONF_QUALITY)
1690                if isinstance(config_quality, str):
1691                    quality = config_quality
1692        is_lossless = quality in YANDEX_MUSIC_LOSSLESS_QUALITIES
1693        base = dict(PCM_LOSSLESS_PARAMS if is_lossless else PCM_LOSSY_PARAMS)
1694        # Promote auto-base from the real stream details when available.
1695        # Validate the hint against the same allow-lists we use for explicit
1696        # config overrides — a Yandex API hiccup that returns an unsupported
1697        # rate (or 0) must not poison the AudioSource provider_mapping or the
1698        # outer ffmpeg input_format.
1699        if hint is not None:
1700            if hint.sample_rate and str(hint.sample_rate) in _VALID_SAMPLE_RATES:
1701                base["sample_rate"] = hint.sample_rate
1702            if hint.bit_depth and str(hint.bit_depth) in _VALID_BIT_DEPTHS:
1703                base["bit_depth"] = hint.bit_depth
1704
1705        # Apply config overrides. MA's ConfigEntry options constrain the UI to
1706        # known-good strings, but a stale persisted value or hand-edited config
1707        # could still surface something unparsable or off-list — fall back to
1708        # the auto-detected base with a warning instead of crashing the load.
1709        sample_rate = base["sample_rate"]
1710        bit_depth = base["bit_depth"]
1711        explicit_rate = False
1712        if self._cfg_sample_rate != OUTPUT_AUTO:
1713            if self._cfg_sample_rate in _VALID_SAMPLE_RATES:
1714                sample_rate = int(self._cfg_sample_rate)
1715                explicit_rate = True
1716            else:
1717                self.logger.warning(
1718                    "Invalid %s=%r; falling back to auto-detected %d Hz",
1719                    CONF_OUTPUT_SAMPLE_RATE,
1720                    self._cfg_sample_rate,
1721                    sample_rate,
1722                )
1723        # Snap the auto / hint / floor rate to a value the target player accepts
1724        # so the declared format matches what MA's AudioSource passthrough picks
1725        # and no second resampling ffmpeg is spawned. A valid explicit override
1726        # is delivered verbatim and is never snapped.
1727        if not explicit_rate:
1728            sample_rate = self._snap_rate_to_player(sample_rate)
1729        if self._cfg_bit_depth != OUTPUT_AUTO:
1730            if self._cfg_bit_depth in _VALID_BIT_DEPTHS:
1731                bit_depth = int(self._cfg_bit_depth)
1732            else:
1733                self.logger.warning(
1734                    "Invalid %s=%r; falling back to auto-detected %d-bit",
1735                    CONF_OUTPUT_BIT_DEPTH,
1736                    self._cfg_bit_depth,
1737                    bit_depth,
1738                )
1739
1740        content_type = ContentType.PCM_S24LE if bit_depth == 24 else ContentType.PCM_S16LE
1741        new_params: dict[str, Any] = {
1742            "content_type": content_type,
1743            "sample_rate": sample_rate,
1744            "bit_depth": bit_depth,
1745            "channels": 2,
1746        }
1747
1748        # Warn if format changes while a player is actively streaming — the
1749        # active session keeps using its frozen snapshot; the new format takes
1750        # effect on the next session.
1751        old = self._normalized_params
1752        if self._in_use_by_player and (
1753            old.get("content_type") != content_type
1754            or old.get("sample_rate") != sample_rate
1755            or old.get("bit_depth") != bit_depth
1756        ):
1757            self.logger.warning(
1758                "Normalization format changed while streaming — new format "
1759                "(%s/%dHz/%dbit) will apply on next session",
1760                content_type.value,
1761                sample_rate,
1762                bit_depth,
1763            )
1764
1765        self._normalized_params = new_params
1766        # Fresh copy for each caller so no shared mutable state
1767        self._normalized_format = make_pcm_format(self._normalized_params)
1768        # rebuild the AudioSource so its ProviderMapping carries the new audio_format
1769        self._audio_source = self._build_audio_source()
1770        self.logger.debug(
1771            "Normalization format: %s/%dHz/%dbit",
1772            self._normalized_format.content_type.value,
1773            self._normalized_format.sample_rate,
1774            self._normalized_format.bit_depth,
1775        )
1776
1777    def _update_source_capabilities(self) -> None:
1778        """Rebuild AudioSource so capability flags reflect linked provider availability."""
1779        self._audio_source = self._build_audio_source()
1780        # The session publishes the controls from the object it holds, so hand it the
1781        # rebuilt one: the new capability flags reach the UI without waiting for the
1782        # source to be selected again.
1783        if not self._in_use_by_player:
1784            return
1785        self.mass.players.refresh_source(self._in_use_by_player, self._audio_source)
1786
1787    def _build_audio_source(self) -> AudioSource:
1788        """Construct the AudioSource MediaItem with current capability flags."""
1789        has_provider = self._yandex_provider is not None
1790        return AudioSource(
1791            item_id=AUDIO_SOURCE_ID,
1792            provider=self.instance_id,
1793            name=self.name,
1794            provider_mappings={
1795                ProviderMapping(
1796                    item_id=AUDIO_SOURCE_ID,
1797                    provider_domain=self.domain,
1798                    provider_instance=self.instance_id,
1799                    # Fresh AudioFormat copy — `self._normalized_format` is a
1800                    # shared mutable that MA's ffmpeg sets `codec_type` on
1801                    # in-place. Sharing it would let that mutation leak into
1802                    # the rebuilt AudioSource and any future stream-details.
1803                    audio_format=make_pcm_format(self._normalized_params),
1804                )
1805            },
1806            can_play_pause=has_provider,
1807            can_seek=has_provider,
1808            can_next_previous=has_provider,
1809            exclusive=True,
1810            allow_external_trigger=True,
1811        )
1812
1813    # ------------------------------------------------------------------
1814    # Playback control callbacks
1815    # ------------------------------------------------------------------
1816
1817    def _best_duration_ms(self) -> int:
1818        """Return the best known duration: actual from stream, or Ynison state as fallback."""
1819        if self._actual_duration_ms > 0:
1820            return self._actual_duration_ms
1821        if self._ynison:
1822            return self._ynison.state.duration_ms
1823        return 0
1824
1825    def _require_connected_ynison(self) -> YnisonClient:
1826        """
1827        Return the live Ynison client or raise an MA player-control error.
1828
1829        :raises UnsupportedFeaturedException: When the provider's Ynison
1830            client has not been initialised yet (pre-`handle_async_init`
1831            or post-`unload`).
1832        :raises PlayerCommandFailed: When the Ynison WebSocket is currently
1833            disconnected (e.g. mid-reconnect after a transient network
1834            error). Surface to MA so the UI shows a clear failure toast
1835            instead of accepting the command and stalling.
1836        """
1837        if not self._ynison:
1838            raise UnsupportedFeaturedException("Ynison client not initialized")
1839        if not self._ynison.connected:
1840            raise PlayerCommandFailed("Ynison WebSocket disconnected")
1841        return self._ynison
1842
1843    async def _on_play(self) -> None:
1844        """Handle play command — send resume to Ynison."""
1845        client = self._require_connected_ynison()
1846        if not self._idempotent("on_play", None):
1847            return
1848        state = client.state
1849        try:
1850            await self._send_progress_to_ynison(
1851                progress_ms=state.progress_ms,
1852                duration_ms=self._best_duration_ms(),
1853                paused=False,
1854                strict=True,
1855            )
1856        except YnisonSendError as exc:
1857            raise PlayerCommandFailed("Ynison send failed") from exc
1858
1859    async def _on_pause(self) -> None:
1860        """Handle pause command — send pause to Ynison."""
1861        client = self._require_connected_ynison()
1862        if not self._idempotent("on_pause", None):
1863            return
1864        state = client.state
1865        try:
1866            await self._send_progress_to_ynison(
1867                progress_ms=state.progress_ms,
1868                duration_ms=self._best_duration_ms(),
1869                paused=True,
1870                strict=True,
1871            )
1872        except YnisonSendError as exc:
1873            raise PlayerCommandFailed("Ynison send failed") from exc
1874
1875    # Entity types that use server-side "radio" queue replenishment.
1876    # Currently only RADIO (personal wave, genre stations).
1877    # Add "WAVE" here if/when Yandex supports it via the same
1878    # rotor_station_tracks API.
1879    _RADIO_ENTITY_TYPES: ClassVar[set[str]] = {"RADIO"}
1880
1881    def _maybe_prefetch(
1882        self,
1883        current_index: int,
1884        playable_list: list[dict[str, Any]],
1885        entity_id: str,
1886        entity_type: str,
1887    ) -> None:
1888        """Kick off background prefetch when nearing the end of the queue."""
1889        if entity_type not in self._RADIO_ENTITY_TYPES:
1890            return
1891        if not self._yandex_provider or not playable_list:
1892            return
1893        # second-to-last or last — trigger prefetch near end of queue
1894        if current_index < len(playable_list) - 2:
1895            return
1896        # Already prefetched or prefetch in progress
1897        if self._prefetched_list is not None:
1898            return
1899        if self._prefetch_task and not self._prefetch_task.done():
1900            return
1901
1902        self.logger.info(
1903            "Pre-fetching tracks (at index %d/%d, entity=%s)",
1904            current_index,
1905            len(playable_list),
1906            entity_id[:40] if entity_id else "<none>",
1907        )
1908
1909        async def _do_prefetch() -> None:
1910            result = await self._replenish_radio_queue(entity_id, entity_type, playable_list)
1911            if result:
1912                self._prefetched_list = result
1913                # Push expanded queue to Ynison immediately so the YM app
1914                # sees upcoming tracks and enables the "next" button.
1915                await self._update_queue_list(result)
1916
1917        self._prefetch_task = self.mass.create_task(_do_prefetch())
1918
1919    async def _signal_track_completion(self) -> None:
1920        """
1921        Signal that the current track finished playing.
1922
1923        Ynison is a state-sync protocol — the active device must advance
1924        current_playable_index itself.
1925
1926        If the next index is within the playable list, we advance immediately.
1927        If we're at the end (typical for RADIO/wave with short queues),
1928        we fetch more tracks via the Yandex Music API, append them to the
1929        playable_list, and then advance.
1930        """
1931        if not self._ynison:
1932            return
1933        state = self._ynison.state
1934        duration = self._best_duration_ms()
1935        queue = state.player_state.get("player_queue", {})
1936        current_index = queue.get("current_playable_index", 0)
1937        playable_list = queue.get("playable_list", [])
1938        entity_type = queue.get("entity_type", "")
1939        entity_id = queue.get("entity_id", "")
1940        next_index = current_index + 1
1941
1942        self.logger.info(
1943            "Track finished at index %d/%d (entity=%s type=%s), "
1944            "advancing to index %d (duration=%dms)",
1945            current_index,
1946            len(playable_list),
1947            entity_id[:40] if entity_id else "<none>",
1948            entity_type,
1949            next_index,
1950            duration,
1951        )
1952        self._actual_duration_ms = 0
1953
1954        # 1. Report that playback reached the end.
1955        # Echo tracking is handled by _send_progress_to_ynison.
1956        # `strict=True`: a dropped end-of-track signal stalls the YM app on
1957        # the just-finished track. We log and continue — the reconnect is
1958        # already scheduled and the queue-advance below sees the same WS state
1959        # — but we don't reraise (this is end-of-stream, there's no command to
1960        # fail back to the user).
1961        try:
1962            await self._send_progress_to_ynison(
1963                progress_ms=duration, duration_ms=duration, paused=False, strict=True
1964            )
1965        except YnisonSendError:
1966            self.logger.warning(
1967                "Track-completion signal dropped (Ynison transport failure); "
1968                "queue advance will retry once the WS reconnects",
1969                exc_info=True,
1970            )
1971
1972        if next_index < len(playable_list):
1973            # 2a. Queue has room — advance immediately.
1974            # Clear stale prefetch data so _maybe_prefetch can trigger for
1975            # the new queue tail on subsequent state updates.
1976            self._prefetched_list = None
1977            await self._advance_queue_index(next_index)
1978        elif entity_type in self._RADIO_ENTITY_TYPES:
1979            # 2b. At end of RADIO queue — use prefetched data or fetch now
1980            expanded: list[dict[str, Any]] | None = None
1981            if self._prefetched_list:
1982                self.logger.info("Using pre-fetched queue (%d items)", len(self._prefetched_list))
1983                expanded = self._prefetched_list
1984                self._prefetched_list = None
1985            elif self._prefetch_task and not self._prefetch_task.done():
1986                self.logger.info("Waiting for in-flight prefetch...")
1987                await self._prefetch_task
1988                expanded = self._prefetched_list
1989                self._prefetched_list = None
1990            else:
1991                expanded = await self._replenish_radio_queue(entity_id, entity_type, playable_list)
1992            if expanded and next_index < len(expanded):
1993                await self._advance_queue_index(next_index, expanded_list=expanded)
1994            elif expanded:
1995                self.logger.warning(
1996                    "Expanded queue has %d items but next_index=%d — re-fetching",
1997                    len(expanded),
1998                    next_index,
1999                )
2000                fresh = await self._replenish_radio_queue(entity_id, entity_type, expanded)
2001                if fresh and next_index < len(fresh):
2002                    await self._advance_queue_index(next_index, expanded_list=fresh)
2003                else:
2004                    self.logger.warning("Still cannot advance after re-fetch")
2005            else:
2006                self.logger.warning(
2007                    "Could not replenish queue (entity=%s type=%s), cannot advance",
2008                    entity_id,
2009                    entity_type,
2010                )
2011        else:
2012            self.logger.info(
2013                "End of non-radio queue (entity=%s type=%s), playback complete",
2014                entity_id[:40] if entity_id else "<none>",
2015                entity_type,
2016            )
2017
2018    async def _replenish_radio_queue(
2019        self,
2020        entity_id: str,
2021        entity_type: str,
2022        playable_list: list[dict[str, Any]],
2023    ) -> list[dict[str, Any]] | None:
2024        """
2025        Fetch more tracks from Yandex Music API and return expanded playable_list.
2026
2027        The active device is responsible for replenishing RADIO/wave queues.
2028        Ynison only syncs state — it does NOT generate new tracks.
2029        """
2030        if not self._yandex_provider:
2031            self.logger.warning("No yandex_music provider available for radio replenishment")
2032            return None
2033
2034        # Determine the last track ID for pagination
2035        last_track_id: str | None = None
2036        if playable_list:
2037            last_track_id = playable_list[-1].get("playable_id")
2038
2039        self.logger.info(
2040            "Fetching more tracks for %s station %s (queue=%s)",
2041            entity_type,
2042            entity_id,
2043            last_track_id,
2044        )
2045
2046        try:
2047            tracks, batch_id = await self._yandex_provider.get_rotor_station_tracks(
2048                entity_id, queue=last_track_id
2049            )
2050        except Exception:
2051            self.logger.exception("Failed to fetch radio tracks for %s", entity_id)
2052            return None
2053
2054        if not tracks:
2055            self.logger.warning("No tracks returned for station %s", entity_id)
2056            return None
2057
2058        # Determine the 'from' field from existing items
2059        from_field = ""
2060        if playable_list:
2061            from_field = playable_list[0].get("from", "")
2062
2063        # Convert tracks to Ynison playable_list format
2064        new_items: list[dict[str, Any]] = []
2065        for track in tracks:
2066            album_id = ""
2067            if hasattr(track, "albums") and track.albums:
2068                album_id = str(track.albums[0].id) if track.albums[0].id else ""
2069            cover = ""
2070            if hasattr(track, "cover_uri") and track.cover_uri:
2071                cover = track.cover_uri
2072            new_items.append(
2073                {
2074                    "playable_id": str(track.id),
2075                    "album_id_optional": album_id,
2076                    "playable_type": "TRACK",
2077                    "from": from_field,
2078                    "title": track.title or "",
2079                    "cover_url_optional": cover,
2080                }
2081            )
2082
2083        self.logger.info(
2084            "Fetched %d new tracks for station %s (batch=%s)",
2085            len(new_items),
2086            entity_id,
2087            batch_id,
2088        )
2089
2090        return list(playable_list) + new_items
2091
2092    async def _advance_queue_index(
2093        self,
2094        next_index: int,
2095        *,
2096        expanded_list: list[dict[str, Any]] | None = None,
2097    ) -> None:
2098        """
2099        Send update_player_state to advance the queue to next_index.
2100
2101        If expanded_list is provided, it replaces the playable_list
2102        (used after radio queue replenishment).
2103
2104        Waits up to 10 s for reconnection if Ynison is temporarily
2105        disconnected (e.g. after a transient error).
2106        """
2107        if not self._ynison:
2108            return
2109        if not self._ynison.connected:
2110            self.logger.info("Waiting for Ynison reconnection before advancing queue…")
2111            for _ in range(10):
2112                await asyncio.sleep(1)
2113                if not self._ynison or self._ynison.connected:
2114                    break
2115            if not self._ynison or not self._ynison.connected:
2116                self.logger.warning("Cannot advance queue — Ynison still disconnected")
2117                return
2118        state = self._ynison.state
2119        queue = state.player_state.get("player_queue", {})
2120        device_id = self._ynison.device_id
2121        new_state = dict(state.player_state)
2122        new_state["player_queue"] = dict(queue)
2123        new_state["player_queue"]["current_playable_index"] = next_index
2124        new_state["player_queue"]["version"] = make_version_block(device_id)
2125        if expanded_list is not None:
2126            new_state["player_queue"]["playable_list"] = expanded_list
2127        new_state["status"] = dict(new_state.get("status", {}))
2128        new_state["status"]["progress_ms"] = "0"
2129        new_state["status"]["duration_ms"] = "0"
2130        new_state["status"]["paused"] = False
2131        new_state["status"]["version"] = make_version_block(device_id)
2132        # `strict=True`: a dropped queue-advance leaves `_wait_for_track_change`
2133        # spinning for its full 30 s timeout. Log and return — the next
2134        # reconnect-broadcast picks up our authored version block and resyncs.
2135        try:
2136            await self._ynison.update_player_state(player_state=new_state, strict=True)
2137        except YnisonSendError:
2138            self.logger.warning(
2139                "Queue-advance dropped (Ynison transport failure); "
2140                "stream will stall until reconnect-broadcast resyncs",
2141                exc_info=True,
2142            )
2143
2144    async def _update_queue_list(self, expanded_list: list[dict[str, Any]]) -> None:
2145        """
2146        Push an expanded playable_list to Ynison without changing index or progress.
2147
2148        Called right after prefetch completes so the YM app sees upcoming
2149        tracks and enables the "next" button.
2150        """
2151        if not self._ynison or not self._ynison.connected:
2152            return
2153        state = self._ynison.state
2154        queue = state.player_state.get("player_queue", {})
2155        device_id = self._ynison.device_id
2156        new_state = dict(state.player_state)
2157        new_state["player_queue"] = dict(queue)
2158        new_state["player_queue"]["playable_list"] = expanded_list
2159        new_state["player_queue"]["version"] = make_version_block(device_id)
2160        await self._ynison.update_player_state(player_state=new_state)
2161
2162    async def _on_next(self) -> None:
2163        """Handle next track command — signal track end so Yandex advances."""
2164        self._require_connected_ynison()
2165        await self._signal_track_completion()
2166
2167    async def _on_previous(self) -> None:
2168        """Handle previous track command — update queue index in Ynison."""
2169        client = self._require_connected_ynison()
2170        queue = client.state.player_state.get("player_queue", {})
2171        current_index = queue.get("current_playable_index", 0)
2172        if current_index > 0:
2173            self._actual_duration_ms = 0
2174            await self._advance_queue_index(current_index - 1)
2175
2176    async def _on_seek(self, position: int) -> None:
2177        """
2178        Handle seek command — send position update to Ynison.
2179
2180        :param position: Position in seconds from Music Assistant.
2181        """
2182        client = self._require_connected_ynison()
2183        seek_ms = position * 1000
2184        state = client.state
2185        try:
2186            await self._send_progress_to_ynison(
2187                progress_ms=seek_ms,
2188                duration_ms=self._best_duration_ms(),
2189                paused=state.is_paused,
2190                strict=True,
2191            )
2192        except YnisonSendError as exc:
2193            # Do not mutate `_seek_position_ms` / `_seek_grace_until` on failure
2194            # — local stream state must not drift past a send that never landed.
2195            raise PlayerCommandFailed("Ynison send failed") from exc
2196        # Also trigger local stream restart so seek takes effect
2197        # immediately without waiting for the Ynison echo.
2198        self._seek_position_ms = seek_ms
2199        self._seek_grace_until = time.monotonic() + _ECHO_GRACE_PERIOD
2200        self._track_changed_event.set()
2201