music-assistant-server

1"""Playback session coordinator for Sendspin players."""
2
3from __future__ import annotations
4
5import asyncio
6import time
7from collections import deque
8from contextlib import suppress
9from dataclasses import dataclass, field
10from typing import TYPE_CHECKING, Any, cast
11from uuid import UUID, uuid4
12
13from aiosendspin.models.types import AudioCodec as SendspinAudioCodec
14from aiosendspin.server.audio import AudioFormat as SendspinAudioFormat
15from aiosendspin.server.push_stream import MAIN_CHANNEL, PushStream, StreamStoppedError
16from aiosendspin.server.roles.player.v1 import PlayerV1Role
17from music_assistant_models.enums import ContentType, MediaType
18from music_assistant_models.media_items.audio_format import AudioFormat
19
20from music_assistant.constants import CONF_OUTPUT_CHANNELS
21from music_assistant.controllers.streams.audio_processing import get_media_session_id
22from music_assistant.helpers.audio import iter_pcm_slices
23from music_assistant.helpers.ffmpeg import FFMpeg
24from music_assistant.models.player import PlayerMedia
25from music_assistant.providers.sendspin.bridge_role import (
26    BRIDGE_BIT_DEPTH,
27    BRIDGE_CHANNELS,
28    BRIDGE_SAMPLE_RATE,
29    BridgePlayerRole,
30)
31
32if TYPE_CHECKING:
33    from music_assistant.helpers.dsp import ComplexFilter
34
35    from .player import SendspinPlayer
36    from .provider import SendspinProvider
37
38
39# Default session PCM format (MA-side and wire) used until _run_playback picks a
40# leader-driven rate. Same sample format expressed in both MA and Sendspin types.
41_DEFAULT_PCM_FORMAT = AudioFormat(
42    content_type=ContentType.PCM_F32LE,
43    sample_rate=48000,
44    bit_depth=32,
45    channels=2,
46)
47_DEFAULT_SENDSPIN_PCM_FORMAT = SendspinAudioFormat(
48    sample_rate=48000,
49    bit_depth=32,
50    channels=2,
51    sample_type="float",
52)
53# Media types whose upstream feeds at realtime rate, so the Sendspin queue cannot
54# grow after playback begins, so their send-ahead stays at the min_buffer_ms floor.
55# Buffered types (tracks, podcasts, etc.) race ahead and fill the queue naturally, so
56# their send-ahead may extend to a larger required_lead_time_ms without lasting cost.
57_LIVE_MEDIA_TYPES: frozenset[MediaType] = frozenset(
58    {
59        MediaType.RADIO,
60        MediaType.AUDIO_SOURCE,
61        MediaType.PLUGIN_SOURCE,
62        MediaType.FLOW_STREAM,
63    }
64)
65
66
67# Sample rate ceiling for lossy output codecs — anything above is wasted bandwidth.
68_LOSSY_MAX_SAMPLE_RATE = 48000
69# Max PCM slice fed to the producer per iteration.
70_PRODUCER_SLICE_US = 100_000
71# Max pending chunks between producer and committer before the producer blocks.
72_PRODUCER_BACKLOG_SIZE = 64
73# Backpressure threshold: push stream sleeps when buffered audio exceeds this.
74_PRODUCER_BUFFER_LIMIT_US = 30_000_000
75# Start join promotion once catchup processor lag is within this window of the history tail.
76_JOIN_PROMOTE_ARM_WINDOW_US = 2_000_000
77# Accept catchup output within this margin of the promotion target.
78_JOIN_PROMOTE_TOLERANCE_US = 50_000
79# Abort join catchup if promotion hasn't completed within this.
80_JOIN_PROMOTION_TIMEOUT_S = 15.0
81# Retain committed history this far behind real-time for late-join backfill.
82# This pre-history also warms up ffmpeg's internal filter buffers so the DSP
83# output has settled by the time the member's channel goes live.
84_HISTORY_KEEP_PAST_US = 1_000_000
85
86
87class _BufferedFfmpegProcessor:
88    """FFmpeg wrapper with small output carry-over buffer and duration-based reads."""
89
90    def __init__(self, ffmpeg: FFMpeg, audio_format: AudioFormat) -> None:
91        self._ffmpeg = ffmpeg
92        self._output_buffer = bytearray()
93        bytes_per_sample = max(1, int(audio_format.bit_depth // 8))
94        self._sample_rate = int(audio_format.sample_rate)
95        self._frame_size = bytes_per_sample * int(audio_format.channels)
96        self._bytes_per_second = self._sample_rate * self._frame_size
97        # ~25ms worth of audio per read syscall.
98        self._read_quantum_bytes = max(1, int(self._bytes_per_second * 0.025))
99        self._produced_output_us = 0
100        self._pending_skip_bytes = 0
101
102    async def start(self) -> None:
103        await self._ffmpeg.start()
104
105    async def close(self) -> None:
106        await self._ffmpeg.close()
107
108    async def push(self, pcm: bytes) -> None:
109        await self._ffmpeg.write(pcm)
110
111    async def write_eof(self) -> None:
112        """Signal no more input, causing ffmpeg to flush its internal buffers."""
113        await self._ffmpeg.write_eof()
114
115    @property
116    def produced_output_us(self) -> int:
117        """Return cumulative output duration currently drained from ffmpeg."""
118        return self._produced_output_us
119
120    async def read_duration_us(self, duration_us: int) -> bytes:
121        """Block-read exactly `duration_us` worth of processed PCM from ffmpeg."""
122        target_bytes = self._target_bytes_for_duration_us(duration_us)
123        if target_bytes == 0:
124            return b""
125
126        while len(self._output_buffer) < target_bytes:
127            missing = target_bytes - len(self._output_buffer)
128            read_size = max(self._read_quantum_bytes, missing)
129            chunk = await self._ffmpeg.readexactly(read_size)
130            if not chunk:
131                break
132            chunk = self._consume_pending_skip(chunk)
133            if chunk:
134                self._output_buffer.extend(chunk)
135
136        out = bytes(self._output_buffer[:target_bytes])
137        del self._output_buffer[:target_bytes]
138        return out
139
140    async def drain_available(self) -> int:
141        """
142        Non-blocking drain of ffmpeg stdout into internal buffer.
143
144        Returns cumulative produced output duration in microseconds.
145        """
146        while True:
147            try:
148                # 1ms timeout: non-blocking check for available data.
149                chunk = await asyncio.wait_for(
150                    self._ffmpeg.read(self._read_quantum_bytes),
151                    timeout=0.001,
152                )
153            except TimeoutError:
154                break
155            if not chunk:
156                break
157            self._produced_output_us += self._duration_us_for_bytes(len(chunk))
158            chunk = self._consume_pending_skip(chunk)
159            if chunk:
160                self._output_buffer.extend(chunk)
161            if len(chunk) < self._read_quantum_bytes:
162                break
163        return self._produced_output_us
164
165    async def drain_forever(self) -> None:
166        """Continuously drain ffmpeg stdout into internal buffer until EOF."""
167        while True:
168            chunk = await self._ffmpeg.read(self._read_quantum_bytes)
169            if not chunk:
170                break
171            self._produced_output_us += self._duration_us_for_bytes(len(chunk))
172            chunk = self._consume_pending_skip(chunk)
173            if chunk:
174                self._output_buffer.extend(chunk)
175
176    def pop_duration_us(self, duration_us: int) -> bytes | None:
177        """Pop exactly `duration_us` from already buffered output, or None if insufficient."""
178        target_bytes = self._target_bytes_for_duration_us(duration_us)
179        if target_bytes == 0:
180            return b""
181        if len(self._output_buffer) < target_bytes:
182            return None
183        out = bytes(self._output_buffer[:target_bytes])
184        del self._output_buffer[:target_bytes]
185        return out
186
187    def buffered_duration_us(self) -> int:
188        """Return buffered output duration currently available for immediate pop."""
189        return self._duration_us_for_bytes(len(self._output_buffer))
190
191    def pop_duration_us_or_pad(self, duration_us: int, pad_tolerance_us: int) -> bytes | None:
192        """Pop target duration; if short within tolerance, pad tail with silence."""
193        target_bytes = self._target_bytes_for_duration_us(duration_us)
194        if target_bytes == 0:
195            return b""
196        available = len(self._output_buffer)
197        if available >= target_bytes:
198            out = bytes(self._output_buffer[:target_bytes])
199            del self._output_buffer[:target_bytes]
200            return out
201        short_bytes = target_bytes - available
202        short_us = self._duration_us_for_bytes(short_bytes)
203        if short_us > max(0, pad_tolerance_us):
204            return None
205        out = bytes(self._output_buffer)
206        self._output_buffer.clear()
207        self._pending_skip_bytes += short_bytes
208        return out + (b"\x00" * short_bytes)
209
210    def pad_and_skip(self, duration_us: int) -> bytes:
211        """Return silence PCM and skip the equivalent from upcoming ffmpeg output."""
212        target_bytes = self._target_bytes_for_duration_us(duration_us)
213        if target_bytes <= 0:
214            return b""
215        # Drop buffered output with stale source positions.
216        leftover = len(self._output_buffer)
217        self._output_buffer.clear()
218        self._pending_skip_bytes += max(0, target_bytes - leftover)
219        return b"\x00" * target_bytes
220
221    def _consume_pending_skip(self, chunk: bytes) -> bytes:
222        # Drops bytes pop_duration_us_or_pad replaced with silence to keep timeline aligned.
223        if self._pending_skip_bytes <= 0 or not chunk:
224            return chunk
225        skip = min(self._pending_skip_bytes, len(chunk))
226        self._pending_skip_bytes -= skip
227        return chunk[skip:]
228
229    def _duration_us_for_bytes(self, byte_count: int) -> int:
230        if byte_count <= 0 or self._sample_rate <= 0 or self._frame_size <= 0:
231            return 0
232        frames = byte_count // self._frame_size
233        if frames <= 0:
234            return 0
235        return int((frames * 1_000_000) / self._sample_rate)
236
237    def _target_bytes_for_duration_us(self, duration_us: int) -> int:
238        """Convert duration to frame-aligned PCM byte count."""
239        if duration_us <= 0 or self._sample_rate <= 0 or self._frame_size <= 0:
240            return 0
241        samples = max(0, int((duration_us * self._sample_rate + 500_000) / 1_000_000))
242        return samples * self._frame_size
243
244
245@dataclass(slots=True)
246class _HistoryChunk:
247    start_time_us: int
248    duration_us: int
249    pcm: bytes
250
251
252@dataclass(slots=True)
253class _PendingChunk:
254    pcm: bytes
255    duration_us: int
256
257
258@dataclass(slots=True)
259class _JoinCatchupState:
260    """
261    Per-member state for a join-catchup processor replaying history through DSP.
262
263    The processor is fed historical + live PCM via ``input_queue``.  Once its
264    output catches up to the live stream (within tolerance), it is promoted to
265    the member's live pipeline.  See ``_inject_ready_join_historical`` for the
266    full promotion lifecycle.
267    """
268
269    processor: _BufferedFfmpegProcessor
270    input_queue: asyncio.Queue[bytes | None]
271    writer_task: asyncio.Task[None]
272    drainer_task: asyncio.Task[None]
273    snapshot_task: asyncio.Task[None] | None = None
274    # Timeline position of the first history chunk fed into the processor.
275    first_history_start_us: int | None = None
276    # Timeline position up to which PCM has been enqueued into the processor.
277    fed_until_us: int | None = None
278    # End of the history snapshot taken when catchup started.
279    history_end_us: int | None = None
280    # Locked target: once set, promotion fires when output reaches this point.
281    promotion_target_end_us: int | None = None
282    # Monotonic time when promotion was armed, used for timeout detection.
283    promotion_armed_monotonic_s: float | None = None
284    write_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
285
286
287@dataclass(slots=True)
288class _PipelineConfig:
289    requires_transform: bool
290    output_channels: str
291    filter_params: tuple[str | ComplexFilter, ...]
292
293    @property
294    def signature(self) -> tuple[bool, str, tuple[str | ComplexFilter, ...]]:
295        return (self.requires_transform, self.output_channels, self.filter_params)
296
297
298@dataclass(slots=True)
299class _MemberPipeline:
300    player_id: str
301    channel_id: UUID
302    config: _PipelineConfig
303    processor: _BufferedFfmpegProcessor | None = None
304    ready: bool = False
305
306
307class SendspinPlaybackSession:
308    """
309    Coordinates playback for a Sendspin player group leader.
310
311    The push stream supports multi-channel audio: members that need per-player
312    DSP (EQ, channel mixing, output routing) each get a dedicated ffmpeg
313    processor and a separate channel. Members without DSP share MAIN_CHANNEL
314    and receive the raw PCM directly.
315
316    Playback runs as two concurrent coroutines inside ``_run_playback``:
317
318    * **Producer** -- reads PCM from the MA stream, slices it into fixed-size
319      chunks, queues them, and writes each slice into per-member ffmpeg
320      processors (transform push) in parallel.
321    * **Consumer** -- dequeues chunks, reads the corresponding transformed
322      output from each processor (transform read), prepares all channels on
323      the push stream, commits audio, and applies backpressure via
324      ``sleep_to_limit_buffer``.
325
326    When a new member joins mid-playback, a *join-catchup* processor replays
327    committed history through the member's DSP chain so it can be promoted
328    to the live pipeline without an audible gap.
329    """
330
331    def __init__(self, player: SendspinPlayer) -> None:
332        """Initialize session coordinator bound to the owning player."""
333        self.player = player
334        self.playback_task: asyncio.Task[None] | None = None
335        self.pending_join_members: set[str] = set()
336        self._state_lock = asyncio.Lock()
337        self._members: set[str] = set()
338        self._member_pipelines: dict[str, _MemberPipeline] = {}
339        self._push_stream: PushStream | None = None
340        self._playback_running = False
341        self._producer_eof_sent = False
342        self._timeline_start_us: int | None = None
343        self._first_commit_monotonic_us: int | None = None
344        self._produced_audio_us = 0
345        self._history: deque[_HistoryChunk] = deque()
346        self._join_catchup: dict[str, _JoinCatchupState] = {}
347        self._pipeline_config_cache: dict[str, _PipelineConfig] = {}
348        self._preassigned_channels: dict[str, UUID] = {}
349        self._mapping_dirty = True
350        self._cancel_requested = False
351        # PCM formats are session-scoped and refreshed at the start of every
352        # _run_playback: the wire/MA-side rate is taken from the leader player's
353        # preferred format (capped at 48 kHz for lossy codecs), F32 is always used
354        # internally for DSP headroom.
355        self._pcm_format: AudioFormat = _DEFAULT_PCM_FORMAT
356        self._sendspin_pcm_format: SendspinAudioFormat = _DEFAULT_SENDSPIN_PCM_FORMAT
357        self._queue_id: str | None = None
358        self._queue_session_id: str | None = None
359
360    def flow_track_anchor_us(self, track_start_offset_us: int) -> int | None:
361        """
362        Server-clock time of the current flow track's file-position 0.
363
364        ``track_start_offset_us`` is the current track's start offset within the
365        flow stream (minus its file seek), so beats timed from the track file
366        map onto the shared audio timeline regardless of queue position. Returns
367        None until the first chunk commits and the timeline anchor is known.
368        """
369        if self._timeline_start_us is None:
370            return None
371        return self._timeline_start_us + track_start_offset_us
372
373    # -- Public API ------------------------------------------------------------
374
375    async def transfer_to(self, new_player: SendspinPlayer) -> None:
376        """
377        Transfer session ownership to a new player.
378
379        Used during dynamic leader switching to keep the push stream alive
380        while the old leader is removed from the sendspin group. The PushStream
381        and all internal state (pipelines, history, join-catchup) stay intact;
382        only the owning player reference is updated.
383
384        Cleans up the old leader's pipeline/channel state so its FFmpeg
385        processor is released.
386
387        :param new_player: The SendspinPlayer that will take over as session owner.
388        """
389        old_leader_id = self.player.player_id
390        self.player = new_player
391        # Release the old leader's DSP pipeline -- it's no longer in the group
392        # and _refresh_member_mappings won't touch it since it only iterates
393        # current members + the (new) leader.
394        async with self._state_lock:
395            pipeline = self._member_pipelines.pop(old_leader_id, None)
396            self._pipeline_config_cache.pop(old_leader_id, None)
397            self._preassigned_channels.pop(old_leader_id, None)
398            self._mapping_dirty = True
399        if pipeline is not None and pipeline.processor is not None:
400            await self._close_member_ffmpeg(pipeline.processor)
401
402    async def cancel(self, reason: str) -> None:
403        """Cancel and await the active playback task, if any."""
404        task = self.playback_task
405        if task is None:
406            return
407        if task.done():
408            if self.playback_task is task:
409                self.playback_task = None
410            return
411        self.player.logger.debug("Cancelling playback task (%s)", reason)
412        self._cancel_requested = True
413        task.cancel()
414        with suppress(asyncio.CancelledError, Exception):
415            await task
416        if self.playback_task is task:
417            self.playback_task = None
418
419    async def start(self, media: PlayerMedia, restart: bool = False) -> None:
420        """Start background playback for `media`."""
421        active_task = self.playback_task
422        if active_task is not None and not active_task.done():
423            if not restart:
424                raise RuntimeError("playback already active")
425            await self.cancel("restart requested")
426        self._cancel_requested = False
427        self.playback_task = asyncio.create_task(self._run_playback(media))
428
429    async def close(self) -> None:
430        """Stop playback and release all managed resources."""
431        await self.cancel("session close")
432        self.pending_join_members.clear()
433        async with self._state_lock:
434            self._members.clear()
435            self._mapping_dirty = True
436        await self._clear_member_pipelines()
437        await self._clear_join_catchup()
438        async with self._state_lock:
439            self._history.clear()
440            self._produced_audio_us = 0
441            self._timeline_start_us = None
442            self._first_commit_monotonic_us = None
443            self._pipeline_config_cache.clear()
444            self._preassigned_channels.clear()
445
446    async def add_member(self, player_id: str) -> None:
447        """Add a member to the group with DSP-aware lifecycle handling."""
448        async with self._state_lock:
449            if player_id in self._members:
450                return
451            self.pending_join_members.add(player_id)
452            # Preserve any channel pre-resolved during add_client so join-time
453            # role requirements and prepared audio stay on the same channel.
454            self._preassigned_channels.setdefault(player_id, uuid4())
455        try:
456            await self._start_join_catchup(player_id)
457        except Exception:
458            async with self._state_lock:
459                self.pending_join_members.discard(player_id)
460            await self._release_player_channel(player_id)
461            raise
462        # Promote to full member even if already pending to avoid losing
463        # the join when a cancelled task clears our pending flag.
464        async with self._state_lock:
465            if player_id not in self.pending_join_members:
466                return
467            self._members.add(player_id)
468            self._mapping_dirty = True
469            self.pending_join_members.discard(player_id)
470
471    async def remove_member(self, player_id: str) -> None:
472        """Remove a member from the group and clean up per-member playback state."""
473        async with self._state_lock:
474            self.pending_join_members.discard(player_id)
475            self._members.discard(player_id)
476            self._mapping_dirty = True
477            self._pipeline_config_cache.pop(player_id, None)
478            self._preassigned_channels.pop(player_id, None)
479        await self._stop_join_catchup(player_id)
480        await self._release_player_channel(player_id)
481
482    async def sync_members(self, member_ids: set[str]) -> None:
483        """Reconcile session members to exactly the provided set."""
484        async with self._state_lock:
485            current_members = set(self._members)
486            stale_pending = self.pending_join_members - member_ids
487        for player_id in stale_pending:
488            await self.remove_member(player_id)
489        for player_id in current_members - member_ids:
490            await self.remove_member(player_id)
491        for player_id in member_ids - current_members:
492            await self.add_member(player_id)
493
494    # -- Helpers ---------------------------------------------------------------
495
496    def _attach_task_exception_logger(self, task: asyncio.Task[Any], name: str) -> None:
497        """Log unhandled exception from background task when it finishes."""
498
499        def _done_callback(done_task: asyncio.Task[Any]) -> None:
500            if done_task.cancelled():
501                return
502            with suppress(Exception):
503                exc = done_task.exception()
504                if exc is not None:
505                    self.player.logger.exception(
506                        "Background task failed: %s",
507                        name,
508                        exc_info=exc,
509                    )
510
511        task.add_done_callback(_done_callback)
512
513    def _get_join_readiness(self) -> tuple[bool, str | None]:
514        """Check whether live join DSP preparation can be performed right now."""
515        if self._playback_running and self._push_stream is not None:
516            return (True, None)
517        return (False, "no active stream context")
518
519    # -- Snapshot helper -------------------------------------------------------
520
521    async def _snapshot_active_pipelines(
522        self,
523    ) -> tuple[set[str], tuple[tuple[str, _MemberPipeline], ...]]:
524        """Return (join_pending_ids, active_pipelines) under lock."""
525        async with self._state_lock:
526            members = self._members
527            leader_id = self.player.player_id
528            return set(self._join_catchup), tuple(
529                (mid, p)
530                for mid, p in self._member_pipelines.items()
531                if mid in members or mid == leader_id
532            )
533
534    # -- Join catchup ----------------------------------------------------------
535
536    async def _start_join_catchup(self, player_id: str) -> None:  # noqa: PLR0915
537        """Start dedicated join catchup processor fed from committed history."""
538        async with self._state_lock:
539            playback_active = self._playback_running and self._push_stream is not None
540        if not playback_active:
541            return
542
543        pipeline = await self._sync_member_pipeline(player_id)
544        if not pipeline.config.requires_transform:
545            return
546
547        await self._stop_join_catchup(player_id)
548
549        ffmpeg_obj = self._create_member_ffmpeg(pipeline.config.filter_params)
550        processor = _BufferedFfmpegProcessor(ffmpeg_obj, self._pcm_format)
551        await processor.start()
552        # Bounded queue sized to hold the full buffer duration with some headroom.
553        queue_size = (_PRODUCER_BUFFER_LIMIT_US // _PRODUCER_SLICE_US) + _PRODUCER_BACKLOG_SIZE
554        input_queue: asyncio.Queue[bytes | None] = asyncio.Queue(maxsize=queue_size)
555
556        async with self._state_lock:
557            history_snapshot = list(self._history)
558        if not history_snapshot:
559            await processor.close()
560            return
561        history_end_us = history_snapshot[-1].start_time_us + history_snapshot[-1].duration_us
562        writer_task: asyncio.Task[None] | None = None
563        drainer_task: asyncio.Task[None] | None = None
564        snapshot_task: asyncio.Task[None] | None = None
565        state: _JoinCatchupState | None = None
566        registered = False
567
568        async def _writer() -> None:
569            while True:
570                chunk = await input_queue.get()
571                if chunk is None:
572                    return
573                await processor.push(chunk)
574
575        async def _drainer() -> None:
576            await processor.drain_forever()
577
578        try:
579            async with self._state_lock:
580                writer_task = asyncio.create_task(_writer())
581                drainer_task = asyncio.create_task(_drainer())
582                self._attach_task_exception_logger(writer_task, f"join_writer_{player_id}")
583                self._attach_task_exception_logger(drainer_task, f"join_drainer_{player_id}")
584
585                state = _JoinCatchupState(
586                    processor=processor,
587                    input_queue=input_queue,
588                    writer_task=writer_task,
589                    drainer_task=drainer_task,
590                    history_end_us=history_end_us,
591                )
592                self._join_catchup[player_id] = state
593                snapshot_task = asyncio.create_task(
594                    self._feed_join_history(player_id, processor, history_snapshot)
595                )
596                self._attach_task_exception_logger(snapshot_task, f"join_snapshot_{player_id}")
597                state.snapshot_task = snapshot_task
598                registered = True
599        except BaseException:
600            if not registered:
601                # If registered, cleanup is handled via _stop_join_catchup/_clear_join_catchup.
602                async with self._state_lock:
603                    current = self._join_catchup.get(player_id)
604                    if current is state:
605                        self._join_catchup.pop(player_id, None)
606                for task in (snapshot_task, drainer_task, writer_task):
607                    if task is None:
608                        continue
609                    task.cancel()
610                    with suppress(asyncio.CancelledError, Exception):
611                        await task
612                with suppress(Exception):
613                    await processor.close()
614            raise
615
616    async def _feed_join_history(
617        self,
618        player_id: str,
619        processor: _BufferedFfmpegProcessor,
620        history_snapshot: list[_HistoryChunk],
621    ) -> None:
622        """Feed historical PCM into a join-catchup processor."""
623        async with self._state_lock:
624            state = self._join_catchup.get(player_id)
625        if state is None or state.processor is not processor:
626            return
627        async with state.write_lock:
628            first_history_start_us: int | None = None
629            previous_end_us: int | None = None
630            for hist_chunk in history_snapshot:
631                if first_history_start_us is None:
632                    first_history_start_us = hist_chunk.start_time_us
633                    async with self._state_lock:
634                        current = self._join_catchup.get(player_id)
635                        if current is not None and current.processor is processor:
636                            current.first_history_start_us = first_history_start_us
637                            current.fed_until_us = first_history_start_us
638                if previous_end_us is not None and hist_chunk.start_time_us > previous_end_us:
639                    gap_us = hist_chunk.start_time_us - previous_end_us
640                    silence = self._silence_for_duration_us(gap_us)
641                    if silence:
642                        await self._enqueue_join_pcm(state, silence)
643                await self._enqueue_join_pcm(state, hist_chunk.pcm)
644                previous_end_us = hist_chunk.start_time_us + hist_chunk.duration_us
645                async with self._state_lock:
646                    current = self._join_catchup.get(player_id)
647                    if current is not None and current.processor is processor:
648                        current.fed_until_us = previous_end_us
649
650    async def _stop_join_catchup(self, player_id: str) -> None:
651        """Stop and remove dedicated join catchup processor for one player."""
652        async with self._state_lock:
653            state = self._join_catchup.pop(player_id, None)
654        if state is None:
655            return
656        if state.snapshot_task is not None:
657            state.snapshot_task.cancel()
658            with suppress(asyncio.CancelledError, Exception):
659                await state.snapshot_task
660        state.writer_task.cancel()
661        with suppress(asyncio.CancelledError, Exception):
662            await state.writer_task
663        state.drainer_task.cancel()
664        with suppress(asyncio.CancelledError, Exception):
665            await state.drainer_task
666        with suppress(Exception):
667            await state.processor.close()
668
669    async def _promote_join_catchup_processor(
670        self,
671        player_id: str,
672        pipeline: _MemberPipeline,
673        target_end_us: int,
674    ) -> None:
675        """Promote join catchup processor to the member's live DSP processor."""
676        old_processor: _BufferedFfmpegProcessor | None = None
677        async with self._state_lock:
678            state = self._join_catchup.pop(player_id, None)
679            if state is None:
680                return
681            old_processor = pipeline.processor
682            pipeline.processor = state.processor
683        if state.snapshot_task is not None:
684            state.snapshot_task.cancel()
685            with suppress(asyncio.CancelledError, Exception):
686                await state.snapshot_task
687        # Let writer flush queued PCM before handoff; cancel if queue is full.
688        try:
689            state.input_queue.put_nowait(None)
690        except asyncio.QueueFull:
691            state.writer_task.cancel()
692        with suppress(asyncio.CancelledError, Exception):
693            await state.writer_task
694        state.drainer_task.cancel()
695        with suppress(asyncio.CancelledError, Exception):
696            await state.drainer_task
697        if self._producer_eof_sent and pipeline.config.requires_transform:
698            with suppress(Exception):
699                await pipeline.processor.write_eof()
700        if old_processor is not None and old_processor is not state.processor:
701            with suppress(Exception):
702                await old_processor.close()
703
704    async def _clear_join_catchup(self) -> None:
705        """Stop and remove all dedicated join catchup processors."""
706        async with self._state_lock:
707            player_ids = list(self._join_catchup.keys())
708        for player_id in player_ids:
709            await self._stop_join_catchup(player_id)
710
711    async def _release_player_channel(self, player_id: str) -> None:
712        """Release per-member channel/DSP state for a removed member."""
713        async with self._state_lock:
714            pipeline = self._member_pipelines.pop(player_id, None)
715            self._preassigned_channels.pop(player_id, None)
716        if pipeline is None or pipeline.processor is None:
717            return
718        await self._close_member_ffmpeg(pipeline.processor)
719
720    # -- Playback pipeline -----------------------------------------------------
721
722    async def _run_playback(self, media: PlayerMedia) -> None:  # noqa: PLR0915
723        """
724        Run the playback pipeline for a single media session.
725
726        Pulls PCM from the MA stream, feeds main + per-member DSP channels into the
727        Sendspin push stream, and commits audio continuously. Supports dynamic group
728        membership changes and late-join historical backfill while running.
729        """
730        # refresh the session PCM format from the leader's preferred output before
731        # building any pipelines; member ffmpeg pipelines and pre-computed filter
732        # params depend on this rate so the cache must also be cleared
733        self._pcm_format, self._sendspin_pcm_format = self._select_session_pcm_formats()
734        self._queue_id = media.source_id
735        self._queue_session_id = get_media_session_id(media)
736        self._pipeline_config_cache.clear()
737        self.player.logger.debug(
738            "Sendspin session PCM format: %d Hz / F32",
739            self._pcm_format.sample_rate,
740        )
741        push_stream = self._create_push_stream()
742        is_live = media.media_type in _LIVE_MEDIA_TYPES
743        push_stream.set_live_source(is_live)
744        async with self._state_lock:
745            self._push_stream = push_stream
746            self._playback_running = True
747            self._producer_eof_sent = False
748            self._history.clear()
749            self._produced_audio_us = 0
750            self._timeline_start_us = None
751            self._first_commit_monotonic_us = None
752            self._mapping_dirty = True
753        # Bounded queue between producer (stream reader) and consumer (committer).
754        pending_chunks: asyncio.Queue[_PendingChunk | None] = asyncio.Queue(
755            maxsize=_PRODUCER_BACKLOG_SIZE
756        )
757        # Shadow deque mirroring pending_chunks for join-catchup backlog peeking.
758        pending_backlog: deque[_PendingChunk] = deque()
759        pending_duration_us = 0
760        last_elapsed_update_s = 0.0
761
762        async def _produce_pending_chunks() -> None:
763            nonlocal pending_duration_us
764            audio_source = self.player.mass.streams.get_stream(
765                media, self._pcm_format, self.player.player_id
766            )
767            completed = False
768            try:
769                async for chunk in audio_source:
770                    if not chunk:
771                        continue
772                    for slice_chunk in iter_pcm_slices(
773                        chunk, self._pcm_format, target_duration_ms=_PRODUCER_SLICE_US // 1000
774                    ):
775                        if not slice_chunk:
776                            continue
777                        duration_us = self._duration_us(slice_chunk, self._pcm_format)
778                        if duration_us <= 0:
779                            continue
780                        await self._refresh_member_mappings()
781                        pending = _PendingChunk(pcm=slice_chunk, duration_us=duration_us)
782                        await pending_chunks.put(pending)
783                        pending_backlog.append(pending)
784                        pending_duration_us += duration_us
785                        join_pending_ids, pipelines = await self._snapshot_active_pipelines()
786                        transform_pipelines: list[_MemberPipeline] = []
787                        for member_id, pipeline in pipelines:
788                            if not pipeline.config.requires_transform:
789                                continue
790                            if member_id in join_pending_ids:
791                                continue
792                            transform_pipelines.append(pipeline)
793                        results = await asyncio.gather(
794                            *(
795                                self._transform_member_chunk(pipeline, slice_chunk)
796                                for pipeline in transform_pipelines
797                            ),
798                            return_exceptions=True,
799                        )
800                        for pipeline, result in zip(transform_pipelines, results, strict=True):
801                            if isinstance(result, BaseException):
802                                self.player.logger.warning(
803                                    "Transform push failed for channel %s: %s",
804                                    pipeline.channel_id,
805                                    result,
806                                )
807                completed = True
808            finally:
809                if not completed:
810                    close_task = asyncio.create_task(audio_source.aclose())
811                    try:
812                        await asyncio.shield(close_task)
813                    except asyncio.CancelledError:
814                        await close_task
815                        raise
816
817        async def _commit_pending_chunks() -> None:
818            nonlocal pending_duration_us, last_elapsed_update_s
819            while True:
820                pending = await pending_chunks.get()
821                if pending is None:
822                    break
823                pending_backlog.popleft()
824                pending_duration_us = max(0, pending_duration_us - pending.duration_us)
825                await self._inject_ready_join_historical(push_stream, pending_backlog, pending.pcm)
826                push_stream.prepare_audio(
827                    pending.pcm, self._sendspin_pcm_format, channel_id=MAIN_CHANNEL
828                )
829                join_pending_ids, pipelines = await self._snapshot_active_pipelines()
830                transform_pipelines: list[_MemberPipeline] = []
831                for member_id, pipeline in pipelines:
832                    if not pipeline.config.requires_transform:
833                        continue
834                    if member_id in join_pending_ids:
835                        continue
836                    transform_pipelines.append(pipeline)
837                transformed_chunks = await asyncio.gather(
838                    *(
839                        self._read_member_chunk(pipeline, pending.duration_us)
840                        for pipeline in transform_pipelines
841                    ),
842                    return_exceptions=True,
843                )
844                for pipeline, transformed_chunk in zip(
845                    transform_pipelines, transformed_chunks, strict=True
846                ):
847                    if isinstance(transformed_chunk, BaseException):
848                        self.player.logger.warning(
849                            "Transform read failed for channel %s: %s",
850                            pipeline.channel_id,
851                            transformed_chunk,
852                        )
853                        continue
854                    if transformed_chunk is None:
855                        continue
856                    push_stream.prepare_audio(
857                        transformed_chunk,
858                        self._sendspin_pcm_format,
859                        channel_id=pipeline.channel_id,
860                    )
861                try:
862                    commit_start_us = await push_stream.commit_audio()
863                except StreamStoppedError:
864                    # Stream stopped since it was replaced by another stream
865                    self.player.logger.debug("Stopping commit loop due to stopped push stream")
866                    break
867                await push_stream.sleep_to_limit_buffer(_PRODUCER_BUFFER_LIMIT_US)
868                commit_now_us = push_stream.now_us()
869                committed_history_chunk = _HistoryChunk(
870                    start_time_us=int(commit_start_us),
871                    duration_us=pending.duration_us,
872                    pcm=pending.pcm,
873                )
874                async with self._state_lock:
875                    if self._timeline_start_us is None:
876                        self._timeline_start_us = int(commit_start_us)
877                    if self._first_commit_monotonic_us is None:
878                        self._first_commit_monotonic_us = commit_now_us
879                    self._history.append(committed_history_chunk)
880                    self._produced_audio_us += pending.duration_us
881                    self._prune_history_locked(commit_now_us)
882                await self._fanout_history_chunk_to_join_processors(committed_history_chunk)
883                if self._timeline_start_us is not None:
884                    elapsed_real_s = max(0.0, (commit_now_us - self._timeline_start_us) / 1_000_000)
885                    if elapsed_real_s - last_elapsed_update_s >= 1.0:
886                        last_elapsed_update_s = elapsed_real_s
887                        self.player._attr_elapsed_time = elapsed_real_s
888                        self.player._attr_elapsed_time_last_updated = time.time()
889                        self.player.update_state()
890
891        commit_task = asyncio.create_task(_commit_pending_chunks())
892        self._attach_task_exception_logger(commit_task, "commit_pending_chunks")
893        producer_stopped_cleanly = False
894        try:
895            await _produce_pending_chunks()
896            producer_stopped_cleanly = True
897        finally:
898            if producer_stopped_cleanly and not self._cancel_requested and not commit_task.done():
899                # Mark EOF so that catchup processors promoted after this
900                # point also get flushed (see _promote_join_catchup_processor).
901                self._producer_eof_sent = True
902                # Signal EOF to transform pipelines so ffmpeg flushes its
903                # internal buffers instead of blocking on the last read.
904                _, pipelines = await self._snapshot_active_pipelines()
905                for _, pipeline in pipelines:
906                    if pipeline.processor is not None and pipeline.config.requires_transform:
907                        with suppress(Exception):
908                            await pipeline.processor.write_eof()
909                # Producer finished normally; send a None sentinel so the
910                # consumer exits cleanly.  The queue may be full, so retry
911                # with a deadline before falling back to cancellation.
912                sentinel_sent = False
913                deadline = time.monotonic() + 1.0
914                while not sentinel_sent and not commit_task.done():
915                    try:
916                        pending_chunks.put_nowait(None)
917                        sentinel_sent = True
918                    except asyncio.QueueFull:
919                        if time.monotonic() >= deadline:
920                            break
921                        await asyncio.sleep(0.01)
922                if sentinel_sent:
923                    with suppress(asyncio.CancelledError, Exception):
924                        await commit_task
925                else:
926                    commit_task.cancel()
927                    with suppress(asyncio.CancelledError, Exception):
928                        await commit_task
929            else:
930                commit_task.cancel()
931                with suppress(asyncio.CancelledError, Exception):
932                    await commit_task
933            # On clean EOF, wait for clients to finish playing their
934            # buffered audio before sending stream/end (which clears
935            # client buffers per the Sendspin spec). Skip this when a
936            # new playback is superseding this one, so skipping tracks is still fast.
937            if producer_stopped_cleanly and not self._cancel_requested:
938                try:
939                    await self._wait_for_buffer_drain()
940                except asyncio.CancelledError:
941                    # New playback interrupted the drain — treat as
942                    # non-clean stop so we skip group.stop() below
943                    # and let the new playback handle the transition.
944                    producer_stopped_cleanly = False
945            with suppress(Exception):
946                # Same condition as the group.stop() below, so we snapshot on exactly the
947                # paths where a group STOP - and therefore a freeze - is already emitted.
948                self._stop_push_stream(
949                    snapshot_progress=producer_stopped_cleanly and not self._cancel_requested
950                )
951            await self._clear_join_catchup()
952            await self._clear_member_pipelines()
953            async with self._state_lock:
954                self._push_stream = None
955                self._playback_running = False
956                self._timeline_start_us = None
957                self._first_commit_monotonic_us = None
958                self._produced_audio_us = 0
959                self._history.clear()
960                # Drop cached DSP decisions so next playback reflects latest config.
961                self._pipeline_config_cache.clear()
962            # Only emit a group STOP when MA stream playback reached natural EOF.
963            # Skip this on cancellation/error paths to avoid stop-event races with transitions.
964            if producer_stopped_cleanly and not self._cancel_requested:
965                with suppress(Exception):
966                    await self.player.api.group.stop()
967
968    # -- Join injection --------------------------------------------------------
969
970    async def _inject_ready_join_historical(
971        self,
972        push_stream: PushStream,
973        pending_backlog: deque[_PendingChunk],
974        current_pcm: bytes,
975    ) -> bool:
976        """
977        Inject join-catchup historical audio once processor output reaches history end.
978
979        Join promotion lifecycle:
980        1. A catchup processor is fed historical PCM and new commits in parallel.
981        2. Once the processor's output lag falls within _JOIN_PROMOTE_ARM_WINDOW_US
982           of the history tail, promotion is "armed" and a target end timestamp is locked.
983        3. Once output reaches the target (within _JOIN_PROMOTE_TOLERANCE_US), the
984           catchup processor is promoted to the member's live DSP pipeline.
985        4. If promotion doesn't complete within _JOIN_PROMOTION_TIMEOUT_S, it's aborted.
986        """
987        injected_any = False
988        async with self._state_lock:
989            items = list(self._join_catchup.items())
990        for player_id, state in items:
991            produced_output_us = state.processor.produced_output_us
992            async with self._state_lock:
993                current = self._join_catchup.get(player_id)
994                if current is None or current.processor is not state.processor:
995                    continue
996                first_history_start_us = current.first_history_start_us
997                fed_until_us = current.fed_until_us
998                history_end_us = current.history_end_us
999                promotion_target_end_us = current.promotion_target_end_us
1000                promotion_armed_monotonic_s = current.promotion_armed_monotonic_s
1001            if first_history_start_us is None or fed_until_us is None or history_end_us is None:
1002                continue
1003            max_ready_end_us = min(
1004                fed_until_us,
1005                first_history_start_us + max(0, produced_output_us),
1006            )
1007            if promotion_target_end_us is None:
1008                lag_to_tail_us = history_end_us - max_ready_end_us
1009                if lag_to_tail_us > _JOIN_PROMOTE_ARM_WINDOW_US:
1010                    continue
1011                async with self._state_lock:
1012                    current = self._join_catchup.get(player_id)
1013                    if current is None or current.processor is not state.processor:
1014                        continue
1015                    if current.promotion_target_end_us is None:
1016                        current.promotion_target_end_us = history_end_us
1017                        current.promotion_armed_monotonic_s = time.monotonic()
1018                    promotion_target_end_us = current.promotion_target_end_us
1019                    promotion_armed_monotonic_s = current.promotion_armed_monotonic_s
1020            target_end_us = promotion_target_end_us
1021            if (
1022                promotion_armed_monotonic_s is not None
1023                and time.monotonic() - promotion_armed_monotonic_s > _JOIN_PROMOTION_TIMEOUT_S
1024            ):
1025                self.player.logger.error(
1026                    "Join promotion timed out for %s after %.1fs; dropping join catchup",
1027                    player_id,
1028                    _JOIN_PROMOTION_TIMEOUT_S,
1029                )
1030                await self._stop_join_catchup(player_id)
1031                continue
1032            if max_ready_end_us + _JOIN_PROMOTE_TOLERANCE_US < target_end_us:
1033                continue
1034            inject_duration_us = target_end_us - first_history_start_us
1035            transformed_history = state.processor.pop_duration_us_or_pad(
1036                inject_duration_us, _JOIN_PROMOTE_TOLERANCE_US
1037            )
1038            if transformed_history is None:
1039                continue
1040            transformed_history = await self._pad_history_to_live_tail(
1041                state, target_end_us, transformed_history
1042            )
1043            pipeline = await self._sync_member_pipeline(player_id)
1044            # Split the blob into slices so push_stream can yield between encodes.
1045            frame_stride = (
1046                self._sendspin_pcm_format.bit_depth // 8
1047            ) * self._sendspin_pcm_format.channels
1048            slice_bytes = (
1049                int(self._sendspin_pcm_format.sample_rate * _PRODUCER_SLICE_US / 1_000_000)
1050                * frame_stride
1051            )
1052            for offset in range(0, len(transformed_history), slice_bytes):
1053                push_stream.prepare_historical_audio(
1054                    transformed_history[offset : offset + slice_bytes],
1055                    self._sendspin_pcm_format,
1056                    channel_id=pipeline.channel_id,
1057                    start_time_us=first_history_start_us if offset == 0 else None,
1058                )
1059            await self._prefeed_pending_backlog_for_join(state, current_pcm, pending_backlog)
1060            await self._promote_join_catchup_processor(player_id, pipeline, target_end_us)
1061            injected_any = True
1062        return injected_any
1063
1064    async def _pad_history_to_live_tail(
1065        self,
1066        state: _JoinCatchupState,
1067        target_end_us: int,
1068        transformed_history: bytes,
1069    ) -> bytes:
1070        """Append silence so joiner's channel_timing aligns with the live tail at promotion."""
1071        # target_end_us is locked from earlier and may lag the live tail by seconds.
1072        async with self._state_lock:
1073            live_tail_us = (
1074                self._history[-1].start_time_us + self._history[-1].duration_us
1075                if self._history
1076                else target_end_us
1077            )
1078        promotion_lag_us = max(0, live_tail_us - target_end_us)
1079        if promotion_lag_us <= 0:
1080            return transformed_history
1081        return transformed_history + state.processor.pad_and_skip(promotion_lag_us)
1082
1083    async def _prefeed_pending_backlog_for_join(
1084        self,
1085        state: _JoinCatchupState,
1086        current_pcm: bytes,
1087        pending_backlog: deque[_PendingChunk],
1088    ) -> None:
1089        """
1090        Push current chunk + queued pending chunks into join processor before promotion.
1091
1092        Between the last committed chunk and the next commit, there may be
1093        chunks already queued by the producer that the catchup processor hasn't
1094        seen yet.  Feeding them now avoids a gap in transformed audio after
1095        promotion.
1096        """
1097        await self._enqueue_join_pcm(state, current_pcm)
1098        for item in list(pending_backlog):
1099            await self._enqueue_join_pcm(state, item.pcm)
1100
1101    async def _fanout_history_chunk_to_join_processors(self, hist_chunk: _HistoryChunk) -> None:
1102        """Feed newly committed history chunk into all active join-catchup processors."""
1103        async with self._state_lock:
1104            items = list(self._join_catchup.items())
1105        for player_id, state in items:
1106            async with state.write_lock:
1107                # Read current state under lock.
1108                async with self._state_lock:
1109                    current = self._join_catchup.get(player_id)
1110                    if current is None or current.processor is not state.processor:
1111                        continue
1112                    previous_end_us = current.fed_until_us
1113                    first_history_start_us = current.first_history_start_us
1114                # Initialize first_history_start_us if this is the first chunk.
1115                if first_history_start_us is None:
1116                    first_history_start_us = hist_chunk.start_time_us
1117                    previous_end_us = first_history_start_us
1118                # Fill timeline gaps with silence.
1119                if previous_end_us is not None and hist_chunk.start_time_us > previous_end_us:
1120                    gap_us = hist_chunk.start_time_us - previous_end_us
1121                    silence = self._silence_for_duration_us(gap_us)
1122                    if silence:
1123                        await self._enqueue_join_pcm(state, silence)
1124                await self._enqueue_join_pcm(state, hist_chunk.pcm)
1125                # Write updated state back under lock.
1126                new_end_us = hist_chunk.start_time_us + hist_chunk.duration_us
1127                async with self._state_lock:
1128                    current = self._join_catchup.get(player_id)
1129                    if current is not None and current.processor is state.processor:
1130                        if current.first_history_start_us is None:
1131                            current.first_history_start_us = first_history_start_us
1132                        if current.fed_until_us is None:
1133                            current.fed_until_us = first_history_start_us
1134                        current.fed_until_us = new_end_us
1135                        current.history_end_us = new_end_us
1136
1137    async def _enqueue_join_pcm(
1138        self,
1139        state: _JoinCatchupState,
1140        pcm: bytes,
1141    ) -> None:
1142        """
1143        Enqueue PCM into a joining member writer queue.
1144
1145        Bails out immediately if the writer task is dead to avoid blocking
1146        the commit loop on a queue with no consumer.
1147        """
1148        if state.writer_task.done():
1149            return
1150        try:
1151            state.input_queue.put_nowait(pcm)
1152        except asyncio.QueueFull:
1153            if state.writer_task.done():
1154                return
1155            await state.input_queue.put(pcm)
1156
1157    # -- Member pipeline management --------------------------------------------
1158
1159    async def _refresh_member_mappings(self) -> None:
1160        """Re-evaluate per-member channel mapping and DSP requirements."""
1161        async with self._state_lock:
1162            if not self._mapping_dirty:
1163                return
1164            member_ids = tuple(self._members)
1165            self._mapping_dirty = False
1166        for member_id in member_ids:
1167            await self._sync_member_pipeline(member_id)
1168        # Keep leader pipeline in sync so leader DSP can be applied when required.
1169        await self._sync_member_pipeline(self.player.player_id)
1170
1171    async def _sync_member_pipeline(self, player_id: str) -> _MemberPipeline:
1172        """Create/update pipeline state for one member from current MA config."""
1173        config = self._get_pipeline_config_cached(player_id)
1174        release_processor: _BufferedFfmpegProcessor | None = None
1175        start_processor: _BufferedFfmpegProcessor | None = None
1176        async with self._state_lock:
1177            current = self._member_pipelines.get(player_id)
1178            if current is not None and current.config.signature == config.signature:
1179                return current
1180            if current and current.config.requires_transform:
1181                channel_id = current.channel_id if config.requires_transform else MAIN_CHANNEL
1182                release_processor = current.processor
1183            elif config.requires_transform:
1184                channel_id = self._get_or_create_preassigned_channel(player_id)
1185            else:
1186                channel_id = MAIN_CHANNEL
1187                self._preassigned_channels.pop(player_id, None)
1188            processor: _BufferedFfmpegProcessor | None = None
1189            if config.requires_transform:
1190                ffmpeg_obj = self._create_member_ffmpeg(config.filter_params)
1191                processor = _BufferedFfmpegProcessor(ffmpeg_obj, self._pcm_format)
1192                start_processor = processor
1193            pipeline = _MemberPipeline(
1194                player_id=player_id,
1195                channel_id=channel_id,
1196                config=config,
1197                processor=processor,
1198            )
1199            self._member_pipelines[player_id] = pipeline
1200        if start_processor is not None:
1201            try:
1202                await start_processor.start()
1203            except Exception as err:
1204                async with self._state_lock:
1205                    if (
1206                        self._member_pipelines.get(player_id) is not None
1207                        and self._member_pipelines[player_id].processor is start_processor
1208                    ):
1209                        self._member_pipelines.pop(player_id, None)
1210                with suppress(Exception):
1211                    await self._close_member_ffmpeg(start_processor)
1212                raise RuntimeError(f"Failed to start member DSP ffmpeg for {player_id}") from err
1213        if release_processor is not None:
1214            await self._close_member_ffmpeg(release_processor)
1215        return pipeline
1216
1217    def _get_pipeline_config_cached(
1218        self,
1219        player_id: str,
1220        *,
1221        force_refresh: bool = False,
1222    ) -> _PipelineConfig:
1223        """Return cached pipeline config for a player, calculating on cache miss."""
1224        if not force_refresh and (cached := self._pipeline_config_cache.get(player_id)) is not None:
1225            return cached
1226        config = self._read_pipeline_config(player_id)
1227        self._pipeline_config_cache[player_id] = config
1228        return config
1229
1230    def _read_pipeline_config(self, player_id: str) -> _PipelineConfig:
1231        """Read MA config and determine if member needs a dedicated DSP channel."""
1232        dsp_config = self.player.mass.config.get_player_dsp_config(player_id)
1233        dsp_enabled = bool(dsp_config.enabled)
1234        raw_output_channels = self.player.mass.config.get_raw_player_config_value(
1235            player_id,
1236            CONF_OUTPUT_CHANNELS,
1237            "stereo",
1238        )
1239        output_channels = str(raw_output_channels or "stereo").strip().lower()
1240        if output_channels not in {"stereo", "left", "right", "mono"}:
1241            output_channels = "stereo"
1242        try:
1243            output_format = self._get_member_output_format(player_id)
1244            output_plan = self.player.mass.streams.audio.get_player_output_plan(
1245                player_id,
1246                self._pcm_format,
1247                output_format,
1248                handoff_format=self._pcm_format,
1249            )
1250            filter_params = tuple(output_plan.filter_params)
1251        except Exception:
1252            filter_params = ()
1253            output_plan = None
1254        # a ComplexFilter (e.g. convolution) is never a plain string, so it always counts
1255        custom_filter_graph = any(
1256            not isinstance(param, str) or param.strip() for param in filter_params
1257        )
1258        requires_transform = dsp_enabled or output_channels != "stereo" or custom_filter_graph
1259        if (
1260            output_plan is not None
1261            and self._queue_id is not None
1262            and self._queue_session_id is not None
1263        ):
1264            self.player.mass.streams.audio_processing.update_output(
1265                output_plan.output_details.player_ids[0],
1266                output_plan,
1267                queue_id=self._queue_id,
1268                session_id=self._queue_session_id,
1269            )
1270        return _PipelineConfig(
1271            requires_transform=requires_transform,
1272            output_channels=output_channels,
1273            filter_params=filter_params,
1274        )
1275
1276    def _select_session_pcm_formats(self) -> tuple[AudioFormat, SendspinAudioFormat]:
1277        """
1278        Pick the session PCM format (MA-side + wire) from the leader's preferred format.
1279
1280        F32 is always used for DSP headroom. The sample rate follows the leader's
1281        preferred rate but is capped at 48 kHz when the leader's output codec is
1282        lossy — higher rates yield no perceivable quality gain there. Member
1283        clients with a different preferred rate up/down-sample in their own DSP
1284        step on the receiving side.
1285        """
1286        leader_output = self._get_member_output_format(self.player.player_id)
1287        sample_rate = int(leader_output.sample_rate) or _DEFAULT_PCM_FORMAT.sample_rate
1288        if leader_output.content_type in (ContentType.OPUS, ContentType.MP3, ContentType.AAC):
1289            sample_rate = min(sample_rate, _LOSSY_MAX_SAMPLE_RATE)
1290        pcm_format = AudioFormat(
1291            content_type=ContentType.PCM_F32LE,
1292            sample_rate=sample_rate,
1293            bit_depth=32,
1294            channels=2,
1295        )
1296        sendspin_pcm_format = SendspinAudioFormat(
1297            sample_rate=sample_rate,
1298            bit_depth=32,
1299            channels=2,
1300            sample_type="float",
1301        )
1302        return pcm_format, sendspin_pcm_format
1303
1304    def _get_member_output_format(self, player_id: str) -> AudioFormat:
1305        """
1306        Return the actual output AudioFormat for a group member.
1307
1308        Derives the format from the member's sendspin player role (preferred codec
1309        and format), falling back to the internal PCM format if unavailable.
1310        """
1311        provider = cast("SendspinProvider", self.player.provider)
1312        client = provider.server_api.get_client(player_id)
1313        if client is not None:
1314            for role in client.roles_by_family("player"):
1315                if isinstance(role, PlayerV1Role):
1316                    preferred_fmt = role.preferred_format
1317                    preferred_codec = role.preferred_codec
1318                    if preferred_fmt is not None and preferred_codec is not None:
1319                        if preferred_codec == SendspinAudioCodec.FLAC:
1320                            content_type = ContentType.FLAC
1321                        elif preferred_codec == SendspinAudioCodec.OPUS:
1322                            content_type = ContentType.OPUS
1323                        else:
1324                            content_type = ContentType.from_bit_depth(preferred_fmt.bit_depth)
1325                        return AudioFormat(
1326                            content_type=content_type,
1327                            sample_rate=preferred_fmt.sample_rate,
1328                            bit_depth=preferred_fmt.bit_depth,
1329                            channels=preferred_fmt.channels,
1330                        )
1331                elif isinstance(role, BridgePlayerRole):
1332                    fmt = role.preferred_format
1333                    if fmt is not None:
1334                        return AudioFormat(
1335                            content_type=ContentType.from_bit_depth(fmt.bit_depth),
1336                            sample_rate=fmt.sample_rate,
1337                            bit_depth=fmt.bit_depth,
1338                            channels=fmt.channels,
1339                        )
1340                    return AudioFormat(
1341                        content_type=ContentType.from_bit_depth(BRIDGE_BIT_DEPTH),
1342                        sample_rate=BRIDGE_SAMPLE_RATE,
1343                        bit_depth=BRIDGE_BIT_DEPTH,
1344                        channels=BRIDGE_CHANNELS,
1345                    )
1346        return self._pcm_format
1347
1348    def _get_or_create_preassigned_channel(self, player_id: str) -> UUID:
1349        """Return stable dedicated channel id for transform-required player."""
1350        if (channel_id := self._preassigned_channels.get(player_id)) is not None:
1351            return channel_id
1352        channel_id = uuid4()
1353        self._preassigned_channels[player_id] = channel_id
1354        return channel_id
1355
1356    # -- FFmpeg lifecycle ------------------------------------------------------
1357
1358    def _create_member_ffmpeg(self, filter_params: tuple[str | ComplexFilter, ...]) -> FFMpeg:
1359        """Create per-member FFMpeg for DSP pipeline."""
1360        return FFMpeg(
1361            audio_input="-",
1362            input_format=self._pcm_format,
1363            output_format=self._pcm_format,
1364            filter_params=list(filter_params),
1365        )
1366
1367    async def _transform_member_chunk(self, pipeline: _MemberPipeline, chunk: bytes) -> None:
1368        """Push one PCM chunk into a member DSP pipeline."""
1369        processor = pipeline.processor
1370        if processor is None:
1371            return
1372        await processor.push(chunk)
1373
1374    async def _read_member_chunk(
1375        self,
1376        pipeline: _MemberPipeline,
1377        duration_us: int,
1378    ) -> bytes | None:
1379        """Read one transformed chunk from a member DSP pipeline."""
1380        processor = pipeline.processor
1381        if processor is None or duration_us <= 0:
1382            return b""
1383        transformed = await processor.read_duration_us(duration_us)
1384        if not transformed:
1385            return None
1386        pipeline.ready = True
1387        return bytes(transformed)
1388
1389    async def _close_member_ffmpeg(self, processor: _BufferedFfmpegProcessor) -> None:
1390        """Close an ffmpeg processor, suppressing errors."""
1391        with suppress(Exception):
1392            await processor.close()
1393
1394    async def _clear_member_pipelines(self) -> None:
1395        """Release all member pipeline resources."""
1396        async with self._state_lock:
1397            pipelines = list(self._member_pipelines.values())
1398            self._member_pipelines.clear()
1399        for pipeline in pipelines:
1400            if pipeline.processor is not None:
1401                await self._close_member_ffmpeg(pipeline.processor)
1402
1403    # -- Push stream -----------------------------------------------------------
1404
1405    def _create_push_stream(self) -> PushStream:
1406        """Create PushStream with channel resolver for per-member routing."""
1407        return self.player.api.group.start_stream(channel_resolver=self._resolve_channel_for_player)
1408
1409    async def _wait_for_buffer_drain(self) -> None:
1410        """
1411        Wait for clients to finish playing buffered audio.
1412
1413        Called before stopping the push stream on natural EOF to prevent
1414        stream/end from clearing client buffers while audio is still playing.
1415
1416        Uses the push stream's public backpressure API with a zero buffer
1417        target to sleep until the clock catches up with all committed audio.
1418        Each internal sleep is capped at 1 second, so we loop until drained.
1419
1420        Raises asyncio.CancelledError if a new playback request interrupts.
1421        """
1422        ps = self._push_stream
1423        if ps is None or ps.is_stopped:
1424            return
1425        self.player.logger.debug("Waiting for client buffer drain before stream/end")
1426        # Safety timeout: never wait longer than the max buffer depth.
1427        deadline = time.monotonic() + (_PRODUCER_BUFFER_LIMIT_US / 1_000_000)
1428        while time.monotonic() < deadline:
1429            t0 = time.monotonic()
1430            await ps.sleep_to_limit_buffer(0)
1431            # sleep_to_limit_buffer returns immediately when the clock has
1432            # caught up with all committed audio (nothing left to drain).
1433            if time.monotonic() - t0 < 0.05:
1434                break
1435        self.player.logger.debug("Client buffer drain complete")
1436
1437    def _stop_push_stream(self, *, snapshot_progress: bool = False) -> None:
1438        """
1439        Stop the active PushStream.
1440
1441        :param snapshot_progress: Freeze the group's playback progress first. Pass this
1442            only for a natural end of stream, never for one being superseded.
1443        """
1444        ps = self._push_stream
1445        if ps is None or ps.is_stopped:
1446            return
1447        if snapshot_progress and (metadata_role := self.player._metadata_role) is not None:
1448            # The group can only resolve the live position while the stream is up; once
1449            # it is down the freeze can just re-emit the last anchor that was pushed.
1450            metadata_role.freeze_progress()
1451        ps.stop()
1452
1453    def _resolve_channel_for_player(self, player_id: str) -> UUID:
1454        """Channel resolver callback for per-player routing."""
1455        pipeline = self._member_pipelines.get(player_id)
1456        if pipeline is not None:
1457            return pipeline.channel_id
1458        # Force a fresh config read for pending/unknown joiners so the very
1459        # first resolution (triggered by add_client) uses up-to-date DSP settings.
1460        force = player_id not in self._members and player_id != self.player.player_id
1461        config = self._get_pipeline_config_cached(player_id, force_refresh=force)
1462        if not config.requires_transform:
1463            return MAIN_CHANNEL
1464        return self._get_or_create_preassigned_channel(player_id)
1465
1466    # -- History ---------------------------------------------------------------
1467
1468    def _prune_history_locked(self, now_monotonic_us: int) -> None:
1469        """Drop old history chunks that are fully in the past."""
1470        if self._timeline_start_us is None or self._first_commit_monotonic_us is None:
1471            return
1472        elapsed_real_us = max(0, now_monotonic_us - self._first_commit_monotonic_us)
1473        source_now_us = self._timeline_start_us + elapsed_real_us
1474        cutoff_us = source_now_us - _HISTORY_KEEP_PAST_US
1475        while self._history and (
1476            self._history[0].start_time_us + self._history[0].duration_us <= cutoff_us
1477        ):
1478            self._history.popleft()
1479
1480    # -- PCM utilities ---------------------------------------------------------
1481
1482    @staticmethod
1483    def _duration_us(audio: bytes, audio_format: AudioFormat) -> int:
1484        """Compute chunk duration from PCM payload size."""
1485        bytes_per_sample = max(1, int(audio_format.bit_depth // 8))
1486        bytes_per_second = (
1487            int(audio_format.sample_rate) * bytes_per_sample * int(audio_format.channels)
1488        )
1489        if bytes_per_second <= 0:
1490            return 0
1491        return int((len(audio) / bytes_per_second) * 1_000_000)
1492
1493    def _silence_for_duration_us(self, duration_us: int) -> bytes:
1494        """Generate silent PCM with frame-aligned duration for the current session format."""
1495        if duration_us <= 0:
1496            return b""
1497        bytes_per_sample = max(1, int(self._pcm_format.bit_depth // 8))
1498        frame_size = bytes_per_sample * int(self._pcm_format.channels)
1499        samples = max(0, round((duration_us / 1_000_000) * int(self._pcm_format.sample_rate)))
1500        return b"\x00" * (samples * frame_size)
1501