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