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