music-assistant-server

1"""
2Unit tests for the Sendspin -> AirPlay bridge timing.
3
4Cover ten things, with the Sendspin clock mocked via ``ManualClock`` so the
5tests are deterministic and independent of the host wall-clock:
6
7* the clock-domain conversion turning a Sendspin audible instant (Sendspin's own
8  monotonic clock) into the unix epoch ms used by the START command, and back;
9* the startup lead reported to Sendspin, which decides how far ahead of the
10  audible instant it schedules the first chunk;
11* the start anchor: byte 0 is anchored to the first chunk Sendspin delivers, so a
12  fresh track keeps position 0 and a late joiner lands at the group's live position;
13* anchoring against the binary: the commanded instant honours the join headroom,
14  the receiver's clock-ready projection and the content Sendspin already
15  scheduled, and the content is then mapped onto the instant the binary acked;
16* the timeline alignment that keeps every chunk at the byte offset its timestamp
17  claims, so a discontinuity in the Sendspin timeline does not shift the device
18  off the group's clock for the rest of the stream;
19* the playout shift the binary reports after a PCM starvation, which moves the
20  anchor so the device does not stay behind the group once it re-anchors itself;
21* the write pacing that keeps the device buffered a bounded amount ahead of real
22  time so a late-join catch-up backlog is not dumped into the CLI;
23* the warm handover: a running, connected stream is kept (not torn down) across
24  a new Sendspin stream and rides the persistent-stdin flush-refill (FLUSH +
25  re-anchoring START) instead of a cold reconnect -- with flush-timeout and
26  superseded-task fallback, and the supersession handling that keeps a stale
27  start from spawning a process or touching the stream a newer one owns;
28* the recovery from a transport lost mid-stream: the dead CLI is released and
29  re-anchored on the group's live timeline. Every give-up then takes the speaker
30  out of the Sendspin session, so the player stops reporting playback nobody can
31  hear, and a bounded re-join brings back one that was only briefly away.
32"""
33
34import asyncio
35from collections.abc import Coroutine
36from typing import cast
37from unittest.mock import AsyncMock, MagicMock, patch
38
39import pytest
40from aiosendspin.clock import ManualClock
41from aiosendspin.server.roles import AudioChunk
42from music_assistant_models.enums import PlaybackState
43
44from music_assistant.providers.airplay.constants import (
45    AIRPLAY_CLOCK_READY_LEAD_MS,
46    AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS,
47    AIRPLAY_SPLICE_LEAD_MARGIN_MS,
48    ClockReadiness,
49    StreamingProtocol,
50)
51from music_assistant.providers.airplay.player import AirPlayPlayer
52from music_assistant.providers.airplay.sendspin_bridge import (
53    BRIDGE_COLD_START_LEAD_MS,
54    BRIDGE_MIN_BUFFER_MS,
55    BRIDGE_TRANSPORT_RECOVERY_GUARD_SECONDS,
56    BRIDGE_WARM_START_LEAD_MS,
57    MAX_DEVICE_BUFFER_SECONDS,
58    MAX_HELD_AUDIO_US,
59    MAX_HELD_CHUNKS,
60    PAD_BLOCK_FRAMES,
61    SILENCE_BLOCK,
62    SendspinAirPlayBridge,
63    SendspinBridgeManager,
64    device_buffer_ahead_seconds,
65    sendspin_audible_instant_to_unix_ms,
66    unix_ms_to_sendspin_audible_instant,
67)
68from music_assistant.providers.airplay.stream_session import AirPlayStreamSession
69from music_assistant.providers.sendspin.bridge_role import (
70    BRIDGE_BYTES_PER_SAMPLE,
71    BRIDGE_CHANNELS,
72    BRIDGE_SAMPLE_RATE,
73    BridgePlayerRole,
74)
75
76BRIDGE_BYTES_PER_SECOND = BRIDGE_SAMPLE_RATE * BRIDGE_CHANNELS * BRIDGE_BYTES_PER_SAMPLE
77BRIDGE_BYTES_PER_FRAME = BRIDGE_CHANNELS * BRIDGE_BYTES_PER_SAMPLE
78
79# A large, arbitrary Sendspin monotonic-clock epoch (microseconds). Real
80# monotonic clocks start from an unspecified point (e.g. host boot), so the
81# conversion must never depend on this value.
82SENDSPIN_EPOCH_US = 5_000_000_000_000  # ~57.8 days of monotonic uptime
83UNIX_NOW_S = 1_784_000_000.0  # fixed unix wall-clock reading for the tests
84UNIX_NOW_MS = int(UNIX_NOW_S * 1000)
85COLD_LEAD_MS = BRIDGE_COLD_START_LEAD_MS
86WARM_LEAD_MS = BRIDGE_WARM_START_LEAD_MS
87# Patched with zero delays so the re-join backoff runs instantly while the
88# attempt-count and give-up logic around it stays real.
89_NO_REJOIN_DELAYS = "music_assistant.providers.airplay.sendspin_bridge.BRIDGE_REJOIN_ATTEMPT_DELAYS"
90
91
92def _audible_instant_us(clock: ManualClock, lead_ms: int) -> int:
93    """Return a sample Sendspin audible instant that far ahead of now (exercises the mapping)."""
94    return clock.now_us() + lead_ms * 1_000
95
96
97def _unix_at(sendspin_us: int) -> float:
98    """Model a constant-offset, same-rate Sendspin<->unix relationship."""
99    return UNIX_NOW_S + (sendspin_us - SENDSPIN_EPOCH_US) / 1_000_000
100
101
102def test_maps_future_delta_to_unix_now_plus_lead() -> None:
103    """An instant a lead ahead maps to unix_now + that lead (in ms)."""
104    clock = ManualClock(now_us_value=SENDSPIN_EPOCH_US)
105    drop_until = _audible_instant_us(clock, COLD_LEAD_MS)
106
107    start_unix_ms = sendspin_audible_instant_to_unix_ms(drop_until, clock.now_us(), UNIX_NOW_S)
108
109    assert start_unix_ms == int(UNIX_NOW_S * 1000) + COLD_LEAD_MS
110
111
112def test_standing_clock_offset_cancels_out() -> None:
113    """
114    The absolute Sendspin epoch must not affect the result.
115
116    Two wildly different monotonic epochs, with the same future delta and the
117    same unix reading, must yield the exact same start instant. This is what
118    makes the naive ``now/now`` subtraction correct: only the delta transfers
119    between the clocks, so any standing offset cancels.
120    """
121    clock_a = ManualClock(now_us_value=SENDSPIN_EPOCH_US)
122    clock_b = ManualClock(now_us_value=SENDSPIN_EPOCH_US + 987_654_321_000)
123
124    result_a = sendspin_audible_instant_to_unix_ms(
125        _audible_instant_us(clock_a, COLD_LEAD_MS), clock_a.now_us(), UNIX_NOW_S
126    )
127    result_b = sendspin_audible_instant_to_unix_ms(
128        _audible_instant_us(clock_b, COLD_LEAD_MS), clock_b.now_us(), UNIX_NOW_S
129    )
130
131    assert result_a == result_b
132
133
134def test_derived_start_equals_sendspin_audible_instant_in_unix() -> None:
135    """
136    The derived start lands on the unix time that coincides with the Sendspin instant.
137
138    Models the two clocks as running at the same rate with a constant offset
139    (unix = anchor + (sendspin_us - epoch)/1e6). The bridge only ever reads the
140    two clocks together, so the result must land exactly on the unix time that
141    coincides with the Sendspin audible instant, for any offset and any lead.
142    """
143    for lead_ms in (WARM_LEAD_MS, COLD_LEAD_MS):
144        clock = ManualClock(now_us_value=SENDSPIN_EPOCH_US)
145        drop_until = _audible_instant_us(clock, lead_ms)
146        # Some real time passes between setting the anchor and starting the CLI.
147        clock.advance_us(40_000)  # 40 ms of setup churn (cleanup, task hop)
148        sendspin_now = clock.now_us()
149        unix_now = _unix_at(sendspin_now)
150
151        start_unix_ms = sendspin_audible_instant_to_unix_ms(drop_until, sendspin_now, unix_now)
152
153        assert start_unix_ms == int(_unix_at(drop_until) * 1000)
154
155
156def test_scheduling_gap_between_reads_shrinks_lead_not_target() -> None:
157    """
158    A gap before CLI start shrinks the remaining lead but keeps the audible instant fixed.
159
160    Computing the anchor immediately vs after a 400 ms gap must resolve to the
161    same unix instant, because the future delta is recomputed against the same
162    (advanced) Sendspin clock and unix reading.
163    """
164    clock = ManualClock(now_us_value=SENDSPIN_EPOCH_US)
165    drop_until = _audible_instant_us(clock, COLD_LEAD_MS)
166
167    immediate = sendspin_audible_instant_to_unix_ms(drop_until, clock.now_us(), UNIX_NOW_S)
168
169    gap_s = 0.4
170    clock.advance_us(int(gap_s * 1_000_000))
171    delayed = sendspin_audible_instant_to_unix_ms(drop_until, clock.now_us(), UNIX_NOW_S + gap_s)
172
173    assert immediate == delayed
174    # And the remaining lead really did shrink by the gap.
175    remaining_lead_ms = delayed - int((UNIX_NOW_S + gap_s) * 1000)
176    assert remaining_lead_ms == COLD_LEAD_MS - int(gap_s * 1000)
177
178
179def test_anchor_already_in_the_past_maps_to_a_past_unix_instant() -> None:
180    """
181    An audible instant behind 'now' yields a unix ms before the unix reading.
182
183    This is the setup-outran-the-lead edge case: the value stays a faithful
184    projection (negative lead) rather than being clamped here, so the anchor
185    math can see it and raise the start to the join floor itself.
186    """
187    clock = ManualClock(now_us_value=SENDSPIN_EPOCH_US)
188    audible_in_the_past = clock.now_us() - 300_000  # 300 ms ago
189
190    start_unix_ms = sendspin_audible_instant_to_unix_ms(
191        audible_in_the_past, clock.now_us(), UNIX_NOW_S
192    )
193
194    assert start_unix_ms == int(UNIX_NOW_S * 1000) - 300
195    assert start_unix_ms < int(UNIX_NOW_S * 1000)
196
197
198# --- Start anchor: fresh keeps position 0, late join lands at live position ---
199
200
201def _make_bridge(
202    clock_now_us: int,
203    protocol: StreamingProtocol = StreamingProtocol.AIRPLAY2,
204    sync_adjust: int = 0,
205) -> SendspinAirPlayBridge:
206    """Build a bridge with mocked provider/player/server and a ManualClock."""
207    provider = MagicMock()
208    provider.mass = MagicMock()
209    # Real values: the decision is handed to the CLI verbatim and the group's is
210    # compared against it, both of which a MagicMock would answer truthily
211    # whatever was resolved. None models a group with no live decision.
212    provider.bridge_manager.resolve_shared_ptp = MagicMock(return_value=False)
213    provider.bridge_manager.group_shared_ptp = MagicMock(return_value=None)
214    airplay_player = MagicMock()
215    airplay_player.player_id = "apc43875e9e53a"
216    airplay_player.display_name = "Test Player"
217    airplay_player.protocol = protocol
218    # A real None: the start path stops whatever stream the player already holds
219    # before it spawns a process, and a bare MagicMock reads as a live session.
220    airplay_player.stream = None
221    # A real lock: it is what keeps the bridge and native spawn paths off the
222    # same receiver, and `async with` on a MagicMock does not even work.
223    airplay_player.stream_spawn_lock = asyncio.Lock()
224    # A real int: the anchor math guards sync_adjust with isinstance(..., int), so
225    # a MagicMock would silently read as 0 and pass the test for the wrong reason.
226    airplay_player.config.get_value = MagicMock(return_value=sync_adjust)
227    sendspin_server = MagicMock()
228    sendspin_server.clock = ManualClock(now_us_value=clock_now_us)
229    bridge = SendspinAirPlayBridge(provider, airplay_player, sendspin_server)
230    bridge._is_streaming = True
231    return bridge
232
233
234def _pcm_chunk(timestamp_us: int, duration_us: int = 100_000) -> AudioChunk:
235    """Build a silent PCM AudioChunk at a Sendspin timestamp."""
236    frames = int(duration_us * BRIDGE_SAMPLE_RATE / 1_000_000)
237    data = b"\x00" * (frames * BRIDGE_CHANNELS * BRIDGE_BYTES_PER_SAMPLE)
238    return AudioChunk(
239        data=data, timestamp_us=timestamp_us, duration_us=duration_us, byte_count=len(data)
240    )
241
242
243def test_fresh_start_anchors_to_first_chunk_and_keeps_intro() -> None:
244    """
245    A fresh track's opening is kept: byte 0 anchors to the first chunk, not now+lead.
246
247    Models the clip scenario where the first delivered chunk (file position 0)
248    is scheduled earlier than ``clock.now() + the bridge lead``. Anchoring to
249    ``now + lead`` would drop everything before it -- the intro. The chunk
250    timestamp must win, and its audio must reach the CLI, not be dropped.
251    """
252    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
253    now_plus_lead = SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
254    first_chunk_ts = SENDSPIN_EPOCH_US + 250_000  # position 0, only 250 ms ahead of now
255    assert first_chunk_ts < now_plus_lead
256
257    with patch.object(bridge, "_start_protocol_from_chunk", MagicMock()):
258        bridge._on_audio_chunk(_pcm_chunk(first_chunk_ts))
259
260    assert bridge._drop_until_us == first_chunk_ts
261    # Held while the anchor is negotiated, then queued -- never discarded.
262    _settle_anchor(bridge)
263    assert not bridge._write_queue.empty()
264
265
266def test_late_join_anchors_to_catchup_target_live_position() -> None:
267    """
268    A late joiner lands at the group's current position, not at track zero.
269
270    After minutes of playback the first delivered chunk is the catch-up target
271    (playhead + the bridge lead), far from the track start. The anchor must follow that
272    chunk so the joiner maps onto the live timeline instead of restarting at 0.
273    """
274    playhead_us = SENDSPIN_EPOCH_US + 600_000_000  # 600 s into the session
275    bridge = _make_bridge(clock_now_us=playhead_us)
276    catchup_target_ts = playhead_us + COLD_LEAD_MS * 1_000
277
278    with patch.object(bridge, "_start_protocol_from_chunk", MagicMock()):
279        bridge._on_audio_chunk(_pcm_chunk(catchup_target_ts))
280
281    assert bridge._drop_until_us == catchup_target_ts
282    # The anchor tracks the advanced playhead, not a fresh now+lead-from-zero.
283    assert bridge._drop_until_us > SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
284
285
286# --- Timeline alignment: a discontinuity must not shift the device off the group clock ---
287
288
289def _drain_queued_bytes(bridge: SendspinAirPlayBridge) -> int:
290    """Return the number of audio bytes handed to the CLI writer, emptying the queue."""
291    total = 0
292    while not bridge._write_queue.empty():
293        data = bridge._write_queue.get_nowait()
294        if data is not None:
295            total += len(data)
296    return total
297
298
299def _settle_anchor(bridge: SendspinAirPlayBridge) -> None:
300    """Model the binary acking exactly the anchor asked for: replay what was held."""
301    bridge._anchor_settled = True
302    held = list(bridge._held_chunks)
303    bridge._held_chunks.clear()
304    bridge._held_us = 0
305    for chunk in held:
306        bridge._align_chunk(chunk)
307
308
309def _start_stream_at(bridge: SendspinAirPlayBridge, first_chunk_ts: int) -> None:
310    """Feed the anchoring first chunk so the bridge is aligned and streaming."""
311    with patch.object(bridge, "_start_protocol_from_chunk", MagicMock()):
312        bridge._on_audio_chunk(_pcm_chunk(first_chunk_ts))
313    # The mocked task reports done() truthy by default, which the chunk handler
314    # reads as a failed protocol start; model a start still in flight instead.
315    cast("MagicMock", bridge._airplay_stream_start_task).done.return_value = False
316    _settle_anchor(bridge)
317
318
319def _expected_frames(bridge: SendspinAirPlayBridge, timeline_end_us: int) -> int:
320    """Frames the CLI stream must hold for its cursor to sit at a timeline instant."""
321    return round((timeline_end_us - bridge._drop_until_us) * BRIDGE_SAMPLE_RATE / 1_000_000)
322
323
324def test_timeline_gap_is_padded_with_silence() -> None:
325    """
326    A hole in the Sendspin timeline is filled so the device stays on the group clock.
327
328    Sendspin rebases the shared timeline forward when audio production stalls,
329    delivering no audio for the skipped span. The CLI plays its byte stream at a
330    fixed rate from an anchor that is never revised, so writing the next chunk
331    straight after the previous one would leave this device permanently ahead of
332    the group by the size of the hole.
333    """
334    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
335    first_ts = SENDSPIN_EPOCH_US + 250_000
336    _start_stream_at(bridge, first_ts)
337    _drain_queued_bytes(bridge)
338
339    gap_us = 415_711
340    next_ts = first_ts + 100_000 + gap_us
341    bridge._on_audio_chunk(_pcm_chunk(next_ts))
342
343    expected = _expected_frames(bridge, next_ts + 100_000)
344    assert bridge._queued_frames == expected
345    assert (
346        _drain_queued_bytes(bridge)
347        == (expected - _expected_frames(bridge, first_ts + 100_000))
348        * BRIDGE_CHANNELS
349        * BRIDGE_BYTES_PER_SAMPLE
350    )
351
352
353def test_overlapping_chunk_head_is_trimmed() -> None:
354    """A chunk reaching back behind the write cursor keeps only its unwritten tail."""
355    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
356    first_ts = SENDSPIN_EPOCH_US + 250_000
357    _start_stream_at(bridge, first_ts)
358    _drain_queued_bytes(bridge)
359
360    overlap_us = 40_000
361    next_ts = first_ts + 100_000 - overlap_us
362    bridge._on_audio_chunk(_pcm_chunk(next_ts))
363
364    assert bridge._queued_frames == _expected_frames(bridge, next_ts + 100_000)
365
366
367def test_chunk_entirely_behind_the_cursor_is_dropped() -> None:
368    """Audio already written is not queued a second time."""
369    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
370    first_ts = SENDSPIN_EPOCH_US + 250_000
371    _start_stream_at(bridge, first_ts)
372    _drain_queued_bytes(bridge)
373    cursor_frames = bridge._queued_frames
374
375    bridge._on_audio_chunk(_pcm_chunk(first_ts + 10_000, duration_us=50_000))
376
377    assert bridge._queued_frames == cursor_frames
378    assert _drain_queued_bytes(bridge) == 0
379
380
381@pytest.mark.parametrize("server_side", [False, True])
382def test_stream_start_resets_the_write_cursor(server_side: bool) -> None:
383    """
384    Both stream-start entry points rewind the cursor so the next chunk re-anchors byte 0.
385
386    A cursor carried over from the previous stream would place the first chunk of
387    the new one far behind the write position and get it trimmed away as already
388    written, and a settled-anchor flag carried over would let the new stream's
389    chunks be placed against the previous stream's anchor.
390    """
391    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
392    _start_stream_at(bridge, SENDSPIN_EPOCH_US + 250_000)
393    bridge._held_chunks.append(_pcm_chunk(SENDSPIN_EPOCH_US + 250_000))
394    assert bridge._queued_frames > 0
395
396    if server_side:
397        bridge._on_stream_start(MagicMock())
398    else:
399        bridge._on_bridge_stream_start()
400
401    assert bridge._queued_frames == 0
402    assert bridge._drop_until_us == 0
403    assert bridge._anchor_settled is False
404    assert not bridge._held_chunks
405
406
407def test_contiguous_chunks_are_written_untouched() -> None:
408    """Normal playback queues exactly its own audio -- no padding, no trimming."""
409    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
410    first_ts = SENDSPIN_EPOCH_US + 250_000
411    _start_stream_at(bridge, first_ts)
412    _drain_queued_bytes(bridge)
413
414    for index in range(1, 20):
415        bridge._on_audio_chunk(_pcm_chunk(first_ts + index * 100_000))
416
417    assert bridge._queued_frames == _expected_frames(bridge, first_ts + 20 * 100_000)
418    assert _drain_queued_bytes(bridge) == 19 * 100_000 * BRIDGE_BYTES_PER_SECOND // 1_000_000
419
420
421# --- Write pacing: bound the device buffer so a catch-up backlog is not dumped ---
422
423
424def test_device_buffer_ahead_seconds_tracks_write_cursor() -> None:
425    """The buffered-ahead measure follows byte 0 = start anchor, +1 s per second written."""
426    start_unix_ms = 1_784_000_000_000
427    now = start_unix_ms / 1000
428
429    assert device_buffer_ahead_seconds(start_unix_ms, 0, BRIDGE_BYTES_PER_SECOND, now) == 0.0
430    one_second = BRIDGE_BYTES_PER_SECOND
431    ahead = device_buffer_ahead_seconds(start_unix_ms, one_second, BRIDGE_BYTES_PER_SECOND, now)
432    assert abs(ahead - 1.0) < 1e-9
433
434
435def test_late_join_backlog_trips_pacing_bound_but_steady_feed_does_not() -> None:
436    """A ~27 s catch-up backlog exceeds the bound; a few seconds of steady audio stays under it."""
437    start_unix_ms = 1_784_000_000_000
438    now = start_unix_ms / 1000
439
440    backlog_ahead = device_buffer_ahead_seconds(
441        start_unix_ms, 27 * BRIDGE_BYTES_PER_SECOND, BRIDGE_BYTES_PER_SECOND, now
442    )
443    assert backlog_ahead > MAX_DEVICE_BUFFER_SECONDS
444
445    steady_ahead = device_buffer_ahead_seconds(
446        start_unix_ms, 3 * BRIDGE_BYTES_PER_SECOND, BRIDGE_BYTES_PER_SECOND, now
447    )
448    assert steady_ahead < MAX_DEVICE_BUFFER_SECONDS
449
450
451async def test_failed_cli_write_does_not_advance_pacing_cursor() -> None:
452    """A dropped write cannot move the pacing cursor past audio the CLI never received."""
453    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
454    stream = MagicMock()
455    stream.write_audio = AsyncMock(side_effect=[OSError("write failed"), None])
456    stream.write_audio_eof = AsyncMock()
457    bridge._airplay_stream = stream
458    bridge._airplay_stream_ready.set()
459    bridge._start_unix_ms = int(UNIX_NOW_S * 1000)
460    bridge._write_queue.put_nowait(b"first")
461    bridge._write_queue.put_nowait(b"second")
462    bridge._write_queue.put_nowait(None)
463
464    with patch(
465        "music_assistant.providers.airplay.sendspin_bridge.device_buffer_ahead_seconds",
466        return_value=0.0,
467    ) as buffer_ahead:
468        await bridge._cli_writer()
469
470    assert [call.args[1] for call in buffer_ahead.call_args_list] == [0, 0]
471    assert stream.write_audio.await_count == 2
472
473
474# --- Commanded cold start and warm handover ------------------------------------
475
476
477def _make_anchor_stream(
478    *,
479    ready_at_unix_ms: int | None = None,
480    ack: int | None = None,
481    warm_lead_ms: int = 0,
482    flushed_head_unix_ms: int = 0,
483    audio_pending_ms: int = 0,
484) -> MagicMock:
485    """
486    Build an AirPlayStream mock the anchor math can run against.
487
488    The bridge reads these off the stream and does arithmetic on them, so they
489    must be real numbers: the anchor compares ``warm_lead_ms`` /
490    ``flushed_head_unix_ms`` / ``audio_pending_ms`` with ``> 0`` and the shift
491    fold subtracts ``cumulative_shift_seconds``, none of which a bare MagicMock
492    can answer (every one of them is truthy).
493
494    :param ack: Instant the binary acks the START at. None acks the commanded
495        instant, as a feasible one is.
496    """
497
498    async def _ack_start(start_unix_ms: int = 0, **_kwargs: object) -> int:
499        return start_unix_ms if ack is None else ack
500
501    stream = MagicMock()
502    stream.cumulative_shift_seconds = 0.0
503    stream.connect = AsyncMock()
504    stream.wait_for_connection = AsyncMock()
505    stream.stop = AsyncMock()
506    stream.flush = AsyncMock(return_value=True)
507    stream.wait_clock_ready = AsyncMock(return_value=(ClockReadiness.PROJECTED, ready_at_unix_ms))
508    stream.start = AsyncMock(side_effect=_ack_start)
509    stream.warm_lead_ms = warm_lead_ms
510    stream.flushed_head_unix_ms = flushed_head_unix_ms
511    stream.audio_pending_ms = audio_pending_ms
512    return stream
513
514
515async def test_cold_start_connects_then_anchors_first_start() -> None:
516    """A fresh bridge stream anchors its first START only after the CLI connects."""
517    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
518    bridge._drop_until_us = SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
519    bridge._airplay_stream_start_task = asyncio.current_task()
520    commanded = UNIX_NOW_MS + COLD_LEAD_MS
521    stream = _make_anchor_stream(ack=commanded)
522
523    with (
524        patch(
525            "music_assistant.providers.airplay.sendspin_bridge.AirPlayStream",
526            return_value=stream,
527        ),
528        patch(
529            "music_assistant.providers.airplay.sendspin_bridge.time.time",
530            return_value=UNIX_NOW_S,
531        ),
532    ):
533        await bridge._start_protocol_from_chunk()
534
535    stream.connect.assert_awaited_once_with(False)
536    stream.wait_for_connection.assert_awaited_once_with()
537    stream.start.assert_awaited_once_with(commanded, join=True)
538    assert bridge._airplay_stream is stream
539    assert bridge.airplay_player.stream is stream
540    assert bridge._started is True
541    assert bridge._airplay_stream_ready.is_set()
542
543
544async def test_a_fresh_process_releases_a_foreign_mute_latch_before_it_connects() -> None:
545    """
546    A cold start releases a foreign mute latch before it connects.
547
548    Connecting is what carries that state to the device, so releasing the latch
549    after it would not be heard until the next command.
550    """
551    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
552    bridge._drop_until_us = SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
553    bridge._airplay_stream_start_task = asyncio.current_task()
554    stream = _make_anchor_stream()
555    order: list[str] = []
556    cast("MagicMock", bridge.airplay_player).release_foreign_mute_latch = MagicMock(
557        side_effect=lambda: order.append("release_mute")
558    )
559    stream.connect = AsyncMock(side_effect=lambda *_a, **_kw: order.append("connect"))
560
561    with (
562        patch(
563            "music_assistant.providers.airplay.sendspin_bridge.AirPlayStream",
564            return_value=stream,
565        ),
566        patch(
567            "music_assistant.providers.airplay.sendspin_bridge.time.time",
568            return_value=UNIX_NOW_S,
569        ),
570    ):
571        await bridge._start_protocol_from_chunk()
572
573    assert order == ["release_mute", "connect"]
574
575
576async def test_a_kept_process_keeps_a_mute_latch_owned_by_another_control() -> None:
577    """
578    A warm handover does not release a foreign mute latch.
579
580    Only a connect re-sends VOLUME=, so releasing the latch over a kept process
581    would clear it while the device is still muted on its own end, leaving the
582    stream silent with nothing to say so.
583    """
584    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
585    kept_stream = _make_anchor_stream()
586    bridge._airplay_stream = kept_stream
587    bridge.airplay_player.stream = kept_stream
588    bridge._started = True
589
590    # the whole warm restart, from the Sendspin stream-start callback to the handover
591    bridge._on_bridge_stream_start()
592    assert bridge._airplay_stream is kept_stream
593    bridge._drop_until_us = SENDSPIN_EPOCH_US
594    bridge._airplay_stream_start_task = asyncio.current_task()
595
596    with patch(
597        "music_assistant.providers.airplay.sendspin_bridge.time.time",
598        return_value=UNIX_NOW_S,
599    ):
600        await bridge._start_protocol_from_chunk()
601
602    kept_stream.flush.assert_awaited_once_with()
603    cast("MagicMock", bridge.airplay_player).release_foreign_mute_latch.assert_not_called()
604
605
606async def test_a_failed_warm_handover_releases_the_latch_before_its_cold_retry() -> None:
607    """
608    The cold retry after a failed warm handover still releases a foreign mute latch.
609
610    That retry spawns a fresh process, which is sent whatever volume and mute it
611    finds on connect, so a mute latch the parent no longer owns would start it silent.
612    """
613    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
614    bridge._drop_until_us = SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
615    bridge._airplay_stream_start_task = asyncio.current_task()
616    kept_stream = _make_anchor_stream()
617    kept_stream.flush = AsyncMock(return_value=False)
618    bridge._airplay_stream = kept_stream
619    cold_stream = _make_anchor_stream()
620
621    with (
622        patch(
623            "music_assistant.providers.airplay.sendspin_bridge.AirPlayStream",
624            return_value=cold_stream,
625        ),
626        patch(
627            "music_assistant.providers.airplay.sendspin_bridge.time.time",
628            return_value=UNIX_NOW_S,
629        ),
630    ):
631        await bridge._start_protocol_from_chunk()
632
633    cold_stream.connect.assert_awaited_once_with(False)
634    cast("MagicMock", bridge.airplay_player).release_foreign_mute_latch.assert_called_once_with()
635
636
637async def test_a_superseded_cold_start_never_reaches_the_receiver() -> None:
638    """
639    A cold start that already lost the race bails out before it spawns anything.
640
641    Connecting first would pay a full process spawn and session setup only to
642    kill it again, put a second session on a receiver the newer start is about
643    to claim, and overwrite the shared-clock decision of the process that start
644    is really running.
645    """
646    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
647    bridge._drop_until_us = SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
648    # the decision the newer start recorded for the process it is spawning
649    bridge._use_shared_ptp = True
650    # a different task owns the bridge: this cold start is stale
651    bridge._airplay_stream_start_task = MagicMock()
652    stream = _make_anchor_stream()
653
654    with (
655        patch(
656            "music_assistant.providers.airplay.sendspin_bridge.AirPlayStream",
657            return_value=stream,
658        ),
659        patch(
660            "music_assistant.providers.airplay.sendspin_bridge.time.time",
661            return_value=UNIX_NOW_S,
662        ),
663    ):
664        await bridge._start_protocol_from_chunk()
665
666    stream.connect.assert_not_awaited()
667    stream.stop.assert_not_awaited()
668    assert bridge._use_shared_ptp is True
669
670
671async def test_a_superseded_start_leaves_the_kept_stream_untouched() -> None:
672    """
673    A start that lost the race never flushes the stream the newer one kept.
674
675    Arming the bridge keeps a warm-eligible stream alive, so the stale and the
676    newer start find the same instance; flushing it here would cut into the
677    audio the newer start is anchoring on it.
678    """
679    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
680    kept_stream = _make_anchor_stream()
681    bridge._airplay_stream = kept_stream
682    # a different task owns the bridge: this start is stale
683    bridge._airplay_stream_start_task = MagicMock()
684
685    await bridge._start_protocol_from_chunk()
686
687    kept_stream.flush.assert_not_awaited()
688    kept_stream.stop.assert_not_awaited()
689    assert bridge._airplay_stream is kept_stream
690
691
692async def test_a_start_superseded_during_the_warm_fallback_spawns_nothing() -> None:
693    """
694    Losing the race while releasing the kept stream still stops short of the receiver.
695
696    A failed warm handover tears the kept stream down before it falls back to a
697    cold start, and that teardown is long enough for a newer start to claim the
698    bridge in the meantime.
699    """
700    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
701    bridge._drop_until_us = SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
702    bridge._airplay_stream_start_task = asyncio.current_task()
703    kept_stream = _make_anchor_stream()
704    kept_stream.flush = AsyncMock(return_value=False)
705    bridge._airplay_stream = kept_stream
706    cold_stream = _make_anchor_stream()
707
708    async def stop(**_kwargs: object) -> None:
709        # a newer stream start claimed the bridge while the kept stream went down
710        bridge._airplay_stream_start_task = MagicMock()
711
712    kept_stream.stop = AsyncMock(side_effect=stop)
713
714    with patch(
715        "music_assistant.providers.airplay.sendspin_bridge.AirPlayStream",
716        return_value=cold_stream,
717    ):
718        await bridge._start_protocol_from_chunk()
719
720    cold_stream.connect.assert_not_awaited()
721    cold_stream.stop.assert_not_awaited()
722
723
724async def test_cold_start_superseded_while_connecting_stops_its_transport() -> None:
725    """A cold stream superseded while its process comes up is torn down again."""
726    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
727    bridge._drop_until_us = SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
728    bridge._airplay_stream_start_task = asyncio.current_task()
729    stream = _make_anchor_stream()
730
731    async def wait_for_connection() -> None:
732        # a newer stream start claimed the bridge while the process came up
733        bridge._airplay_stream_start_task = MagicMock()
734
735    stream.wait_for_connection = AsyncMock(side_effect=wait_for_connection)
736
737    assert await bridge._start_cold_stream(stream) is False
738
739    stream.start.assert_not_awaited()
740    stream.stop.assert_awaited_once_with(force=True)
741
742
743async def test_cold_start_superseded_during_the_anchor_stops_its_transport() -> None:
744    """
745    A cold stream superseded while the binary holds its ack is torn down.
746
747    The anchor publishes the stream before commanding START, so a supersession
748    inside it would otherwise leave a live cliairplay attached to the receiver
749    with nobody owning it.
750    """
751    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
752    bridge._drop_until_us = SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
753    bridge._airplay_stream_start_task = asyncio.current_task()
754    stream = _make_anchor_stream()
755
756    async def start(start_unix_ms: int, **_kwargs: object) -> int:
757        # A newer stream start claimed the bridge while the binary held its ack.
758        bridge._airplay_stream_start_task = MagicMock()
759        return start_unix_ms
760
761    stream.start = AsyncMock(side_effect=start)
762
763    with patch(
764        "music_assistant.providers.airplay.sendspin_bridge.time.time",
765        return_value=UNIX_NOW_S,
766    ):
767        assert await bridge._start_cold_stream(stream) is False
768
769    stream.stop.assert_awaited_once_with(force=True)
770    assert bridge._airplay_stream is None
771    assert bridge.airplay_player.stream is None
772
773
774async def test_superseded_cold_stream_teardown_spares_the_newer_owner() -> None:
775    """A newer start's published stream survives the stale cold stream's teardown."""
776    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
777    bridge._drop_until_us = SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
778    bridge._airplay_stream_start_task = asyncio.current_task()
779    stream = _make_anchor_stream()
780    newer_stream = _make_anchor_stream()
781
782    async def start(start_unix_ms: int, **_kwargs: object) -> int:
783        bridge._airplay_stream_start_task = MagicMock()
784        bridge._airplay_stream = newer_stream
785        bridge.airplay_player.stream = newer_stream
786        return start_unix_ms
787
788    stream.start = AsyncMock(side_effect=start)
789
790    with patch(
791        "music_assistant.providers.airplay.sendspin_bridge.time.time",
792        return_value=UNIX_NOW_S,
793    ):
794        assert await bridge._start_cold_stream(stream) is False
795
796    stream.stop.assert_awaited_once_with(force=True)
797    newer_stream.stop.assert_not_awaited()
798    assert bridge._airplay_stream is newer_stream
799    assert bridge.airplay_player.stream is newer_stream
800
801
802# --- Displacing a native session the bridge does not own ---
803
804
805def _make_native_stream(session: MagicMock | None = None) -> MagicMock:
806    """Build a live native AirPlayStream mock the bridge has to displace."""
807    stream = MagicMock()
808    stream.stop = AsyncMock()
809    stream.session = session
810    return stream
811
812
813async def test_cold_start_stops_a_native_session_before_it_spawns_a_process() -> None:
814    """
815    A native session on the same player is released before the bridge connects.
816
817    Two cliairplay processes on one receiver reset each other's control
818    channel, so the displaced transport has to be gone before the new
819    pair-setup; the session bookkeeping follows it.
820    """
821    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
822    bridge._drop_until_us = SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
823    bridge._airplay_stream_start_task = asyncio.current_task()
824    session = MagicMock()
825    order: list[str] = []
826    session.remove_client = AsyncMock(side_effect=lambda *_a, **_kw: order.append("remove_client"))
827    native_stream = _make_native_stream(session)
828    native_stream.stop = AsyncMock(side_effect=lambda **_kw: order.append("stop"))
829    bridge.airplay_player.stream = native_stream
830    stream = _make_anchor_stream()
831    stream.connect = AsyncMock(side_effect=lambda *_a, **_kw: order.append("connect"))
832
833    with (
834        patch(
835            "music_assistant.providers.airplay.sendspin_bridge.AirPlayStream",
836            return_value=stream,
837        ),
838        patch(
839            "music_assistant.providers.airplay.sendspin_bridge.time.time",
840            return_value=UNIX_NOW_S,
841        ),
842    ):
843        await bridge._start_protocol_from_chunk()
844
845    assert order == ["stop", "remove_client", "connect"]
846    assert session.remove_client.await_args.args[0] is bridge.airplay_player
847    assert bridge.airplay_player.stream is stream
848
849
850class _ReceiverProcesses:
851    """
852    Counts the cli processes live on one receiver.
853
854    Two of them reset each other's RTSP channel and both sessions die, so
855    ``peak`` is the value the collision tests assert on.
856    """
857
858    def __init__(self) -> None:
859        self.live = 0
860        self.peak = 0
861
862    def spawn(self, *_args: object, **_kwargs: object) -> None:
863        """Record a cli process pairing with the receiver."""
864        self.live += 1
865        self.peak = max(self.peak, self.live)
866
867    def kill(self, *_args: object, **_kwargs: object) -> None:
868        """Record a cli process letting go of the receiver."""
869        self.live = max(0, self.live - 1)
870
871
872async def _run_native_start(player: AirPlayPlayer, stream: MagicMock) -> None:
873    """Run the real native start path for a player against a throwaway session."""
874    pcm_format = MagicMock(sample_rate=44100, bit_depth=16, channels=2)
875    session = AirPlayStreamSession(MagicMock(), [player], pcm_format, MagicMock(elapsed_time=0))
876    with (
877        patch.object(session, "_start_player_ffmpeg", new_callable=AsyncMock),
878        patch(
879            "music_assistant.providers.airplay.stream_session.AirPlayStream",
880            return_value=stream,
881        ),
882    ):
883        await session._start_client(player, False)
884
885
886async def _bridge_start_racing_a_native_start(
887    interleave_at: str,
888) -> tuple[SendspinAirPlayBridge, _ReceiverProcesses, MagicMock, asyncio.Task[None]]:
889    """
890    Drive a bridge start and let a native start in at one of its awaits.
891
892    :param interleave_at: "displacement" starts the native path while the bridge
893        releases the session it is displacing, "connect" starts it while the
894        bridge's own cold connect is in flight.
895    :return: The bridge, its receiver process counter, the stream the bridge
896        started and the still-pending native start task.
897    """
898    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
899    bridge._drop_until_us = SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
900    bridge._airplay_stream_start_task = asyncio.current_task()
901    player = bridge.airplay_player
902    processes = _ReceiverProcesses()
903    native_task: asyncio.Task[None] | None = None
904
905    native_stream = _make_native_stream()
906    native_stream.running = False
907    native_stream.connect = AsyncMock(side_effect=processes.spawn)
908    native_stream.stop = AsyncMock(side_effect=processes.kill)
909
910    async def _start_native() -> None:
911        """Fire the native start and let it run until it can get no further."""
912        nonlocal native_task
913        native_task = asyncio.create_task(_run_native_start(player, native_stream))
914        # Three passes through the loop: enough for the native start to reach
915        # the spawn lock and, without one, to spawn its process outright.
916        for _ in range(3):
917            await asyncio.sleep(0)
918
919    displaced_stream = _make_native_stream(MagicMock(remove_client=AsyncMock()))
920    displaced_stream.running = True
921    processes.spawn()
922
923    async def _stop_displaced(**_kwargs: object) -> None:
924        displaced_stream.running = False
925        processes.kill()
926        if interleave_at == "displacement":
927            await _start_native()
928
929    displaced_stream.stop = AsyncMock(side_effect=_stop_displaced)
930    player.stream = displaced_stream
931
932    bridge_stream = _make_anchor_stream()
933    # A real bool: the native start behind the lock only stops what it reads as
934    # live, and this stream is what it finds published once the bridge is done.
935    bridge_stream.running = True
936    bridge_stream.stop = AsyncMock(side_effect=processes.kill)
937
938    async def _connect(*_args: object, **_kwargs: object) -> None:
939        processes.spawn()
940        if interleave_at == "connect":
941            await _start_native()
942
943    bridge_stream.connect = AsyncMock(side_effect=_connect)
944
945    with (
946        patch(
947            "music_assistant.providers.airplay.sendspin_bridge.AirPlayStream",
948            return_value=bridge_stream,
949        ),
950        patch(
951            "music_assistant.providers.airplay.sendspin_bridge.time.time",
952            return_value=UNIX_NOW_S,
953        ),
954    ):
955        await bridge._start_protocol_from_chunk()
956
957    assert native_task is not None
958    return bridge, processes, bridge_stream, native_task
959
960
961@pytest.mark.parametrize("interleave_at", ["displacement", "connect"])
962async def test_a_native_start_cannot_slip_into_a_bridge_start(interleave_at: str) -> None:
963    """
964    A native start racing a bridge start waits for it instead of joining it.
965
966    Both paths spawn a cli process for the same receiver, and neither the
967    displacement nor the cold connect is instant. Without the shared spawn lock
968    the native start publishes its own stream inside one of those awaits and the
969    bridge then connects a second process over it - two sessions on one receiver,
970    with the native one orphaned by the bridge's later publication.
971    """
972    bridge, processes, bridge_stream, native_task = await _bridge_start_racing_a_native_start(
973        interleave_at
974    )
975
976    # the bridge start finished on its own process, with the native one still
977    # queued behind the lock rather than paired with the same receiver
978    assert processes.peak == 1
979    assert not native_task.done()
980    assert bridge.airplay_player.stream is bridge_stream
981
982    await native_task
983
984    # the native start then displaces the bridge in turn, still one at a time
985    assert processes.peak == 1
986    assert processes.live == 1
987    bridge_stream.stop.assert_awaited()
988
989
990async def test_a_bridge_start_cannot_slip_into_a_native_start() -> None:
991    """
992    A bridge start racing a native start waits for it instead of joining it.
993
994    The mirror of the case above: the native path's own displacement stop is a
995    real await, and a bridge start landing inside it would connect its own
996    process and publish over a native start still in flight.
997    """
998    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
999    bridge._drop_until_us = SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
1000    player = bridge.airplay_player
1001    processes = _ReceiverProcesses()
1002    bridge_task: asyncio.Task[None] | None = None
1003
1004    bridge_stream = _make_anchor_stream()
1005    bridge_stream.running = True
1006    bridge_stream.connect = AsyncMock(side_effect=processes.spawn)
1007    bridge_stream.stop = AsyncMock(side_effect=processes.kill)
1008
1009    displaced_stream = _make_native_stream()
1010    displaced_stream.running = True
1011    processes.spawn()
1012
1013    async def _stop_displaced(**_kwargs: object) -> None:
1014        """Fire the bridge start from inside the native path's displacement stop."""
1015        nonlocal bridge_task
1016        displaced_stream.running = False
1017        processes.kill()
1018        if bridge_task is not None:
1019            # Only the first call is the interleaving under test. Without the
1020            # spawn lock the bridge start finds this same stream still published
1021            # and re-enters here; starting another bridge task would supersede
1022            # the one under test and the race would go unnoticed.
1023            return
1024        bridge_task = asyncio.create_task(bridge._start_protocol_from_chunk())
1025        # published before the task runs: the start path compares itself against
1026        # this handle to tell whether it still owns the bridge
1027        bridge._airplay_stream_start_task = bridge_task
1028        for _ in range(3):
1029            await asyncio.sleep(0)
1030
1031    displaced_stream.stop = AsyncMock(side_effect=_stop_displaced)
1032    player.stream = displaced_stream
1033
1034    native_stream = _make_native_stream()
1035    native_stream.running = True
1036    native_stream.connect = AsyncMock(side_effect=processes.spawn)
1037    native_stream.stop = AsyncMock(side_effect=processes.kill)
1038
1039    with (
1040        patch(
1041            "music_assistant.providers.airplay.sendspin_bridge.AirPlayStream",
1042            return_value=bridge_stream,
1043        ),
1044        patch(
1045            "music_assistant.providers.airplay.sendspin_bridge.time.time",
1046            return_value=UNIX_NOW_S,
1047        ),
1048    ):
1049        await _run_native_start(player, native_stream)
1050
1051        # the native start finished on its own process, with the bridge still
1052        # queued behind the lock
1053        assert processes.peak == 1
1054        assert bridge_task is not None
1055        assert not bridge_task.done()
1056        assert cast("MagicMock", player).stream is native_stream
1057
1058        await bridge_task
1059
1060    # the bridge then displaces the native session in turn, still one at a time
1061    assert processes.peak == 1
1062    assert cast("MagicMock", player).stream is bridge_stream
1063
1064
1065def _run_bridge_tasks_for_real(bridge: SendspinAirPlayBridge) -> list[asyncio.Task[None]]:
1066    """
1067    Let the bridge schedule real tasks instead of MagicMock ones.
1068
1069    :return: The list the scheduled tasks are appended to, to await at the end
1070        of the test.
1071    """
1072    loop = asyncio.get_running_loop()
1073    started: list[asyncio.Task[None]] = []
1074
1075    def create_task(
1076        coro: Coroutine[None, None, None], *, eager_start: bool = True, **_kwargs: object
1077    ) -> asyncio.Task[None]:
1078        # mirrors mass.create_task, whose default eager start would run the
1079        # coroutine to its first await before this returns
1080        task = asyncio.Task(coro, loop=loop, eager_start=eager_start)
1081        started.append(task)
1082        return task
1083
1084    cast("MagicMock", bridge.mass).create_task = create_task
1085    return started
1086
1087
1088async def test_a_native_start_waits_for_the_bridge_to_let_go_of_the_receiver() -> None:
1089    """
1090    A native start after a stop waits for the bridge process to actually be gone.
1091
1092    A stop (stop_streaming, or a give-up through _abandon_streaming) hands the
1093    transport to the cleanup path, which only kills the process a few awaits
1094    later. A play arriving in between would otherwise pair its own process with
1095    a receiver the old one is still holding.
1096    """
1097    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1098    player = bridge.airplay_player
1099    processes = _ReceiverProcesses()
1100    scheduled = _run_bridge_tasks_for_real(bridge)
1101
1102    bridge_stream = _make_anchor_stream()
1103
1104    async def _slow_stop(**_kwargs: object) -> None:
1105        # a real teardown writes ACTION=STOP, drops the command pipe and kills
1106        # the process, none of which is instant
1107        for _ in range(4):
1108            await asyncio.sleep(0)
1109        processes.kill()
1110
1111    bridge_stream.stop = AsyncMock(side_effect=_slow_stop)
1112    bridge._airplay_stream = bridge_stream
1113    player.stream = bridge_stream
1114    processes.spawn()
1115
1116    native_stream = _make_native_stream()
1117    native_stream.running = False
1118    native_stream.connect = AsyncMock(side_effect=processes.spawn)
1119    native_stream.stop = AsyncMock(side_effect=processes.kill)
1120
1121    bridge._release_streaming()
1122    # the reference is already gone while the process is not
1123    assert cast("MagicMock", player).stream is None
1124    assert processes.live == 1
1125
1126    await _run_native_start(player, native_stream)
1127
1128    assert processes.peak == 1
1129    assert cast("MagicMock", player).stream is native_stream
1130    await asyncio.gather(*scheduled, return_exceptions=True)
1131    assert processes.live == 1
1132
1133
1134async def test_a_teardown_does_not_lose_the_receiver_to_a_queued_native_start() -> None:
1135    """
1136    A native start already queued for the lock still finds the process to displace.
1137
1138    The lock is handed out in arrival order, so a teardown that only queues for
1139    it after cancelling the start holding it lands behind a native start that
1140    was already waiting. That start has to see the stream the teardown has not
1141    released yet, or it reads the speaker as free and pairs a second process
1142    with it.
1143    """
1144    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1145    bridge._drop_until_us = SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
1146    player = bridge.airplay_player
1147    processes = _ReceiverProcesses()
1148    scheduled = _run_bridge_tasks_for_real(bridge)
1149
1150    kept_stream = _make_kept_stream()
1151    flush_gate = asyncio.Event()
1152
1153    async def _blocking_flush(**_kwargs: object) -> bool:
1154        await flush_gate.wait()
1155        return True
1156
1157    kept_stream.flush = AsyncMock(side_effect=_blocking_flush)
1158    kept_stream.stop = AsyncMock(side_effect=lambda **_kw: processes.kill())
1159    bridge._airplay_stream = kept_stream
1160    # a legacy RAOP process carries no timing decision, so it stays warm eligible
1161    bridge._use_shared_ptp = None
1162    player.stream = kept_stream
1163    processes.spawn()
1164
1165    warm_task = asyncio.create_task(bridge._start_protocol_from_chunk())
1166    bridge._airplay_stream_start_task = warm_task
1167    for _ in range(5):
1168        await asyncio.sleep(0)
1169    assert player.stream_spawn_lock.locked()
1170
1171    native_stream = _make_native_stream()
1172    native_stream.running = False
1173    native_stream.connect = AsyncMock(side_effect=processes.spawn)
1174    native_stream.stop = AsyncMock(side_effect=processes.kill)
1175    native_task = asyncio.create_task(_run_native_start(player, native_stream))
1176    for _ in range(3):
1177        await asyncio.sleep(0)
1178    assert not native_task.done()
1179
1180    # the speaker is stopped: the teardown cancels the warm start that holds the
1181    # lock, and can only queue for it behind the native start already waiting
1182    bridge._release_streaming()
1183
1184    await asyncio.gather(warm_task, native_task, *scheduled, return_exceptions=True)
1185
1186    assert processes.peak == 1
1187    assert cast("MagicMock", player).stream is native_stream
1188
1189
1190async def test_a_native_start_joins_a_teardown_already_under_way() -> None:
1191    """
1192    A start waits out a stop that is already running on the stream it displaces.
1193
1194    A stream stops reporting itself as running the moment its own stop() begins,
1195    which is before the STOP write and the process kill. Reading that as "gone"
1196    would let the start pair a process with a receiver the old one has not let
1197    go of yet.
1198    """
1199    player = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US).airplay_player
1200    processes = _ReceiverProcesses()
1201
1202    # mid-teardown: no longer reporting itself as running, process still alive
1203    stopping_stream = _make_native_stream()
1204    stopping_stream.running = False
1205    processes.spawn()
1206
1207    async def _finish_teardown(**_kwargs: object) -> None:
1208        for _ in range(4):
1209            await asyncio.sleep(0)
1210        processes.kill()
1211
1212    stopping_stream.stop = AsyncMock(side_effect=_finish_teardown)
1213    player.stream = stopping_stream
1214
1215    native_stream = _make_native_stream()
1216    native_stream.running = False
1217    native_stream.connect = AsyncMock(side_effect=processes.spawn)
1218    native_stream.stop = AsyncMock(side_effect=processes.kill)
1219
1220    await _run_native_start(player, native_stream)
1221
1222    stopping_stream.stop.assert_awaited_once()
1223    assert processes.peak == 1
1224    assert cast("MagicMock", player).stream is native_stream
1225
1226
1227async def test_a_displaced_stream_that_cannot_be_stopped_blocks_the_start() -> None:
1228    """
1229    A transport the bridge cannot release stops it from spawning a second process.
1230
1231    Carrying on would put the new cli process on a receiver that is still
1232    serving the old one, which is the collision this guard exists to prevent.
1233    """
1234    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1235    bridge._drop_until_us = SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
1236    bridge._airplay_stream_start_task = asyncio.current_task()
1237    native_stream = _make_native_stream()
1238    native_stream.stop = AsyncMock(side_effect=RuntimeError("device unreachable"))
1239    bridge.airplay_player.stream = native_stream
1240    stream = _make_anchor_stream()
1241
1242    with (
1243        patch(
1244            "music_assistant.providers.airplay.sendspin_bridge.AirPlayStream",
1245            return_value=stream,
1246        ),
1247        patch(
1248            "music_assistant.providers.airplay.sendspin_bridge.time.time",
1249            return_value=UNIX_NOW_S,
1250        ),
1251    ):
1252        await bridge._start_protocol_from_chunk()
1253
1254    stream.connect.assert_not_awaited()
1255    # the session we could not stop stays published, so it is not lost track of
1256    assert bridge.airplay_player.stream is native_stream
1257
1258
1259async def test_failed_session_bookkeeping_still_lets_the_start_proceed() -> None:
1260    """The transport is what matters: a failed client removal does not block the start."""
1261    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1262    bridge._drop_until_us = SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
1263    bridge._airplay_stream_start_task = asyncio.current_task()
1264    session = MagicMock()
1265    session.remove_client = AsyncMock(side_effect=RuntimeError("session already gone"))
1266    bridge.airplay_player.stream = _make_native_stream(session)
1267    stream = _make_anchor_stream()
1268
1269    with (
1270        patch(
1271            "music_assistant.providers.airplay.sendspin_bridge.AirPlayStream",
1272            return_value=stream,
1273        ),
1274        patch(
1275            "music_assistant.providers.airplay.sendspin_bridge.time.time",
1276            return_value=UNIX_NOW_S,
1277        ),
1278    ):
1279        await bridge._start_protocol_from_chunk()
1280
1281    stream.connect.assert_awaited_once()
1282    assert bridge.airplay_player.stream is stream
1283
1284
1285async def test_deferred_teardown_spares_a_native_stream_that_replaced_the_bridge() -> None:
1286    """
1287    A teardown that fires after the native path took the speaker leaves it alone.
1288
1289    The grace window between a Sendspin stream ending and its cleanup is long
1290    enough for native playback to start, and unpublishing its stream here would
1291    strand the process behind it.
1292    """
1293    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1294    bridge_stream = MagicMock()
1295    bridge_stream.stop = AsyncMock()
1296    bridge._airplay_stream = bridge_stream
1297    native_stream = _make_native_stream()
1298    bridge.airplay_player.stream = native_stream
1299
1300    await bridge._stop_streaming()
1301
1302    assert bridge.airplay_player.stream is native_stream
1303    native_stream.stop.assert_not_awaited()
1304    cast("MagicMock", bridge.airplay_player).set_state_from_stream.assert_not_called()
1305
1306
1307async def test_cold_start_stops_a_sessionless_native_stream_directly() -> None:
1308    """A displaced stream with no session behind it is stopped on its own."""
1309    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1310    bridge._drop_until_us = SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
1311    bridge._airplay_stream_start_task = asyncio.current_task()
1312    native_stream = _make_native_stream()
1313    bridge.airplay_player.stream = native_stream
1314    stream = _make_anchor_stream()
1315
1316    with (
1317        patch(
1318            "music_assistant.providers.airplay.sendspin_bridge.AirPlayStream",
1319            return_value=stream,
1320        ),
1321        patch(
1322            "music_assistant.providers.airplay.sendspin_bridge.time.time",
1323            return_value=UNIX_NOW_S,
1324        ),
1325    ):
1326        await bridge._start_protocol_from_chunk()
1327
1328    native_stream.stop.assert_awaited_once_with(force=True)
1329    assert bridge.airplay_player.stream is stream
1330
1331
1332async def test_warm_reuse_never_displaces_the_stream_it_reuses() -> None:
1333    """The stream a warm handover rides is the bridge's own, so it is not stopped."""
1334    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1335    bridge._drop_until_us = SENDSPIN_EPOCH_US + WARM_LEAD_MS * 1_000
1336    bridge._airplay_stream_start_task = asyncio.current_task()
1337    kept_stream = _make_anchor_stream(ack=UNIX_NOW_MS + WARM_LEAD_MS)
1338    bridge._airplay_stream = kept_stream
1339    bridge.airplay_player.stream = kept_stream
1340
1341    with patch(
1342        "music_assistant.providers.airplay.sendspin_bridge.time.time",
1343        return_value=UNIX_NOW_S,
1344    ):
1345        await bridge._start_protocol_from_chunk()
1346
1347    kept_stream.stop.assert_not_awaited()
1348    kept_stream.flush.assert_awaited_once()
1349    assert bridge.airplay_player.stream is kept_stream
1350
1351
1352@pytest.mark.parametrize("server_side", [False, True])
1353def test_stream_start_callbacks_leave_a_native_session_published(server_side: bool) -> None:
1354    """
1355    Both Sendspin stream-start callbacks leave a native stream on the player.
1356
1357    Dropping the reference there would strand its cli process: the start task
1358    stops what the player still points at, and only that.
1359    """
1360    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1361    native_stream = _make_native_stream()
1362    bridge.airplay_player.stream = native_stream
1363
1364    if server_side:
1365        bridge._on_stream_start(MagicMock())
1366    else:
1367        bridge._on_bridge_stream_start()
1368
1369    assert bridge._airplay_stream is None
1370    assert bridge.airplay_player.stream is native_stream
1371
1372
1373# --- Teardown player state reset ---
1374
1375
1376async def test_bridge_teardown_resets_the_players_stream_state() -> None:
1377    """A torn-down bridge stream leaves the AirPlay player IDLE at position 0."""
1378    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1379    stream = MagicMock()
1380    stream.stop = AsyncMock()
1381    bridge._airplay_stream = stream
1382    bridge.airplay_player.stream = stream
1383
1384    await bridge._stop_streaming()
1385
1386    stream.stop.assert_awaited_once_with(force=True)
1387    cast("MagicMock", bridge.airplay_player).set_state_from_stream.assert_called_once_with(
1388        state=PlaybackState.IDLE, elapsed_time=0
1389    )
1390
1391
1392async def test_bridge_teardown_spares_a_newer_streams_state() -> None:
1393    """Cleanup of a superseded stream never resets state a newer stream owns."""
1394    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1395    old_stream = MagicMock()
1396    old_stream.stop = AsyncMock()
1397    bridge.airplay_player.stream = MagicMock()
1398
1399    await bridge._cleanup_old_stream(old_stream, None, None)
1400
1401    cast("MagicMock", bridge.airplay_player).set_state_from_stream.assert_not_called()
1402
1403
1404# --- Anchoring: command an instant the device can hit, then honour the ack ---
1405
1406
1407def _prepare_anchor(
1408    bridge: SendspinAirPlayBridge, stream: MagicMock, first_chunk_lead_ms: int
1409) -> None:
1410    """Wire a bridge so ``_anchor_stream`` can be awaited directly on ``stream``."""
1411    bridge._drop_until_us = bridge.sendspin_server.clock.now_us() + first_chunk_lead_ms * 1_000
1412    bridge._airplay_stream = stream
1413    bridge._airplay_stream_start_task = asyncio.current_task()
1414
1415
1416async def _anchor(bridge: SendspinAirPlayBridge, stream: MagicMock, *, warm: bool = False) -> bool:
1417    """Run ``_anchor_stream`` with the unix clock pinned to UNIX_NOW_S."""
1418    with patch(
1419        "music_assistant.providers.airplay.sendspin_bridge.time.time",
1420        return_value=UNIX_NOW_S,
1421    ):
1422        return await bridge._anchor_stream(stream, warm=warm)
1423
1424
1425def _commanded_instant(stream: MagicMock) -> int:
1426    """Return the instant the START command carried."""
1427    return int(stream.start.await_args.args[0])
1428
1429
1430async def test_anchor_floors_at_the_join_headroom() -> None:
1431    """
1432    A Sendspin lead shorter than the join floor is raised to it.
1433
1434    AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS is the least the anchor may sit ahead of
1435    now, so a shorter lead is floored rather than honoured.
1436    """
1437    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1438    stream = _make_anchor_stream()
1439    _prepare_anchor(bridge, stream, first_chunk_lead_ms=250)
1440
1441    assert await _anchor(bridge, stream) is True
1442
1443    assert _commanded_instant(stream) == UNIX_NOW_MS + AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS
1444
1445
1446async def test_anchor_reports_leftover_audio_pending_on_stdin() -> None:
1447    """
1448    Audio still pending when the anchor is commanded is named as the offset it causes.
1449
1450    The writer is gated until the anchor settles, so anything pending was left
1451    behind by an earlier stream and the START anchors it as this one's first
1452    sample. The cursor only counts what the bridge queued itself, so no later
1453    realignment can see the resulting offset, which leaves this warning as the
1454    one place it surfaces.
1455    """
1456    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1457    stream = _make_anchor_stream(audio_pending_ms=92)
1458    _prepare_anchor(bridge, stream, first_chunk_lead_ms=WARM_LEAD_MS)
1459
1460    with patch.object(bridge.logger, "warning") as warning:
1461        assert await _anchor(bridge, stream, warm=True) is True
1462
1463    pending = [call for call in warning.call_args_list if "pending" in call.args[0]]
1464    assert len(pending) == 1
1465    assert pending[0].args[2] == 92
1466
1467
1468async def test_anchor_stays_quiet_when_stdin_was_left_empty() -> None:
1469    """An anchor commanded against empty stdin reports nothing."""
1470    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1471    stream = _make_anchor_stream()
1472    _prepare_anchor(bridge, stream, first_chunk_lead_ms=WARM_LEAD_MS)
1473
1474    with patch.object(bridge.logger, "warning") as warning:
1475        assert await _anchor(bridge, stream, warm=True) is True
1476
1477    assert not [call for call in warning.call_args_list if "pending" in call.args[0]]
1478
1479
1480async def test_anchor_follows_the_clock_ready_projection() -> None:
1481    """A receiver that projects a later readiness pushes the anchor out past it."""
1482    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1483    ready_at = UNIX_NOW_MS + 3200
1484    stream = _make_anchor_stream(ready_at_unix_ms=ready_at)
1485    _prepare_anchor(bridge, stream, first_chunk_lead_ms=250)
1486
1487    assert await _anchor(bridge, stream) is True
1488
1489    assert _commanded_instant(stream) == ready_at + AIRPLAY_CLOCK_READY_LEAD_MS
1490
1491
1492async def test_anchor_still_starts_a_receiver_whose_clock_stalled() -> None:
1493    """
1494    A stalled receiver is anchored anyway, unlike a late joiner, and warned about.
1495
1496    A joiner is dropped because the session plays on without it, while here
1497    dropping would stop the speaker, and the binary's stall report is a
1498    diagnosis a receiver can still come good from.
1499    """
1500    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1501    stream = _make_anchor_stream()
1502    stream.wait_clock_ready = AsyncMock(return_value=(ClockReadiness.STALLED, 0))
1503    _prepare_anchor(bridge, stream, first_chunk_lead_ms=250)
1504
1505    with patch.object(bridge.logger, "warning") as warning:
1506        assert await _anchor(bridge, stream) is True
1507
1508    assert _commanded_instant(stream) == UNIX_NOW_MS + AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS
1509    assert bridge._started is True
1510    assert len([call for call in warning.call_args_list if "PTP clock" in call.args[0]]) == 1
1511
1512
1513async def test_anchor_never_precedes_content_already_scheduled() -> None:
1514    """
1515    A buffered source keeps its intro: the anchor lands on the first chunk we hold.
1516
1517    Sendspin can schedule the first sample much further out than the device
1518    needs. Anchoring on the floor instead would place byte 0 in the middle of
1519    the audio already delivered and throw away everything before it.
1520    """
1521    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1522    first_chunk_us = SENDSPIN_EPOCH_US + 6_000_000
1523    stream = _make_anchor_stream(ack=UNIX_NOW_MS + 6000)
1524    _prepare_anchor(bridge, stream, first_chunk_lead_ms=6000)
1525    bridge._held_chunks.append(_pcm_chunk(first_chunk_us))
1526
1527    assert await _anchor(bridge, stream) is True
1528
1529    assert _commanded_instant(stream) == UNIX_NOW_MS + 6000
1530    # Nothing skipped, and the held opening reached the writer intact.
1531    assert bridge._drop_until_us == first_chunk_us
1532    assert _drain_queued_bytes(bridge) == 100_000 * BRIDGE_BYTES_PER_SECOND // 1_000_000
1533
1534
1535async def test_writer_stays_blocked_until_the_start_is_acked() -> None:
1536    """
1537    The writer is released only once the content is mapped onto the acked instant.
1538
1539    Feeding the CLI before the ack would place bytes against an anchor the
1540    binary has not confirmed and may still correct forward.
1541    """
1542    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1543    stream = _make_anchor_stream()
1544    _prepare_anchor(bridge, stream, first_chunk_lead_ms=250)
1545    ack_gate = asyncio.Event()
1546    start_called = asyncio.Event()
1547
1548    async def start(start_unix_ms: int, **_kwargs: object) -> int:
1549        start_called.set()
1550        await ack_gate.wait()
1551        return start_unix_ms
1552
1553    stream.start = AsyncMock(side_effect=start)
1554    anchor_task = asyncio.create_task(_anchor(bridge, stream))
1555    bridge._airplay_stream_start_task = cast("asyncio.Task[None]", anchor_task)
1556    await start_called.wait()
1557
1558    assert not bridge._airplay_stream_ready.is_set()
1559    ack_gate.set()
1560    assert await anchor_task is True
1561    assert bridge._airplay_stream_ready.is_set()
1562
1563
1564async def test_chunks_arriving_before_the_ack_are_held_and_replayed_in_order() -> None:
1565    """Audio delivered while the anchor is outstanding is queued once, in order."""
1566    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1567    stream = _make_anchor_stream()
1568    _prepare_anchor(bridge, stream, first_chunk_lead_ms=2500)
1569    first_chunk_us = SENDSPIN_EPOCH_US + 2_500_000
1570    ack_gate = asyncio.Event()
1571    start_called = asyncio.Event()
1572
1573    async def start(start_unix_ms: int, **_kwargs: object) -> int:
1574        start_called.set()
1575        await ack_gate.wait()
1576        return start_unix_ms
1577
1578    stream.start = AsyncMock(side_effect=start)
1579    anchor_task = asyncio.create_task(_anchor(bridge, stream))
1580    bridge._airplay_stream_start_task = cast("asyncio.Task[None]", anchor_task)
1581    await start_called.wait()
1582
1583    for index in range(4):
1584        bridge._on_audio_chunk(_pcm_chunk(first_chunk_us + index * 100_000))
1585    assert len(bridge._held_chunks) == 4
1586    assert bridge._write_queue.empty()
1587
1588    ack_gate.set()
1589    assert await anchor_task is True
1590
1591    # Four contiguous 100 ms chunks, replayed without padding or trimming.
1592    assert bridge._queued_frames == _expected_frames(bridge, first_chunk_us + 400_000)
1593    assert _drain_queued_bytes(bridge) == 4 * 100_000 * BRIDGE_BYTES_PER_SECOND // 1_000_000
1594
1595
1596def test_held_backlog_is_capped_and_drops_the_oldest() -> None:
1597    """An anchor that never settles cannot grow the hold without bound."""
1598    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1599    chunk_us = 100_000
1600    over_cap = MAX_HELD_AUDIO_US // chunk_us + 50
1601
1602    for index in range(over_cap):
1603        bridge._hold_chunk(_pcm_chunk(SENDSPIN_EPOCH_US + index * chunk_us))
1604
1605    assert sum(chunk.duration_us for chunk in bridge._held_chunks) <= MAX_HELD_AUDIO_US
1606    # The running total the cap is measured against tracks the deque exactly.
1607    assert bridge._held_us == sum(chunk.duration_us for chunk in bridge._held_chunks)
1608    # The oldest went, the newest stayed.
1609    assert bridge._held_chunks[0].timestamp_us > SENDSPIN_EPOCH_US
1610    assert bridge._held_chunks[-1].timestamp_us == SENDSPIN_EPOCH_US + (over_cap - 1) * chunk_us
1611
1612
1613def test_held_backlog_is_capped_by_chunk_count() -> None:
1614    """
1615    A run of zero-duration chunks is bounded by the count cap.
1616
1617    Such chunks carry no duration at all, so the µs cap can never trip on them
1618    however many arrive.
1619    """
1620    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1621
1622    for index in range(MAX_HELD_CHUNKS + 50):
1623        bridge._hold_chunk(_pcm_chunk(SENDSPIN_EPOCH_US + index, duration_us=0))
1624
1625    assert len(bridge._held_chunks) == MAX_HELD_CHUNKS
1626    assert bridge._held_us == 0
1627    assert bridge._held_chunks[-1].timestamp_us == SENDSPIN_EPOCH_US + MAX_HELD_CHUNKS + 49
1628
1629
1630async def test_corrected_ack_rebases_the_content_onto_the_acked_instant() -> None:
1631    """An instant the binary moved forward re-bases the anchor, cursor and pacing base."""
1632    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1633    correction_ms = 3000
1634    acked = UNIX_NOW_MS + AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS + correction_ms
1635    stream = _make_anchor_stream(ack=acked)
1636    _prepare_anchor(bridge, stream, first_chunk_lead_ms=250)
1637    bridge._queued_frames = 12_345
1638
1639    assert await _anchor(bridge, stream) is True
1640
1641    assert _commanded_instant(stream) == UNIX_NOW_MS + AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS
1642    assert bridge._drop_until_us == SENDSPIN_EPOCH_US + (acked - UNIX_NOW_MS) * 1_000
1643    assert bridge._queued_frames == 0
1644    assert bridge._start_unix_ms == acked
1645
1646
1647async def test_content_before_the_acked_instant_is_dropped_and_trimmed() -> None:
1648    """Held audio the acked anchor moved past is dropped, the straddling chunk trimmed."""
1649    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1650    acked = UNIX_NOW_MS + 2600
1651    stream = _make_anchor_stream(ack=acked)
1652    _prepare_anchor(bridge, stream, first_chunk_lead_ms=250)
1653    anchor_us = SENDSPIN_EPOCH_US + 2_600_000
1654    for offset in (-200_000, -50_000, 50_000):
1655        bridge._held_chunks.append(_pcm_chunk(anchor_us + offset))
1656
1657    assert await _anchor(bridge, stream) is True
1658
1659    # Fully-behind chunk gone, the straddling one keeps its 50 ms tail, and the
1660    # last chunk continues contiguously: 150 ms of audio in total.
1661    assert bridge._queued_frames == _expected_frames(bridge, anchor_us + 150_000)
1662    assert _drain_queued_bytes(bridge) == bridge._queued_frames * BRIDGE_BYTES_PER_FRAME
1663
1664
1665async def test_sync_adjust_shifts_the_command_but_not_the_content_mapping() -> None:
1666    """The device's own offset rides on the command; the group timeline stays untouched."""
1667    adjust_ms = 300
1668    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US, sync_adjust=adjust_ms)
1669    anchor_ms = UNIX_NOW_MS + AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS
1670    stream = _make_anchor_stream(ack=anchor_ms + adjust_ms)
1671    _prepare_anchor(bridge, stream, first_chunk_lead_ms=250)
1672
1673    assert await _anchor(bridge, stream) is True
1674
1675    assert _commanded_instant(stream) == anchor_ms + adjust_ms
1676    # Pacing tracks the real wall-clock instant of byte 0 (adjust included)...
1677    assert bridge._start_unix_ms == anchor_ms + adjust_ms
1678    # ...while the content is placed on the group timeline, without it.
1679    assert bridge._drop_until_us == SENDSPIN_EPOCH_US + AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS * 1_000
1680
1681
1682async def test_ack_earlier_than_commanded_is_trusted_verbatim() -> None:
1683    """
1684    An acked instant before the commanded one is used as-is, never clamped up.
1685
1686    Clamping to the commanded instant would map the content onto a moment the
1687    binary is not rendering at and put the device ahead of the rest of the group.
1688    """
1689    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1690    acked = UNIX_NOW_MS + AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS - 400
1691    stream = _make_anchor_stream(ack=acked)
1692    _prepare_anchor(bridge, stream, first_chunk_lead_ms=250)
1693
1694    assert await _anchor(bridge, stream) is True
1695
1696    assert _commanded_instant(stream) > acked
1697    assert bridge._start_unix_ms == acked
1698    assert bridge._drop_until_us == SENDSPIN_EPOCH_US + (acked - UNIX_NOW_MS) * 1_000
1699
1700
1701def test_first_chunk_after_an_anchor_is_not_reported_as_drift() -> None:
1702    """The gap between a fresh anchor and its first chunk is placement, not drift."""
1703    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1704    bridge._drop_until_us = SENDSPIN_EPOCH_US
1705    bridge._queued_frames = 0
1706
1707    bridge._align_chunk(_pcm_chunk(SENDSPIN_EPOCH_US + 900_000))
1708
1709    cast("MagicMock", bridge.logger).warning.assert_not_called()
1710
1711    # A cursor that already advanced can drift, and that is still reported.
1712    bridge._align_chunk(_pcm_chunk(SENDSPIN_EPOCH_US + 2_000_000))
1713
1714    cast("MagicMock", bridge.logger).warning.assert_called_once()
1715
1716
1717def test_a_long_timeline_gap_is_padded_from_one_shared_block() -> None:
1718    """
1719    A long hole is queued as repeats of one silence block, not as a single buffer.
1720
1721    Sendspin rebases the shared timeline forward when audio production stalls,
1722    which can open a hole of tens of seconds. Building that as one bytes object
1723    puts megabytes on the event loop in a single synchronous allocation.
1724    """
1725    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1726    first_ts = SENDSPIN_EPOCH_US + 250_000
1727    _start_stream_at(bridge, first_ts)
1728    _drain_queued_bytes(bridge)
1729
1730    gap_us = 60_000_000
1731    next_ts = first_ts + 100_000 + gap_us
1732    bridge._on_audio_chunk(_pcm_chunk(next_ts))
1733
1734    queued: list[bytes] = []
1735    while not bridge._write_queue.empty():
1736        block = bridge._write_queue.get_nowait()
1737        assert block is not None
1738        queued.append(block)
1739
1740    pad, data = queued[:-1], queued[-1]
1741    gap_frames = round(gap_us * BRIDGE_SAMPLE_RATE / 1_000_000)
1742    assert len(pad) == gap_frames // PAD_BLOCK_FRAMES
1743    # Identity, not equality: the whole hole costs one allocation.
1744    assert all(block is SILENCE_BLOCK for block in pad)
1745    # The hole is still filled exactly, so the device stays on the group's clock.
1746    assert sum(len(block) for block in pad) == gap_frames * BRIDGE_BYTES_PER_FRAME
1747    assert len(data) == 100_000 * BRIDGE_BYTES_PER_SECOND // 1_000_000
1748
1749
1750# --- Playout shift: the binary's own mid-stream re-anchors move the mapping ---
1751
1752
1753def _shifted_bridge(shift_seconds: float) -> tuple[SendspinAirPlayBridge, MagicMock, int]:
1754    """
1755    Start a streaming bridge whose CLI reports a mid-stream playout shift.
1756
1757    :param shift_seconds: Cumulative shift the binary reports since its START.
1758    :return: The bridge, its stream mock and the first chunk's timestamp.
1759    """
1760    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1761    first_ts = SENDSPIN_EPOCH_US + 250_000
1762    _start_stream_at(bridge, first_ts)
1763    bridge._start_unix_ms = UNIX_NOW_MS
1764    _drain_queued_bytes(bridge)
1765    stream = MagicMock()
1766    # A real float: the fold subtracts it from the applied baseline, which a bare
1767    # MagicMock cannot answer.
1768    stream.cumulative_shift_seconds = shift_seconds
1769    bridge._airplay_stream = stream
1770    return bridge, stream, first_ts
1771
1772
1773def test_reported_reanchor_moves_the_anchor_and_the_pacing_start() -> None:
1774    """
1775    A starvation re-anchor makes every byte audible later, so the anchor follows it.
1776
1777    cliairplay shifts its playout forward when it runs out of PCM on stdin.
1778    Leaving the bridge's mapping where it was would place all following content
1779    ahead of where the device actually plays it, for the rest of the stream.
1780    """
1781    shift_s = 1.5
1782    bridge, _, first_ts = _shifted_bridge(shift_s)
1783    anchor_before = bridge._drop_until_us
1784
1785    bridge._on_audio_chunk(_pcm_chunk(first_ts + 100_000))
1786
1787    assert bridge._drop_until_us == anchor_before + round(shift_s * 1_000_000)
1788    # The write pacing measures from the same instant, so it moves with it.
1789    assert bridge._start_unix_ms == UNIX_NOW_MS + round(shift_s * 1000)
1790
1791
1792def test_reported_reanchor_skips_content_until_the_device_catches_up() -> None:
1793    """The device plays the shift late, so exactly that much content is dropped."""
1794    shift_s = 0.5
1795    bridge, _, first_ts = _shifted_bridge(shift_s)
1796
1797    fed_us = 1_000_000
1798    for index in range(fed_us // 100_000):
1799        bridge._on_audio_chunk(_pcm_chunk(first_ts + 100_000 * (index + 1)))
1800
1801    assert bridge._queued_frames == _expected_frames(bridge, first_ts + 100_000 + fed_us)
1802    assert _drain_queued_bytes(bridge) == round(
1803        (fed_us / 1_000_000 - shift_s) * BRIDGE_BYTES_PER_SECOND
1804    )
1805
1806
1807def test_absorbing_a_reanchor_is_not_reported_as_timeline_drift() -> None:
1808    """The trim that works a folded shift off is a correction, not Sendspin drift."""
1809    bridge, _, first_ts = _shifted_bridge(0.5)
1810    logger = cast("MagicMock", bridge.logger)
1811
1812    # Straddles the shifted anchor, so it is trimmed rather than dropped outright.
1813    bridge._on_audio_chunk(_pcm_chunk(first_ts + 550_000))
1814
1815    assert bridge._queued_frames == _expected_frames(bridge, first_ts + 650_000)
1816    # Only the re-anchor itself is reported; the trim it caused stays quiet.
1817    logger.warning.assert_called_once()
1818    logger.warning.reset_mock()
1819
1820    # Back on the timeline: the shift is worked off and nothing is realigned.
1821    bridge._on_audio_chunk(_pcm_chunk(first_ts + 650_000))
1822    assert not bridge._absorbing_shift
1823    logger.warning.assert_not_called()
1824
1825    # A real discontinuity is reported again.
1826    bridge._on_audio_chunk(_pcm_chunk(first_ts + 1_500_000))
1827    logger.warning.assert_called_once()
1828
1829
1830def test_a_cursor_off_the_frame_grid_still_absorbs_quietly() -> None:
1831    """
1832    The trim stays quiet however the cursor happens to sit when the shift lands.
1833
1834    ``_align_chunk`` re-targets each chunk against the anchor independently, so
1835    the cursor routinely rests a frame either side of the timeline. The trim a
1836    fold asks for is a correction at any of those offsets, never Sendspin drift.
1837    """
1838    bridge, stream, first_ts = _shifted_bridge(0.0)
1839    logger = cast("MagicMock", bridge.logger)
1840
1841    # Play on a while first, so the cursor sits well past the anchor the way it
1842    # does mid-stream when a starvation hits.
1843    for index in range(1, 11):
1844        bridge._on_audio_chunk(_pcm_chunk(first_ts + 100_000 * index))
1845    logger.warning.assert_not_called()
1846
1847    # A frame past the timeline: the trim the fold asks for then runs one frame
1848    # deeper than the shift itself.
1849    bridge._queued_frames += 1
1850    stream.cumulative_shift_seconds = 0.5
1851    bridge._on_audio_chunk(_pcm_chunk(first_ts + 1_100_000))
1852
1853    # Only the re-anchor itself is reported.
1854    logger.warning.assert_called_once()
1855
1856
1857def test_an_absorption_spanning_chunks_reports_only_the_reanchor() -> None:
1858    """
1859    A shift takes several chunks to trim off, and stays one report throughout.
1860
1861    The binary keeps reporting the same running total while the trim works, so
1862    every chunk until the cursor is back on the timeline realigns by design.
1863    """
1864    bridge, stream, first_ts = _shifted_bridge(0.0)
1865    logger = cast("MagicMock", bridge.logger)
1866    for index in range(1, 11):
1867        bridge._on_audio_chunk(_pcm_chunk(first_ts + 100_000 * index))
1868    logger.warning.assert_not_called()
1869
1870    stream.cumulative_shift_seconds = 0.5
1871    # Five chunks of content are trimmed away before the cursor catches up.
1872    for index in range(11, 17):
1873        bridge._on_audio_chunk(_pcm_chunk(first_ts + 100_000 * index))
1874
1875    logger.warning.assert_called_once()
1876    assert not bridge._absorbing_shift
1877
1878
1879def test_an_unchanged_reanchor_total_is_folded_only_once() -> None:
1880    """The binary reports a running total, so only what is new moves the anchor."""
1881    bridge, _, first_ts = _shifted_bridge(0.5)
1882    bridge._on_audio_chunk(_pcm_chunk(first_ts + 600_000))
1883    anchor_after_fold = bridge._drop_until_us
1884    assert anchor_after_fold == first_ts + 500_000
1885
1886    bridge._on_audio_chunk(_pcm_chunk(first_ts + 700_000))
1887
1888    assert bridge._drop_until_us == anchor_after_fold
1889
1890
1891def test_a_reset_reanchor_total_rebaselines_without_moving_the_anchor() -> None:
1892    """
1893    A total that went backwards means a START already replaced the mapping.
1894
1895    The binary zeroes its running total on every START, so the drop is not the
1896    device un-shifting: the bridge takes the new baseline and leaves the anchor
1897    to the START that set it.
1898    """
1899    bridge, stream, first_ts = _shifted_bridge(0.5)
1900    bridge._on_audio_chunk(_pcm_chunk(first_ts + 600_000))
1901    anchor_after_fold = bridge._drop_until_us
1902    assert anchor_after_fold == first_ts + 500_000
1903
1904    stream.cumulative_shift_seconds = 0.0
1905    bridge._on_audio_chunk(_pcm_chunk(first_ts + 700_000))
1906
1907    assert bridge._drop_until_us == anchor_after_fold
1908    assert bridge._applied_shift_seconds == 0.0
1909
1910
1911def test_a_reset_reanchor_total_ends_the_absorption() -> None:
1912    """
1913    A total that went backwards leaves no correction outstanding to stay quiet for.
1914
1915    The START that zeroed the total replaced the mapping the trim was working
1916    against, so a realignment after it is Sendspin timeline drift again and
1917    worth reporting.
1918    """
1919    bridge, stream, first_ts = _shifted_bridge(0.5)
1920    logger = cast("MagicMock", bridge.logger)
1921    # Mid-absorption: the trim has not caught the cursor up to the timeline yet.
1922    bridge._on_audio_chunk(_pcm_chunk(first_ts + 550_000))
1923    logger.warning.reset_mock()
1924
1925    stream.cumulative_shift_seconds = 0.0
1926    bridge._on_audio_chunk(_pcm_chunk(first_ts + 600_000))
1927
1928    assert not bridge._absorbing_shift
1929    logger.warning.assert_called_once()
1930
1931
1932async def test_anchoring_clears_the_folded_shift_baseline() -> None:
1933    """A START re-anchors the binary from scratch, so the fold starts over with it."""
1934    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1935    stream = _make_anchor_stream(ack=UNIX_NOW_MS + COLD_LEAD_MS)
1936    _prepare_anchor(bridge, stream, COLD_LEAD_MS)
1937    bridge._applied_shift_seconds = 1.5
1938    bridge._absorbing_shift = True
1939
1940    assert await _anchor(bridge, stream)
1941
1942    assert bridge._applied_shift_seconds == 0.0
1943    assert not bridge._absorbing_shift
1944
1945
1946@pytest.mark.parametrize(
1947    ("warm_lead_ms", "flushed_head_offset_ms", "adjust_ms", "expected_anchor_offset_ms"),
1948    [
1949        (4000, 0, 0, 4000 + AIRPLAY_SPLICE_LEAD_MARGIN_MS),
1950        (4000, 0, -600, 4600 + AIRPLAY_SPLICE_LEAD_MARGIN_MS),
1951        (0, 5000, 0, 5000 + AIRPLAY_SPLICE_LEAD_MARGIN_MS),
1952    ],
1953)
1954async def test_warm_anchor_clears_the_receivers_queued_audio(
1955    warm_lead_ms: int,
1956    flushed_head_offset_ms: int,
1957    adjust_ms: int,
1958    expected_anchor_offset_ms: int,
1959) -> None:
1960    """
1961    A warm re-anchor lands beyond the audio the receiver still has queued.
1962
1963    A negative sync_adjust moves the commanded instant earlier and eats into that
1964    lead, so it is added back to the requirement.
1965    """
1966    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US, sync_adjust=adjust_ms)
1967    stream = _make_anchor_stream(
1968        warm_lead_ms=warm_lead_ms,
1969        flushed_head_unix_ms=UNIX_NOW_MS + flushed_head_offset_ms if flushed_head_offset_ms else 0,
1970    )
1971    _prepare_anchor(bridge, stream, first_chunk_lead_ms=250)
1972
1973    assert await _anchor(bridge, stream, warm=True) is True
1974
1975    assert _commanded_instant(stream) == UNIX_NOW_MS + expected_anchor_offset_ms + adjust_ms
1976
1977
1978async def test_superseded_during_the_ack_mutates_nothing() -> None:
1979    """A newer stream start taking over while the ack is outstanding wins untouched."""
1980    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
1981    stream = _make_anchor_stream()
1982    _prepare_anchor(bridge, stream, first_chunk_lead_ms=250)
1983    original_drop_until = bridge._drop_until_us
1984    bridge._held_chunks.append(_pcm_chunk(original_drop_until))
1985
1986    async def start(start_unix_ms: int, **_kwargs: object) -> int:
1987        # A newer stream start claimed the bridge while the binary held its ack.
1988        bridge._airplay_stream_start_task = MagicMock()
1989        return start_unix_ms
1990
1991    stream.start = AsyncMock(side_effect=start)
1992
1993    assert await _anchor(bridge, stream) is False
1994
1995    assert bridge._drop_until_us == original_drop_until
1996    assert bridge._start_unix_ms == 0
1997    assert bridge._started is False
1998    assert bridge._anchor_settled is False
1999    assert len(bridge._held_chunks) == 1
2000    assert not bridge._airplay_stream_ready.is_set()
2001
2002
2003def test_unix_to_sendspin_instant_round_trips() -> None:
2004    """The two clock-domain helpers are exact inverses of each other."""
2005    clock = ManualClock(now_us_value=SENDSPIN_EPOCH_US)
2006    for lead_ms in (-300, 0, WARM_LEAD_MS, COLD_LEAD_MS):
2007        audible_us = _audible_instant_us(clock, lead_ms)
2008        unix_ms = sendspin_audible_instant_to_unix_ms(audible_us, clock.now_us(), UNIX_NOW_S)
2009
2010        assert unix_ms == UNIX_NOW_MS + lead_ms
2011        assert (
2012            unix_ms_to_sendspin_audible_instant(unix_ms, clock.now_us(), UNIX_NOW_S) == audible_us
2013        )
2014
2015
2016# --- Warm handover: a kept stream survives a new stream start and rides flush-refill ---
2017
2018
2019def _make_kept_stream(
2020    *, running: bool = True, connected: bool = True, ended_cleanly: bool = False
2021) -> MagicMock:
2022    """
2023    Build a mock AirPlayStream reporting the given running/connected state.
2024
2025    :param running: Whether the cli process behind the stream is still alive.
2026    :param connected: Whether the device connection has been established.
2027    :param ended_cleanly: Whether the binary reported the end of the stream
2028        itself. A real bool: a bare MagicMock reads as a clean end, which the
2029        loss check treats as no loss at all.
2030    """
2031    stream = MagicMock()
2032    stream.running = running
2033    stream.connected = connected
2034    stream.ended_cleanly = ended_cleanly
2035    # A real float: the shift fold subtracts it from the applied baseline, which
2036    # a bare MagicMock cannot answer.
2037    stream.cumulative_shift_seconds = 0.0
2038    return stream
2039
2040
2041def test_on_bridge_stream_start_keeps_warm_eligible_stream() -> None:
2042    """A running, connected AirPlay 2 stream survives a new Sendspin stream start."""
2043    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2044    kept_stream = _make_kept_stream()
2045    bridge._airplay_stream = kept_stream
2046    bridge.airplay_player.stream = kept_stream
2047    bridge._started = True
2048
2049    bridge._on_bridge_stream_start()
2050
2051    assert bridge._airplay_stream is kept_stream
2052    assert bridge.airplay_player.stream is kept_stream
2053    assert bridge._stream_is_warm_eligible()
2054
2055
2056def test_on_bridge_stream_start_keeps_raop_stream() -> None:
2057    """A started legacy RAOP stream is eligible for warm Sendspin flush-refill."""
2058    bridge = _make_bridge(
2059        clock_now_us=SENDSPIN_EPOCH_US,
2060        protocol=StreamingProtocol.RAOP,
2061    )
2062    old_stream = _make_kept_stream()
2063    bridge._airplay_stream = old_stream
2064    bridge.airplay_player.stream = old_stream
2065    bridge._started = True
2066
2067    bridge._on_bridge_stream_start()
2068
2069    assert bridge._airplay_stream is old_stream
2070    assert bridge.airplay_player.stream is old_stream
2071
2072
2073def test_sendspin_callbacks_keep_raop_stream_until_warm_handover() -> None:
2074    """Both Sendspin start callbacks preserve a reusable legacy RAOP session."""
2075    bridge = _make_bridge(
2076        clock_now_us=SENDSPIN_EPOCH_US,
2077        protocol=StreamingProtocol.RAOP,
2078    )
2079    kept_stream = _make_kept_stream()
2080    bridge._airplay_stream = kept_stream
2081    bridge.airplay_player.stream = kept_stream
2082    bridge._started = True
2083
2084    bridge._on_stream_start(MagicMock())
2085    bridge._on_bridge_stream_start()
2086
2087    assert bridge._airplay_stream is kept_stream
2088    assert bridge.airplay_player.stream is kept_stream
2089
2090
2091def test_on_bridge_stream_start_replaces_uncommitted_stream() -> None:
2092    """A connected stream cannot be retained before its first START."""
2093    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2094    old_stream = _make_kept_stream()
2095    bridge._airplay_stream = old_stream
2096    bridge.airplay_player.stream = old_stream
2097
2098    bridge._on_bridge_stream_start()
2099
2100    assert bridge._airplay_stream is None
2101    # the player keeps the reference until the scheduled teardown has the process
2102    # off the receiver, so nothing reads the speaker as free while it is not
2103    assert bridge.airplay_player.stream is old_stream  # type: ignore[unreachable]
2104
2105
2106def test_on_stream_start_keeps_warm_eligible_stream() -> None:
2107    """The Sendspin-server-side stream-start callback also keeps a warm-eligible stream."""
2108    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2109    kept_stream = _make_kept_stream()
2110    bridge._airplay_stream = kept_stream
2111    bridge.airplay_player.stream = kept_stream
2112    bridge._started = True
2113
2114    bridge._on_stream_start(MagicMock())
2115
2116    assert bridge._airplay_stream is kept_stream
2117    assert bridge.airplay_player.stream is kept_stream
2118
2119
2120async def test_warm_stream_flushes_and_reanchors_on_kept_instance() -> None:
2121    """A warm handover flushes and re-anchors START on the same stream instance."""
2122    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2123    bridge._drop_until_us = SENDSPIN_EPOCH_US + WARM_LEAD_MS * 1_000
2124    commanded = UNIX_NOW_MS + WARM_LEAD_MS
2125    kept_stream = _make_anchor_stream(ack=commanded)
2126    bridge._airplay_stream = kept_stream
2127    bridge._airplay_stream_start_task = asyncio.current_task()
2128
2129    with patch(
2130        "music_assistant.providers.airplay.sendspin_bridge.time.time",
2131        return_value=UNIX_NOW_S,
2132    ):
2133        committed = await bridge._start_warm_stream(kept_stream)
2134
2135    assert committed is True
2136    assert bridge._airplay_stream is kept_stream  # no new instance was built
2137    kept_stream.flush.assert_awaited_once_with()
2138    kept_stream.start.assert_awaited_once_with(commanded, join=True)
2139    assert bridge._started is True
2140    assert bridge._airplay_stream_ready.is_set()
2141
2142
2143async def test_warm_stream_flush_timeout_falls_back_to_cold() -> None:
2144    """A flush that is never acknowledged never re-anchors and falls back to cold."""
2145    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2146    kept_stream = _make_anchor_stream()
2147    kept_stream.flush = AsyncMock(return_value=False)
2148    bridge._airplay_stream = kept_stream
2149    bridge._airplay_stream_start_task = asyncio.current_task()
2150
2151    committed = await bridge._start_warm_stream(kept_stream)
2152
2153    assert committed is False
2154    kept_stream.start.assert_not_awaited()
2155    assert bridge._started is False
2156
2157
2158async def test_warm_stream_superseded_before_start_does_not_anchor() -> None:
2159    """If a newer stream start already owns the bridge, the stale flush never re-anchors."""
2160    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2161    kept_stream = _make_anchor_stream()
2162    bridge._airplay_stream = kept_stream
2163    # Simulate a newer stream start having already replaced the tracked task.
2164    bridge._airplay_stream_start_task = MagicMock()
2165
2166    committed = await bridge._start_warm_stream(kept_stream)
2167
2168    assert committed is False
2169    kept_stream.start.assert_not_awaited()
2170
2171
2172async def test_warm_stream_cancellation_propagates() -> None:
2173    """Cancellation while flushing propagates without re-anchoring."""
2174    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2175    kept_stream = _make_anchor_stream()
2176    flush_waiting = asyncio.Event()
2177
2178    async def flush(*_args: object, **_kwargs: object) -> bool:
2179        bridge._airplay_stream_start_task = asyncio.current_task()
2180        flush_waiting.set()
2181        await asyncio.Event().wait()
2182        return True
2183
2184    kept_stream.flush = AsyncMock(side_effect=flush)
2185    bridge._airplay_stream = kept_stream
2186
2187    warm_task = asyncio.create_task(bridge._start_warm_stream(kept_stream))
2188    await flush_waiting.wait()
2189    warm_task.cancel()
2190    with pytest.raises(asyncio.CancelledError):
2191        await warm_task
2192
2193    kept_stream.start.assert_not_awaited()
2194
2195
2196async def test_warm_handover_superseded_during_the_anchor_keeps_the_stream() -> None:
2197    """
2198    A superseded warm handover leaves the kept stream to the newer start.
2199
2200    The anchor reports the same False for "superseded" as for a genuine failure,
2201    so without an ownership re-check the stale task would stop the very transport
2202    the newer start decided to keep and put a second cliairplay on the receiver.
2203    """
2204    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2205    bridge._drop_until_us = SENDSPIN_EPOCH_US + WARM_LEAD_MS * 1_000
2206    bridge._airplay_stream_start_task = asyncio.current_task()
2207    kept_stream = _make_anchor_stream()
2208    bridge._airplay_stream = kept_stream
2209    bridge.airplay_player.stream = kept_stream
2210    cold_stream = _make_anchor_stream()
2211
2212    async def start(start_unix_ms: int, **_kwargs: object) -> int:
2213        # A newer stream start claimed the bridge while the binary held its ack.
2214        bridge._airplay_stream_start_task = MagicMock()
2215        return start_unix_ms
2216
2217    kept_stream.start = AsyncMock(side_effect=start)
2218
2219    with (
2220        patch(
2221            "music_assistant.providers.airplay.sendspin_bridge.AirPlayStream",
2222            return_value=cold_stream,
2223        ),
2224        patch(
2225            "music_assistant.providers.airplay.sendspin_bridge.time.time",
2226            return_value=UNIX_NOW_S,
2227        ),
2228    ):
2229        await bridge._start_protocol_from_chunk()
2230
2231    kept_stream.stop.assert_not_awaited()
2232    cold_stream.connect.assert_not_awaited()
2233    assert bridge._airplay_stream is kept_stream
2234    assert bridge.airplay_player.stream is kept_stream
2235
2236
2237async def test_superseded_start_failure_leaves_the_newer_stream_alone() -> None:
2238    """
2239    A stale start that fails must not take the newer stream's state with it.
2240
2241    The receiver is busy precisely because the newer start just claimed it, so a
2242    superseded cold connect failing is the ordinary outcome. Running the recovery
2243    would drop the newer stream's held backlog, release its writer before its own
2244    anchor is settled and schedule its teardown.
2245    """
2246    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2247    bridge._drop_until_us = SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
2248    bridge._airplay_stream_start_task = asyncio.current_task()
2249    bridge._hold_chunk(_pcm_chunk(SENDSPIN_EPOCH_US))
2250    newer_stream = _make_anchor_stream()
2251    stale_stream = _make_anchor_stream()
2252
2253    async def connect(_use_shared_ptp: bool | None) -> None:
2254        # The newer start won the receiver, so this one cannot have it.
2255        bridge._airplay_stream_start_task = MagicMock()
2256        bridge._airplay_stream = newer_stream
2257        bridge.airplay_player.stream = newer_stream
2258        raise OSError("device busy")
2259
2260    stale_stream.connect = AsyncMock(side_effect=connect)
2261
2262    with patch(
2263        "music_assistant.providers.airplay.sendspin_bridge.AirPlayStream",
2264        return_value=stale_stream,
2265    ):
2266        await bridge._start_protocol_from_chunk()
2267
2268    assert bridge._is_streaming is True
2269    assert len(bridge._held_chunks) == 1
2270    assert not bridge._airplay_stream_ready.is_set()
2271    assert bridge._airplay_stream is newer_stream
2272    newer_stream.stop.assert_not_awaited()
2273    # No teardown was scheduled for the newer stream's resources.
2274    cast("MagicMock", bridge.mass).create_task.assert_not_called()
2275
2276
2277# --- Startup lead: how far ahead Sendspin schedules the first chunk ---
2278
2279
2280def _make_timed_bridge() -> tuple[SendspinAirPlayBridge, MagicMock]:
2281    """Return a bridge with a mocked bridge role attached, plus that role."""
2282    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2283    role = MagicMock()
2284    bridge._bridge_role = role
2285    return bridge, role
2286
2287
2288def test_bridge_timing_reports_the_cold_lead_without_a_warm_stream() -> None:
2289    """Without a reusable stream the lead has to cover a full process spawn and connect."""
2290    bridge, role = _make_timed_bridge()
2291
2292    bridge._refresh_bridge_timing()
2293
2294    role.set_timing.assert_called_once_with(
2295        required_lead_time_ms=BRIDGE_COLD_START_LEAD_MS, min_buffer_ms=BRIDGE_MIN_BUFFER_MS
2296    )
2297
2298
2299def test_bridge_timing_reports_the_warm_lead_for_a_reusable_stream() -> None:
2300    """A kept, connected, already-anchored stream pays no connect, so it needs less lead."""
2301    bridge, role = _make_timed_bridge()
2302    bridge._airplay_stream = _make_kept_stream()
2303    bridge._started = True
2304
2305    bridge._refresh_bridge_timing()
2306
2307    role.set_timing.assert_called_once_with(
2308        required_lead_time_ms=BRIDGE_WARM_START_LEAD_MS, min_buffer_ms=BRIDGE_MIN_BUFFER_MS
2309    )
2310
2311
2312def test_bridge_timing_is_a_noop_without_a_bridge_role() -> None:
2313    """Timing can be refreshed before registration completes, with nothing to push it to."""
2314    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2315    assert bridge._bridge_role is None
2316
2317    bridge._refresh_bridge_timing()
2318
2319
2320def test_stream_start_reads_the_lead_before_rewinding_the_stream_state() -> None:
2321    """
2322    The warm/cold decision is taken while the previous stream's state is intact.
2323
2324    ``_on_stream_start`` rewinds the per-stream state; reading the timing after
2325    that rewind would report the cold lead for every start, including the warm
2326    handovers that need none of that budget.
2327    """
2328    bridge, role = _make_timed_bridge()
2329    kept_stream = _make_kept_stream()
2330    bridge._airplay_stream = kept_stream
2331    bridge.airplay_player.stream = kept_stream
2332    bridge._started = True
2333    observed: list[tuple[bool, object]] = []
2334    refresh = bridge._refresh_bridge_timing
2335
2336    def record() -> None:
2337        observed.append((bridge._started, bridge._airplay_stream))
2338        refresh()
2339
2340    with patch.object(bridge, "_refresh_bridge_timing", record):
2341        bridge._on_stream_start(MagicMock())
2342
2343    assert observed == [(True, kept_stream)]
2344    role.set_timing.assert_called_once_with(
2345        required_lead_time_ms=BRIDGE_WARM_START_LEAD_MS, min_buffer_ms=BRIDGE_MIN_BUFFER_MS
2346    )
2347
2348
2349# --- Mid-stream transport loss: re-anchoring, and giving up when it keeps dropping ---
2350
2351
2352def _make_completed_start_task(*, failed: bool = False) -> MagicMock:
2353    """
2354    Build a start-task mock the chunk handler reads as a finished protocol start.
2355
2356    Every predicate must answer a real bool: a bare MagicMock reports itself as
2357    cancelled, which the handler reads as a failed start.
2358    """
2359    task = MagicMock()
2360    task.done.return_value = True
2361    task.cancelled.return_value = failed
2362    task.exception.return_value = None
2363    return task
2364
2365
2366def _make_anchored_bridge(
2367    *, running: bool, ended_cleanly: bool = False
2368) -> tuple[SendspinAirPlayBridge, MagicMock]:
2369    """
2370    Return a bridge anchored on a transport in the given running state, plus that transport.
2371
2372    :param running: Whether the cli process behind the transport is still alive.
2373    :param ended_cleanly: Whether the binary reported the end of the stream
2374        itself, which stops the transport without losing it.
2375    """
2376    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2377    stream = _make_kept_stream(running=running, ended_cleanly=ended_cleanly)
2378    bridge._airplay_stream = stream
2379    bridge.airplay_player.stream = stream
2380    bridge._airplay_stream_start_task = _make_completed_start_task()
2381    bridge._started = True
2382    bridge._anchor_settled = True
2383    bridge._drop_until_us = SENDSPIN_EPOCH_US
2384    return bridge, stream
2385
2386
2387def test_lost_transport_rearms_a_cold_start_on_the_current_chunk() -> None:
2388    """
2389    A transport that died mid-stream is released and re-anchored on the live timeline.
2390
2391    The CLI accepts and discards writes once its process is gone, so the loss is
2392    only visible on the stream itself. The chunk that exposes it is also the one
2393    the fresh transport anchors to, which is where the group is playing now.
2394    """
2395    bridge, dead_stream = _make_anchored_bridge(running=False)
2396    chunk_ts = SENDSPIN_EPOCH_US + 30_000_000
2397
2398    bridge._on_audio_chunk(_pcm_chunk(chunk_ts))
2399
2400    assert bridge._airplay_stream is None
2401    # released from the bridge, but still published until the teardown that owns
2402    # it has the dead process cleared
2403    assert bridge.airplay_player.stream is dead_stream
2404    assert dead_stream.superseded is True
2405    assert bridge._started is False
2406    assert bridge._anchor_settled is False
2407    # a fresh start is armed and anchored where the group is playing right now
2408    assert bridge._drop_until_us == chunk_ts
2409    # the chunk is held until the new anchor is acked, not placed against the dead one
2410    assert len(bridge._held_chunks) == 1
2411
2412
2413def test_lost_transport_and_its_writer_are_torn_down() -> None:
2414    """The dead transport and the writer feeding it are handed to the cleanup path."""
2415    bridge, dead_stream = _make_anchored_bridge(running=False)
2416    writer_task = MagicMock()
2417    bridge._writer_task = writer_task
2418    start_task = bridge._airplay_stream_start_task
2419
2420    with patch.object(bridge, "_cleanup_old_stream", MagicMock()) as cleanup:
2421        bridge._on_audio_chunk(_pcm_chunk(SENDSPIN_EPOCH_US + 1_000_000))
2422
2423    assert cleanup.call_args.args[:3] == (dead_stream, writer_task, start_task)
2424
2425
2426def test_live_transport_keeps_streaming_untouched() -> None:
2427    """A running transport is left alone: chunks keep flowing to the same stream."""
2428    bridge, stream = _make_anchored_bridge(running=True)
2429    start_task = bridge._airplay_stream_start_task
2430
2431    bridge._on_audio_chunk(_pcm_chunk(SENDSPIN_EPOCH_US + 1_000_000))
2432
2433    assert bridge._airplay_stream is stream
2434    assert bridge._airplay_stream_start_task is start_task
2435    assert bridge._started is True
2436    assert not bridge._write_queue.empty()
2437
2438
2439def test_transport_is_not_judged_while_a_start_is_in_flight() -> None:
2440    """
2441    A start owns its transport, so a stream it is tearing down is not a loss.
2442
2443    A warm handover that fails stops the kept stream before dropping it, leaving
2444    a window where the bridge still points at a stopped stream. Restarting from
2445    that window would fight the start already falling back to a cold reconnect.
2446    """
2447    bridge, stopped_stream = _make_anchored_bridge(running=False)
2448    cast("MagicMock", bridge._airplay_stream_start_task).done.return_value = False
2449
2450    bridge._on_audio_chunk(_pcm_chunk(SENDSPIN_EPOCH_US + 1_000_000))
2451
2452    assert bridge._airplay_stream is stopped_stream
2453    assert bridge._started is True
2454
2455
2456def test_unanchored_transport_is_not_treated_as_a_loss() -> None:
2457    """
2458    A stream that never anchored is the start's to report, not a mid-stream loss.
2459
2460    Recovery re-joins the group where the current chunk sits, which only means
2461    anything once an anchor existed. A start that finished without one has
2462    already taken the bridge out of streaming through its own failure path.
2463    """
2464    bridge, stopped_stream = _make_anchored_bridge(running=False)
2465    start_task = bridge._airplay_stream_start_task
2466    bridge._started = False
2467
2468    bridge._on_audio_chunk(_pcm_chunk(SENDSPIN_EPOCH_US + 1_000_000))
2469
2470    assert bridge._airplay_stream is stopped_stream
2471    assert bridge._airplay_stream_start_task is start_task
2472
2473
2474def test_a_stream_the_native_path_took_over_is_not_recovered() -> None:
2475    """
2476    A transport the bridge no longer owns is not the bridge's to restart.
2477
2478    The native path stops (or replaces) the player's stream without telling the
2479    bridge, which reads its own stopped stream as a crash. Recovering would put
2480    a second cli process on the same receiver and let the cold start publish its
2481    stream over the native session's.
2482    """
2483    bridge, stopped_stream = _make_anchored_bridge(running=False)
2484    # the native path took the player over and left the bridge holding a stream
2485    # that is no longer the player's
2486    cast("MagicMock", bridge.airplay_player).stream = _make_kept_stream()
2487
2488    with patch.object(bridge, "_restart_transport", MagicMock()) as restart:
2489        bridge._on_audio_chunk(_pcm_chunk(SENDSPIN_EPOCH_US + 1_000_000))
2490
2491    restart.assert_not_called()
2492    assert bridge._airplay_stream is stopped_stream
2493
2494
2495def test_a_stream_the_binary_ended_itself_is_not_a_loss() -> None:
2496    """
2497    A cli process that reported the end of the stream did not lose its transport.
2498
2499    The stderr loop also ends on a clean [STATUS] eof or the binary's idle cap,
2500    which stops the stream exactly like a crash does. Restarting one of those
2501    spawns a process for audio that is already over, and two such restarts
2502    inside the guard window take the speaker out of the group for good. Its
2503    counterpart is test_lost_transport_rearms_a_cold_start_on_the_current_chunk,
2504    where the same stopped stream ended without saying so.
2505    """
2506    bridge, ended_stream = _make_anchored_bridge(running=False, ended_cleanly=True)
2507
2508    with patch.object(bridge, "_restart_transport", MagicMock()) as restart:
2509        bridge._on_audio_chunk(_pcm_chunk(SENDSPIN_EPOCH_US + 1_000_000))
2510
2511    restart.assert_not_called()
2512    assert bridge._airplay_stream is ended_stream
2513    assert bridge._started is True
2514
2515
2516def test_restarting_the_transport_drops_a_deferred_teardown() -> None:
2517    """
2518    A teardown deferred by an earlier stream end must not fire into the new transport.
2519
2520    The restart arms a transport that pending timer knows nothing about, so it is
2521    cancelled along with the stream it was scheduled for.
2522    """
2523    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2524
2525    bridge._restart_transport()
2526
2527    cast("MagicMock", bridge.mass).cancel_timer.assert_called_once_with(bridge._teardown_timer_id)
2528
2529
2530def test_a_grace_timer_that_already_fired_spares_the_restarted_stream() -> None:
2531    """
2532    A teardown whose timer fired before the restart cancelled it leaves the new stream alone.
2533
2534    cancel_timer cannot recall a handle that already fired, so a stream arriving
2535    at the very end of the grace window still gets the call. Reading the live
2536    fields there would cancel that stream's writer and drain its queue, leaving
2537    the speaker silent for the whole track -- and the warm restart keeps the same
2538    stream object, so telling the two apart by the stream alone cannot work.
2539    """
2540    bridge, stream = _make_anchored_bridge(running=True)
2541    bridge._writer_task = MagicMock()
2542
2543    bridge._on_bridge_stream_end()
2544    # the next stream arrives and rides the kept process, cancelling a timer that
2545    # has already fired
2546    bridge._restart_transport()
2547    new_writer_task = bridge._writer_task
2548    bridge._write_queue.put_nowait(b"\x00" * BRIDGE_BYTES_PER_FRAME)
2549
2550    bridge._deferred_cleanup()
2551
2552    assert bridge._airplay_stream is stream
2553    assert bridge._writer_task is new_writer_task
2554    assert not bridge._write_queue.empty()
2555
2556
2557async def test_the_cleanup_a_start_waits_on_cannot_cancel_it() -> None:
2558    """
2559    A start waiting for the pending teardown is not among the handles it cancels.
2560
2561    _start_protocol_from_chunk and _cli_writer both await _cleanup_task before
2562    touching the transport. A teardown reading the live fields when it finally
2563    ran would find the waiting start there and cancel it, killing the stream it
2564    was clearing the way for.
2565    """
2566    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2567    bridge._is_streaming = False
2568    stream = _make_kept_stream()
2569    stream.stop = AsyncMock()
2570    bridge._airplay_stream = stream
2571    bridge._airplay_stream_start_task = _make_completed_start_task()
2572
2573    bridge._schedule_cleanup()
2574    teardown = cast("MagicMock", bridge.mass).create_task.call_args.args[0]
2575    # the start that arrives next publishes itself and then awaits the teardown
2576    start = cast("asyncio.Task[None]", asyncio.current_task())
2577    bridge._airplay_stream_start_task = start
2578
2579    await teardown
2580
2581    # the teardown ran against what the bridge held when it was scheduled
2582    stream.stop.assert_awaited_once_with(force=True)
2583    assert start.cancelling() == 0
2584
2585
2586def test_a_new_sendspin_stream_restores_the_recovery_budget() -> None:
2587    """
2588    Every Sendspin stream starts with a full recovery budget.
2589
2590    A loss on the previous stream says nothing about the device's health on this
2591    one; carrying the stamp over would abandon a speaker on its very first loss.
2592    """
2593    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2594    bridge._last_transport_recovery = 100.0
2595
2596    bridge._on_stream_start(MagicMock())
2597
2598    assert bridge._last_transport_recovery is None
2599
2600
2601def test_a_stream_start_on_an_unavailable_player_still_restores_the_budget() -> None:
2602    """
2603    The recovery budget is settled before any early return can skip it.
2604
2605    The stream-start callback bails out when the player is unavailable, but the
2606    role-side entry point has no such gate; leaving the verdict of the previous
2607    stream in place would abandon the speaker on the next stream's first loss.
2608    """
2609    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2610    cast("MagicMock", bridge.airplay_player).available = False
2611    bridge._last_transport_recovery = 100.0
2612
2613    bridge._on_stream_start(MagicMock())
2614
2615    assert bridge._last_transport_recovery is None
2616
2617
2618def test_second_transport_loss_within_the_guard_window_gives_up() -> None:
2619    """A device dropping its transport again right away is abandoned, not re-anchored."""
2620    bridge, _ = _make_anchored_bridge(running=False)
2621
2622    with (
2623        patch(
2624            "music_assistant.providers.airplay.sendspin_bridge.time.monotonic",
2625            side_effect=[100.0, 100.0 + BRIDGE_TRANSPORT_RECOVERY_GUARD_SECONDS - 1],
2626        ),
2627        patch.object(bridge, "_restart_transport", MagicMock()) as restart,
2628        patch.object(bridge, "_abandon_streaming", MagicMock()) as abandon,
2629    ):
2630        assert bridge._recover_transport() is True
2631        assert bridge._recover_transport() is False
2632
2633    restart.assert_called_once_with()
2634    abandon.assert_called_once_with()
2635
2636
2637def test_transport_loss_after_the_guard_window_recovers_again() -> None:
2638    """A single blip hours apart is a new incident, not a flapping device."""
2639    bridge, _ = _make_anchored_bridge(running=False)
2640
2641    with (
2642        patch(
2643            "music_assistant.providers.airplay.sendspin_bridge.time.monotonic",
2644            side_effect=[100.0, 100.0 + BRIDGE_TRANSPORT_RECOVERY_GUARD_SECONDS + 1],
2645        ),
2646        patch.object(bridge, "_restart_transport", MagicMock()) as restart,
2647        patch.object(bridge, "_abandon_streaming", MagicMock()) as abandon,
2648    ):
2649        assert bridge._recover_transport() is True
2650        assert bridge._recover_transport() is True
2651
2652    assert restart.call_count == 2
2653    abandon.assert_not_called()
2654
2655
2656def test_giving_up_does_not_queue_the_chunk_that_exposed_the_loss() -> None:
2657    """
2658    The chunk that trips the give-up is dropped, not written into the dead stream.
2659
2660    Giving up leaves the anchor and the stream reference untouched, so a chunk
2661    that carried on through the handler would still be placed and queued.
2662    """
2663    bridge, _ = _make_anchored_bridge(running=False)
2664    bridge._last_transport_recovery = 100.0
2665
2666    with (
2667        patch(
2668            "music_assistant.providers.airplay.sendspin_bridge.time.monotonic",
2669            return_value=100.0 + BRIDGE_TRANSPORT_RECOVERY_GUARD_SECONDS - 1,
2670        ),
2671        patch.object(bridge, "_restart_transport", MagicMock()) as restart,
2672    ):
2673        bridge._on_audio_chunk(_pcm_chunk(SENDSPIN_EPOCH_US + 1_000_000))
2674
2675    restart.assert_not_called()
2676    assert bridge._write_queue.empty()
2677
2678
2679async def test_a_failed_protocol_start_leaves_the_session() -> None:
2680    """
2681    A cold start that raised takes the speaker out of the group it cannot play in.
2682
2683    Whether the start was the stream's first or a replacement for a transport
2684    that died, the outcome is the same silence; leaving is what stops the player
2685    reporting playback nobody can hear.
2686    """
2687    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2688    bridge._airplay_stream_start_task = asyncio.current_task()
2689    stream = _make_anchor_stream()
2690    stream.connect = AsyncMock(side_effect=OSError("no route to device"))
2691
2692    with (
2693        patch(
2694            "music_assistant.providers.airplay.sendspin_bridge.AirPlayStream",
2695            return_value=stream,
2696        ),
2697        patch.object(bridge, "_leave_sendspin_session", MagicMock()) as leave,
2698    ):
2699        await bridge._start_protocol_from_chunk()
2700
2701    assert bridge._is_streaming is False
2702    leave.assert_called_once_with()
2703
2704
2705async def test_losing_a_speaker_for_good_runs_the_whole_chain() -> None:
2706    """
2707    End to end: a transport dies, the reconnect is refused, the speaker leaves the group.
2708
2709    Every step here is the real one -- detection, the recovery decision, the
2710    re-arm and the cold start -- so a give-up swallowed anywhere along that
2711    chain shows up as a speaker that stays silently "playing" instead of
2712    dropping out.
2713    """
2714    bridge, _ = _make_anchored_bridge(running=False)
2715    stream = _make_anchor_stream()
2716    stream.connect = AsyncMock(side_effect=OSError("device gone"))
2717
2718    with (
2719        patch(
2720            "music_assistant.providers.airplay.sendspin_bridge.AirPlayStream",
2721            return_value=stream,
2722        ),
2723        patch.object(bridge, "_leave_sendspin_session", MagicMock()) as leave,
2724    ):
2725        # the chunk that exposes the loss re-arms and anchors a replacement
2726        bridge._on_audio_chunk(_pcm_chunk(SENDSPIN_EPOCH_US + 1_000_000))
2727        assert bridge._airplay_stream_start_task is not None
2728        leave.assert_not_called()
2729        # run the replacement start the chunk handler scheduled
2730        start = cast("MagicMock", bridge.mass).create_task.call_args.args[0]
2731        bridge._airplay_stream_start_task = asyncio.current_task()
2732        await start
2733
2734    assert bridge._is_streaming is False
2735    leave.assert_called_once_with()
2736
2737
2738def test_abandoning_streaming_stops_the_feed_and_leaves_the_session() -> None:
2739    """Giving up stops accepting chunks, unblocks the writer and leaves the session."""
2740    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2741    bridge._held_chunks.append(_pcm_chunk(SENDSPIN_EPOCH_US))
2742    bridge._held_us = 100_000
2743
2744    with patch.object(bridge, "_leave_sendspin_session", MagicMock()) as leave:
2745        bridge._abandon_streaming()
2746
2747    assert bridge._is_streaming is False
2748    assert not bridge._held_chunks
2749    assert bridge._held_us == 0
2750    assert bridge._airplay_stream_ready.is_set()
2751    # scheduled, not merely constructed: an unscheduled coroutine never leaves
2752    leave.assert_called_once_with()
2753    scheduled = [call.args[0] for call in cast("MagicMock", bridge.mass).create_task.call_args_list]
2754    assert leave.return_value in scheduled
2755
2756
2757def test_a_flapping_device_is_taken_out_of_the_sendspin_session() -> None:
2758    """
2759    Only a device that cannot hold a transport is dropped from the group.
2760
2761    Its silence is real and permanent, so the visible player must stop reporting
2762    playback; the rest of the group keeps going without it.
2763    """
2764    bridge, _ = _make_anchored_bridge(running=False)
2765
2766    with (
2767        patch(
2768            "music_assistant.providers.airplay.sendspin_bridge.time.monotonic",
2769            side_effect=[100.0, 100.0 + BRIDGE_TRANSPORT_RECOVERY_GUARD_SECONDS - 1],
2770        ),
2771        patch.object(bridge, "_restart_transport", MagicMock()),
2772        patch.object(bridge, "_leave_sendspin_session", MagicMock()) as leave,
2773    ):
2774        assert bridge._recover_transport() is True
2775        # the first loss is recoverable, so the speaker keeps its place
2776        leave.assert_not_called()
2777        assert bridge._recover_transport() is False
2778
2779    # scheduled, not merely constructed: an unscheduled coroutine never leaves
2780    leave.assert_called_once_with()
2781    scheduled = [call.args[0] for call in cast("MagicMock", bridge.mass).create_task.call_args_list]
2782    assert leave.return_value in scheduled
2783
2784
2785def test_failed_start_task_gives_up_on_the_stream() -> None:
2786    """A protocol start that failed stops the feed and drops out of the group."""
2787    bridge, _ = _make_anchored_bridge(running=True)
2788    bridge._airplay_stream_start_task = _make_completed_start_task(failed=True)
2789
2790    with patch.object(bridge, "_leave_sendspin_session", MagicMock()) as leave:
2791        bridge._on_audio_chunk(_pcm_chunk(SENDSPIN_EPOCH_US + 1_000_000))
2792
2793    assert bridge._is_streaming is False
2794    leave.assert_called_once_with()
2795
2796
2797async def test_writer_readiness_timeout_gives_up_on_the_stream() -> None:
2798    """
2799    A protocol that never becomes ready stops the feed and drops out of the group.
2800
2801    A transport that hangs instead of failing renders the same silence as one
2802    that refused the connection, so it is given up on the same way.
2803    """
2804    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2805    bridge._writer_task = asyncio.current_task()
2806    bridge._airplay_stream_ready = MagicMock(wait=AsyncMock(side_effect=TimeoutError))
2807
2808    with patch.object(bridge, "_leave_sendspin_session", MagicMock()) as leave:
2809        await bridge._cli_writer()
2810
2811    assert bridge._is_streaming is False
2812    leave.assert_called_once_with()
2813
2814
2815async def test_a_stale_writer_cannot_give_up_on_a_newer_stream() -> None:
2816    """
2817    Only the writer still feeding the bridge may abandon it.
2818
2819    A writer left behind by a slow teardown speaks for a stream that is already
2820    gone; letting it give up would stop, and un-group, its successor.
2821    """
2822    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2823    # a newer stream owns the bridge; this writer is the previous stream's
2824    bridge._writer_task = MagicMock()
2825    bridge._airplay_stream_ready = MagicMock(wait=AsyncMock(side_effect=TimeoutError))
2826
2827    with patch.object(bridge, "_leave_sendspin_session", MagicMock()) as leave:
2828        await bridge._cli_writer()
2829
2830    assert bridge._is_streaming is True
2831    leave.assert_not_called()
2832
2833
2834def _make_grouped_client(*, group_members: int = 2, has_active_stream: bool = False) -> MagicMock:
2835    """
2836    Build a bridge client mock that reads as having left a shared group.
2837
2838    Quiescing moves the client on to a solo group, exactly as the real one does,
2839    so a caller reading ``client.group`` after leaving no longer sees the group
2840    that was left.
2841
2842    :param group_members: Members in the group the client lands in after
2843        leaving; more than one means it was grouped again meanwhile.
2844    :param has_active_stream: Whether that group is playing something of its own.
2845    """
2846    client = MagicMock()
2847    client.group.clients = [MagicMock() for _ in range(group_members)]
2848    client.group.has_active_stream = has_active_stream
2849
2850    async def _quiesce() -> str:
2851        client.group = MagicMock(clients=[client], has_active_stream=False)
2852        return "group-1"
2853
2854    # a real group id: leaving a shared group is what earns a re-join, and None
2855    # (a solo group, which leaving simply stops) must stay distinguishable
2856    client.quiesce_to_solo_stopped = AsyncMock(side_effect=_quiesce)
2857    return client
2858
2859
2860async def test_leaving_a_shared_group_lines_up_a_rejoin() -> None:
2861    """
2862    A bridge taken out of a shared group is given an attempt to come back.
2863
2864    The group it left is captured before quiescing, because that is what moves
2865    the client into a solo group of its own.
2866    """
2867    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2868    client = _make_grouped_client()
2869    left_group = client.group
2870    bridge._sendspin_client = client
2871
2872    with patch.object(bridge, "_rejoin_attempts", MagicMock()) as rejoin:
2873        await bridge._leave_sendspin_session()
2874
2875    rejoin.assert_called_once_with(left_group)
2876
2877
2878async def test_leaving_a_solo_group_has_nothing_to_rejoin() -> None:
2879    """
2880    A solo bridge is stopped by leaving, so there is no group to return to.
2881
2882    Quiescing reports that by returning no previous group; scheduling a re-join
2883    against the group it is already alone in would put it back on PLAYING.
2884    """
2885    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2886    client = _make_grouped_client()
2887    client.quiesce_to_solo_stopped = AsyncMock(return_value=None)
2888    bridge._sendspin_client = client
2889
2890    with patch.object(bridge, "_rejoin_attempts", MagicMock()) as rejoin:
2891        await bridge._leave_sendspin_session()
2892
2893    rejoin.assert_not_called()
2894
2895
2896async def test_a_speaker_that_fails_again_right_after_a_rejoin_stays_out() -> None:
2897    """
2898    A speaker that keeps dropping out cannot cycle in and out of its group.
2899
2900    Re-joining re-runs the stream start that just failed, and a device that
2901    accepts a START before dying would otherwise earn a fresh attempt every
2902    time round, churning CLI processes and group membership indefinitely.
2903    """
2904    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2905    bridge._sendspin_client = _make_grouped_client()
2906    bridge._last_rejoin = 100.0
2907
2908    with (
2909        patch(
2910            "music_assistant.providers.airplay.sendspin_bridge.time.monotonic",
2911            return_value=100.0 + BRIDGE_TRANSPORT_RECOVERY_GUARD_SECONDS - 1,
2912        ),
2913        patch.object(bridge, "_rejoin_attempts", MagicMock()) as rejoin,
2914    ):
2915        await bridge._leave_sendspin_session()
2916
2917    rejoin.assert_not_called()
2918
2919
2920async def test_a_speaker_that_held_its_place_earns_another_rejoin() -> None:
2921    """A device that played on for a while before failing is worth bringing back again."""
2922    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2923    bridge._sendspin_client = _make_grouped_client()
2924    bridge._last_rejoin = 100.0
2925
2926    with (
2927        patch(
2928            "music_assistant.providers.airplay.sendspin_bridge.time.monotonic",
2929            return_value=100.0 + BRIDGE_TRANSPORT_RECOVERY_GUARD_SECONDS + 1,
2930        ),
2931        patch.object(bridge, "_rejoin_attempts", MagicMock()) as rejoin,
2932    ):
2933        await bridge._leave_sendspin_session()
2934
2935    rejoin.assert_called_once()
2936
2937
2938async def test_the_rejoin_window_is_measured_from_the_actual_rejoin() -> None:
2939    """
2940    The guard is stamped where the speaker rejoins, not where the attempt was scheduled.
2941
2942    Stamping at schedule time would tie the guard to the backoff: longer delays
2943    would put the stamp far enough in the past for the window to have expired by
2944    the time the re-joined speaker fails, letting the cycle run again.
2945    """
2946    bridge, _, group = _make_rejoin_bridge()
2947
2948    with (
2949        patch(_NO_REJOIN_DELAYS, (0,)),
2950        patch(
2951            "music_assistant.providers.airplay.sendspin_bridge.time.monotonic",
2952            return_value=1234.0,
2953        ),
2954    ):
2955        await bridge._rejoin_attempts(group)
2956
2957    group.add_client.assert_awaited_once()
2958    assert bridge._last_rejoin == 1234.0
2959
2960
2961async def test_a_failed_rejoin_never_stamps_the_window() -> None:
2962    """A speaker that never made it back has not held a place to be judged on."""
2963    bridge, _, group = _make_rejoin_bridge()
2964    group.add_client = AsyncMock(side_effect=OSError("group is gone"))
2965
2966    with patch(_NO_REJOIN_DELAYS, (0,)):
2967        await bridge._rejoin_attempts(group)
2968
2969    assert bridge._last_rejoin is None
2970
2971
2972async def test_a_give_up_inside_the_window_drops_a_pending_rejoin() -> None:
2973    """
2974    Leaving the speaker out means dropping the attempt that would put it back.
2975
2976    A schedule left running would contradict the decision this give-up just
2977    made, and re-add a speaker that was meant to stay out.
2978    """
2979    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
2980    bridge._sendspin_client = _make_grouped_client()
2981    bridge._last_rejoin = 100.0
2982    pending = MagicMock()
2983    pending.done.return_value = False
2984    bridge._rejoin_task = pending
2985
2986    with patch(
2987        "music_assistant.providers.airplay.sendspin_bridge.time.monotonic",
2988        return_value=100.0 + BRIDGE_TRANSPORT_RECOVERY_GUARD_SECONDS - 1,
2989    ):
2990        await bridge._leave_sendspin_session()
2991
2992    assert bridge._rejoin_task is None
2993    pending.cancel.assert_called_once_with()  # type: ignore[unreachable]
2994
2995
2996def _make_rejoin_bridge(
2997    *, group_members: int = 1, has_active_stream: bool = False
2998) -> tuple[SendspinAirPlayBridge, MagicMock, MagicMock]:
2999    """
3000    Build a bridge in the state a give-up leaves behind, with its client and lost group.
3001
3002    The AirPlay stream is cleared explicitly: a give-up tears it down, and a
3003    bridge still pointing at one reads as a speaker streaming outside the
3004    bridge, which is itself a reason not to re-join.
3005
3006    :param group_members: Members of the group the client sits in now.
3007    :param has_active_stream: Whether that group is playing something of its own.
3008    """
3009    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3010    cast("MagicMock", bridge.airplay_player).stream = None
3011    client = _make_grouped_client(group_members=group_members, has_active_stream=has_active_stream)
3012    bridge._sendspin_client = client
3013    group = MagicMock()
3014    group.clients = [MagicMock()]
3015    group.add_client = AsyncMock()
3016    return bridge, client, group
3017
3018
3019async def test_a_rejoin_puts_the_bridge_back_into_the_group_it_left() -> None:
3020    """The bridge re-joins through the ordinary group-add, which re-runs the stream start."""
3021    bridge, client, group = _make_rejoin_bridge()
3022
3023    with patch(_NO_REJOIN_DELAYS, (0,)):
3024        await bridge._rejoin_attempts(group)
3025
3026    group.add_client.assert_awaited_once_with(client)
3027
3028
3029async def test_a_rejoin_leaves_a_regrouped_speaker_alone() -> None:
3030    """
3031    A speaker grouped again meanwhile is never pulled out of where it was put.
3032
3033    The re-join answers a failure, not the user; landing anywhere other than the
3034    solo group the give-up left means someone else has since decided otherwise.
3035    """
3036    bridge, _, group = _make_rejoin_bridge(group_members=2)
3037
3038    with patch(_NO_REJOIN_DELAYS, (0,)):
3039        await bridge._rejoin_attempts(group)
3040
3041    group.add_client.assert_not_awaited()
3042
3043
3044async def test_a_rejoin_leaves_a_speaker_playing_on_its_own_alone() -> None:
3045    """
3046    A speaker started on its own meanwhile keeps that playback.
3047
3048    Its solo group has one member, so membership alone cannot tell it apart from
3049    the group the give-up left it in -- but adding a client to another group
3050    stops the group it came from, which here is the user's own playback.
3051    """
3052    bridge, _, group = _make_rejoin_bridge(has_active_stream=True)
3053
3054    with patch(_NO_REJOIN_DELAYS, (0,)):
3055        await bridge._rejoin_attempts(group)
3056
3057    group.add_client.assert_not_awaited()
3058
3059
3060async def test_a_rejoin_leaves_a_natively_streaming_speaker_alone() -> None:
3061    """
3062    A speaker taken over by native AirPlay is not dragged back into Sendspin.
3063
3064    Re-joining restarts the bridge transport, which would tear down a session
3065    the bridge does not own.
3066    """
3067    bridge, _, group = _make_rejoin_bridge()
3068    cast("MagicMock", bridge.airplay_player).stream = MagicMock()
3069
3070    with patch(_NO_REJOIN_DELAYS, (0,)):
3071        await bridge._rejoin_attempts(group)
3072
3073    group.add_client.assert_not_awaited()
3074
3075
3076async def test_an_offline_speaker_is_looked_for_again_before_giving_up() -> None:
3077    """
3078    A speaker missing from discovery is never re-joined, but is looked for again.
3079
3080    A rebooting device is absent from discovery for a while after it starts
3081    answering, so abandoning on the first look would spend the whole re-join
3082    budget inside the window where such a device is always missing. Running out
3083    of attempts, rather than returning on the first one, is what shows the later
3084    look happened.
3085    """
3086    bridge, _, group = _make_rejoin_bridge()
3087    cast("MagicMock", bridge.airplay_player).available = False
3088    logger = MagicMock()
3089    bridge.logger = logger
3090
3091    with patch(_NO_REJOIN_DELAYS, (0, 0)):
3092        await bridge._rejoin_attempts(group)
3093
3094    group.add_client.assert_not_awaited()
3095    assert logger.debug.call_count == 2
3096    # the give-up is only reached once the attempts run out
3097    logger.warning.assert_called_once()
3098
3099
3100async def test_a_rejoin_is_abandoned_when_the_group_is_gone() -> None:
3101    """
3102    A group everyone else has left is not a group to return to.
3103
3104    Its object outlives the members holding it, so adding the bridge back would
3105    strand it alone in a group nothing streams to.
3106    """
3107    bridge, _, group = _make_rejoin_bridge()
3108    group.clients = []
3109
3110    with patch(_NO_REJOIN_DELAYS, (0,)):
3111        await bridge._rejoin_attempts(group)
3112
3113    group.add_client.assert_not_awaited()
3114
3115
3116async def test_a_rejoin_that_keeps_failing_gives_up() -> None:
3117    """Every attempt is tried, and a speaker that never returns leaves the player idle."""
3118    bridge, _, group = _make_rejoin_bridge()
3119    group.add_client = AsyncMock(side_effect=OSError("group is gone"))
3120
3121    with patch(_NO_REJOIN_DELAYS, (0, 0)):
3122        await bridge._rejoin_attempts(group)
3123
3124    assert group.add_client.await_count == 2
3125
3126
3127async def test_a_new_stream_supersedes_a_pending_rejoin() -> None:
3128    """Joining a session by any means makes the pending re-join stale."""
3129    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3130    pending = MagicMock()
3131    pending.done.return_value = False
3132    bridge._rejoin_task = pending
3133
3134    bridge._on_stream_start(MagicMock())
3135
3136    assert bridge._rejoin_task is None
3137    pending.cancel.assert_called_once_with()  # type: ignore[unreachable]
3138
3139
3140async def test_a_rejoin_never_cancels_itself() -> None:
3141    """
3142    The re-join survives the stream start it causes.
3143
3144    Adding the bridge back to the group runs the stream-start path that clears
3145    stale schedules, and that path cannot be allowed to kill the attempt making
3146    the call.
3147    """
3148    bridge, client, group = _make_rejoin_bridge()
3149
3150    async def _add_client(_client: MagicMock) -> None:
3151        bridge._rejoin_task = asyncio.current_task()
3152        bridge._on_stream_start(MagicMock())
3153
3154    group.add_client = AsyncMock(side_effect=_add_client)
3155
3156    with patch(_NO_REJOIN_DELAYS, (0,)):
3157        await bridge._rejoin_attempts(group)
3158
3159    group.add_client.assert_awaited_once_with(client)
3160
3161
3162async def test_stopping_the_bridge_drops_a_pending_rejoin() -> None:
3163    """An unloaded bridge has no group to return to."""
3164    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3165    pending = MagicMock()
3166    pending.done.return_value = False
3167    bridge._rejoin_task = pending
3168
3169    await bridge.stop()
3170
3171    assert bridge._rejoin_task is None
3172    pending.cancel.assert_called_once_with()  # type: ignore[unreachable]
3173
3174
3175async def test_leaving_the_session_quiesces_the_bridge_client() -> None:
3176    """The bridge leaves a shared group (or stops a solo one) but stays registered."""
3177    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3178    client = _make_grouped_client()
3179    bridge._sendspin_client = client
3180
3181    await bridge._leave_sendspin_session()
3182
3183    client.quiesce_to_solo_stopped.assert_awaited_once_with()
3184    # staying registered is what keeps the player around for the next stream
3185    cast("MagicMock", bridge.sendspin_server).remove_client.assert_not_called()
3186
3187
3188async def test_leaving_the_session_without_a_client_is_a_noop() -> None:
3189    """
3190    Giving up before registration completed has no session to leave.
3191
3192    The call has to return without touching anything: swallowing an error from
3193    an absent client would look identical from the outside, so the absence of a
3194    complaint is what distinguishes the two.
3195    """
3196    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3197    logger = MagicMock()
3198    bridge.logger = logger
3199    assert bridge._sendspin_client is None
3200
3201    await bridge._leave_sendspin_session()
3202
3203    logger.warning.assert_not_called()
3204
3205
3206# --- An explicit stop: end playback now, without lining up a return ------------
3207
3208
3209def _bridge_manager_for(bridge: SendspinAirPlayBridge) -> SendspinBridgeManager:
3210    """Return a bridge manager holding the given bridge under its player id."""
3211    manager = SendspinBridgeManager(cast("MagicMock", bridge.provider))
3212    manager._bridges[bridge.airplay_player.player_id] = bridge
3213    return manager
3214
3215
3216async def test_an_explicit_stop_tears_the_transport_down_at_once() -> None:
3217    """
3218    A stop the user asked for stops the speaker now, not after the grace window.
3219
3220    A Sendspin stream ending defers the teardown so the next track can ride the
3221    warm binary; nothing follows a stop, and the device holds seconds of audio,
3222    so deferring there just plays out what the user asked to end.
3223    """
3224    bridge, stream = _make_anchored_bridge(running=True)
3225    writer_task = MagicMock()
3226    bridge._writer_task = writer_task
3227    start_task = bridge._airplay_stream_start_task
3228    manager = _bridge_manager_for(bridge)
3229
3230    with (
3231        patch.object(bridge, "_cleanup_old_stream", MagicMock()) as cleanup,
3232        patch.object(bridge, "_leave_sendspin_session", MagicMock()),
3233    ):
3234        assert manager.stop_streaming(bridge.airplay_player.player_id) is True
3235
3236    assert cleanup.call_args.args[:3] == (stream, writer_task, start_task)
3237    assert bridge._is_streaming is False
3238    assert bridge._airplay_stream is None
3239    # no grace window is armed: that is what the teardown would have waited out
3240    cast("MagicMock", bridge.mass).call_later.assert_not_called()
3241
3242
3243async def test_an_explicit_stop_leaves_the_session_without_a_return() -> None:
3244    """
3245    Stopping takes the speaker out of the session, and it stays out.
3246
3247    Sendspin reports playback from the group's state, so a stopped bridge that
3248    stayed in would hold the visible player on PLAYING. The re-join exists to
3249    recover a speaker that dropped out by itself; a user who stopped one has not
3250    asked for it back.
3251    """
3252    bridge, _ = _make_anchored_bridge(running=True)
3253    manager = _bridge_manager_for(bridge)
3254
3255    with patch.object(bridge, "_leave_sendspin_session", MagicMock()) as leave:
3256        manager.stop_streaming(bridge.airplay_player.player_id)
3257
3258    leave.assert_called_once_with(rejoin=False)
3259    # scheduled, not merely constructed: an unscheduled coroutine never leaves
3260    scheduled = [call.args[0] for call in cast("MagicMock", bridge.mass).create_task.call_args_list]
3261    assert leave.return_value in scheduled
3262
3263
3264async def test_a_stop_of_an_idle_bridge_keeps_its_place_in_the_group() -> None:
3265    """
3266    A bridge with nothing playing has no session to leave.
3267
3268    Its group is not reporting playback through this speaker, so quiescing it out
3269    would only cost a grouped-but-idle player its membership on a stop command.
3270    """
3271    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3272    bridge._is_streaming = False
3273    manager = _bridge_manager_for(bridge)
3274
3275    with patch.object(bridge, "_leave_sendspin_session", MagicMock()) as leave:
3276        assert manager.stop_streaming(bridge.airplay_player.player_id) is True
3277
3278    leave.assert_not_called()
3279
3280
3281async def test_a_stop_never_reaches_a_player_without_a_bridge() -> None:
3282    """An unbridged player is left to the caller's own stop path."""
3283    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3284
3285    assert _bridge_manager_for(bridge).stop_streaming("apother") is False
3286
3287
3288# --- A stop on the Sendspin side skips the warm grace (support#6195) -----------
3289
3290
3291def test_a_sendspin_stop_skips_the_grace_and_tears_the_transport_down() -> None:
3292    """
3293    A stop reaching the bridge through its role kills the CLI at once.
3294
3295    The stream end that precedes it arms the warm grace window, during which
3296    the device plays out the seconds it holds buffered. The explicit-stop
3297    signal cancels that window and hands the transport straight to the
3298    cleanup path.
3299    """
3300    bridge, stream = _make_anchored_bridge(running=True)
3301    writer_task = MagicMock()
3302    bridge._writer_task = writer_task
3303    start_task = bridge._airplay_stream_start_task
3304
3305    with patch.object(bridge, "_cleanup_old_stream", MagicMock()) as cleanup:
3306        bridge._on_bridge_stream_end()
3307        bridge._on_bridge_explicit_stop()
3308
3309    cast("MagicMock", bridge.mass).cancel_timer.assert_called_with(bridge._teardown_timer_id)
3310    assert cleanup.call_args.args[:3] == (stream, writer_task, start_task)
3311    assert bridge._airplay_stream is None
3312
3313
3314def test_a_sendspin_stop_keeps_the_bridges_seat_in_the_group() -> None:
3315    """
3316    A group-wide stop leaves the membership alone.
3317
3318    The group's own STOPPED state is what the visible player reports, and the
3319    next play on the group must include this speaker -- unlike a stop aimed at
3320    the AirPlay player itself, there is no session to leave here.
3321    """
3322    bridge, _ = _make_anchored_bridge(running=True)
3323
3324    with (
3325        patch.object(bridge, "_cleanup_old_stream", MagicMock()),
3326        patch.object(bridge, "_leave_sendspin_session", MagicMock()) as leave,
3327    ):
3328        bridge._on_bridge_stream_end()
3329        bridge._on_bridge_explicit_stop()
3330
3331    leave.assert_not_called()
3332
3333
3334def test_an_explicit_stop_spares_a_stream_that_already_took_over() -> None:
3335    """A play racing the stop owns the transport; the stop must not kill it."""
3336    bridge, stream = _make_anchored_bridge(running=True)
3337    bridge._is_streaming = True
3338
3339    with patch.object(bridge, "_schedule_cleanup", MagicMock()) as schedule:
3340        bridge._on_bridge_explicit_stop()
3341
3342    schedule.assert_not_called()
3343    assert bridge._airplay_stream is stream
3344
3345
3346def test_an_explicit_stop_with_nothing_held_is_a_no_op() -> None:
3347    """The ungroup that follows a stop finds the transport already gone."""
3348    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3349    bridge._is_streaming = False
3350
3351    with patch.object(bridge, "_schedule_cleanup", MagicMock()) as schedule:
3352        bridge._on_bridge_explicit_stop()
3353
3354    schedule.assert_not_called()
3355
3356
3357def test_a_stream_end_with_no_transport_arms_no_grace_timer() -> None:
3358    """
3359    A stream end that left nothing behind has nothing to keep warm or defer.
3360
3361    Removing the member from its Sendspin group ends the stream for its roles
3362    a second time after the stop already tore the transport down; re-arming
3363    the timer there would only reschedule an empty teardown.
3364    """
3365    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3366
3367    bridge._on_bridge_stream_end()
3368
3369    cast("MagicMock", bridge.mass).call_later.assert_not_called()
3370    assert bridge._is_streaming is False
3371
3372
3373# --- One shared-PTP decision per Sendspin group --------------------------------
3374
3375
3376def _group_bridges(*bridges: SendspinAirPlayBridge, daemon_ready: bool) -> SendspinBridgeManager:
3377    """
3378    Put the given bridges in one Sendspin group behind a shared bridge manager.
3379
3380    :param bridges: Bridges to place in the group.
3381    :param daemon_ready: What the shared PTP daemon answers a fresh resolve.
3382    """
3383    provider = MagicMock()
3384    provider.ptp_daemon_ready = daemon_ready
3385    manager = SendspinBridgeManager(provider)
3386    provider.bridge_manager = manager
3387    group = MagicMock()
3388    for index, bridge in enumerate(bridges):
3389        bridge.provider = provider
3390        bridge._sendspin_client = MagicMock()
3391        bridge._sendspin_client.group = group
3392        manager._bridges[f"player{index}"] = bridge
3393    return manager
3394
3395
3396def test_the_first_group_member_asks_the_daemon() -> None:
3397    """With no live decision in the group, the daemon's readiness decides."""
3398    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3399    _group_bridges(bridge, daemon_ready=True)
3400
3401    assert bridge._resolve_shared_ptp() is True
3402
3403
3404@pytest.mark.parametrize(("live_decision", "daemon_ready"), [(True, False), (False, True)])
3405def test_a_later_member_adopts_the_groups_live_decision(
3406    live_decision: bool, daemon_ready: bool
3407) -> None:
3408    """
3409    A member starting later joins on the clock the group is already running.
3410
3411    Bridges in one group can start minutes apart, so what the daemon answers at
3412    the second start says nothing about the source the first member's process
3413    was spawned against. Parametrised both ways so the decision is proven to
3414    follow the sibling rather than the daemon.
3415    """
3416    playing = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3417    joiner = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3418    _group_bridges(playing, joiner, daemon_ready=daemon_ready)
3419    playing._use_shared_ptp = live_decision
3420
3421    assert joiner._resolve_shared_ptp() is live_decision
3422
3423
3424def test_a_warm_member_still_speaks_for_the_group() -> None:
3425    """
3426    A process kept for a warm reuse keeps deciding for its group.
3427
3428    Its Sendspin stream ended, but the next one rides that same cli process with
3429    the flag it was spawned with, so a sibling cold-starting alongside it has to
3430    match that flag rather than resolve against the daemon.
3431    """
3432    warm = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3433    joiner = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3434    _group_bridges(warm, joiner, daemon_ready=False)
3435    warm._use_shared_ptp = True
3436    warm._airplay_stream = _make_kept_stream()
3437    warm._started = True
3438    warm._is_streaming = False
3439
3440    assert warm.active_shared_ptp is True
3441    assert joiner._resolve_shared_ptp() is True
3442
3443
3444def test_an_idle_member_does_not_decide() -> None:
3445    """A bridge with no cli process left leaves the group to resolve fresh."""
3446    idle = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3447    starter = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3448    _group_bridges(idle, starter, daemon_ready=True)
3449    idle._use_shared_ptp = False
3450    idle._is_streaming = False
3451
3452    assert idle.active_shared_ptp is None
3453    assert starter._resolve_shared_ptp() is True
3454
3455
3456def test_another_groups_decision_is_not_adopted() -> None:
3457    """Only members of the same Sendspin group share one timing source."""
3458    stranger = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3459    starter = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3460    _group_bridges(stranger, starter, daemon_ready=False)
3461    stranger._use_shared_ptp = True
3462    # the stranger moved on to a group of its own
3463    stranger_client = MagicMock()
3464    stranger_client.group = MagicMock()
3465    stranger._sendspin_client = stranger_client
3466
3467    assert starter._resolve_shared_ptp() is False
3468
3469
3470def test_a_raop_member_carries_no_decision() -> None:
3471    """A legacy RAOP process has no shared-clock flag to hand its group."""
3472    raop = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US, protocol=StreamingProtocol.RAOP)
3473    _group_bridges(raop, daemon_ready=True)
3474
3475    assert raop._resolve_shared_ptp() is None
3476
3477
3478async def test_a_cold_start_spawns_the_cli_with_the_groups_decision() -> None:
3479    """The adopted decision reaches the cli process and is recorded on the bridge."""
3480    playing = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3481    joiner = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3482    _group_bridges(playing, joiner, daemon_ready=False)
3483    playing._use_shared_ptp = True
3484    joiner._drop_until_us = SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
3485    joiner._airplay_stream_start_task = asyncio.current_task()
3486    stream = _make_anchor_stream(ack=UNIX_NOW_MS + COLD_LEAD_MS)
3487
3488    with patch(
3489        "music_assistant.providers.airplay.sendspin_bridge.time.time",
3490        return_value=UNIX_NOW_S,
3491    ):
3492        assert await joiner._start_cold_stream(stream) is True
3493
3494    stream.connect.assert_awaited_once_with(True)
3495    assert joiner.active_shared_ptp is True
3496
3497
3498async def test_a_daemon_lost_mid_start_cannot_split_the_group() -> None:
3499    """
3500    Members starting together agree even when the daemon goes away between them.
3501
3502    The first member records its decision before it awaits its connect, so the
3503    second one finds it however the daemon answers by the time it resolves.
3504    """
3505    first = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3506    second = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3507    manager = _group_bridges(first, second, daemon_ready=True)
3508    first_stream = _make_anchor_stream(ack=UNIX_NOW_MS + COLD_LEAD_MS)
3509    second_stream = _make_anchor_stream(ack=UNIX_NOW_MS + COLD_LEAD_MS)
3510
3511    async def connect(_use_shared_ptp: bool | None) -> None:
3512        # the daemon dies while the first member is still connecting
3513        cast("MagicMock", manager.provider).ptp_daemon_ready = False
3514        await asyncio.sleep(0)
3515
3516    first_stream.connect = AsyncMock(side_effect=connect)
3517
3518    async def cold_start(bridge: SendspinAirPlayBridge, stream: MagicMock) -> None:
3519        bridge._drop_until_us = SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000
3520        bridge._airplay_stream_start_task = asyncio.current_task()
3521        await bridge._start_cold_stream(stream)
3522
3523    with patch(
3524        "music_assistant.providers.airplay.sendspin_bridge.time.time",
3525        return_value=UNIX_NOW_S,
3526    ):
3527        await asyncio.gather(cold_start(first, first_stream), cold_start(second, second_stream))
3528
3529    assert first.active_shared_ptp is True
3530    assert second.active_shared_ptp is True
3531    second_stream.connect.assert_awaited_once_with(True)
3532
3533
3534async def test_a_torn_down_bridge_stops_deciding() -> None:
3535    """
3536    The decision dies with the cli process it was spawned for.
3537
3538    A new Sendspin stream arms the bridge before it resolves, so a decision left
3539    behind by the torn-down process would be handed to the group on its behalf.
3540    """
3541    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3542    bridge._use_shared_ptp = True
3543
3544    await bridge._stop_streaming()
3545    bridge._on_stream_start(MagicMock())
3546
3547    assert bridge._is_streaming is True
3548    assert bridge.active_shared_ptp is None
3549
3550
3551def _make_warm_bridge(
3552    *,
3553    use_shared_ptp: bool | None,
3554    protocol: StreamingProtocol = StreamingProtocol.AIRPLAY2,
3555) -> SendspinAirPlayBridge:
3556    """
3557    Build a bridge holding a connected, anchored cli process on the given flag.
3558
3559    :param use_shared_ptp: The shared-PTP flag its process was spawned with,
3560        None for a process that carries no such decision.
3561    :param protocol: The streaming protocol the bridged player speaks.
3562    """
3563    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US, protocol=protocol)
3564    bridge._airplay_stream = _make_kept_stream()
3565    bridge._started = True
3566    bridge._use_shared_ptp = use_shared_ptp
3567    return bridge
3568
3569
3570def test_a_regrouped_warm_process_is_not_reused() -> None:
3571    """
3572    A process whose flag no longer matches its group has to be respawned.
3573
3574    The flag is baked in at spawn, so reusing the process would keep the bridge
3575    on the clock its old group ran on. Its start lead has to report the cold
3576    figure too, or the respawn lands past the audio Sendspin already scheduled.
3577    """
3578    regrouped = _make_warm_bridge(use_shared_ptp=False)
3579    playing = _make_warm_bridge(use_shared_ptp=True)
3580    _group_bridges(regrouped, playing, daemon_ready=True)
3581    regrouped._bridge_role = MagicMock()
3582
3583    assert regrouped._stream_is_warm_eligible() is True
3584    assert regrouped._can_reuse_stream_warm() is False
3585
3586    regrouped._refresh_bridge_timing()
3587
3588    regrouped._bridge_role.set_timing.assert_called_once_with(
3589        required_lead_time_ms=BRIDGE_COLD_START_LEAD_MS, min_buffer_ms=BRIDGE_MIN_BUFFER_MS
3590    )
3591
3592
3593def test_a_warm_process_matching_its_group_is_reused() -> None:
3594    """A group already on the process's flag costs it no respawn."""
3595    warm = _make_warm_bridge(use_shared_ptp=True)
3596    playing = _make_warm_bridge(use_shared_ptp=True)
3597    _group_bridges(warm, playing, daemon_ready=False)
3598
3599    assert warm._can_reuse_stream_warm() is True
3600
3601
3602def test_a_group_without_a_live_decision_reuses_the_warm_process() -> None:
3603    """
3604    A bridge whose group has no other live decision keeps its process.
3605
3606    Its own process is the group's decision, so a daemon that changed state
3607    since must not churn the transport on every track change.
3608    """
3609    solo = _make_warm_bridge(use_shared_ptp=True)
3610    _group_bridges(solo, daemon_ready=False)
3611
3612    assert solo._can_reuse_stream_warm() is True
3613
3614
3615def test_a_raop_member_keeps_its_warm_process_beside_an_ap2_member() -> None:
3616    """
3617    A RAOP process is never respawned over a group's shared-clock decision.
3618
3619    It carries no such decision of its own, and no respawn could give it one, so
3620    comparing it against an AirPlay 2 sibling's would cost the group a cold
3621    reconnect (and its longer start lead) on every track change for nothing.
3622    """
3623    raop = _make_warm_bridge(use_shared_ptp=None, protocol=StreamingProtocol.RAOP)
3624    ap2 = _make_warm_bridge(use_shared_ptp=True)
3625    _group_bridges(raop, ap2, daemon_ready=True)
3626    raop._bridge_role = MagicMock()
3627
3628    assert raop._can_reuse_stream_warm() is True
3629
3630    raop._refresh_bridge_timing()
3631
3632    raop._bridge_role.set_timing.assert_called_once_with(
3633        required_lead_time_ms=BRIDGE_WARM_START_LEAD_MS, min_buffer_ms=BRIDGE_MIN_BUFFER_MS
3634    )
3635
3636
3637async def test_the_real_chunk_path_records_the_decision_it_spawns_with() -> None:
3638    """
3639    Driving the bridge the way Sendspin does still records what the CLI got.
3640
3641    The start path tells whether it still owns the bridge by comparing itself
3642    against the task handle the chunk handler publishes, so the start task must
3643    not run before that handle is set. Started eagerly it would read None on its
3644    very first check and give up as if a newer start had claimed the bridge.
3645    """
3646    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3647    _group_bridges(bridge, daemon_ready=True)
3648    stream = _make_anchor_stream(ack=UNIX_NOW_MS + COLD_LEAD_MS)
3649    started: list[asyncio.Task[None]] = []
3650
3651    async def connect(_use_shared_ptp: bool | None) -> None:
3652        # a real connect does I/O, so the task suspends here
3653        await asyncio.sleep(0)
3654
3655    stream.connect = AsyncMock(side_effect=connect)
3656
3657    loop = asyncio.get_running_loop()
3658
3659    def create_task(
3660        coro: Coroutine[None, None, None], *, eager_start: bool = True, **_kwargs: object
3661    ) -> asyncio.Task[None]:
3662        # mirrors mass.create_task, whose default eager start would run the
3663        # coroutine to its first await before this returns
3664        task = asyncio.Task(coro, loop=loop, eager_start=eager_start)
3665        started.append(task)
3666        return task
3667
3668    cast("MagicMock", bridge.mass).create_task = create_task
3669
3670    with (
3671        patch(
3672            "music_assistant.providers.airplay.sendspin_bridge.AirPlayStream",
3673            return_value=stream,
3674        ),
3675        patch(
3676            "music_assistant.providers.airplay.sendspin_bridge.time.time",
3677            return_value=UNIX_NOW_S,
3678        ),
3679    ):
3680        bridge._on_audio_chunk(_pcm_chunk(SENDSPIN_EPOCH_US + COLD_LEAD_MS * 1_000))
3681        await asyncio.gather(*started)
3682
3683    stream.connect.assert_awaited_once_with(True)
3684    assert bridge.active_shared_ptp is True
3685
3686
3687@pytest.mark.parametrize("arm", ["sendspin_stream_start", "transport_restart"])
3688def test_a_released_process_stops_deciding_for_its_group(arm: str) -> None:
3689    """
3690    A process the bridge is about to tear down no longer speaks for its group.
3691
3692    Arming the bridge for its next stream happens well before that stream
3693    resolves, so a decision left over from the released process would be handed
3694    to a sibling resolving in between - and after a regroup it is the wrong one.
3695    """
3696    regrouped = _make_warm_bridge(use_shared_ptp=False)
3697    playing = _make_warm_bridge(use_shared_ptp=True)
3698    _group_bridges(regrouped, playing, daemon_ready=True)
3699
3700    if arm == "sendspin_stream_start":
3701        regrouped._on_stream_start(MagicMock())
3702    else:
3703        regrouped._on_bridge_stream_start()
3704
3705    assert regrouped._is_streaming is True
3706    assert regrouped.active_shared_ptp is None
3707    # the sibling still holding a live process keeps deciding for the group
3708    assert playing.active_shared_ptp is True
3709
3710
3711def test_an_abandoned_process_stops_deciding_for_its_group() -> None:
3712    """Giving up on a transport takes its decision out of the group with it."""
3713    abandoned = _make_warm_bridge(use_shared_ptp=True)
3714    _group_bridges(abandoned, daemon_ready=True)
3715
3716    abandoned._abandon_streaming()
3717
3718    assert abandoned._use_shared_ptp is None
3719    assert abandoned.active_shared_ptp is None
3720
3721
3722# --- Volume/mute moved on the AirPlay side, fed back into the bridge role ------
3723
3724
3725def _make_bridge_with_role(
3726    volume: int | None = 40, muted: bool = False
3727) -> tuple[SendspinAirPlayBridge, BridgePlayerRole]:
3728    """Build a bridge whose (real) role is wired to its mocked AirPlay player."""
3729    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3730    role = BridgePlayerRole(client=MagicMock())
3731    role.set_callbacks(
3732        on_audio_chunk=bridge._on_audio_chunk,
3733        on_volume_change=bridge._on_volume_change,
3734        on_mute_change=bridge._on_mute_change,
3735        on_stream_start=bridge._on_bridge_stream_start,
3736        on_stream_end=bridge._on_bridge_stream_end,
3737        initial_volume=volume or 25,
3738        initial_muted=muted,
3739    )
3740    bridge._bridge_role = role
3741    player = cast("MagicMock", bridge.airplay_player)
3742    player.volume_level = volume
3743    player.volume_muted = muted
3744    return bridge, role
3745
3746
3747async def test_registration_seeds_the_role_with_the_state_the_speaker_is_in() -> None:
3748    """
3749    A bridge (re)registering adopts the volume and mute the speaker is already at.
3750
3751    A bridge is torn down and rebuilt on a config change, so starting from a
3752    fixed unmuted default would re-create the divergence on every rebuild.
3753    """
3754    bridge = _make_bridge(clock_now_us=SENDSPIN_EPOCH_US)
3755    player = cast("MagicMock", bridge.airplay_player)
3756    player.volume_level = 35
3757    player.volume_muted = True
3758    role = MagicMock()
3759    server = cast("MagicMock", bridge.sendspin_server)
3760    server.register_external_player.return_value.roles_by_family.return_value = [role]
3761
3762    await bridge.start()
3763
3764    assert role.set_callbacks.call_args.kwargs["initial_volume"] == 35
3765    assert role.set_callbacks.call_args.kwargs["initial_muted"] is True
3766
3767
3768def test_device_volume_feedback_reaches_the_visible_player() -> None:
3769    """
3770    A volume the device reports itself is adopted by the role, not sent back to it.
3771
3772    The role is what the parent's volume resolves to while the bridge streams, so
3773    it has to follow the speaker; forwarding it would hand the device back the
3774    value it just reported.
3775    """
3776    bridge, role = _make_bridge_with_role(volume=40)
3777    player = cast("MagicMock", bridge.airplay_player)
3778    player.volume_level = 55
3779
3780    bridge.sync_role_volume_state()
3781
3782    assert role.get_player_volume() == 55
3783    player.volume_set.assert_not_called()
3784
3785
3786def test_a_mute_latched_on_the_airplay_side_shows_through_the_bridge() -> None:
3787    """
3788    A mute applied on the AirPlay side is visible on the Sendspin player.
3789
3790    While it is latched the AirPlay player swallows every volume command, so a
3791    bridge still reporting the speaker as unmuted leaves nothing to explain why
3792    it is silent - or to unmute it with.
3793    """
3794    bridge, role = _make_bridge_with_role(muted=False)
3795    player = cast("MagicMock", bridge.airplay_player)
3796    player.volume_muted = True
3797
3798    bridge.sync_role_volume_state()
3799
3800    assert role.get_player_muted() is True
3801    player.volume_mute.assert_not_called()
3802
3803
3804def test_an_airplay_volume_change_is_routed_to_that_player_s_bridge() -> None:
3805    """A state update carrying a volume or mute change lands on the right bridge."""
3806    bridge, role = _make_bridge_with_role(volume=40)
3807    manager = _bridge_manager_for(bridge)
3808    player = cast("MagicMock", bridge.airplay_player)
3809    player.volume_level = 55
3810    player.volume_muted = True
3811
3812    manager._on_player_state_updated(
3813        player, {"volume_level": (40, 55), "volume_muted": (False, True)}
3814    )
3815
3816    assert role.get_player_volume() == 55
3817    assert role.get_player_muted() is True
3818
3819
3820def test_state_updates_without_a_volume_change_leave_the_role_alone() -> None:
3821    """
3822    Every player's state update passes through, so unrelated ones do no work.
3823
3824    The callback runs for the whole player graph on every tick, including the
3825    position updates of a playing queue.
3826    """
3827    bridge, role = _make_bridge_with_role(volume=40)
3828    manager = _bridge_manager_for(bridge)
3829    player = cast("MagicMock", bridge.airplay_player)
3830    player.volume_level = 55
3831
3832    manager._on_player_state_updated(player, {"playback_state": ("idle", "playing")})
3833
3834    assert role.get_player_volume() == 40
3835
3836
3837def test_a_player_without_a_bridge_is_ignored() -> None:
3838    """A volume change on a player this manager knows nothing about leaves bridges alone."""
3839    bridge, role = _make_bridge_with_role(volume=40)
3840    manager = _bridge_manager_for(bridge)
3841    other_player = MagicMock()
3842    other_player.player_id = "ap0011223344ff"
3843    other_player.volume_level = 55
3844
3845    manager._on_player_state_updated(other_player, {"volume_level": (40, 55)})
3846
3847    assert role.get_player_volume() == 40
3848
3849
3850def test_the_manager_listens_for_the_state_updates_it_routes() -> None:
3851    """
3852    The bridged AirPlay players are watched for the whole life of the manager.
3853
3854    A protocol player emits no PLAYER_UPDATED event, so the controller's internal
3855    state-update subscription is the only way their volume changes are seen.
3856    """
3857    manager = SendspinBridgeManager(MagicMock())
3858
3859    cast("MagicMock", manager.mass).players.subscribe_player_state_update.assert_called_once_with(
3860        manager._on_player_state_updated
3861    )
3862
3863
3864def test_a_volume_set_through_the_bridge_settles_in_one_pass() -> None:
3865    """
3866    A volume coming down from Sendspin is not announced again on its way back.
3867
3868    The player ends up holding what the role handed it, so reading that state back
3869    has to compare equal - otherwise every command would bounce between the two.
3870    """
3871    bridge, role = _make_bridge_with_role(volume=40)
3872    player = cast("MagicMock", bridge.airplay_player)
3873    client = cast("MagicMock", role._client)
3874
3875    role.set_player_volume(70)
3876    player.volume_set.assert_called_once_with(70)
3877    # the AirPlay player records the level it was handed
3878    player.volume_level = 70
3879    client.reset_mock()
3880
3881    bridge.sync_role_volume_state()
3882
3883    assert role.get_player_volume() == 70
3884    client._signal_event.assert_not_called()
3885
3886
3887def test_a_volume_from_the_role_is_not_resolved_a_second_time() -> None:
3888    """
3889    A volume the role delivers reaches the speaker as-is, not via the controller.
3890
3891    The level arrives on the device's own scale, already resolved and scaled where
3892    it came from Music Assistant, so sending it back through the controller would
3893    resolve it a second time and return over the same route.
3894    """
3895    bridge, role = _make_bridge_with_role(volume=40)
3896    player = cast("MagicMock", bridge.airplay_player)
3897    players_ctrl = cast("MagicMock", bridge.mass).players
3898
3899    role.set_player_volume(60)
3900    role.set_player_mute(True)
3901
3902    player.volume_set.assert_called_once_with(60)
3903    player.volume_mute.assert_called_once_with(True)
3904    players_ctrl.cmd_volume_set.assert_not_called()
3905    players_ctrl.cmd_volume_mute.assert_not_called()
3906