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