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