music-assistant-server

1"""
2Spotify Soloist backend for the Spotify Connect provider.
3
4Soloist is Spotify's official headless Connect client for Linux. The daemon is
5installed/updated through ``SoloistBinaryManager``, plays its audio into a
6private PulseAudio capture sink (read back by MA as a named pipe) and is driven
7over its local WebSocket API (``SoloistClient``).
8
9SECURITY NOTE: the daemon takes the user's personal API key on its command
10line; nothing in this module may ever log the process argv.
11"""
12
13from __future__ import annotations
14
15import asyncio
16import re
17from asyncio import FIRST_COMPLETED
18from contextlib import suppress
19from dataclasses import replace
20from pathlib import Path
21from typing import TYPE_CHECKING, Any, Final
22
23from aiohttp import ClientError
24from music_assistant_models.enums import ContentType, RepeatMode, StreamType
25from music_assistant_models.errors import AudioError
26from music_assistant_models.media_items import AudioFormat
27
28from music_assistant.helpers.process import AsyncProcess
29from music_assistant.helpers.pulse_capture import (
30    CAPTURE_CHANNELS,
31    CAPTURE_SAMPLE_RATE,
32    PipeSink,
33    get_pulse_capture_server,
34)
35from music_assistant.providers.spotify_connect.base import (
36    AUDIO_QUALITY_LOSSLESS,
37    SpotifyConnectBackend,
38    spotify_source_audio_format,
39)
40from music_assistant.providers.spotify_connect.models import (
41    BackendEvent,
42    BackendEventType,
43    BackendPlaybackOptions,
44    BackendQueueEntry,
45    BackendQueueState,
46    BackendStreamSource,
47    BackendTrackMetadata,
48    QueueEntrySource,
49)
50
51from .prefs import write_audio_prefs
52from .runtime import (
53    EXIT_CODE_BUILD_EXPIRED,
54    BuildExpiredError,
55    SoloistAuthState,
56    SoloistBinaryManager,
57    SoloistClient,
58    SoloistContextChanged,
59    SoloistDeviceChanged,
60    SoloistError,
61    SoloistErrorMessage,
62    SoloistOptionsChanged,
63    SoloistPlaybackState,
64    SoloistPositionSync,
65    SoloistQueueChanged,
66    SoloistTrackChanged,
67    SoloistVolumeChanged,
68)
69
70if TYPE_CHECKING:
71    import logging
72
73    from music_assistant.helpers.pulse_capture import PulseCaptureServer
74    from music_assistant.mass import MusicAssistant
75    from music_assistant.providers.spotify_connect.models import (
76        AudioChunkReader,
77        BackendEventCallback,
78    )
79
80    from .runtime import SoloistEntity, SoloistEvent, SoloistPlaybackOptions, SoloistQueueEntry
81
82# How Spotify-side volume changes are handled (see set_volume).
83VOLUME_MODE_PLAYER_ONLY: Final = "player_only"
84VOLUME_MODE_SYNC_SPOTIFY: Final = "sync_spotify"
85
86# Bounded soloist playback cache (docs: 0 = unlimited, otherwise at least 100 MB).
87CACHE_SIZE_MB: Final = 512
88
89# Supervised-restart policy: give up after this many consecutive daemon
90# failures; the counter resets once the events websocket delivers again.
91MAX_RESTART_ATTEMPTS: Final = 5
92RESTART_DELAY_S: Final = 2
93
94# How long the daemon's log reader may keep draining buffered output after the
95# process itself exited.
96DAEMON_LOG_DRAIN_TIMEOUT_S: Final = 5
97
98# Proactive binary refresh interval: soloist builds expire 90 days after their
99# build date, so a daily check swaps in a fresh build long before a long-lived
100# instance would hit the expiry.
101BINARY_REFRESH_INTERVAL_S: Final = 24 * 3600
102
103# How often the supervisor checks whether the pulse daemon restarted (which
104# invalidates the capture sink), so the sink is replaced proactively instead
105# of on the (side-effect-free) stream request.
106GENERATION_WATCH_INTERVAL_S: Final = 5
107
108# item uri prefixes Spotify never serves losslessly, whatever the tier is set to.
109_SPOKEN_URI_PREFIXES: Final = ("spotify:episode:", "spotify:chapter:")
110
111# playback_state/playback_changed status values mapped to normalized events;
112# undocumented values degrade to OTHER.
113_STATUS_EVENTS: Final[dict[str, BackendEventType]] = {
114    "playing": BackendEventType.PLAYING,
115    "paused": BackendEventType.PAUSED,
116    "buffering": BackendEventType.BUFFERING,
117    "stopped": BackendEventType.STOPPED,
118    "idle": BackendEventType.STOPPED,
119}
120
121
122class SoloistBackend(SpotifyConnectBackend):
123    """
124    Spotify Connect backend wrapping a supervised Spotify Soloist daemon.
125
126    The daemon plays into a private PulseAudio capture sink whose FIFO is
127    consumed by the streams controller as a named pipe; control and state flow
128    over the daemon's local WebSocket API.
129    """
130
131    def __init__(  # noqa: PLR0913
132        self,
133        mass: MusicAssistant,
134        *,
135        identity_key: str,
136        publish_name: str,
137        name: str,
138        logger: logging.Logger,
139        event_callback: BackendEventCallback,
140        api_key: str,
141        consent: bool,
142        volume_mode: str = VOLUME_MODE_PLAYER_ONLY,
143        crossfade_ms: int = 0,
144        loudness_normalization: bool = True,
145        audio_quality: str = AUDIO_QUALITY_LOSSLESS,
146    ) -> None:
147        """
148        Initialize the backend (cheap; the daemon is launched in ``start``).
149
150        :param mass: The MusicAssistant instance.
151        :param identity_key: Unique identity of this daemon (one per connected
152            player); keys the data/cache dirs and the capture sink name.
153        :param publish_name: Device name advertised to the Spotify app.
154        :param name: Display name of the owning provider instance (log messages).
155        :param logger: Logger to use for diagnostics.
156        :param event_callback: Awaited with a normalized BackendEvent for every
157            state change the daemon reports.
158        :param api_key: The user's personal Spotify Soloist API key (secret,
159            kept out of all logs).
160        :param consent: Whether the user consented to downloading the soloist
161            binary from Spotify's CDN.
162        :param volume_mode: VOLUME_MODE_PLAYER_ONLY to let MA/the player own the
163            volume exclusively, VOLUME_MODE_SYNC_SPOTIFY to mirror the Spotify
164            app's volume onto the MA player.
165        :param crossfade_ms: Crossfade duration between tracks in milliseconds
166            (0 disables crossfade).
167        :param loudness_normalization: Whether Spotify's loudness normalization
168            should be applied to the audio.
169        :param audio_quality: Ceiling for the streaming quality Spotify is asked
170            to deliver (one of the AUDIO_QUALITY_* tiers).
171        """
172        self.mass = mass
173        self.logger = logger
174        self.name = name
175        self._publish_name = publish_name
176        self._event_callback = event_callback
177        self._api_key = api_key
178        self._consent = consent
179        self._volume_mode = volume_mode
180        self._crossfade_ms = crossfade_ms
181        self._loudness_normalization = loudness_normalization
182        self._audio_quality = audio_quality
183        self._data_dir = Path(mass.storage_path) / "spotify_connect" / identity_key / "soloist-data"
184        self._cache_dir = Path(mass.cache_path) / identity_key / "soloist-cache"
185        # PA sink names end up in space-delimited module arguments and env vars
186        self._sink_prefix = re.sub(r"[^A-Za-z0-9_.-]", "_", identity_key)
187        self._binary: Path | None = None
188        # digest of the build the running daemon was spawned from; the shared
189        # install can move ahead of it when a sibling instance updates first
190        self._build_sha: str | None = None
191        self._server: PulseCaptureServer | None = None
192        self._sink: PipeSink | None = None
193        self._sink_generation: int = -1
194        # serializes sink (re)creation against teardown in stop()
195        self._sink_lock = asyncio.Lock()
196        self._client: SoloistClient | None = None
197        self._stop_called: bool = False
198        self._daemon_task: asyncio.Task[None] | None = None
199        self._events_task: asyncio.Task[None] | None = None
200        self._refresh_task: asyncio.Task[None] | None = None
201        self._watcher_task: asyncio.Task[None] | None = None
202        self._proc: AsyncProcess | None = None
203        self._restart_error_count = 0
204        # set when a daemon close is intentional (sink replaced), so the
205        # supervisor respawns immediately instead of counting a failure
206        self._respawn_requested: bool = False
207        # serializes all volume handling (sink compensation + event forwarding)
208        self._volume_lock = asyncio.Lock()
209        # serializes the two-command repeat sequences of concurrent set_repeat calls
210        self._repeat_lock = asyncio.Lock()
211        # last volume reported by the daemon (None until the first event)
212        self._spotify_volume: int | None = None
213        # guards the player_only 100%-pin so overlapping resets are not issued
214        self._pin_in_flight: bool = False
215        self._last_context_uri: str | None = None
216        self._last_track_uri: str | None = None
217        # the capture sink delivers fixed s32le/44.1kHz/2ch PCM — that is what
218        # actually arrives on the named pipe, and every decision about the bytes
219        # follows it because it is reported as the decoded format
220        self._capture_format = AudioFormat(
221            content_type=ContentType.PCM_S32LE,
222            codec_type=ContentType.PCM_S32LE,
223            sample_rate=CAPTURE_SAMPLE_RATE,
224            bit_depth=32,
225            channels=CAPTURE_CHANNELS,
226        )
227        self._tier_format = spotify_source_audio_format(
228            audio_quality, lossless=audio_quality == AUDIO_QUALITY_LOSSLESS
229        )
230        # spoken content is Ogg Vorbis whatever the tier says; on the lossy tiers
231        # that is the tier format itself
232        self._spoken_format = (
233            spotify_source_audio_format(audio_quality, lossless=False)
234            if audio_quality == AUDIO_QUALITY_LOSSLESS
235            else self._tier_format
236        )
237
238    @property
239    def audio_format(self) -> AudioFormat:
240        """Return the source audio format (advertised to clients for display)."""
241        # a Connect session plays whatever the Spotify app picked, so the uri of the
242        # item playing when the stream starts is the only media-type signal there is;
243        # an unknown item is treated as music, the dominant case for a ceiling claim
244        if self._last_track_uri and self._last_track_uri.startswith(_SPOKEN_URI_PREFIXES):
245            return self._spoken_format
246        return self._tier_format
247
248    @property
249    def decoded_audio_format(self) -> AudioFormat:
250        """Return the PCM format the capture sink's named pipe delivers."""
251        # a copy per stream: the core mirrors what ffmpeg probes onto this object,
252        # which must not land on the one format every stream shares
253        return replace(self._capture_format)
254
255    @property
256    def stream_ends_on_pause(self) -> bool:
257        """The pipe sink never signals end of stream; the provider stops the player."""
258        # the sink renders only while a client is connected: silence while the daemon
259        # holds its stream open, nothing at all once the daemon drops it. A reader
260        # therefore sees neither audio nor EOF, so MA has to end the stream itself.
261        return False
262
263    @property
264    def supports_queue_control(self) -> bool:
265        """The soloist session implements the queue verbs and queue/options events."""
266        return True
267
268    async def start(self) -> None:
269        """Start the backend and its supervised soloist daemon."""
270        # setup errors (unsupported platform, missing consent, download failure,
271        # expired build) propagate so the provider load fails with a clear error
272        manager = SoloistBinaryManager(self.mass)
273        self._binary = await manager.ensure_fresh(self._consent)
274        self._build_sha = manager.diagnostics().get("sha256")
275        self._server = await get_pulse_capture_server(self.mass).acquire()
276        try:
277            await self._ensure_fresh_sink()
278
279            def _prepare_dirs() -> None:
280                self._data_dir.mkdir(parents=True, exist_ok=True)
281                # the data dir persists the Spotify device identity and login
282                # session; keep it readable by the MA user only
283                self._data_dir.chmod(0o700)
284                self._cache_dir.mkdir(parents=True, exist_ok=True)
285
286            await asyncio.to_thread(_prepare_dirs)
287            self._client = SoloistClient(self.mass, self._data_dir, self.logger)
288            # Two self-healing supervisors: one keeps the daemon process alive,
289            # the other keeps the events websocket connected (reconnecting
290            # across daemon restarts).
291            self._daemon_task = self.mass.create_task(self._daemon_runner())
292            self._events_task = self.mass.create_task(self._events_runner())
293            # Two housekeeping loops: a daily binary refresh (builds expire 90
294            # days after their build date) and a watcher replacing the capture
295            # sink after a pulse daemon restart.
296            self._refresh_task = self.mass.create_task(self._binary_refresh_loop())
297            self._watcher_task = self.mass.create_task(self._generation_watcher())
298        except BaseException:
299            # a failed startup aborts the provider load before unload() would
300            # ever run — release everything acquired so far ourselves
301            with suppress(Exception):
302                await self.stop()
303            raise
304
305    async def stop(self) -> None:
306        """Stop the daemon, its supervisors and the capture resources (idempotent)."""
307        self._stop_called = True
308        for task in (self._events_task, self._daemon_task, self._refresh_task, self._watcher_task):
309            if task and not task.done():
310                task.cancel()
311                with suppress(asyncio.CancelledError):
312                    await task
313        self._events_task = None
314        self._daemon_task = None
315        self._refresh_task = None
316        self._watcher_task = None
317        if (proc := self._proc) is not None:
318            self._proc = None
319            await proc.close()
320        # the lock lets an in-flight _ensure_fresh_sink finish before the
321        # capture resources go away (later callers fail on the stop flag)
322        async with self._sink_lock:
323            if (sink := self._sink) is not None:
324                self._sink = None
325                await sink.unload()
326            if (server := self._server) is not None:
327                self._server = None
328                await server.release()
329
330    async def get_stream_source(self) -> BackendStreamSource:
331        """
332        Return the NAMED_PIPE stream source delivering the capture sink's PCM.
333
334        Side-effect-free (it also runs from queue preload): the sink is only
335        read here — (re)creation is owned by the daemon supervisor and the
336        generation watcher.
337
338        ffmpeg reads the FIFO directly and is the single pacing owner:
339        ``-readrate 1`` paces the read to realtime with a small initial burst
340        as jitter headroom — nothing else may sleep-pace this audio path.
341
342        :raises AudioError: No usable capture sink is currently available.
343        """
344        sink = self._sink
345        server = self._server
346        if sink is None or server is None or self._sink_generation != server.generation:
347            raise AudioError("Spotify Connect capture sink is not available")
348        return BackendStreamSource(
349            stream_type=StreamType.NAMED_PIPE,
350            path=str(sink.fifo_path),
351            extra_input_args=["-readrate", "1", "-readrate_initial_burst", "0.5"],
352        )
353
354    def get_audio_reader(self) -> AudioChunkReader | None:
355        """Return None: the audio is delivered through the NAMED_PIPE stream source."""
356        return None
357
358    async def play(self, uri: str, *, skip_to_uri: str | None = None) -> None:
359        """
360        Start playing a Spotify URI/context, making this device the active one.
361
362        :param uri: Spotify URI (track, album, playlist, ...) — typically a context.
363        :param skip_to_uri: Ignored — soloist's play command has no skip-to-track
364            option, so playback starts at the beginning of the context.
365        """
366        assert self._client is not None
367        # a bare play starts local playback without a Connect transfer, leaving
368        # the Spotify apps unaware; claim active device status first
369        await self._client.activate(await_result=True)
370        await self._client.play(uri)
371        if skip_to_uri:
372            self.logger.debug(
373                "skip_to_uri is not supported by soloist; starting %s from the beginning", uri
374            )
375
376    async def resume(self) -> None:
377        """Resume playback on the active session."""
378        assert self._client is not None
379        # re-claim active device status first (idempotent when already active):
380        # a resume after a deactivate would otherwise start local playback
381        # without a Connect transfer, leaving the Spotify apps unaware
382        await self._client.activate(await_result=True)
383        await self._client.resume()
384
385    async def pause(self) -> None:
386        """Pause playback on the active session."""
387        assert self._client is not None
388        await self._client.pause()
389
390    async def deactivate(self) -> None:
391        """Release this device as the active Spotify Connect device."""
392        assert self._client is not None
393        # pause first so the session's resume position is preserved; tolerate
394        # a rejected or unacknowledged pause and still give up the device
395        with suppress(SoloistError, TimeoutError):
396            await self._client.pause(await_result=True)
397        await self._client.deactivate()
398
399    async def next(self) -> None:
400        """Skip to the next track."""
401        assert self._client is not None
402        await self._client.skip_next()
403
404    async def previous(self) -> None:
405        """Skip to the previous track (or rewind the current one)."""
406        assert self._client is not None
407        await self._client.skip_prev()
408
409    async def seek(self, position_ms: int) -> None:
410        """
411        Seek to an absolute position in the current track.
412
413        :param position_ms: Target position in milliseconds.
414        """
415        assert self._client is not None
416        await self._client.seek(position_ms)
417
418    async def set_volume(self, volume: int) -> None:
419        """
420        Set the Spotify-side playback volume.
421
422        :param volume: Absolute volume as a 0-100 percentage. In player-only
423            mode the value is ignored and the daemon stays pinned at 100%.
424        """
425        assert self._client is not None
426        if self._volume_mode == VOLUME_MODE_SYNC_SPOTIFY:
427            # the daemon echoes the change back as a volume_changed event; the
428            # provider's dedupe keeps that echo from bouncing back to the player
429            async with self._volume_lock:
430                await self._client.set_volume(volume)
431            return
432        # player_only: the MA player owns the volume, so the daemon is pinned at
433        # 100% (unity PCM into the sink); only (re)send the pin when it is known
434        # (or possibly) off
435        async with self._volume_lock:
436            if self._spotify_volume == 100:
437                return
438            await self._client.set_volume(100)
439
440    async def add_to_queue(self, uri: str) -> None:
441        """
442        Add a track to the session's play queue.
443
444        :param uri: Spotify track URI to queue.
445        """
446        assert self._client is not None
447        await self._client.add_to_queue(uri)
448
449    async def set_shuffle(self, enabled: bool) -> None:
450        """
451        Enable or disable shuffle on the active session.
452
453        :param enabled: True to enable shuffle, False to disable it.
454        """
455        assert self._client is not None
456        await self._client.set_shuffle(enabled)
457
458    async def set_repeat(self, repeat: RepeatMode) -> None:
459        """
460        Set the repeat mode on the active session.
461
462        Awaits the engine's acknowledgement of both underlying commands, so
463        this call can block and raise; never call it from the backend event
464        callback (the acknowledgements arrive on the same loop and the wait
465        could only time out).
466
467        :param repeat: OFF for no repeat, ONE for the current track, ALL for
468            the playing context.
469        """
470        assert self._client is not None
471        if repeat == RepeatMode.UNKNOWN:
472            raise ValueError("cannot apply an unknown repeat mode")
473        # soloist models repeat as two independent booleans (track/context) with
474        # both-true undefined: the lock serializes concurrent callers and the
475        # awaited acks order the pair, so one flag is always cleared before the
476        # other is raised
477        async with self._repeat_lock:
478            if repeat == RepeatMode.ONE:
479                await self._client.set_repeat_context(False, await_result=True)
480                await self._client.set_repeat_track(True, await_result=True)
481                return
482            await self._client.set_repeat_track(False, await_result=True)
483            await self._client.set_repeat_context(repeat == RepeatMode.ALL, await_result=True)
484
485    async def request_queue(self, limit: int = 10) -> None:
486        """
487        Ask the session to (re)emit its queue view (arrives as a QUEUE_CHANGED event).
488
489        :param limit: Maximum number of upcoming entries the snapshot should
490            include.
491        """
492        assert self._client is not None
493        await self._client.get_queue(limit)
494
495    async def _ensure_fresh_sink(self) -> PipeSink:
496        """
497        Return the capture sink, replacing it when the pulse daemon restarted.
498
499        Only called from the supervisor paths (daemon spawn), never from the
500        side-effect-free stream request.
501
502        :raises AudioError: The backend is (being) stopped.
503        """
504        async with self._sink_lock:
505            # checked under the lock: a concurrent stop() sets the flag before
506            # it waits for the lock to tear down the capture resources
507            if self._stop_called or self._server is None:
508                raise AudioError("Spotify Connect backend is stopping")
509            if self._sink is not None and self._sink_generation == self._server.generation:
510                return self._sink
511            if self._sink is not None:
512                # sinks do not survive a daemon restart; this only cleans up the FIFO
513                await self._sink.unload()
514            self._sink = await PipeSink.create(self._server, self._sink_prefix)
515            self._sink_generation = self._server.generation
516            # the daemon plays into the sink named in its spawn env (PULSE_SINK),
517            # so a running process must respawn against the recreated sink
518            await self._close_daemon_for_respawn()
519            return self._sink
520
521    async def _recover_sink(self) -> None:
522        """
523        Fail-closed recovery: drop the sink and daemon so the supervisor rebuilds both.
524
525        Used when the current sink can no longer be trusted (its compensation
526        state is unknown after a failed volume call, or the pulse daemon
527        restarted underneath it): audio through such a sink risks a stale
528        reciprocal gain of up to 100x, so both the sink and the daemon are
529        torn down and recreated by the daemon supervisor.
530        """
531        async with self._sink_lock:
532            if self._stop_called:
533                return
534            if (sink := self._sink) is not None:
535                self._sink = None
536                with suppress(Exception):
537                    await sink.unload()
538            await self._close_daemon_for_respawn()
539
540    async def _close_daemon_for_respawn(self) -> None:
541        """Close a running daemon so its supervisor respawns it right away (not a failure)."""
542        if (proc := self._proc) is not None:
543            self._respawn_requested = True
544            await proc.close()
545
546    async def _generation_watcher(self) -> None:
547        """Proactively replace the capture sink when the pulse daemon restarted."""
548        while True:
549            await asyncio.sleep(GENERATION_WATCH_INTERVAL_S)
550            server = self._server
551            if server is None or self._sink is None:
552                continue
553            if self._sink_generation != server.generation:
554                self.logger.info("Pulse daemon restart detected; recreating the capture sink")
555                await self._recover_sink()
556
557    def _daemon_args(self) -> list[str]:
558        """
559        Build the soloist daemon's argv.
560
561        SECURITY: the argv carries the user's API key — it must never be logged
562        or end up in any error message.
563        """
564        assert self._binary is not None
565        return [
566            str(self._binary),
567            "--device-name",
568            self._publish_name,
569            "--api-key",
570            self._api_key,
571            "--data-dir",
572            str(self._data_dir),
573            "--cache-dir",
574            str(self._cache_dir),
575            # bounded playback cache (0 would be unlimited)
576            "--cache-size",
577            str(CACHE_SIZE_MB),
578            # start at 100%: MA (or the sink compensation) owns the real volume
579            "--initial-volume",
580            "100",
581            # local WebSocket API on a free loopback port; the daemon publishes
582            # the actual endpoint in its data dir where SoloistClient finds it
583            "--ws",
584            "127.0.0.1:0",
585        ]
586
587    def _write_audio_prefs(self) -> None:
588        """Write the configured audio behavior into the engine's prefs stores (blocking)."""
589        # best-effort here: nothing outside this backend depends on the result
590        write_audio_prefs(
591            self._data_dir,
592            self.logger,
593            crossfade_ms=self._crossfade_ms,
594            loudness_normalization=self._loudness_normalization,
595            audio_quality=self._audio_quality,
596        )
597
598    async def _daemon_runner(self) -> None:
599        """Run and supervise the soloist daemon, restarting (and refreshing) as needed."""
600        # Loop forever; stop() cancels this task and the explicit stop-check below
601        # handles a graceful exit without a restart.
602        while True:
603            proc: AsyncProcess | None = None
604            returncode: int | None = None
605            try:
606                # the sink may have been replaced (pulse restart) while the
607                # daemon was down; never spawn against a stale sink
608                sink = await self._ensure_fresh_sink()
609                server = self._server
610                assert server is not None  # guaranteed by _ensure_fresh_sink
611                # the engine only reads its prefs at startup (and scrubs foreign
612                # keys from the global store when it rewrites it), so refresh the
613                # audio settings on every spawn, while the daemon is down
614                await asyncio.to_thread(self._write_audio_prefs)
615                # the explicit process name keeps AsyncProcess logging free of
616                # the argv (which carries the API key)
617                self._proc = proc = AsyncProcess(
618                    self._daemon_args(),
619                    # the daemon writes all of its logging to stdout and only
620                    # ever puts argument-parsing complaints on stderr, so the
621                    # two are merged into one captured stream. Capturing is
622                    # what makes the redaction below reachable at all: an
623                    # unset stdout is inherited, which would leak the daemon's
624                    # output straight to the server console instead.
625                    stdout=True,
626                    stderr=asyncio.subprocess.STDOUT,
627                    name=f"soloist[{self.name}]",
628                    env=server.child_env(sink.sink_name),
629                )
630                await proc.start()
631                self.logger.info("Started Spotify Connect background daemon [%s]", self.name)
632                await self._reset_volume_state(sink)
633                await self._await_daemon_exit(proc)
634            except asyncio.CancelledError:
635                raise
636            except Exception as err:
637                self.logger.warning("soloist daemon error [%s]: %s", self.name, err)
638            finally:
639                if proc:
640                    await proc.close()
641                    returncode = proc.returncode
642                # The daemon — and thus the Spotify session — is gone. Tell the
643                # provider so a dead/restarting daemon isn't treated as active and
644                # controllable; a fresh 'active' event re-establishes it on reconnect.
645                self._proc = None
646                try:
647                    await self._event_callback(BackendEvent(BackendEventType.CONNECTION_LOST))
648                except Exception:
649                    # never let a callback error replace a propagating
650                    # cancellation or kill the daemon supervisor
651                    self.logger.exception("Error while handling daemon exit")
652            if self._stop_called:
653                break
654            if self._respawn_requested:
655                # intentional close (the sink was replaced): respawn right away,
656                # this is not a daemon failure
657                self._respawn_requested = False
658                continue
659            self.logger.info("Spotify Connect background daemon stopped for %s", self.name)
660            if returncode == EXIT_CODE_BUILD_EXPIRED and not await self._refresh_expired_binary():
661                return
662            self._restart_error_count += 1
663            if self._restart_error_count >= MAX_RESTART_ATTEMPTS:
664                await self._event_callback(
665                    BackendEvent(
666                        BackendEventType.FATAL_ERROR,
667                        # fatal errors are plain (non-localized) strings for now,
668                        # matching the go-librespot backend
669                        error="soloist daemon failed to start multiple times.",
670                    )
671                )
672                return
673            await asyncio.sleep(RESTART_DELAY_S)
674
675    async def _await_daemon_exit(self, proc: AsyncProcess) -> None:
676        """
677        Wait for the daemon to exit, forwarding its log for as long as it runs.
678
679        :param proc: The running daemon process.
680        """
681        # The log is drained by a side task rather than inline: a close() from
682        # one of the other supervisors (sink replacement, binary refresh) locks
683        # readers out of the process streams for good, so waiting on the reader
684        # would hang here and the daemon would never be respawned.
685        log_task = asyncio.create_task(self._log_daemon_output(proc))
686        wait_task = asyncio.create_task(proc.wait())
687        try:
688            # Watch both: nothing else drains the daemon's stdout, so a reader
689            # that died would leave the daemon blocked on a full pipe and this
690            # wait would never return.
691            await asyncio.wait((wait_task, log_task), return_when=FIRST_COMPLETED)
692            reader_error = (
693                log_task.exception() if log_task.done() and not log_task.cancelled() else None
694            )
695            if reader_error is not None:
696                self.logger.error(
697                    "soloist log reader failed [%s]: %s; restarting the daemon",
698                    self.name,
699                    reader_error,
700                )
701                await proc.close()
702            await wait_task
703            # an exited daemon still has its last (often most telling) lines in
704            # the stream buffer; the shield keeps the reader alive across the
705            # timeout so it can drain them
706            with suppress(TimeoutError):
707                await asyncio.wait_for(asyncio.shield(log_task), DAEMON_LOG_DRAIN_TIMEOUT_S)
708        finally:
709            # a reader locked out by a close() from another supervisor never
710            # ends on its own; joining it consumes its outcome the way the
711            # other process readers in the codebase do
712            for task in (wait_task, log_task):
713                task.cancel()
714            for task in (wait_task, log_task):
715                with suppress(asyncio.CancelledError, Exception):
716                    await task
717
718    async def _log_daemon_output(self, proc: AsyncProcess) -> None:
719        """
720        Forward the daemon's log lines to our logger until its output ends.
721
722        :param proc: The running daemon process.
723        """
724        async for line in proc.iter_stdout():
725            # the third-party binary's own output may echo argv (which carries
726            # the api key), so redact it before logging
727            text = line.replace(self._api_key, "<redacted>") if self._api_key else line
728            self.logger.debug("[%s] %s", self.name, text)
729
730    async def _reset_volume_state(self, sink: PipeSink) -> None:
731        """
732        Realign the volume bookkeeping with a freshly spawned daemon.
733
734        :param sink: The capture sink the daemon plays into.
735        """
736        # the daemon always starts at --initial-volume 100; without this reset a
737        # stale reciprocal sink compensation from before a crash would amplify
738        # and clip the captured audio (sync_spotify mode)
739        async with self._volume_lock:
740            self._spotify_volume = 100
741            try:
742                await sink.set_volume(100)
743            except Exception as err:
744                # fail closed: a stale reciprocal gain may still be active on
745                # the sink — recreate sink and daemon rather than clipping
746                self.logger.warning(
747                    "Failed to reset capture sink volume (%s); recreating capture sink", err
748                )
749                await self._recover_sink()
750
751    async def _refresh_expired_binary(self) -> bool:
752        """
753        Replace the expired soloist build before the next daemon restart.
754
755        :return: True when the supervisor may restart the daemon, False when a
756            fatal error was reported and the supervisor must stop.
757        """
758        self.logger.warning("soloist build expired; looking for a replacement build")
759        try:
760            # force: the daemon itself reported expiry, so the recently-verified
761            # fast path must not hand back the same binary
762            manager = SoloistBinaryManager(self.mass)
763            self._binary = await manager.ensure_fresh(self._consent, force=True)
764            self._build_sha = manager.diagnostics().get("sha256")
765        except BuildExpiredError:
766            await self._event_callback(
767                BackendEvent(
768                    BackendEventType.FATAL_ERROR,
769                    error=(
770                        "The Spotify Soloist build expired and no replacement could be "
771                        "installed. Check the server's internet connection and reload "
772                        "this provider."
773                    ),
774                )
775            )
776            return False
777        except SoloistError as err:
778            # transient refresh problem: keep restarting, bounded by the failure cap
779            self.logger.warning("Unable to refresh the expired soloist build: %s", err)
780        return True
781
782    async def _binary_refresh_loop(self) -> None:
783        """Periodically refresh the soloist binary ahead of its 90-day build expiry."""
784        while True:
785            await asyncio.sleep(BINARY_REFRESH_INTERVAL_S)
786            try:
787                manager = SoloistBinaryManager(self.mass)
788                # a replaced build keeps the same install path, so compare the
789                # install metadata's digest against the build this daemon runs
790                # (a sibling instance may have updated the shared install)
791                binary = await manager.ensure_fresh(self._consent)
792                new_sha = manager.diagnostics().get("sha256")
793                if binary == self._binary and new_sha == self._build_sha:
794                    continue
795                self.logger.info("A fresh soloist build was installed; restarting the daemon")
796                self._binary = binary
797                self._build_sha = new_sha
798                await self._close_daemon_for_respawn()
799            except asyncio.CancelledError:
800                raise
801            except Exception as err:
802                self.logger.warning("Periodic soloist binary refresh failed: %s", err)
803
804    async def _events_runner(self) -> None:
805        """Keep the soloist events websocket connected, reconnecting as needed."""
806        assert self._client is not None
807        while not self._stop_called:
808            try:
809                if not await self._client.wait_until_ready():
810                    await asyncio.sleep(RESTART_DELAY_S)
811                    continue
812                await self._client.listen_events(self._handle_event)
813            except asyncio.CancelledError:
814                raise
815            except (TimeoutError, OSError, ClientError, SoloistError) as err:
816                # ordinary connection drop; the loop reconnects quietly
817                self.logger.debug("soloist events websocket dropped: %s", err)
818            except Exception:
819                # a defect in event handling must not kill the control plane,
820                # but unlike a connection drop it has to surface loudly
821                self.logger.exception("Unexpected error while handling soloist events")
822            if not self._stop_called:
823                await asyncio.sleep(RESTART_DELAY_S)
824
825    async def _handle_event(self, event: SoloistEvent) -> None:
826        """Adapt a raw soloist event and emit its normalized counterpart."""
827        self.logger.debug("Received %s event [%s]", event.type, self.name)
828        # A delivered event means the websocket — and thus the daemon — is
829        # healthy: reset the restart backoff counter the daemon supervisor uses.
830        # (The endpoint files the events runner polls can be stale leftovers
831        # from a previous run, so the reset cannot happen on wait_until_ready.)
832        self._restart_error_count = 0
833        if isinstance(event.data, SoloistVolumeChanged):
834            await self._handle_volume_changed(event.data.volume)
835            return
836        if (
837            isinstance(event.data, SoloistPlaybackState)
838            and event.data.volume is not None
839            and event.data.volume != self._spotify_volume
840        ):
841            # a playback_state snapshot (e.g. right after a websocket reconnect)
842            # carries the daemon's current volume; resync the pin/compensation
843            # before forwarding the playback event itself
844            await self._handle_volume_changed(event.data.volume)
845            if self._sink is None:
846                # fail-closed recovery tore down the capture path; drop this
847                # stale snapshot — the respawned daemon reports fresh state
848                return
849        if (
850            isinstance(event.data, SoloistPlaybackState)
851            and (item := event.data.item) is not None
852            and item.uri != self._last_track_uri
853        ):
854            # the snapshot describes a track we have no metadata for yet (an
855            # already-playing session at (re)connect); emit its metadata so it
856            # does not stay stale until the next track_changed
857            await self._event_callback(
858                self._make_event(BackendEventType.METADATA, metadata=_entity_metadata(item))
859            )
860        if isinstance(event.data, SoloistPlaybackState) and event.data.options is not None:
861            # a state snapshot/delta carrying the playback options doubles as an
862            # options report; emit a separate OPTIONS_CHANGED so consumers only
863            # ever need to watch one event type for options
864            await self._event_callback(
865                self._make_event(
866                    BackendEventType.OPTIONS_CHANGED,
867                    options=_backend_options(event.data.options),
868                )
869            )
870        await self._event_callback(self._translate_event(event))
871
872    async def _handle_volume_changed(self, volume: int) -> None:
873        """
874        Apply a Spotify-side volume change according to the configured volume mode.
875
876        :param volume: The reported volume as a 0-100 percentage.
877        """
878        async with self._volume_lock:
879            self._spotify_volume = volume
880            if self._volume_mode != VOLUME_MODE_SYNC_SPOTIFY:
881                # player_only: MA/the player owns the volume. Keep the daemon
882                # pinned at 100% so the captured PCM stays at unity gain, and
883                # never forward VOLUME events (they would fight the MA volume).
884                if volume != 100 and self._client is not None and not self._pin_in_flight:
885                    self._pin_in_flight = True
886                    try:
887                        await self._client.set_volume(100)
888                    except Exception as err:
889                        self.logger.debug("Failed to reset soloist volume: %s", err)
890                        # mark the volume unknown so the next snapshot
891                        # reporting the same value still retries the pin
892                        self._spotify_volume = None
893                    finally:
894                        self._pin_in_flight = False
895                return
896            # sync_spotify: the daemon attenuates the PCM it plays into the sink
897            # with Spotify's cubic volume curve; undo that with the reciprocal
898            # raw sink gain (sink_pct = 10000 / spotify_pct) so the FIFO always
899            # carries unity-gain audio, then forward the volume for the MA
900            # player to apply (subject to the provider's dedupe/grace policy).
901            # NOTE: the percentage reciprocal is the exact linear inverse
902            # because pulse's software volume is cubic in the percentage too
903            # (pa_sw_volume_to_linear(p) = (p/100)^3) — validated by capture
904            # measurement, do not "fix" this to an explicit cube.
905            if self._sink is not None:
906                # explicit zero handling: no reciprocal exists, silence the sink
907                sink_pct = 0.0 if volume <= 0 else round(10000 / volume, 2)
908                try:
909                    await self._sink.set_volume(sink_pct)
910                except Exception as err:
911                    # fail closed: the compensation state is now unknown, so
912                    # recreate sink and daemon and do NOT forward the volume —
913                    # the player must not adopt a value whose compensation
914                    # never applied
915                    self.logger.warning(
916                        "Failed to set capture sink volume (%s); recreating capture sink", err
917                    )
918                    await self._recover_sink()
919                    return
920            await self._event_callback(
921                self._make_event(BackendEventType.VOLUME, volume=max(0, volume))
922            )
923
924    def _translate_event(self, event: SoloistEvent) -> BackendEvent:
925        """Map a raw soloist event onto the normalized BackendEvent model."""
926        data = event.data
927        if isinstance(data, SoloistAuthState):
928            # a logged-out daemon keeps advertising itself for Connect, so being
929            # signed out is just a session that ended: awaiting a first pairing,
930            # the user signing out, or another account taking the device over
931            return self._make_event(
932                BackendEventType.SESSION_ACTIVE
933                if data.logged_in and data.is_active
934                else BackendEventType.SESSION_INACTIVE
935            )
936        if isinstance(data, SoloistDeviceChanged):
937            return self._make_event(
938                BackendEventType.SESSION_ACTIVE
939                if data.is_active
940                else BackendEventType.SESSION_INACTIVE
941            )
942        if isinstance(data, SoloistPlaybackState):
943            # covers both the playback_state snapshot and the playback_changed delta
944            self._cache_uris(track=data.item, context=data.context)
945            return self._make_event(_STATUS_EVENTS.get(data.status, BackendEventType.OTHER))
946        if isinstance(data, SoloistTrackChanged) and data.item is not None:
947            self._cache_uris(track=data.item)
948            return self._make_event(BackendEventType.METADATA, metadata=_entity_metadata(data.item))
949        if isinstance(data, SoloistPositionSync):
950            return self._make_event(
951                BackendEventType.POSITION, position=data.position.position_ms // 1000
952            )
953        if isinstance(data, SoloistErrorMessage):
954            return self._make_event(BackendEventType.ERROR, error=data.message)
955        if isinstance(data, SoloistQueueChanged):
956            return self._make_event(
957                BackendEventType.QUEUE_CHANGED,
958                queue=BackendQueueState(
959                    previous=_queue_entries(data.previous),
960                    upcoming=_queue_entries(data.upcoming),
961                ),
962            )
963        if isinstance(data, SoloistOptionsChanged):
964            return self._make_event(
965                BackendEventType.OPTIONS_CHANGED, options=_backend_options(data.options)
966            )
967        if isinstance(data, SoloistContextChanged):
968            self._cache_uris(context=data.context)
969        # context changes, command acks and unknown events
970        return self._make_event(BackendEventType.OTHER)
971
972    def _make_event(self, event_type: BackendEventType, **fields: Any) -> BackendEvent:
973        """Build a BackendEvent carrying the latest known context/track uris."""
974        return BackendEvent(
975            event_type,
976            context_uri=self._last_context_uri,
977            track_uri=self._last_track_uri,
978            **fields,
979        )
980
981    def _cache_uris(
982        self, *, track: SoloistEntity | None = None, context: SoloistEntity | None = None
983    ) -> None:
984        """Remember the latest context/track uris seen on the event stream."""
985        if track is not None and track.uri:
986            self._last_track_uri = track.uri
987        if context is not None and context.uri:
988            self._last_context_uri = context.uri
989
990
991def _entity_metadata(item: SoloistEntity) -> BackendTrackMetadata:
992    """
993    Extract normalized track metadata from an entity's decorations.
994
995    Mapped from the observed 1.3.7 wire format: the artist lives under
996    ``creators[].entity``, the album under ``parent.entity`` and artwork
997    under ``visual_identity.cover[]`` — all traversed defensively since the
998    decorations bag is extensible.
999
1000    :param item: The soloist entity (track) to extract the metadata from.
1001    """
1002    decorations = item.decorations or {}
1003    playback = _as_dict(decorations.get("playback"))
1004    duration_ms = playback.get("duration_ms")
1005    return BackendTrackMetadata(
1006        track_uri=item.uri,
1007        title=_identity_title(decorations),
1008        artist=_creator_name(decorations.get("creators")),
1009        album=_nested_entity_name(decorations.get("parent")),
1010        image_url=_cover_url(decorations),
1011        duration=int(duration_ms) // 1000 if isinstance(duration_ms, int | float) else None,
1012    )
1013
1014
1015def _queue_entries(entries: list[SoloistQueueEntry]) -> list[BackendQueueEntry]:
1016    """
1017    Map soloist queue entries onto normalized queue entries.
1018
1019    Entries without a resolvable uri are skipped.
1020
1021    :param entries: The soloist queue entries (previous or upcoming listing).
1022    """
1023    result: list[BackendQueueEntry] = []
1024    for entry in entries:
1025        if entry.item is None or not entry.item.uri:
1026            continue
1027        result.append(
1028            BackendQueueEntry(
1029                uid=entry.uid,
1030                uri=entry.item.uri,
1031                source=QueueEntrySource(entry.source),
1032                name=_identity_title(entry.item.decorations or {}),
1033            )
1034        )
1035    return result
1036
1037
1038# soloist's repeat vocabulary mapped onto the MA repeat modes
1039_REPEAT_MODES: Final = {
1040    "off": RepeatMode.OFF,
1041    "context": RepeatMode.ALL,
1042    "track": RepeatMode.ONE,
1043}
1044
1045
1046def _backend_options(options: SoloistPlaybackOptions) -> BackendPlaybackOptions:
1047    """Map soloist playback options onto the normalized model."""
1048    return BackendPlaybackOptions(
1049        shuffle=options.shuffle,
1050        repeat=_REPEAT_MODES.get(options.repeat, RepeatMode.UNKNOWN),
1051    )
1052
1053
1054def _as_dict(value: Any) -> dict[str, Any]:
1055    """Return the value if it is a dict, an empty dict otherwise."""
1056    return value if isinstance(value, dict) else {}
1057
1058
1059def _as_str(value: Any) -> str | None:
1060    """Return the value if it is a non-empty string, None otherwise."""
1061    return value if isinstance(value, str) and value else None
1062
1063
1064def _identity_title(decorations: dict[str, Any]) -> str | None:
1065    """Return the display title from a decorations bag's identity."""
1066    identity = _as_dict(decorations.get("identity"))
1067    return _as_str(identity.get("name")) or _as_str(identity.get("title"))
1068
1069
1070def _entity_name(value: Any) -> str | None:
1071    """Return the display name of a nested entity-like decoration value."""
1072    if isinstance(value, str):
1073        return value or None
1074    if isinstance(value, dict):
1075        return _as_str(value.get("name")) or _as_str(_as_dict(value.get("identity")).get("name"))
1076    return None
1077
1078
1079def _nested_entity_name(value: Any) -> str | None:
1080    """Return the identity name of a ``{"entity": {...}}`` decoration (album parent)."""
1081    entity = _as_dict(_as_dict(value).get("entity"))
1082    return _entity_name(_as_dict(entity.get("decorations"))) or _entity_name(entity)
1083
1084
1085def _creator_name(value: Any) -> str | None:
1086    """Return the name of the first creator credit (``creators[].entity``)."""
1087    if isinstance(value, list) and value:
1088        return _nested_entity_name(value[0])
1089    return None
1090
1091
1092def _cover_url(decorations: dict[str, Any]) -> str | None:
1093    """
1094    Return the artwork url from ``visual_identity.cover[]``.
1095
1096    Prefers the large rendition; falls back to the last (largest) entry.
1097    """
1098    covers = _as_dict(decorations.get("visual_identity")).get("cover")
1099    if not isinstance(covers, list) or not covers:
1100        return None
1101    for entry in covers:
1102        if isinstance(entry, dict) and entry.get("size") == "large":
1103            if url := _as_str(entry.get("url")):
1104                return url
1105    for entry in reversed(covers):
1106        if isinstance(entry, dict) and (url := _as_str(entry.get("url"))):
1107            return url
1108    return None
1109