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