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