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        source_session = (
1573            self.mass.players.get_audio_source_session(owner_player_id) if owner_player_id else None
1574        )
1575        self._active_player_id = None
1576        self._in_use_by_player = None
1577        self._active_session_id = None
1578        self._stream_stop_event.set()
1579        self._streaming_progress_ms = 0
1580        self._prefetched_list = None
1581        self._command_idempotency.clear()
1582        self._externally_paused = False
1583        if self._prefetch_task and not self._prefetch_task.done():
1584            self._prefetch_task.cancel()
1585
1586        if prev_player_id:
1587            self.logger.debug(
1588                "Playback ended on player %s, clearing active player",
1589                prev_player_id,
1590            )
1591            if owner_player_id:
1592                # give the source back as well as stopping: a session left on the player
1593                # keeps it publishing this source, so its own queue stays unreachable
1594                self.mass.create_task(
1595                    self.mass.players.deselect_source(
1596                        owner_player_id,
1597                        provider_instance_id=self.instance_id,
1598                        source_id=AUDIO_SOURCE_ID,
1599                        playback_session_id=(
1600                            source_session.playback_session_id if source_session else None
1601                        ),
1602                    )
1603                )
1604            self.mass.players.trigger_player_update(prev_player_id)
1605
1606    # ------------------------------------------------------------------
1607    # Yandex Music provider matching
1608    # ------------------------------------------------------------------
1609
1610    def _on_provider_event(self, event: MassEvent) -> None:
1611        """Handle provider added/removed events."""
1612        self.mass.create_task(self._check_yandex_provider_match())
1613
1614    async def _check_yandex_provider_match(self) -> None:
1615        """
1616        Check if a Yandex Music provider is available for audio streaming.
1617
1618        In borrow mode (self._ym_instance_id set), match strictly by instance_id
1619        so that audio and credentials come from the same account. In own mode,
1620        accept any yandex_music music-provider (prior behavior).
1621        """
1622        for provider in self.mass.get_providers():
1623            if provider.domain != "yandex_music" or provider.type != ProviderType.MUSIC:
1624                continue
1625            if self._ym_instance_id is not None and provider.instance_id != self._ym_instance_id:
1626                continue
1627            self.logger.debug("Found Yandex Music provider — enabling playback control")
1628            self._yandex_provider = cast("YandexMusicProviderLike", provider)
1629            self._update_normalized_format()
1630            self._update_source_capabilities()
1631            return
1632
1633        if self._yandex_provider is not None:
1634            self.logger.debug(
1635                "Yandex Music provider no longer available — disabling playback control"
1636            )
1637            self._yandex_provider = None
1638            self._update_source_capabilities()
1639
1640    def _snap_rate_to_player(self, rate: int) -> int:
1641        """
1642        Snap *rate* down to the nearest sample rate the target player accepts.
1643
1644        Best-effort: returns *rate* unchanged when no target player or
1645        supported-rate set can be resolved, and never raises.
1646
1647        :param rate: The sample rate the hint / floor logic chose.
1648        :return: A rate the target player can play (``rate`` itself when it is
1649            already supported or no player is resolvable).
1650        """
1651        # Mirror MA's _select_audio_source_pcm_format so the declared format
1652        # equals what the AudioSource passthrough picks — keeping MA off its
1653        # second resampling ffmpeg.
1654        try:
1655            player_id = self._get_target_player_id()
1656            if not player_id:
1657                return rate
1658            player = self.mass.players.get_player(player_id)
1659            if player is None:
1660                return rate
1661            supported = [sr for sr, _ in player.get_supported_sample_rates()]
1662            if not supported or rate in supported:
1663                return rate
1664            return max((r for r in supported if r <= rate), default=min(supported))
1665        except Exception:
1666            self.logger.debug(
1667                "Could not snap sample rate to player capabilities; keeping %d Hz",
1668                rate,
1669                exc_info=True,
1670            )
1671            return rate
1672
1673    def _update_normalized_format(self, hint: AudioFormat | None = None) -> None:
1674        """
1675        Set PCM normalization profile based on config and YM quality.
1676
1677        Priority: explicit config values > hint from real stream_details >
1678        auto-detection from YM quality. The hint is fed by
1679        ``_prefetch_format_for_track`` when ``CONF_OUTPUT_SAMPLE_RATE`` is
1680        ``auto`` so the AudioSource ``provider_mapping.audio_format`` matches
1681        the actual source rate of the upcoming track. Without a hint, falls
1682        back to YM-quality-based detection (superb/lossless → 24bit/44.1kHz,
1683        else → 16bit/44.1kHz). The resulting auto/hint rate is then snapped
1684        down to the nearest rate the target player supports; a valid explicit
1685        override is delivered verbatim and never snapped.
1686
1687        Creates fresh AudioFormat instances each time to prevent mutation by
1688        MA's FFMpeg._log_reader_task (which sets input_format.codec_type
1689        in-place on the object passed as input_format to the outer ffmpeg).
1690
1691        :param hint: Optional real source AudioFormat (from a stream-details
1692            pre-fetch). Lifts auto mode from the quality-based default to the
1693            track's actual sample rate and bit depth.
1694        """
1695        # Start with auto-detected base from YM quality config
1696        # (yandex_music does not expose get_quality(); read from its ProviderConfig instead)
1697        quality = ""
1698        if self._yandex_provider is not None:
1699            provider_config = getattr(self._yandex_provider, "config", None)
1700            if provider_config is not None and hasattr(provider_config, "get_value"):
1701                config_quality = provider_config.get_value(YANDEX_MUSIC_CONF_QUALITY)
1702                if isinstance(config_quality, str):
1703                    quality = config_quality
1704        is_lossless = quality in YANDEX_MUSIC_LOSSLESS_QUALITIES
1705        base = dict(PCM_LOSSLESS_PARAMS if is_lossless else PCM_LOSSY_PARAMS)
1706        # Promote auto-base from the real stream details when available.
1707        # Validate the hint against the same allow-lists we use for explicit
1708        # config overrides — a Yandex API hiccup that returns an unsupported
1709        # rate (or 0) must not poison the AudioSource provider_mapping or the
1710        # outer ffmpeg input_format.
1711        if hint is not None:
1712            if hint.sample_rate and str(hint.sample_rate) in _VALID_SAMPLE_RATES:
1713                base["sample_rate"] = hint.sample_rate
1714            if hint.bit_depth and str(hint.bit_depth) in _VALID_BIT_DEPTHS:
1715                base["bit_depth"] = hint.bit_depth
1716
1717        # Apply config overrides. MA's ConfigEntry options constrain the UI to
1718        # known-good strings, but a stale persisted value or hand-edited config
1719        # could still surface something unparsable or off-list — fall back to
1720        # the auto-detected base with a warning instead of crashing the load.
1721        sample_rate = base["sample_rate"]
1722        bit_depth = base["bit_depth"]
1723        explicit_rate = False
1724        if self._cfg_sample_rate != OUTPUT_AUTO:
1725            if self._cfg_sample_rate in _VALID_SAMPLE_RATES:
1726                sample_rate = int(self._cfg_sample_rate)
1727                explicit_rate = True
1728            else:
1729                self.logger.warning(
1730                    "Invalid %s=%r; falling back to auto-detected %d Hz",
1731                    CONF_OUTPUT_SAMPLE_RATE,
1732                    self._cfg_sample_rate,
1733                    sample_rate,
1734                )
1735        # Snap the auto / hint / floor rate to a value the target player accepts
1736        # so the declared format matches what MA's AudioSource passthrough picks
1737        # and no second resampling ffmpeg is spawned. A valid explicit override
1738        # is delivered verbatim and is never snapped.
1739        if not explicit_rate:
1740            sample_rate = self._snap_rate_to_player(sample_rate)
1741        if self._cfg_bit_depth != OUTPUT_AUTO:
1742            if self._cfg_bit_depth in _VALID_BIT_DEPTHS:
1743                bit_depth = int(self._cfg_bit_depth)
1744            else:
1745                self.logger.warning(
1746                    "Invalid %s=%r; falling back to auto-detected %d-bit",
1747                    CONF_OUTPUT_BIT_DEPTH,
1748                    self._cfg_bit_depth,
1749                    bit_depth,
1750                )
1751
1752        content_type = ContentType.PCM_S24LE if bit_depth == 24 else ContentType.PCM_S16LE
1753        new_params: dict[str, Any] = {
1754            "content_type": content_type,
1755            "sample_rate": sample_rate,
1756            "bit_depth": bit_depth,
1757            "channels": 2,
1758        }
1759
1760        # Warn if format changes while a player is actively streaming — the
1761        # active session keeps using its frozen snapshot; the new format takes
1762        # effect on the next session.
1763        old = self._normalized_params
1764        if self._in_use_by_player and (
1765            old.get("content_type") != content_type
1766            or old.get("sample_rate") != sample_rate
1767            or old.get("bit_depth") != bit_depth
1768        ):
1769            self.logger.warning(
1770                "Normalization format changed while streaming — new format "
1771                "(%s/%dHz/%dbit) will apply on next session",
1772                content_type.value,
1773                sample_rate,
1774                bit_depth,
1775            )
1776
1777        self._normalized_params = new_params
1778        # Fresh copy for each caller so no shared mutable state
1779        self._normalized_format = make_pcm_format(self._normalized_params)
1780        # rebuild the AudioSource so its ProviderMapping carries the new audio_format
1781        self._audio_source = self._build_audio_source()
1782        self.logger.debug(
1783            "Normalization format: %s/%dHz/%dbit",
1784            self._normalized_format.content_type.value,
1785            self._normalized_format.sample_rate,
1786            self._normalized_format.bit_depth,
1787        )
1788
1789    def _update_source_capabilities(self) -> None:
1790        """Rebuild AudioSource so capability flags reflect linked provider availability."""
1791        self._audio_source = self._build_audio_source()
1792        # The session publishes the controls from the object it holds, so hand it the
1793        # rebuilt one: the new capability flags reach the UI without waiting for the
1794        # source to be selected again.
1795        if not self._in_use_by_player:
1796            return
1797        self.mass.players.refresh_source(self._in_use_by_player, self._audio_source)
1798
1799    def _build_audio_source(self) -> AudioSource:
1800        """Construct the AudioSource MediaItem with current capability flags."""
1801        has_provider = self._yandex_provider is not None
1802        return AudioSource(
1803            item_id=AUDIO_SOURCE_ID,
1804            provider=self.instance_id,
1805            name=self.name,
1806            provider_mappings={
1807                ProviderMapping(
1808                    item_id=AUDIO_SOURCE_ID,
1809                    provider_domain=self.domain,
1810                    provider_instance=self.instance_id,
1811                    # Fresh AudioFormat copy — `self._normalized_format` is a
1812                    # shared mutable that MA's ffmpeg sets `codec_type` on
1813                    # in-place. Sharing it would let that mutation leak into
1814                    # the rebuilt AudioSource and any future stream-details.
1815                    audio_format=make_pcm_format(self._normalized_params),
1816                )
1817            },
1818            can_play_pause=has_provider,
1819            can_seek=has_provider,
1820            can_next_previous=has_provider,
1821            exclusive=True,
1822            allow_external_trigger=True,
1823        )
1824
1825    # ------------------------------------------------------------------
1826    # Playback control callbacks
1827    # ------------------------------------------------------------------
1828
1829    def _best_duration_ms(self) -> int:
1830        """Return the best known duration: actual from stream, or Ynison state as fallback."""
1831        if self._actual_duration_ms > 0:
1832            return self._actual_duration_ms
1833        if self._ynison:
1834            return self._ynison.state.duration_ms
1835        return 0
1836
1837    def _require_connected_ynison(self) -> YnisonClient:
1838        """
1839        Return the live Ynison client or raise an MA player-control error.
1840
1841        :raises UnsupportedFeaturedException: When the provider's Ynison
1842            client has not been initialised yet (pre-`handle_async_init`
1843            or post-`unload`).
1844        :raises PlayerCommandFailed: When the Ynison WebSocket is currently
1845            disconnected (e.g. mid-reconnect after a transient network
1846            error). Surface to MA so the UI shows a clear failure toast
1847            instead of accepting the command and stalling.
1848        """
1849        if not self._ynison:
1850            raise UnsupportedFeaturedException("Ynison client not initialized")
1851        if not self._ynison.connected:
1852            raise PlayerCommandFailed("Ynison WebSocket disconnected")
1853        return self._ynison
1854
1855    async def _on_play(self) -> None:
1856        """Handle play command — send resume to Ynison."""
1857        client = self._require_connected_ynison()
1858        if not self._idempotent("on_play", None):
1859            return
1860        state = client.state
1861        try:
1862            await self._send_progress_to_ynison(
1863                progress_ms=state.progress_ms,
1864                duration_ms=self._best_duration_ms(),
1865                paused=False,
1866                strict=True,
1867            )
1868        except YnisonSendError as exc:
1869            raise PlayerCommandFailed("Ynison send failed") from exc
1870
1871    async def _on_pause(self) -> None:
1872        """Handle pause command — send pause to Ynison."""
1873        client = self._require_connected_ynison()
1874        if not self._idempotent("on_pause", None):
1875            return
1876        state = client.state
1877        try:
1878            await self._send_progress_to_ynison(
1879                progress_ms=state.progress_ms,
1880                duration_ms=self._best_duration_ms(),
1881                paused=True,
1882                strict=True,
1883            )
1884        except YnisonSendError as exc:
1885            raise PlayerCommandFailed("Ynison send failed") from exc
1886
1887    # Entity types that use server-side "radio" queue replenishment.
1888    # Currently only RADIO (personal wave, genre stations).
1889    # Add "WAVE" here if/when Yandex supports it via the same
1890    # rotor_station_tracks API.
1891    _RADIO_ENTITY_TYPES: ClassVar[set[str]] = {"RADIO"}
1892
1893    def _maybe_prefetch(
1894        self,
1895        current_index: int,
1896        playable_list: list[dict[str, Any]],
1897        entity_id: str,
1898        entity_type: str,
1899    ) -> None:
1900        """Kick off background prefetch when nearing the end of the queue."""
1901        if entity_type not in self._RADIO_ENTITY_TYPES:
1902            return
1903        if not self._yandex_provider or not playable_list:
1904            return
1905        # second-to-last or last — trigger prefetch near end of queue
1906        if current_index < len(playable_list) - 2:
1907            return
1908        # Already prefetched or prefetch in progress
1909        if self._prefetched_list is not None:
1910            return
1911        if self._prefetch_task and not self._prefetch_task.done():
1912            return
1913
1914        self.logger.info(
1915            "Pre-fetching tracks (at index %d/%d, entity=%s)",
1916            current_index,
1917            len(playable_list),
1918            entity_id[:40] if entity_id else "<none>",
1919        )
1920
1921        async def _do_prefetch() -> None:
1922            result = await self._replenish_radio_queue(entity_id, entity_type, playable_list)
1923            if result:
1924                self._prefetched_list = result
1925                # Push expanded queue to Ynison immediately so the YM app
1926                # sees upcoming tracks and enables the "next" button.
1927                await self._update_queue_list(result)
1928
1929        self._prefetch_task = self.mass.create_task(_do_prefetch())
1930
1931    async def _signal_track_completion(self) -> None:
1932        """
1933        Signal that the current track finished playing.
1934
1935        Ynison is a state-sync protocol — the active device must advance
1936        current_playable_index itself.
1937
1938        If the next index is within the playable list, we advance immediately.
1939        If we're at the end (typical for RADIO/wave with short queues),
1940        we fetch more tracks via the Yandex Music API, append them to the
1941        playable_list, and then advance.
1942        """
1943        if not self._ynison:
1944            return
1945        state = self._ynison.state
1946        duration = self._best_duration_ms()
1947        queue = state.player_state.get("player_queue", {})
1948        current_index = queue.get("current_playable_index", 0)
1949        playable_list = queue.get("playable_list", [])
1950        entity_type = queue.get("entity_type", "")
1951        entity_id = queue.get("entity_id", "")
1952        next_index = current_index + 1
1953
1954        self.logger.info(
1955            "Track finished at index %d/%d (entity=%s type=%s), "
1956            "advancing to index %d (duration=%dms)",
1957            current_index,
1958            len(playable_list),
1959            entity_id[:40] if entity_id else "<none>",
1960            entity_type,
1961            next_index,
1962            duration,
1963        )
1964        self._actual_duration_ms = 0
1965
1966        # 1. Report that playback reached the end.
1967        # Echo tracking is handled by _send_progress_to_ynison.
1968        # `strict=True`: a dropped end-of-track signal stalls the YM app on
1969        # the just-finished track. We log and continue — the reconnect is
1970        # already scheduled and the queue-advance below sees the same WS state
1971        # — but we don't reraise (this is end-of-stream, there's no command to
1972        # fail back to the user).
1973        try:
1974            await self._send_progress_to_ynison(
1975                progress_ms=duration, duration_ms=duration, paused=False, strict=True
1976            )
1977        except YnisonSendError:
1978            self.logger.warning(
1979                "Track-completion signal dropped (Ynison transport failure); "
1980                "queue advance will retry once the WS reconnects",
1981                exc_info=True,
1982            )
1983
1984        if next_index < len(playable_list):
1985            # 2a. Queue has room — advance immediately.
1986            # Clear stale prefetch data so _maybe_prefetch can trigger for
1987            # the new queue tail on subsequent state updates.
1988            self._prefetched_list = None
1989            await self._advance_queue_index(next_index)
1990        elif entity_type in self._RADIO_ENTITY_TYPES:
1991            # 2b. At end of RADIO queue — use prefetched data or fetch now
1992            expanded: list[dict[str, Any]] | None = None
1993            if self._prefetched_list:
1994                self.logger.info("Using pre-fetched queue (%d items)", len(self._prefetched_list))
1995                expanded = self._prefetched_list
1996                self._prefetched_list = None
1997            elif self._prefetch_task and not self._prefetch_task.done():
1998                self.logger.info("Waiting for in-flight prefetch...")
1999                await self._prefetch_task
2000                expanded = self._prefetched_list
2001                self._prefetched_list = None
2002            else:
2003                expanded = await self._replenish_radio_queue(entity_id, entity_type, playable_list)
2004            if expanded and next_index < len(expanded):
2005                await self._advance_queue_index(next_index, expanded_list=expanded)
2006            elif expanded:
2007                self.logger.warning(
2008                    "Expanded queue has %d items but next_index=%d — re-fetching",
2009                    len(expanded),
2010                    next_index,
2011                )
2012                fresh = await self._replenish_radio_queue(entity_id, entity_type, expanded)
2013                if fresh and next_index < len(fresh):
2014                    await self._advance_queue_index(next_index, expanded_list=fresh)
2015                else:
2016                    self.logger.warning("Still cannot advance after re-fetch")
2017            else:
2018                self.logger.warning(
2019                    "Could not replenish queue (entity=%s type=%s), cannot advance",
2020                    entity_id,
2021                    entity_type,
2022                )
2023        else:
2024            self.logger.info(
2025                "End of non-radio queue (entity=%s type=%s), playback complete",
2026                entity_id[:40] if entity_id else "<none>",
2027                entity_type,
2028            )
2029
2030    async def _replenish_radio_queue(
2031        self,
2032        entity_id: str,
2033        entity_type: str,
2034        playable_list: list[dict[str, Any]],
2035    ) -> list[dict[str, Any]] | None:
2036        """
2037        Fetch more tracks from Yandex Music API and return expanded playable_list.
2038
2039        The active device is responsible for replenishing RADIO/wave queues.
2040        Ynison only syncs state — it does NOT generate new tracks.
2041        """
2042        if not self._yandex_provider:
2043            self.logger.warning("No yandex_music provider available for radio replenishment")
2044            return None
2045
2046        # Determine the last track ID for pagination
2047        last_track_id: str | None = None
2048        if playable_list:
2049            last_track_id = playable_list[-1].get("playable_id")
2050
2051        self.logger.info(
2052            "Fetching more tracks for %s station %s (queue=%s)",
2053            entity_type,
2054            entity_id,
2055            last_track_id,
2056        )
2057
2058        try:
2059            tracks, batch_id = await self._yandex_provider.get_rotor_station_tracks(
2060                entity_id, queue=last_track_id
2061            )
2062        except Exception:
2063            self.logger.exception("Failed to fetch radio tracks for %s", entity_id)
2064            return None
2065
2066        if not tracks:
2067            self.logger.warning("No tracks returned for station %s", entity_id)
2068            return None
2069
2070        # Determine the 'from' field from existing items
2071        from_field = ""
2072        if playable_list:
2073            from_field = playable_list[0].get("from", "")
2074
2075        # Convert tracks to Ynison playable_list format
2076        new_items: list[dict[str, Any]] = []
2077        for track in tracks:
2078            album_id = ""
2079            if hasattr(track, "albums") and track.albums:
2080                album_id = str(track.albums[0].id) if track.albums[0].id else ""
2081            cover = ""
2082            if hasattr(track, "cover_uri") and track.cover_uri:
2083                cover = track.cover_uri
2084            new_items.append(
2085                {
2086                    "playable_id": str(track.id),
2087                    "album_id_optional": album_id,
2088                    "playable_type": "TRACK",
2089                    "from": from_field,
2090                    "title": track.title or "",
2091                    "cover_url_optional": cover,
2092                }
2093            )
2094
2095        self.logger.info(
2096            "Fetched %d new tracks for station %s (batch=%s)",
2097            len(new_items),
2098            entity_id,
2099            batch_id,
2100        )
2101
2102        return list(playable_list) + new_items
2103
2104    async def _advance_queue_index(
2105        self,
2106        next_index: int,
2107        *,
2108        expanded_list: list[dict[str, Any]] | None = None,
2109    ) -> None:
2110        """
2111        Send update_player_state to advance the queue to next_index.
2112
2113        If expanded_list is provided, it replaces the playable_list
2114        (used after radio queue replenishment).
2115
2116        Waits up to 10 s for reconnection if Ynison is temporarily
2117        disconnected (e.g. after a transient error).
2118        """
2119        if not self._ynison:
2120            return
2121        if not self._ynison.connected:
2122            self.logger.info("Waiting for Ynison reconnection before advancing queue…")
2123            for _ in range(10):
2124                await asyncio.sleep(1)
2125                if not self._ynison or self._ynison.connected:
2126                    break
2127            if not self._ynison or not self._ynison.connected:
2128                self.logger.warning("Cannot advance queue — Ynison still disconnected")
2129                return
2130        state = self._ynison.state
2131        queue = state.player_state.get("player_queue", {})
2132        device_id = self._ynison.device_id
2133        new_state = dict(state.player_state)
2134        new_state["player_queue"] = dict(queue)
2135        new_state["player_queue"]["current_playable_index"] = next_index
2136        new_state["player_queue"]["version"] = make_version_block(device_id)
2137        if expanded_list is not None:
2138            new_state["player_queue"]["playable_list"] = expanded_list
2139        new_state["status"] = dict(new_state.get("status", {}))
2140        new_state["status"]["progress_ms"] = "0"
2141        new_state["status"]["duration_ms"] = "0"
2142        new_state["status"]["paused"] = False
2143        new_state["status"]["version"] = make_version_block(device_id)
2144        # `strict=True`: a dropped queue-advance leaves `_wait_for_track_change`
2145        # spinning for its full 30 s timeout. Log and return — the next
2146        # reconnect-broadcast picks up our authored version block and resyncs.
2147        try:
2148            await self._ynison.update_player_state(player_state=new_state, strict=True)
2149        except YnisonSendError:
2150            self.logger.warning(
2151                "Queue-advance dropped (Ynison transport failure); "
2152                "stream will stall until reconnect-broadcast resyncs",
2153                exc_info=True,
2154            )
2155
2156    async def _update_queue_list(self, expanded_list: list[dict[str, Any]]) -> None:
2157        """
2158        Push an expanded playable_list to Ynison without changing index or progress.
2159
2160        Called right after prefetch completes so the YM app sees upcoming
2161        tracks and enables the "next" button.
2162        """
2163        if not self._ynison or not self._ynison.connected:
2164            return
2165        state = self._ynison.state
2166        queue = state.player_state.get("player_queue", {})
2167        device_id = self._ynison.device_id
2168        new_state = dict(state.player_state)
2169        new_state["player_queue"] = dict(queue)
2170        new_state["player_queue"]["playable_list"] = expanded_list
2171        new_state["player_queue"]["version"] = make_version_block(device_id)
2172        await self._ynison.update_player_state(player_state=new_state)
2173
2174    async def _on_next(self) -> None:
2175        """Handle next track command — signal track end so Yandex advances."""
2176        self._require_connected_ynison()
2177        await self._signal_track_completion()
2178
2179    async def _on_previous(self) -> None:
2180        """Handle previous track command — update queue index in Ynison."""
2181        client = self._require_connected_ynison()
2182        queue = client.state.player_state.get("player_queue", {})
2183        current_index = queue.get("current_playable_index", 0)
2184        if current_index > 0:
2185            self._actual_duration_ms = 0
2186            await self._advance_queue_index(current_index - 1)
2187
2188    async def _on_seek(self, position: int) -> None:
2189        """
2190        Handle seek command — send position update to Ynison.
2191
2192        :param position: Position in seconds from Music Assistant.
2193        """
2194        client = self._require_connected_ynison()
2195        seek_ms = position * 1000
2196        state = client.state
2197        try:
2198            await self._send_progress_to_ynison(
2199                progress_ms=seek_ms,
2200                duration_ms=self._best_duration_ms(),
2201                paused=state.is_paused,
2202                strict=True,
2203            )
2204        except YnisonSendError as exc:
2205            # Do not mutate `_seek_position_ms` / `_seek_grace_until` on failure
2206            # — local stream state must not drift past a send that never landed.
2207            raise PlayerCommandFailed("Ynison send failed") from exc
2208        # Also trigger local stream restart so seek takes effect
2209        # immediately without waiting for the Ynison echo.
2210        self._seek_position_ms = seek_ms
2211        self._seek_grace_until = time.monotonic() + _ECHO_GRACE_PERIOD
2212        self._track_changed_event.set()
2213