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