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