music-assistant-server

1"""Unit tests for AirPlay stream session late-join logic."""
2
3import asyncio
4import logging
5import time
6from collections.abc import AsyncGenerator
7from typing import Any
8from unittest.mock import AsyncMock, MagicMock, patch
9
10import pytest
11from music_assistant_models.enums import PlaybackState
12from music_assistant_models.errors import PlayerCommandFailed
13
14from music_assistant.providers.airplay.constants import (
15    AIRPLAY_CLOCK_READY_LEAD_MS,
16    AIRPLAY_CLOCK_READY_TIMEOUT_MS,
17    AIRPLAY_COLD_GROUP_START_LEAD_MS,
18    AIRPLAY_GROUP_START_LEAD_MS,
19    AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS,
20    AIRPLAY_LATE_JOIN_RING_MARGIN_SECONDS,
21    AIRPLAY_LATE_JOIN_RING_MAX_BYTES,
22    AIRPLAY_LATE_JOIN_RING_MIN_SECONDS,
23    AIRPLAY_SPLICE_LEAD_MARGIN_MS,
24    AIRPLAY_START_LEAD_MS,
25    ClockReadiness,
26    StreamingProtocol,
27)
28from music_assistant.providers.airplay.stream_session import AirPlayStreamSession
29
30PCM_SAMPLE_SIZE = 176400  # 44.1kHz / 16-bit / 2ch
31
32
33async def _ack_commanded_instant(start_unix_ms: int = 0, *_args: Any, **_kwargs: Any) -> int:
34    """Ack a START at exactly the instant it was commanded, as a feasible one is."""
35    return start_unix_ms
36
37
38def _stream_defaults(stream: MagicMock) -> MagicMock:
39    """
40    Apply the verified-start API defaults to a mocked stream.
41
42    Every START is acked at the commanded instant, with no warm-lead constraint
43    and no receiver clock projection, so the tests assert the commanded values
44    directly.
45    """
46    stream.start = AsyncMock(side_effect=_ack_commanded_instant)
47    stream.wait_clock_ready = AsyncMock(return_value=(ClockReadiness.UNREPORTED, 0))
48    stream.warm_lead_ms = 0
49    stream.flushed_head_unix_ms = 0
50    return stream
51
52
53def _make_session(
54    start_time: float,
55    seconds_streamed: float,
56) -> AirPlayStreamSession:
57    """
58    Create a stream session for testing.
59
60    :param start_time: The wall-clock time when the stream was started.
61    :param seconds_streamed: How many seconds of audio have been streamed.
62    """
63    prov = MagicMock()
64
65    pcm_format = MagicMock()
66    pcm_format.pcm_sample_size = PCM_SAMPLE_SIZE
67    pcm_format.sample_rate = 44100
68    pcm_format.bit_depth = 16
69    pcm_format.channels = 2
70
71    leader = MagicMock()
72    leader.player_id = "leader"
73    leader.protocol = StreamingProtocol.RAOP
74    # a joinable session has a reference member that is actually playing
75    leader.playback_state = PlaybackState.PLAYING
76    leader.stream = _stream_defaults(MagicMock())
77    leader.stream.running = True
78    leader.stream.connected = True
79    leader.stream.wait_audio_present = AsyncMock(return_value=True)
80    leader.stream.cumulative_shift_seconds = 0.0
81    leader.config.get_value = MagicMock(return_value=0)
82
83    session = AirPlayStreamSession(prov, [leader], pcm_format, MagicMock(elapsed_time=0))
84    session.start_time = start_time
85    session.seconds_streamed = seconds_streamed
86    session.start_unix_ms = 1  # dummy
87
88    return session
89
90
91def _make_late_joiner() -> MagicMock:
92    """Create a mock AirPlay player for late-join testing."""
93    player = MagicMock()
94    player.player_id = "late_joiner"
95    player.protocol = StreamingProtocol.RAOP
96    player.stream = None
97    player.config = MagicMock()
98    player.config.get_value = MagicMock(return_value=0)
99    return player
100
101
102def _setup_stream(player: MagicMock) -> Any:
103    """Return a side_effect callable that sets up the stream mock on the player."""
104
105    def _side_effect(*_args: Any, **_kwargs: Any) -> None:
106        player.stream = _stream_defaults(MagicMock())
107        player.stream.running = True
108        player.stream.connected = True
109        player.stream.wait_for_connection = AsyncMock()
110        player.stream.wait_audio_present = AsyncMock(return_value=True)
111        player.stream.flush = AsyncMock(return_value=True)
112        player.stream.cumulative_shift_seconds = 0.0
113
114    return _side_effect
115
116
117def _captured_start_at(player: MagicMock) -> float:
118    """Return the start time passed to the player's START command."""
119    start_unix_ms = player.stream.start.call_args[0][0]
120    assert isinstance(start_unix_ms, int)
121    return start_unix_ms / 1000
122
123
124@pytest.mark.asyncio
125async def test_initial_client_failure_stops_started_clients() -> None:
126    """A partial group startup failure cancels siblings before session teardown."""
127    session = _make_session(time.time(), 0)
128    first_player: Any = session.sync_clients[0]
129    second_player = MagicMock()
130    session.sync_clients.append(second_player)
131    first_started = asyncio.Event()
132    first_cancelled = asyncio.Event()
133
134    async def audio_source() -> AsyncGenerator[bytes]:
135        yield b""
136
137    async def start_client(player: Any, *_args: Any) -> None:
138        if player is first_player:
139            first_started.set()
140            try:
141                await asyncio.Event().wait()
142            except asyncio.CancelledError:
143                first_cancelled.set()
144                raise
145        await first_started.wait()
146        raise OSError("process failed")
147
148    with (
149        patch.object(
150            session,
151            "_start_client",
152            new_callable=AsyncMock,
153            side_effect=start_client,
154        ),
155        patch.object(session, "stop", new_callable=AsyncMock) as stop_session,
156        pytest.raises(PlayerCommandFailed, match="Playback failed to start"),
157    ):
158        await session.start(audio_source())
159
160    assert first_cancelled.is_set()
161    stop_session.assert_awaited_once()
162
163
164@pytest.mark.asyncio
165@pytest.mark.parametrize(
166    ("first_protocol", "second_protocol"),
167    [
168        (StreamingProtocol.RAOP, StreamingProtocol.RAOP),
169        (StreamingProtocol.AIRPLAY2, StreamingProtocol.AIRPLAY2),
170        (StreamingProtocol.RAOP, StreamingProtocol.AIRPLAY2),
171    ],
172)
173async def test_initial_group_waits_for_every_member_before_shared_start(
174    first_protocol: StreamingProtocol,
175    second_protocol: StreamingProtocol,
176) -> None:
177    """Every member connects before one shared START anchors the group."""
178    session = _make_session(0, 0)
179    first_player: Any = session.sync_clients[0]
180    first_player.player_id = "first"
181    first_player.protocol = first_protocol
182    first_player.config.get_value = MagicMock(return_value=0)
183    second_player = MagicMock(player_id="second")
184    second_player.protocol = second_protocol
185    second_player.config.get_value = MagicMock(return_value=0)
186    session.sync_clients.append(second_player)
187    session.media.elapsed_time = 12
188    operations: list[str] = []
189
190    async def start_client(player: MagicMock, _use_shared_ptp: bool) -> None:
191        stream = _stream_defaults(MagicMock(running=True))
192
193        async def wait_for_connection() -> None:
194            operations.append(f"connected:{player.player_id}")
195
196        async def start(start_unix_ms: int, position_ms: int) -> int:
197            assert position_ms == 12_000
198            operations.append(f"started:{player.player_id}:{start_unix_ms}")
199            return start_unix_ms
200
201        stream.wait_for_connection = AsyncMock(side_effect=wait_for_connection)
202        stream.wait_audio_present = AsyncMock(return_value=True)
203        stream.start = AsyncMock(side_effect=start)
204        player.stream = stream
205
206    with (
207        patch.object(session, "_start_client", side_effect=start_client),
208        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
209        patch.object(session, "_resolve_shared_ptp", new_callable=AsyncMock, return_value=False),
210        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=100.0),
211    ):
212        await session.start(MagicMock())
213
214    last_connected = max(i for i, op in enumerate(operations) if op.startswith("connected:"))
215    first_start = min(i for i, op in enumerate(operations) if op.startswith("started:"))
216    assert last_connected < first_start
217    # start = now (100_000 ms) + the cold group start lead, one shared instant
218    starts = {int(op.rsplit(":", 1)[1]) for op in operations if op.startswith("started:")}
219    assert starts == {100_000 + AIRPLAY_COLD_GROUP_START_LEAD_MS}
220
221
222@pytest.mark.asyncio
223async def test_group_start_never_anchors_before_every_receiver_clock_is_usable() -> None:
224    """A member whose clock lands past the group lead pushes the shared anchor out."""
225    now = 100.0
226    session = _make_session(0, 0)
227    first: Any = session.sync_clients[0]
228    first.protocol = StreamingProtocol.AIRPLAY2
229    first.config.get_value = MagicMock(return_value=0)
230    second = MagicMock(player_id="second", protocol=StreamingProtocol.AIRPLAY2)
231    second.config.get_value = MagicMock(return_value=0)
232    session.sync_clients.append(second)
233    # The slower member's clock lands past the cold group lead, so anchoring on
234    # that lead alone would leave it seated at a different instant to the first.
235    ready_at = int(now * 1000) + AIRPLAY_COLD_GROUP_START_LEAD_MS + 400
236
237    async def start_client(player: MagicMock, _use_shared_ptp: bool) -> None:
238        stream = _stream_defaults(MagicMock(running=True))
239        stream.wait_for_connection = AsyncMock()
240        stream.wait_audio_present = AsyncMock(return_value=True)
241        stream.wait_clock_ready = AsyncMock(
242            return_value=(
243                ClockReadiness.PROJECTED,
244                ready_at if player is second else int(now * 1000),
245            )
246        )
247        player.stream = stream
248
249    with (
250        patch.object(session, "_start_client", side_effect=start_client),
251        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
252        patch.object(session, "_resolve_shared_ptp", new_callable=AsyncMock, return_value=False),
253        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
254    ):
255        await session.start(MagicMock())
256
257    expected = ready_at + AIRPLAY_CLOCK_READY_LEAD_MS
258    for player in (first, second):
259        assert player.stream.start.await_args.args[0] == expected
260
261
262@pytest.mark.asyncio
263async def test_solo_start_waits_for_the_receiver_clock_projection() -> None:
264    """A lone receiver on a cold clock renders silence, so its anchor waits for it too."""
265    now = 100.0
266    ready_at_unix_ms = int(now * 1000) + 5_000
267    session = _make_session(0, 0)
268    player: Any = session.sync_clients[0]
269    player.protocol = StreamingProtocol.AIRPLAY2
270    player.config.get_value = MagicMock(return_value=0)
271
272    async def start_client(_player: MagicMock, _use_shared_ptp: bool) -> None:
273        stream = _stream_defaults(MagicMock(running=True))
274        stream.wait_for_connection = AsyncMock()
275        stream.wait_audio_present = AsyncMock(return_value=True)
276        stream.wait_clock_ready = AsyncMock(
277            return_value=(ClockReadiness.PROJECTED, ready_at_unix_ms)
278        )
279        player.stream = stream
280
281    with (
282        patch.object(session, "_start_client", side_effect=start_client),
283        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
284        patch.object(session, "_resolve_shared_ptp", new_callable=AsyncMock, return_value=False),
285        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
286    ):
287        await session.start(MagicMock())
288
289    player.stream.wait_clock_ready.assert_awaited_once()
290    # the projection is further out than the solo lead, so it carries the anchor
291    assert player.stream.start.await_args.args[0] == ready_at_unix_ms + AIRPLAY_CLOCK_READY_LEAD_MS
292
293
294@pytest.mark.asyncio
295async def test_solo_start_with_a_usable_clock_keeps_the_short_lead() -> None:
296    """A warm clock reports ready with a past instant, so waiting for it costs nothing."""
297    now = 100.0
298    session = _make_session(0, 0)
299    player: Any = session.sync_clients[0]
300    player.protocol = StreamingProtocol.AIRPLAY2
301    player.config.get_value = MagicMock(return_value=0)
302
303    async def start_client(_player: MagicMock, _use_shared_ptp: bool) -> None:
304        stream = _stream_defaults(MagicMock(running=True))
305        stream.wait_for_connection = AsyncMock()
306        stream.wait_audio_present = AsyncMock(return_value=True)
307        # already usable: the projection sits a second in the past
308        stream.wait_clock_ready = AsyncMock(
309            return_value=(ClockReadiness.PROJECTED, int(now * 1000) - 1_000)
310        )
311        player.stream = stream
312
313    with (
314        patch.object(session, "_start_client", side_effect=start_client),
315        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
316        patch.object(session, "_resolve_shared_ptp", new_callable=AsyncMock, return_value=False),
317        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
318    ):
319        await session.start(MagicMock())
320
321    assert player.stream.start.await_args.args[0] == int(now * 1000) + AIRPLAY_START_LEAD_MS
322
323
324@pytest.mark.asyncio
325async def test_group_start_that_never_converges_anchors_where_members_landed() -> None:
326    """A group that keeps correcting is anchored at the members' own last instant."""
327    session = _make_session(0, 0)
328    first: Any = session.sync_clients[0]
329    first.config.get_value = MagicMock(return_value=0)
330    second = MagicMock(player_id="second")
331    second.protocol = StreamingProtocol.RAOP
332    second.config.get_value = MagicMock(return_value=0)
333    session.sync_clients.append(second)
334    commanded: list[int] = []
335
336    async def correcting_start(start_unix_ms: int, _position_ms: int) -> int:
337        # Correct every round, so the loop exhausts without ever converging.
338        commanded.append(start_unix_ms)
339        return start_unix_ms + 100
340
341    async def honoring_start(start_unix_ms: int, _position_ms: int) -> int:
342        return start_unix_ms
343
344    first.stream = _stream_defaults(MagicMock(running=True))
345    first.stream.start = AsyncMock(side_effect=correcting_start)
346    second.stream = _stream_defaults(MagicMock(running=True))
347    second.stream.start = AsyncMock(side_effect=honoring_start)
348
349    await session._start_members(0, 100_000)
350
351    # The retry after the final round is never commanded, so it must not become
352    # the session anchor: that would map every later joiner behind the group.
353    assert session.start_unix_ms == commanded[-1] + 100
354    assert session.start_unix_ms not in commanded
355
356
357@pytest.mark.asyncio
358async def test_corrected_solo_start_adopts_the_instant_without_reanchoring() -> None:
359    """A corrected solo start is anchored at the binary's instant with no re-START."""
360    session = _make_session(0, 0)
361    player: Any = session.sync_clients[0]
362    player.config.get_value = MagicMock(return_value=0)
363    stream = _stream_defaults(MagicMock(running=True))
364    # The binary corrects the commanded instant forward; a re-START would only
365    # re-base reported position on the raw one, so exactly one START may be
366    # commanded and its corrected ack becomes the anchor.
367    stream.start = AsyncMock(return_value=101_500)
368    player.stream = stream
369
370    await session._start_members(0, 100_000)
371
372    stream.start.assert_awaited_once_with(100_000, 0)
373    assert session.start_unix_ms == 101_500
374
375
376@pytest.mark.asyncio
377async def test_group_start_fails_when_a_member_never_acknowledges() -> None:
378    """An unacknowledged member start fails the session instead of recording its instant."""
379    session = _make_session(0, 0)
380    first_player: Any = session.sync_clients[0]
381    first_player.protocol = StreamingProtocol.RAOP
382    second_player = MagicMock(player_id="silent", protocol=StreamingProtocol.RAOP)
383    second_player.config.get_value = MagicMock(return_value=0)
384    session.sync_clients.append(second_player)
385    anchor_before = (session.start_unix_ms, session.start_time)
386
387    async def start_client(player: MagicMock, _use_shared_ptp: bool) -> None:
388        stream = _stream_defaults(MagicMock(running=True))
389        stream.wait_for_connection = AsyncMock()
390        stream.wait_audio_present = AsyncMock(return_value=True)
391        if player is second_player:
392            stream.start = AsyncMock(
393                side_effect=PlayerCommandFailed(
394                    "AirPlay player Player B did not acknowledge its start within 2.0s"
395                )
396            )
397        player.stream = stream
398
399    with (
400        patch.object(session, "_start_client", side_effect=start_client),
401        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
402        patch.object(session, "stop", new_callable=AsyncMock) as stop_session,
403        pytest.raises(PlayerCommandFailed, match="did not acknowledge its start"),
404    ):
405        await session.start(MagicMock())
406
407    # Nothing may be recorded from a round that had no verified answer: an
408    # anchor the group never played is what every later joiner aligns against.
409    assert (session.start_unix_ms, session.start_time) == anchor_before
410    stop_session.assert_awaited_once()
411
412
413@pytest.mark.asyncio
414@pytest.mark.parametrize(
415    "protocol",
416    [StreamingProtocol.RAOP, StreamingProtocol.AIRPLAY2],
417)
418async def test_initial_single_player_starts_after_connect(
419    protocol: StreamingProtocol,
420) -> None:
421    """A standalone player is anchored with a single START once connected."""
422    session = _make_session(0, 0)
423    player: Any = session.sync_clients[0]
424    player.player_id = "solo"
425    player.protocol = protocol
426    player.config.get_value = MagicMock(return_value=0)
427    stream = _stream_defaults(MagicMock(running=True))
428    stream.wait_for_connection = AsyncMock()
429    stream.wait_audio_present = AsyncMock(return_value=True)
430    stream.flush = AsyncMock(return_value=True)
431
432    async def start_client(_player: MagicMock, _use_shared_ptp: bool) -> None:
433        player.stream = stream
434
435    with (
436        patch.object(session, "_start_client", side_effect=start_client),
437        patch.object(session, "_resolve_shared_ptp", new_callable=AsyncMock, return_value=False),
438        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
439        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=200.0),
440    ):
441        await session.start(MagicMock())
442
443    # start = now (200_000 ms) + the solo start lead, position 0
444    stream.start.assert_awaited_once_with(200_000 + AIRPLAY_START_LEAD_MS, 0)
445
446
447@pytest.mark.asyncio
448async def test_initial_connection_failure_never_starts_partial_group() -> None:
449    """If any member fails to connect, no member receives START and the group is stopped."""
450    session = _make_session(0, 0)
451    first_player: Any = session.sync_clients[0]
452    first_player.protocol = StreamingProtocol.RAOP
453    second_player = MagicMock(player_id="second")
454    second_player.protocol = StreamingProtocol.RAOP
455    session.sync_clients.append(second_player)
456
457    async def start_client(player: MagicMock, _use_shared_ptp: bool) -> None:
458        stream = _stream_defaults(MagicMock(running=True))
459        if player is second_player:
460            stream.wait_for_connection = AsyncMock(side_effect=TimeoutError("connect timeout"))
461        else:
462            stream.wait_for_connection = AsyncMock()
463        player.stream = stream
464
465    with (
466        patch.object(session, "_start_client", side_effect=start_client),
467        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
468        patch.object(session, "stop", new_callable=AsyncMock) as stop_session,
469        pytest.raises(PlayerCommandFailed, match="Playback failed to start"),
470    ):
471        await session.start(MagicMock())
472
473    for player in session.sync_clients:
474        stream: Any = player.stream
475        stream.start.assert_not_awaited()
476    stop_session.assert_awaited_once()
477
478
479@pytest.mark.asyncio
480async def test_connection_failure_names_the_member_that_failed(
481    caplog: pytest.LogCaptureFixture,
482) -> None:
483    """The log has to say WHICH speaker failed, and what its binary reported."""
484    session = _make_session(0, 0)
485    prov_logger = logging.getLogger("test.airplay.session")
486    session.prov.logger = prov_logger
487    first_player: Any = session.sync_clients[0]
488    first_player.display_name = "Player A"
489    first_player.protocol = StreamingProtocol.RAOP
490    second_player = MagicMock(player_id="second", display_name="Player B")
491    second_player.protocol = StreamingProtocol.RAOP
492    session.sync_clients.append(second_player)
493
494    async def start_client(player: MagicMock, _use_shared_ptp: bool) -> None:
495        stream = _stream_defaults(MagicMock(running=True))
496        if player is second_player:
497            stream.wait_for_connection = AsyncMock(
498                side_effect=TimeoutError("cliairplay did not connect to Player B: no route")
499            )
500        else:
501            stream.wait_for_connection = AsyncMock()
502        player.stream = stream
503
504    with (
505        caplog.at_level(logging.WARNING),
506        patch.object(session, "_start_client", side_effect=start_client),
507        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
508        patch.object(session, "stop", new_callable=AsyncMock),
509        pytest.raises(PlayerCommandFailed),
510    ):
511        await session.start(MagicMock())
512
513    assert "Player B failed to connect to its device" in caplog.text
514    assert "no route" in caplog.text
515    assert "Player A failed" not in caplog.text
516
517
518@pytest.mark.asyncio
519async def test_unconfirmed_audio_feed_names_the_silent_members() -> None:
520    """A member whose binary never reported audio flowing is named in the failure."""
521    session = _make_session(0, 0)
522    first_player: Any = session.sync_clients[0]
523    first_player.display_name = "Player A"
524    first_player.protocol = StreamingProtocol.RAOP
525    first_player.stream.wait_audio_present = AsyncMock(return_value=True)
526    second_player = MagicMock(player_id="second", display_name="Player B")
527    second_player.protocol = StreamingProtocol.RAOP
528    second_player.stream = _stream_defaults(MagicMock(running=True))
529    second_player.stream.wait_audio_present = AsyncMock(return_value=False)
530    session.sync_clients.append(second_player)
531
532    with pytest.raises(PlayerCommandFailed, match="not confirmed by Player B"):
533        await session._wait_members_audio_present()
534
535
536@pytest.mark.asyncio
537async def test_initial_connection_cancellation_never_starts_group() -> None:
538    """Cancellation while connecting cleans up without anchoring playback."""
539    session = _make_session(0, 0)
540    player: Any = session.sync_clients[0]
541    player.player_id = "solo"
542    player.protocol = StreamingProtocol.RAOP
543    connection_waiting = asyncio.Event()
544    stream = _stream_defaults(MagicMock(running=True))
545
546    async def wait_for_connection() -> None:
547        connection_waiting.set()
548        await asyncio.Event().wait()
549
550    stream.wait_for_connection = AsyncMock(side_effect=wait_for_connection)
551
552    async def start_client(_player: MagicMock, _use_shared_ptp: bool) -> None:
553        player.stream = stream
554
555    with (
556        patch.object(session, "_start_client", side_effect=start_client),
557        patch.object(session, "stop", new_callable=AsyncMock) as stop_session,
558    ):
559        start_task = asyncio.create_task(session.start(MagicMock()))
560        await connection_waiting.wait()
561        start_task.cancel()
562        with pytest.raises(asyncio.CancelledError):
563            await start_task
564
565    stream.start.assert_not_awaited()
566    stop_session.assert_awaited_once()
567
568
569@pytest.mark.parametrize(
570    "protocols",
571    [
572        (StreamingProtocol.RAOP,),
573        (StreamingProtocol.AIRPLAY2,),
574        (StreamingProtocol.RAOP, StreamingProtocol.AIRPLAY2),
575    ],
576)
577def test_warm_replace_supports_every_streaming_protocol(
578    protocols: tuple[StreamingProtocol, ...],
579) -> None:
580    """Connected legacy RAOP, AirPlay 2 and mixed sessions can replace warm."""
581    session = _make_session(0, 0)
582    players: Any = []
583    for index, protocol in enumerate(protocols):
584        player = MagicMock()
585        player.player_id = f"player-{index}"
586        player.protocol = protocol
587        player.stream = _stream_defaults(MagicMock(running=True, connected=True))
588        players.append(player)
589    session.sync_clients = players
590
591    assert session.can_replace(players, session.pcm_format)
592
593
594@pytest.mark.asyncio
595@pytest.mark.parametrize(
596    "protocols",
597    [
598        (StreamingProtocol.AIRPLAY2,),
599        (StreamingProtocol.RAOP,),
600        (StreamingProtocol.AIRPLAY2, StreamingProtocol.AIRPLAY2),
601        (StreamingProtocol.RAOP, StreamingProtocol.RAOP),
602        (StreamingProtocol.RAOP, StreamingProtocol.AIRPLAY2),
603    ],
604)
605async def test_standby_supports_every_connected_streaming_protocol(
606    protocols: tuple[StreamingProtocol, ...],
607) -> None:
608    """Connected legacy RAOP, AirPlay 2 and mixed sessions can enter standby."""
609    session = _make_session(0, 0)
610    players: Any = []
611    for index, protocol in enumerate(protocols):
612        player = MagicMock()
613        player.player_id = f"player-{index}"
614        player.protocol = protocol
615        player.stream = _stream_defaults(MagicMock(running=True, connected=True))
616        player.stream.send_cli_command = AsyncMock(return_value=True)
617        players.append(player)
618    session.sync_clients = players
619
620    assert await session.standby()
621    # the park is carried by the session, so it survives the group shrinking
622    assert session.parked is True
623    for player in players:
624        player.stream.send_cli_command.assert_awaited_once_with("ACTION=STANDBY")
625        player.set_state_from_stream.assert_called_once_with(
626            state=PlaybackState.PAUSED, stream=player.stream
627        )
628
629
630@pytest.mark.asyncio
631@pytest.mark.parametrize(
632    "protocols",
633    [
634        (StreamingProtocol.RAOP,),
635        (StreamingProtocol.AIRPLAY2,),
636        (StreamingProtocol.RAOP, StreamingProtocol.AIRPLAY2),
637    ],
638)
639async def test_standby_resumes_warm_on_existing_streams(
640    protocols: tuple[StreamingProtocol, ...],
641) -> None:
642    """RAOP, AirPlay 2 and mixed sessions resume warm via flush-refill on parked streams."""
643    session = _make_session(0, 0)
644    players: Any = []
645    original_streams: dict[str, MagicMock] = {}
646    for index, protocol in enumerate(protocols):
647        player = MagicMock()
648        player.player_id = f"player-{index}"
649        player.protocol = protocol
650        player.config.get_value = MagicMock(return_value=0)
651        player.stream = _stream_defaults(MagicMock(running=True, connected=True))
652        player.stream.send_cli_command = AsyncMock(return_value=True)
653        player.stream.wait_audio_present = AsyncMock(return_value=True)
654        player.stream.flush = AsyncMock(return_value=True)
655        players.append(player)
656        original_streams[player.player_id] = player.stream
657    session.sync_clients = players
658
659    assert await session.standby()
660    assert session.can_replace(players, session.pcm_format)
661    media = MagicMock(elapsed_time=10)
662    with (
663        patch.object(session, "_start_player_ffmpeg", new_callable=AsyncMock),
664        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
665        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=100.0),
666    ):
667        assert await session.replace(MagicMock(), media)
668
669    # the re-anchor ends the park, so the members are playing again
670    assert session.parked is False
671    # start = now (100_000 ms) + solo/group start lead, position 10s
672    expected_start = 100_000 + (
673        AIRPLAY_START_LEAD_MS if len(protocols) == 1 else AIRPLAY_GROUP_START_LEAD_MS
674    )
675    for player in players:
676        stream = original_streams[player.player_id]
677        assert player.stream is stream
678        stream.send_cli_command.assert_awaited_once_with("ACTION=STANDBY")
679        stream.flush.assert_awaited_once_with()
680        stream.start.assert_awaited_once_with(expected_start, 10_000)
681
682
683@pytest.mark.asyncio
684async def test_warm_replace_flushes_all_before_starting_any() -> None:
685    """A group flushes every member and awaits all acks before any shared START."""
686    session = _make_session(0, 0)
687    players: list[Any] = []
688    release_delayed = asyncio.Event()
689    delayed_waiting = asyncio.Event()
690
691    for player_id in ("first", "delayed"):
692        player = MagicMock(player_id=player_id, protocol=StreamingProtocol.AIRPLAY2)
693        player.config.get_value = MagicMock(return_value=0)
694        stream = _stream_defaults(MagicMock(running=True, connected=True))
695
696        async def flush(*, current_id: str = player_id) -> bool:
697            if current_id == "delayed":
698                delayed_waiting.set()
699                await release_delayed.wait()
700            return True
701
702        stream.flush = AsyncMock(side_effect=flush)
703        stream.wait_audio_present = AsyncMock(return_value=True)
704        player.stream = stream
705        players.append(player)
706    session.sync_clients = players
707
708    with (
709        patch.object(session, "_start_player_ffmpeg", new_callable=AsyncMock),
710        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
711        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=100.0),
712    ):
713        replace_task = asyncio.create_task(session.replace(MagicMock(), MagicMock(elapsed_time=0)))
714        await delayed_waiting.wait()
715        # one member's flush is still pending: no member may have been started yet
716        for player in players:
717            player.stream.start.assert_not_awaited()
718        release_delayed.set()
719        assert await replace_task
720
721    for player in players:
722        player.stream.flush.assert_awaited_once_with()
723        # start = now (100_000 ms) + the group start lead (500 ms), position 0
724        player.stream.start.assert_awaited_once_with(100_000 + AIRPLAY_GROUP_START_LEAD_MS, 0)
725
726
727@pytest.mark.asyncio
728async def test_warm_replace_stops_old_audio_and_ffmpeg_before_flush() -> None:
729    """The old audio feed and ffmpeg are torn down before any FLUSH is sent."""
730    session = _make_session(0, 0)
731    operations: list[str] = []
732    player: Any = session.sync_clients[0]
733    player.config.get_value = MagicMock(return_value=0)
734    stream = player.stream
735    stream.running = True
736    stream.connected = True
737
738    async def flush() -> bool:
739        operations.append("flush")
740        return True
741
742    stream.flush = AsyncMock(side_effect=flush)
743    old_ffmpeg = MagicMock(closed=False)
744
745    async def kill_ffmpeg() -> None:
746        operations.append("ffmpeg-killed")
747
748    old_ffmpeg.kill = AsyncMock(side_effect=kill_ffmpeg)
749    session._player_ffmpeg[player.player_id] = old_ffmpeg
750
751    async def old_audio() -> None:
752        try:
753            await asyncio.Event().wait()
754        except asyncio.CancelledError:
755            operations.append("audio-cancelled")
756            raise
757
758    session._audio_source_task = asyncio.create_task(old_audio())
759    # let old_audio reach its await so the cancellation lands inside its try/except
760    await asyncio.sleep(0)
761
762    with (
763        patch.object(session, "_start_player_ffmpeg", new_callable=AsyncMock),
764        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
765        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=100.0),
766    ):
767        assert await session.replace(MagicMock(), MagicMock(elapsed_time=0))
768
769    assert operations.index("audio-cancelled") < operations.index("flush")
770    assert operations.index("ffmpeg-killed") < operations.index("flush")
771    old_ffmpeg.kill.assert_awaited_once()
772
773
774@pytest.mark.asyncio
775async def test_warm_replace_flush_failure_falls_back_to_cold() -> None:
776    """A member that never acknowledges its flush makes the whole replace fall back."""
777    session = _make_session(0, 0)
778    players: list[Any] = []
779    for player_id, acked in (("ok", True), ("failed", False)):
780        player = MagicMock(player_id=player_id, protocol=StreamingProtocol.RAOP)
781        player.config.get_value = MagicMock(return_value=0)
782        stream = _stream_defaults(MagicMock(running=True, connected=True))
783        stream.flush = AsyncMock(return_value=acked)
784        player.stream = stream
785        players.append(player)
786    session.sync_clients = players
787
788    with (
789        patch.object(session, "_start_player_ffmpeg", new_callable=AsyncMock) as start_ffmpeg,
790        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
791    ):
792        assert await session.replace(MagicMock(), MagicMock(elapsed_time=0)) is False
793
794    # no member is started and no fresh ffmpeg is wired once a flush is unacknowledged
795    for player in players:
796        player.stream.start.assert_not_awaited()
797    start_ffmpeg.assert_not_awaited()
798
799
800@pytest.mark.asyncio
801async def test_warm_replace_start_failure_falls_back_to_cold() -> None:
802    """A failed shared START after flush returns False so the caller restarts cold."""
803    session = _make_session(0, 0)
804    player: Any = session.sync_clients[0]
805    player.config.get_value = MagicMock(return_value=0)
806    stream = player.stream
807    stream.running = True
808    stream.connected = True
809    stream.flush = AsyncMock(return_value=True)
810    stream.start = AsyncMock(side_effect=OSError("start failed"))
811
812    with (
813        patch.object(session, "_start_player_ffmpeg", new_callable=AsyncMock),
814        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
815        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=100.0),
816    ):
817        assert await session.replace(MagicMock(), MagicMock(elapsed_time=0)) is False
818
819
820@pytest.mark.parametrize(
821    ("member_specs", "expected_requirement_ms"),
822    [
823        # the largest member requirement wins, wherever that member sits
824        (((4000, 0), (1500, 0)), 4000),
825        (((1500, 0), (4000, 0)), 4000),
826        # a negative sync_adjust widens that member's own requirement by exactly
827        # the amount it moves that member's commanded instant earlier
828        (((4000, 0), (3600, -600)), 4200),
829    ],
830)
831def test_warm_anchor_clears_the_receivers_queued_audio(
832    member_specs: tuple[tuple[int, int], ...],
833    expected_requirement_ms: int,
834) -> None:
835    """
836    A warm re-anchor lands beyond the audio the receivers still have queued.
837
838    A splice-timeline member honors the commanded instant only when it lands
839    past that member's own queued audio, so the shared anchor has to clear the
840    largest member requirement: anchoring short leaves that member behind the
841    group and every warm start pays a corrective round.
842    """
843    session = _make_session(0, 0)
844    members: list[Any] = [session.sync_clients[0]]
845    second = MagicMock(player_id="second", protocol=StreamingProtocol.AIRPLAY2)
846    second.stream = _stream_defaults(MagicMock(running=True, connected=True))
847    session.sync_clients.append(second)
848    members.append(second)
849    for player, (warm_lead_ms, adjust_ms) in zip(members, member_specs, strict=True):
850        player.stream.warm_lead_ms = warm_lead_ms
851        player.config.get_value = MagicMock(return_value=adjust_ms)
852
853    with patch("music_assistant.providers.airplay.stream_session.time.time", return_value=100.0):
854        anchor = session._anchor_start_unix_ms(warm=True)
855
856    assert anchor == 100_000 + expected_requirement_ms + AIRPLAY_SPLICE_LEAD_MARGIN_MS
857
858
859def test_warm_anchor_clears_the_head_every_flushed_member_froze() -> None:
860    """
861    A warm re-anchor lands beyond the frozen head the flushed members reported.
862
863    A member renders from its own commanded instant (the anchor plus its
864    sync_adjust), so it is that instant, not the bare anchor, that has to clear
865    the head the member froze at flush, with margin for the command round-trip.
866    """
867    session = _make_session(0, 0)
868    first: Any = session.sync_clients[0]
869    first.stream.flushed_head_unix_ms = 102_000
870    first.config.get_value = MagicMock(return_value=0)
871    second = MagicMock(player_id="second", protocol=StreamingProtocol.AIRPLAY2)
872    second.stream = _stream_defaults(MagicMock(running=True, connected=True))
873    second.stream.flushed_head_unix_ms = 105_000
874    second.config.get_value = MagicMock(return_value=300)
875    session.sync_clients.append(second)
876
877    with patch("music_assistant.providers.airplay.stream_session.time.time", return_value=100.0):
878        anchor = session._anchor_start_unix_ms(warm=True)
879
880    # the later head wins, and the offset its member adds comes back out of it
881    assert anchor == 105_000 - 300 + AIRPLAY_SPLICE_LEAD_MARGIN_MS
882
883
884@pytest.mark.asyncio
885async def test_start_player_ffmpeg_wires_persistent_cli_stdin() -> None:
886    """The per-seek ffmpeg is wired to the member's persistent cli stdin fd and tracked."""
887    session = _make_session(0, 0)
888    player: Any = session.sync_clients[0]
889    old_ffmpeg = MagicMock()
890    old_ffmpeg.close = AsyncMock()
891    session._player_ffmpeg[player.player_id] = old_ffmpeg
892
893    stream = _stream_defaults(MagicMock())
894    stream.pcm_format = session.pcm_format
895    cli_proc = MagicMock()
896    cli_proc.proc.stdin.transport.get_extra_info.return_value.fileno.return_value = 77
897    stream._cli_proc = cli_proc
898    player.stream = stream
899    new_ffmpeg = MagicMock()
900    new_ffmpeg.start = AsyncMock(return_value=None)
901
902    with (
903        patch(
904            "music_assistant.providers.airplay.stream_session.get_final_output_format",
905            return_value=MagicMock(),
906        ),
907        patch(
908            "music_assistant.providers.airplay.stream_session.get_media_session_id",
909            return_value="session-id",
910        ),
911        patch(
912            "music_assistant.providers.airplay.stream_session.FFMpeg", return_value=new_ffmpeg
913        ) as ffmpeg_factory,
914    ):
915        await session._start_player_ffmpeg(player, MagicMock())
916
917    # the old ffmpeg is closed, never killing the shared cli stdin
918    old_ffmpeg.close.assert_awaited_once()
919    # the fresh ffmpeg writes into the cli process stdin fd (77)
920    assert ffmpeg_factory.call_args.kwargs["audio_output"] == 77
921    new_ffmpeg.start.assert_awaited_once()
922    assert session._player_ffmpeg[player.player_id] is new_ffmpeg
923
924
925@pytest.mark.asyncio
926@pytest.mark.parametrize("stream_state", ["missing", "stopped", "disconnected"])
927async def test_standby_requires_every_member_running_and_connected(stream_state: str) -> None:
928    """Standby is unavailable when any member lacks a reusable connected session."""
929    session = _make_session(0, 0)
930    ready_player = MagicMock(player_id="ready", protocol=StreamingProtocol.AIRPLAY2)
931    ready_player.stream = _stream_defaults(MagicMock(running=True, connected=True))
932    ready_player.stream.send_cli_command = AsyncMock()
933    unavailable_player = MagicMock(player_id="unavailable", protocol=StreamingProtocol.RAOP)
934    unavailable_player.stream = None
935    if stream_state != "missing":
936        unavailable_player.stream = MagicMock(
937            running=stream_state != "stopped",
938            connected=stream_state != "disconnected",
939        )
940        unavailable_player.stream.send_cli_command = AsyncMock()
941    players: Any = [ready_player, unavailable_player]
942    session.sync_clients = players
943
944    assert await session.standby() is False
945    assert session.can_replace(players, session.pcm_format) is False
946    ready_player.stream.send_cli_command.assert_not_awaited()
947    ready_player.set_state_from_stream.assert_not_called()
948    if unavailable_player.stream:
949        unavailable_player.stream.send_cli_command.assert_not_awaited()
950        unavailable_player.set_state_from_stream.assert_not_called()
951
952
953@pytest.mark.asyncio
954async def test_standby_returns_false_when_command_is_not_delivered() -> None:
955    """Standby fails without changing state for a member that misses the command."""
956    session = _make_session(0, 0)
957    logger = MagicMock()
958    session.prov.logger = logger
959    delivered_player = MagicMock(player_id="delivered", protocol=StreamingProtocol.AIRPLAY2)
960    delivered_player.stream = _stream_defaults(MagicMock(running=True, connected=True))
961    delivered_player.stream.send_cli_command = AsyncMock(return_value=True)
962    dropped_player = MagicMock(player_id="dropped", protocol=StreamingProtocol.RAOP)
963    dropped_player.stream = _stream_defaults(MagicMock(running=True, connected=True))
964    dropped_player.stream.send_cli_command = AsyncMock(return_value=False)
965    pending_player = MagicMock(player_id="pending", protocol=StreamingProtocol.AIRPLAY2)
966    pending_player.stream = _stream_defaults(MagicMock(running=True, connected=True))
967    pending_player.stream.send_cli_command = AsyncMock(return_value=True)
968    session.sync_clients = [delivered_player, dropped_player, pending_player]
969
970    assert await session.standby() is False
971    delivered_player.stream.send_cli_command.assert_awaited_once_with("ACTION=STANDBY")
972    delivered_player.set_state_from_stream.assert_called_once_with(
973        state=PlaybackState.PAUSED, stream=delivered_player.stream
974    )
975    dropped_player.stream.send_cli_command.assert_awaited_once_with("ACTION=STANDBY")
976    dropped_player.set_state_from_stream.assert_not_called()
977    pending_player.stream.send_cli_command.assert_not_awaited()
978    pending_player.set_state_from_stream.assert_not_called()
979    logger.warning.assert_called_once()
980
981
982@pytest.mark.asyncio
983async def test_standby_returns_false_when_command_raises() -> None:
984    """Standby fails without changing state when command delivery raises."""
985    session = _make_session(0, 0)
986    logger = MagicMock()
987    session.prov.logger = logger
988    player = MagicMock(player_id="failed", protocol=StreamingProtocol.AIRPLAY2)
989    player.stream = _stream_defaults(MagicMock(running=True, connected=True))
990    player.stream.send_cli_command = AsyncMock(side_effect=OSError("command pipe failed"))
991    session.sync_clients = [player]
992
993    assert await session.standby() is False
994    player.set_state_from_stream.assert_not_called()
995    logger.warning.assert_called_once()
996
997
998@pytest.mark.asyncio
999async def test_late_join_empty_buffer() -> None:
1000    """Test that with an empty buffer, start_at = start_time + seconds_streamed."""
1001    now = time.time()
1002    start_time = now - 8
1003    seconds_streamed = 12.5
1004    session = _make_session(start_time, seconds_streamed)
1005    player = _make_late_joiner()
1006
1007    with patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start:
1008        mock_start.side_effect = _setup_stream(player)
1009        await session.add_client(player)
1010
1011    assert mock_start.called, "_start_client was never called"
1012    expected = start_time + seconds_streamed
1013    assert abs(_captured_start_at(player) - expected) < 0.1
1014
1015
1016@pytest.mark.asyncio
1017async def test_late_join_caps_prime_at_whole_ring_when_position_predates_it() -> None:
1018    """A due position older than the ring caps the prime at the whole buffer."""
1019    now = 1_000_000.0
1020    # Synthetic anchor in the future forces the due position behind the oldest
1021    # buffered sample: only the whole ring can prime and the anchor is pulled
1022    # to the ring's first sample so content and anchor stay exactly aligned.
1023    seconds_streamed = 12.5
1024    buffer_seconds = 3
1025    start_time = now + 5.0
1026    session = _make_session(start_time, seconds_streamed)
1027    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * buffer_seconds)
1028    player = _make_late_joiner()
1029
1030    written_chunks: list[bytes] = []
1031
1032    async def capture_write(_player: Any, chunk: bytes) -> None:
1033        written_chunks.append(chunk)
1034
1035    with (
1036        patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start,
1037        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1038        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1039    ):
1040        mock_start.side_effect = _setup_stream(player)
1041        await session.add_client(player)
1042
1043    assert mock_start.called, "_start_client was never called"
1044    # the whole ring is primed, nothing is skipped, and the anchor maps to the
1045    # ring's first sample: start_at = start_time + (seconds_streamed - buffer)
1046    assert session._client_skip_bytes[player.player_id] == 0
1047    assert written_chunks, "expected the whole ring to be primed"
1048    assert len(written_chunks[0]) / PCM_SAMPLE_SIZE == pytest.approx(buffer_seconds, abs=0.01)
1049    expected = start_time + (seconds_streamed - buffer_seconds)
1050    assert _captured_start_at(player) == pytest.approx(expected, abs=0.02)
1051
1052
1053@pytest.mark.asyncio
1054async def test_late_join_adds_to_sync_clients() -> None:
1055    """Test that the late joiner is added to sync_clients."""
1056    now = time.time()
1057    session = _make_session(now - 10, 12.5)
1058    player = _make_late_joiner()
1059
1060    with patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start:
1061        mock_start.side_effect = _setup_stream(player)
1062        await session.add_client(player)
1063
1064    assert player in session.sync_clients
1065
1066
1067@pytest.mark.asyncio
1068async def test_late_join_start_failure_stops_client() -> None:
1069    """A late joiner whose START fails is torn down before joining the session."""
1070    session = _make_session(time.time() - 10, 12.5)
1071    player = _make_late_joiner()
1072
1073    def setup_failing_start(*_args: Any, **_kwargs: Any) -> None:
1074        _setup_stream(player)()
1075        player.stream.start = AsyncMock(side_effect=OSError("start failed"))
1076
1077    with (
1078        patch.object(session, "_start_client", side_effect=setup_failing_start),
1079        patch.object(session, "stop_client", new_callable=AsyncMock) as stop_client,
1080    ):
1081        await session.add_client(player)
1082
1083    player.stream.start.assert_awaited_once()
1084    # START precedes the prime feed and the sync_clients append, so a failed
1085    # joiner is stopped without ever having joined the session.
1086    assert player not in session.sync_clients
1087    stop_client.assert_awaited_once_with(player, reason="late joiner start/prime failed")
1088
1089
1090@pytest.mark.asyncio
1091async def test_late_join_unacknowledged_start_stops_client() -> None:
1092    """A joiner whose START is never acked is torn down, never mapped onto that instant."""
1093    session = _make_session(time.time() - 10, 12.5)
1094    player = _make_late_joiner()
1095
1096    def setup_unacknowledged_start(*_args: Any, **_kwargs: Any) -> None:
1097        _setup_stream(player)()
1098        player.stream.start = AsyncMock(
1099            side_effect=PlayerCommandFailed(
1100                "AirPlay player Player B did not acknowledge its start within 5.0s"
1101            )
1102        )
1103
1104    with (
1105        patch.object(session, "_start_client", side_effect=setup_unacknowledged_start),
1106        patch.object(session, "stop_client", new_callable=AsyncMock) as stop_client,
1107    ):
1108        await session.add_client(player)
1109
1110    assert player not in session.sync_clients
1111    assert player.player_id not in session._client_skip_bytes
1112    player.stream.rebase_position.assert_not_called()
1113    stop_client.assert_awaited_once_with(player, reason="late joiner start/prime failed")
1114
1115
1116@pytest.mark.asyncio
1117async def test_late_join_refuses_a_parked_session() -> None:
1118    """A parked (standby) session has no live timeline, so it cannot absorb a joiner."""
1119    session = _make_session(time.time() - 10, 12.5)
1120    reference: Any = session.sync_clients[0]
1121    reference.stream.send_cli_command = AsyncMock(return_value=True)
1122    reference.set_state_from_stream = MagicMock(
1123        side_effect=lambda **kwargs: setattr(reference, "playback_state", kwargs["state"])
1124    )
1125    assert await session.standby()
1126    assert reference.playback_state == PlaybackState.PAUSED
1127    player = _make_late_joiner()
1128
1129    with (
1130        patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start,
1131        patch.object(session, "stop_client", new_callable=AsyncMock) as stop_client,
1132    ):
1133        await session.add_client(player)
1134
1135    # The parked session zeroed seconds_streamed while start_time stayed put, so
1136    # anchoring here maps the joiner onto a timeline nothing is playing.
1137    mock_start.assert_not_called()
1138    stop_client.assert_not_awaited()
1139    assert player not in session.sync_clients
1140
1141
1142@pytest.mark.asyncio
1143async def test_late_join_no_running_session() -> None:
1144    """Test that add_client is a no-op when no session is running."""
1145    now = time.time()
1146    session = _make_session(now - 10, 12.5)
1147    # Make the leader's stream not running
1148    leader = session.sync_clients[0]
1149    leader.stream = _stream_defaults(MagicMock())
1150    leader.stream.running = False
1151    player = _make_late_joiner()
1152
1153    with patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start:
1154        await session.add_client(player)
1155        mock_start.assert_not_called()
1156        assert player not in session.sync_clients
1157
1158
1159@pytest.mark.asyncio
1160async def test_late_join_primes_from_ring_tail_at_headroom() -> None:
1161    """A due position inside the ring primes from the tail and anchors at now + headroom."""
1162    # Freeze time so both the test and the code under test agree on `now`.
1163    now = 1_000_000.0
1164    start_time = now - 0.5
1165    seconds_streamed = 5.0
1166    session = _make_session(start_time, seconds_streamed)
1167    # Fill ring buffer with 5 seconds of non-silent PCM.
1168    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1169    player = _make_late_joiner()
1170
1171    written_chunks: list[bytes] = []
1172
1173    async def capture_write(_player: Any, chunk: bytes) -> None:
1174        written_chunks.append(chunk)
1175
1176    with (
1177        patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start,
1178        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1179        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1180    ):
1181        mock_start.side_effect = _setup_stream(player)
1182        await session.add_client(player)
1183
1184    # start_at is now + min_headroom (the late-join floor); fed_pos_due = 2.0s
1185    assert mock_start.called, "_start_client was never called"
1186    expected_headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1187    assert _captured_start_at(player) - now == pytest.approx(expected_headroom, abs=0.01), (
1188        f"start_at should be at now + min_headroom, "
1189        f"got offset {_captured_start_at(player) - now:.4f}s"
1190    )
1191
1192    # The last 3.0s of the ring is primed (positions 2.0s..5.0s), nothing skipped.
1193    assert session._client_skip_bytes[player.player_id] == 0
1194    assert written_chunks, "No data was written to the player"
1195    remaining_seconds = len(written_chunks[0]) / PCM_SAMPLE_SIZE
1196    expected_primed = seconds_streamed - (expected_headroom + (now - start_time))
1197    assert remaining_seconds == pytest.approx(expected_primed, abs=0.01), (
1198        f"expected {expected_primed:.2f}s primed, got {remaining_seconds:.4f}s"
1199    )
1200
1201
1202@pytest.mark.asyncio
1203async def test_late_join_skips_live_feed_when_anchor_ahead_of_write_head() -> None:
1204    """When the due position is ahead of the write head, skip that many live bytes."""
1205    # Freeze time so both the test and the code under test agree on `now`.
1206    now = 1_000_000.0
1207    # Diagnosed clamp case: now - start_time = 8.84s, seconds_streamed = 10.0s,
1208    # min_headroom = 2.5s (the late-join floor, no readiness projection) and no
1209    # group shift, so the anchor is due at 11.34s of feed, past the 10.0s write
1210    # head.
1211    start_time = now - 8.84
1212    seconds_streamed = 10.0
1213    session = _make_session(start_time, seconds_streamed)
1214    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 10)
1215    player = _make_late_joiner()
1216
1217    written_chunks: list[bytes] = []
1218
1219    async def capture_write(_player: Any, chunk: bytes) -> None:
1220        written_chunks.append(chunk)
1221
1222    with (
1223        patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start,
1224        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1225        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1226    ):
1227        mock_start.side_effect = _setup_stream(player)
1228        await session.add_client(player)
1229
1230    # anchor is now + min_headroom and nothing is primed (position past the head)
1231    expected_headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1232    assert _captured_start_at(player) - now == pytest.approx(expected_headroom, abs=0.01)
1233    assert written_chunks == []
1234    skip_seconds = session._client_skip_bytes[player.player_id] / PCM_SAMPLE_SIZE
1235    assert skip_seconds == pytest.approx(1.34, abs=0.01)
1236
1237
1238@pytest.mark.asyncio
1239async def test_late_join_primes_from_ring_under_group_shift() -> None:
1240    """A reference member that re-anchored later pulls the due position back into the ring."""
1241    # Freeze time so both the test and the code under test agree on `now`.
1242    now = 1_000_000.0
1243    # Same base as the clamp case, but the reference member accumulated a
1244    # +3.039s starvation shift (134020 frames @44100) so the group's effective
1245    # anchor is later: fed_pos_due = 8.30s, back inside the ring. The joiner is
1246    # primed with ~1.7s from the ring tail and skips nothing.
1247    start_time = now - 8.84
1248    seconds_streamed = 10.0
1249    session = _make_session(start_time, seconds_streamed)
1250    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 10)
1251    reference: Any = session.sync_clients[0]
1252    reference.stream.cumulative_shift_seconds = 134020 / 44100
1253    player = _make_late_joiner()
1254
1255    written_chunks: list[bytes] = []
1256
1257    async def capture_write(_player: Any, chunk: bytes) -> None:
1258        written_chunks.append(chunk)
1259
1260    with (
1261        patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start,
1262        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1263        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1264    ):
1265        mock_start.side_effect = _setup_stream(player)
1266        await session.add_client(player)
1267
1268    expected_headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1269    assert _captured_start_at(player) - now == pytest.approx(expected_headroom, abs=0.01)
1270    assert session._client_skip_bytes[player.player_id] == 0
1271    assert written_chunks, "expected a prime write from the ring tail"
1272    primed_seconds = len(written_chunks[0]) / PCM_SAMPLE_SIZE
1273    assert primed_seconds == pytest.approx(1.699, abs=0.01)
1274
1275
1276@pytest.mark.asyncio
1277async def test_late_join_anchors_on_the_reported_clock_readiness(
1278    caplog: pytest.LogCaptureFixture,
1279) -> None:
1280    """A projected readiness instant anchors the join, just past the receiver's clock."""
1281    # Freeze time so both the test and the code under test agree on `now`.
1282    now = 1_000_000.0
1283    session = _make_session(now - 5.0, 5.0)
1284    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1285    session.prov.logger = logging.getLogger("test.airplay.session")
1286    player = _make_late_joiner()
1287    # A cold receiver: its clock is projected usable 3.0s out, well past the floor.
1288    ready_at_unix_ms = int((now + 3.0) * 1000)
1289
1290    def setup_with_projection(*_args: Any, **_kwargs: Any) -> None:
1291        _setup_stream(player)()
1292        player.stream.wait_clock_ready = AsyncMock(
1293            return_value=(ClockReadiness.PROJECTED, ready_at_unix_ms)
1294        )
1295
1296    with (
1297        caplog.at_level(logging.DEBUG),
1298        patch.object(session, "_start_client", side_effect=setup_with_projection),
1299        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1300        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1301    ):
1302        await session.add_client(player)
1303
1304    expected_lead = AIRPLAY_CLOCK_READY_LEAD_MS / 1000
1305    assert _captured_start_at(player) - now == pytest.approx(3.0 + expected_lead, abs=0.01)
1306    player.stream.wait_clock_ready.assert_awaited_once_with(
1307        timeout=AIRPLAY_CLOCK_READY_TIMEOUT_MS / 1000
1308    )
1309    assert "receiver clock usable in 3.00s; anchoring no earlier than that" in caplog.text
1310
1311
1312@pytest.mark.asyncio
1313async def test_late_join_floor_wins_over_a_clock_that_is_already_ready(
1314    caplog: pytest.LogCaptureFixture,
1315) -> None:
1316    """A receiver whose clock is already locked still gets the join floor as its anchor."""
1317    # Freeze time so both the test and the code under test agree on `now`.
1318    now = 1_000_000.0
1319    session = _make_session(now - 5.0, 5.0)
1320    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1321    session.prov.logger = logging.getLogger("test.airplay.session")
1322    player = _make_late_joiner()
1323    # A warm receiver reports a readiness instant that has already passed.
1324    ready_at_unix_ms = int((now - 1.0) * 1000)
1325
1326    def setup_with_projection(*_args: Any, **_kwargs: Any) -> None:
1327        _setup_stream(player)()
1328        player.stream.wait_clock_ready = AsyncMock(
1329            return_value=(ClockReadiness.PROJECTED, ready_at_unix_ms)
1330        )
1331
1332    with (
1333        caplog.at_level(logging.DEBUG),
1334        patch.object(session, "_start_client", side_effect=setup_with_projection),
1335        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1336        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1337    ):
1338        await session.add_client(player)
1339
1340    expected_headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1341    assert _captured_start_at(player) - now == pytest.approx(expected_headroom, abs=0.01)
1342    assert "receiver clock became usable 1.00s ago; anchoring on the join floor" in caplog.text
1343
1344
1345@pytest.mark.asyncio
1346async def test_late_join_falls_back_to_the_floor_without_a_clock_projection() -> None:
1347    """No projection (NTP timing or a silent receiver) anchors on the floor."""
1348    # Freeze time so both the test and the code under test agree on `now`.
1349    now = 1_000_000.0
1350    session = _make_session(now - 5.0, 5.0)
1351    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1352    player = _make_late_joiner()
1353
1354    with (
1355        patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start,
1356        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1357        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1358    ):
1359        mock_start.side_effect = _setup_stream(player)
1360        await session.add_client(player)
1361
1362    # every fallback shape surfaces as "no projection" to the session
1363    player.stream.wait_clock_ready.assert_awaited_once_with(
1364        timeout=AIRPLAY_CLOCK_READY_TIMEOUT_MS / 1000
1365    )
1366    expected_headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1367    assert _captured_start_at(player) - now == pytest.approx(expected_headroom, abs=0.01)
1368    assert player in session.sync_clients
1369
1370
1371@pytest.mark.asyncio
1372@pytest.mark.parametrize(
1373    "readiness",
1374    [ClockReadiness.UNREPORTED, ClockReadiness.NOT_APPLICABLE],
1375    ids=["unreported", "ntp"],
1376)
1377async def test_late_join_without_a_projection_still_joins(readiness: ClockReadiness) -> None:
1378    """A device with no clock to wait for is a fallback, not a reason to refuse it."""
1379    now = 1_000_000.0
1380    session = _make_session(now - 5.0, 5.0)
1381    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1382    player = _make_late_joiner()
1383
1384    def setup_without_projection(*_args: Any, **_kwargs: Any) -> None:
1385        _setup_stream(player)()
1386        player.stream.wait_clock_ready = AsyncMock(return_value=(readiness, 0))
1387
1388    with (
1389        patch.object(session, "_start_client", side_effect=setup_without_projection),
1390        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1391        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1392    ):
1393        await session.add_client(player)
1394
1395    assert player in session.sync_clients
1396
1397
1398@pytest.mark.asyncio
1399async def test_late_joiner_with_a_stalled_clock_is_not_added() -> None:
1400    """A receiver that never answered our clock renders silence, so keep it out of the group."""
1401    now = 1_000_000.0
1402    session = _make_session(now - 5.0, 5.0)
1403    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1404    player = _make_late_joiner()
1405
1406    def setup_stalled(*_args: Any, **_kwargs: Any) -> None:
1407        _setup_stream(player)()
1408        player.stream.wait_clock_ready = AsyncMock(return_value=(ClockReadiness.STALLED, 0))
1409
1410    with (
1411        patch.object(session, "_start_client", side_effect=setup_stalled),
1412        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1413        patch.object(session, "stop_client", new_callable=AsyncMock) as stop_client,
1414        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1415    ):
1416        await session.add_client(player)
1417
1418    player.stream.start.assert_not_awaited()
1419    assert player not in session.sync_clients
1420    stop_client.assert_awaited_once_with(player, reason="receiver clock stalled")
1421
1422
1423@pytest.mark.asyncio
1424async def test_late_join_feed_keeps_flowing_while_waiting_for_clock_readiness() -> None:
1425    """The group keeps being fed while a joiner's receiver clock projection is pending."""
1426    session = _make_session(time.time() - 5, 5.0)
1427    player = _make_late_joiner()
1428    readiness_pending = asyncio.Event()
1429    readiness_released = asyncio.Event()
1430
1431    def setup_pending_projection(*_args: Any, **_kwargs: Any) -> None:
1432        _setup_stream(player)()
1433
1434        async def wait_clock_ready(*_args: Any, **_kwargs: Any) -> tuple[ClockReadiness, int]:
1435            readiness_pending.set()
1436            await readiness_released.wait()
1437            return (ClockReadiness.UNREPORTED, 0)
1438
1439        player.stream.wait_clock_ready = AsyncMock(side_effect=wait_clock_ready)
1440
1441    with (
1442        patch.object(session, "_start_client", side_effect=setup_pending_projection),
1443        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1444    ):
1445        join = asyncio.create_task(session.add_client(player))
1446        await asyncio.wait_for(readiness_pending.wait(), timeout=5)
1447        assert await asyncio.wait_for(
1448            session._write_chunk_to_all_players(b"\x02" * PCM_SAMPLE_SIZE), timeout=5
1449        )
1450        readiness_released.set()
1451        await asyncio.wait_for(join, timeout=5)
1452
1453    assert session.seconds_streamed == pytest.approx(6.0)
1454    assert player in session.sync_clients
1455
1456
1457@pytest.mark.asyncio
1458async def test_late_join_feed_keeps_flowing_while_start_ack_is_outstanding() -> None:
1459    """The group keeps being fed while a join's START ack is outstanding."""
1460    # Freeze time so both the test and the code under test agree on `now`.
1461    now = 1_000_000.0
1462    start_time = now - 5.0
1463    session = _make_session(start_time, 5.0)
1464    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1465    player = _make_late_joiner()
1466    ack_outstanding = asyncio.Event()
1467    ack_released = asyncio.Event()
1468
1469    def setup_deferred_ack(*_args: Any, **_kwargs: Any) -> None:
1470        _setup_stream(player)()
1471
1472        async def start(start_unix_ms: int, *_args: Any, **_kwargs: Any) -> int:
1473            # the binary holds its ack until the receiver clock is verified
1474            ack_outstanding.set()
1475            await ack_released.wait()
1476            return start_unix_ms
1477
1478        player.stream.start = AsyncMock(side_effect=start)
1479
1480    writes: list[tuple[str, int]] = []
1481
1482    async def capture_write(target: Any, chunk: bytes) -> None:
1483        writes.append((target.player_id, len(chunk)))
1484
1485    with (
1486        patch.object(session, "_start_client", side_effect=setup_deferred_ack),
1487        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1488        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1489    ):
1490        join = asyncio.create_task(session.add_client(player))
1491        await asyncio.wait_for(ack_outstanding.wait(), timeout=5)
1492        assert await asyncio.wait_for(
1493            session._write_chunk_to_all_players(b"\x02" * PCM_SAMPLE_SIZE), timeout=5
1494        )
1495        ack_released.set()
1496        await asyncio.wait_for(join, timeout=5)
1497
1498    # that second of feed reached the leader and moved the write head to 6.0s
1499    assert ("leader", PCM_SAMPLE_SIZE) in writes
1500    assert session.seconds_streamed == pytest.approx(6.0)
1501    # the anchor (now + the 2.5s floor) is due at 7.5s of feed, so the joiner
1502    # skips only the 1.5s still to come, not the 2.5s due at the commanded
1503    # mapping: the content is mapped against the head the feed actually reached
1504    skip_seconds = session._client_skip_bytes[player.player_id] / PCM_SAMPLE_SIZE
1505    assert skip_seconds == pytest.approx(1.5, abs=0.01)
1506    assert player in session.sync_clients
1507
1508
1509@pytest.mark.asyncio
1510async def test_late_join_cancelled_while_ack_outstanding_stops_the_client() -> None:
1511    """A join cancelled while its START ack is outstanding never half-joins the session."""
1512    session = _make_session(time.time() - 5, 5.0)
1513    player = _make_late_joiner()
1514    ack_outstanding = asyncio.Event()
1515
1516    def setup_pending_ack(*_args: Any, **_kwargs: Any) -> None:
1517        _setup_stream(player)()
1518
1519        async def start(*_args: Any, **_kwargs: Any) -> None:
1520            ack_outstanding.set()
1521            await asyncio.Event().wait()
1522
1523        player.stream.start = AsyncMock(side_effect=start)
1524
1525    with (
1526        patch.object(session, "_start_client", side_effect=setup_pending_ack),
1527        patch.object(session, "stop_client", new_callable=AsyncMock) as stop_client,
1528    ):
1529        join = asyncio.create_task(session.add_client(player))
1530        await asyncio.wait_for(ack_outstanding.wait(), timeout=5)
1531        join.cancel()
1532        with pytest.raises(asyncio.CancelledError):
1533            await join
1534
1535    assert player not in session.sync_clients
1536    stop_client.assert_awaited_once_with(player, reason="late joiner start cancelled")
1537
1538
1539@pytest.mark.asyncio
1540async def test_late_join_maps_content_from_the_acked_instant() -> None:
1541    """A binary that acks later than commanded gets its content mapped to the acked instant."""
1542    # Freeze time so both the test and the code under test agree on `now`.
1543    now = 1_000_000.0
1544    start_time = now - 5.0
1545    session = _make_session(start_time, 5.0)
1546    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1547    player = _make_late_joiner()
1548    # A sync_adjust rides on top of the commanded instant, so it has to be taken
1549    # back out of the ack before the content is mapped onto it.
1550    adjust_ms = 200
1551    player.config.get_value = MagicMock(return_value=adjust_ms)
1552    deferral_ms = 1500
1553
1554    def setup_deferred_ack(*_args: Any, **_kwargs: Any) -> None:
1555        _setup_stream(player)()
1556
1557        async def start(start_unix_ms: int, _position_ms: int, *, join: bool = False) -> int:
1558            assert join is True
1559            return start_unix_ms + deferral_ms
1560
1561        player.stream.start = AsyncMock(side_effect=start)
1562
1563    written_chunks: list[bytes] = []
1564
1565    async def capture_write(_player: Any, chunk: bytes) -> None:
1566        written_chunks.append(chunk)
1567
1568    with (
1569        patch.object(session, "_start_client", side_effect=setup_deferred_ack),
1570        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1571        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1572    ):
1573        await session.add_client(player)
1574
1575    commanded_ms = int((now + AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000) * 1000) + adjust_ms
1576    assert player.stream.start.await_args.args[0] == commanded_ms
1577    # The ack lands 1.5s later than commanded, so the due position is 9.0s of
1578    # feed instead of 7.5s: the joiner skips 4.0s of the live feed.
1579    assert written_chunks == []
1580    skip_seconds = session._client_skip_bytes[player.player_id] / PCM_SAMPLE_SIZE
1581    assert skip_seconds == pytest.approx(4.0, abs=0.01)
1582    # Progress is reported against the sample that lands on the acked instant,
1583    # not the one that would have landed on the commanded instant.
1584    player.stream.rebase_position.assert_called_once_with(9000)
1585
1586
1587@pytest.mark.asyncio
1588async def test_write_chunk_drains_skip_counter_across_chunks() -> None:
1589    """A per-client skip is consumed across chunks, slicing the partial one."""
1590    session = _make_session(0, 0)
1591    player: Any = session.sync_clients[0]
1592    ffmpeg = MagicMock(closed=False)
1593    ffmpeg.write = AsyncMock()
1594    session._player_ffmpeg[player.player_id] = ffmpeg
1595    # skip 1.5s of a 1s-per-chunk feed
1596    session._client_skip_bytes[player.player_id] = PCM_SAMPLE_SIZE * 3 // 2
1597
1598    chunk_one = b"\x01" * PCM_SAMPLE_SIZE
1599    chunk_two = b"\x02" * PCM_SAMPLE_SIZE
1600    chunk_three = b"\x03" * PCM_SAMPLE_SIZE
1601    for chunk in (chunk_one, chunk_two, chunk_three):
1602        await session._write_chunk_to_player(player, chunk)
1603
1604    written = [call_args.args[0] for call_args in ffmpeg.write.await_args_list]
1605    # first chunk fully consumed, second chunk sliced in half, third whole
1606    assert written == [b"\x02" * (PCM_SAMPLE_SIZE // 2), chunk_three]
1607    assert session._client_skip_bytes[player.player_id] == 0
1608
1609
1610def test_effective_start_time_adds_reference_member_shift() -> None:
1611    """The effective anchor adds the first sync client's accumulated shift."""
1612    session = _make_session(100.0, 0)
1613    reference: Any = session.sync_clients[0]
1614    reference.stream.cumulative_shift_seconds = 1.539
1615    assert session.effective_start_time == pytest.approx(101.539)
1616
1617    # the reference transfers to whichever client is first
1618    other = MagicMock()
1619    other.stream.cumulative_shift_seconds = 0.5
1620    session.sync_clients.insert(0, other)
1621    assert session.effective_start_time == pytest.approx(100.5)
1622
1623    # a missing stream falls back to the raw anchor
1624    other.stream = None
1625    assert session.effective_start_time == pytest.approx(100.0)
1626
1627
1628@pytest.mark.asyncio
1629async def test_stop_client_clears_skip_and_shift_state() -> None:
1630    """Tearing a client down drops its skip counter and resets its playout shift."""
1631    session = _make_session(0, 0)
1632    player = _make_late_joiner()
1633    player.stream = _stream_defaults(MagicMock())
1634    player.stream.session = session
1635    player.stream.stop = AsyncMock()
1636    session._client_skip_bytes[player.player_id] = 12_345
1637
1638    await session.stop_client(player)
1639
1640    assert player.player_id not in session._client_skip_bytes
1641    player.stream.reset_reanchor_shift.assert_called_once_with()
1642    player.stream.stop.assert_awaited_once_with(force=True)
1643
1644
1645@pytest.mark.asyncio
1646async def test_replace_clears_skip_and_shift_state() -> None:
1647    """A warm replace re-anchors everyone, so skip counters and shifts reset."""
1648    session = _make_session(0, 0)
1649    player: Any = session.sync_clients[0]
1650    player.config.get_value = MagicMock(return_value=0)
1651    stream = player.stream
1652    stream.running = True
1653    stream.connected = True
1654    stream.flush = AsyncMock(return_value=True)
1655    stream.wait_audio_present = AsyncMock(return_value=True)
1656    session._client_skip_bytes[player.player_id] = 999
1657
1658    with (
1659        patch.object(session, "_start_player_ffmpeg", new_callable=AsyncMock),
1660        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
1661        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=100.0),
1662    ):
1663        assert await session.replace(MagicMock(), MagicMock(elapsed_time=0))
1664
1665    assert session._client_skip_bytes == {}
1666    stream.reset_reanchor_shift.assert_called()
1667
1668
1669@pytest.mark.asyncio
1670async def test_cleanup_after_removal_skips_idle_when_player_has_new_session_stream() -> None:
1671    """Cleanup must not idle a player that was already re-added to another session."""
1672    now = time.time()
1673    session = _make_session(now - 10, 12.5)
1674    player = _make_late_joiner()
1675    other_session = object()
1676    player.set_state_from_stream = MagicMock()
1677    player.stream = _stream_defaults(MagicMock())
1678    player.stream.session = other_session
1679    session.sync_clients.clear()
1680
1681    with (
1682        patch.object(session, "stop_client", new_callable=AsyncMock),
1683        patch.object(session, "stop", new_callable=AsyncMock),
1684    ):
1685        await session._cleanup_after_removal(player)
1686
1687    player.set_state_from_stream.assert_not_called()
1688
1689
1690@pytest.mark.asyncio
1691async def test_late_join_pads_with_silence_when_the_ring_ran_out_under_a_committed_anchor() -> None:
1692    """A due position lost to the ring after the START is covered with silence, not a moved anchor."""
1693    # Freeze time so both the test and the code under test agree on `now`.
1694    now = 1_000_000.0
1695    start_time = now - 100.0
1696    session = _make_session(start_time, 110.0)
1697    session._pcm_total_fed = int(110.0 * PCM_SAMPLE_SIZE)
1698    # An 8s ring against a 10s write-head lead: wide enough when the anchor is
1699    # planned, too narrow by the time the binary owns the instant.
1700    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 8)
1701    session._pcm_buffer_max = PCM_SAMPLE_SIZE * 8
1702    # Freeze the adaptive sizing so this test covers the cap, not the growth.
1703    session._peak_lead_seconds = 1e6
1704    logger = MagicMock()
1705    session.prov.logger = logger
1706    player = _make_late_joiner()
1707
1708    written: list[tuple[str, bytes]] = []
1709
1710    async def capture_write(target: Any, chunk: bytes) -> None:
1711        written.append((target.player_id, chunk))
1712
1713    def setup_with_feed(*_args: Any, **_kwargs: Any) -> None:
1714        _setup_stream(player)()
1715
1716        async def start(start_unix_ms: int, _position_ms: int, *, join: bool = False) -> int:
1717            assert join is True
1718            # The group keeps being fed while the START ack is outstanding: this
1719            # is what pushes the joiner's due position off the back of the ring.
1720            await session._write_chunk_to_all_players(b"\x02" * int(2.0 * PCM_SAMPLE_SIZE))
1721            return start_unix_ms
1722
1723        player.stream.start = AsyncMock(side_effect=start)
1724
1725    with (
1726        patch.object(session, "_start_client", side_effect=setup_with_feed),
1727        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1728        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1729    ):
1730        await session.add_client(player)
1731
1732    # The anchor is the one that was commanded: an acked instant is never moved.
1733    headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1734    assert _captured_start_at(player) == pytest.approx(now + headroom, abs=0.001)
1735    prime = next(chunk for pid, chunk in written if pid == player.player_id)
1736    # due is 102.5s of feed and the write head reached 112.0s, so 9.5s is owed
1737    # while only 8s survives in the ring: the missing 1.5s opens as silence.
1738    assert len(prime) / PCM_SAMPLE_SIZE == pytest.approx(9.5, abs=0.001)
1739    pad = int(1.5 * PCM_SAMPLE_SIZE)
1740    assert prime[:pad] == bytes(pad), "the missing head must be silence"
1741    assert set(prime[pad:]) == {1, 2}, "the rest must be the buffered feed"
1742    assert pad % session._pcm_frame_size == 0
1743    assert session._client_skip_bytes[player.player_id] == 0
1744    # Position still reports where the GROUP is at that instant, because the
1745    # real content lands exactly where it would have without the shortfall.
1746    player.stream.rebase_position.assert_not_called()
1747    logger.warning.assert_called_once()
1748    assert "silence" in logger.warning.call_args.args[0]
1749
1750
1751@pytest.mark.asyncio
1752async def test_late_join_ring_shortfall_keeps_a_misaligned_ring_head_frame_aligned() -> None:
1753    """A silence pad over a mid-frame ring head still lands the feed on frame boundaries."""
1754    now = 1_000_000.0
1755    session = _make_session(now - 100.0, 110.0)
1756    # Both the write head and the ring head sit mid-frame.
1757    session._pcm_total_fed = int(110.0 * PCM_SAMPLE_SIZE) + 3
1758    session._pcm_buffer = bytearray(b"\x01" * (PCM_SAMPLE_SIZE * 8 + 2))
1759    session._pcm_buffer_max = len(session._pcm_buffer)
1760    session._peak_lead_seconds = 1e6
1761    player = _make_late_joiner()
1762
1763    written: list[tuple[str, bytes]] = []
1764
1765    async def capture_write(target: Any, chunk: bytes) -> None:
1766        written.append((target.player_id, chunk))
1767
1768    def setup_with_feed(*_args: Any, **_kwargs: Any) -> None:
1769        _setup_stream(player)()
1770
1771        async def start(start_unix_ms: int, _position_ms: int, *, join: bool = False) -> int:
1772            assert join is True
1773            await session._write_chunk_to_all_players(b"\x02" * (int(2.0 * PCM_SAMPLE_SIZE) + 1))
1774            return start_unix_ms
1775
1776        player.stream.start = AsyncMock(side_effect=start)
1777
1778    with (
1779        patch.object(session, "_start_client", side_effect=setup_with_feed),
1780        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1781        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1782    ):
1783        await session.add_client(player)
1784
1785    prime = next(chunk for pid, chunk in written if pid == player.player_id)
1786    frame_size = session._pcm_frame_size
1787    # The prime ends exactly at the write head, so its start - and therefore the
1788    # whole padded prime - has to sit on an absolute frame boundary.
1789    assert (session._pcm_total_fed - len(prime)) % frame_size == 0
1790    pad_len = len(prime) - len(prime.lstrip(b"\x00"))
1791    assert pad_len % frame_size == 0
1792
1793
1794def test_ring_grows_to_the_observed_write_head_lead() -> None:
1795    """The ring tracks the largest lead a session shows, above a floor and under a byte cap."""
1796    now = 1_000_000.0
1797    session = _make_session(now - 100.0, 100.0)
1798    assert session._pcm_buffer_max == int(AIRPLAY_LATE_JOIN_RING_MIN_SECONDS * PCM_SAMPLE_SIZE)
1799
1800    with patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now):
1801        # A 4s lead stays under the floor, which keeps the ring where it was.
1802        session.seconds_streamed = 104.0
1803        session._observe_write_head_lead()
1804        assert session._pcm_buffer_max == int(AIRPLAY_LATE_JOIN_RING_MIN_SECONDS * PCM_SAMPLE_SIZE)
1805
1806        # A 15s lead carries the ring past the floor, with the margin on top.
1807        session.seconds_streamed = 115.0
1808        session._observe_write_head_lead()
1809        assert session._peak_lead_seconds == pytest.approx(15.0, abs=0.001)
1810        expected = int((15.0 + AIRPLAY_LATE_JOIN_RING_MARGIN_SECONDS) * PCM_SAMPLE_SIZE)
1811        assert session._pcm_buffer_max == expected
1812
1813        # A lead that falls back never shrinks the ring: the history a joiner
1814        # still needs is already in it.
1815        session.seconds_streamed = 108.0
1816        session._observe_write_head_lead()
1817        assert session._pcm_buffer_max == expected
1818
1819        # Growth is bounded in bytes, so a hi-res rate cannot multiply it out.
1820        session.seconds_streamed = 100.0 + 3600.0
1821        session._observe_write_head_lead()
1822        assert session._pcm_buffer_max == AIRPLAY_LATE_JOIN_RING_MAX_BYTES
1823
1824
1825def test_write_head_lead_is_not_measured_before_the_anchor_arrives() -> None:
1826    """Audio fed inside the start lead is not counted as pipeline depth."""
1827    now = 1_000_000.0
1828    # Anchored 2.5s into the future: nothing is audible yet.
1829    session = _make_session(now + 2.5, 8.0)
1830    with patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now):
1831        session._observe_write_head_lead()
1832    assert session._peak_lead_seconds == 0.0
1833    assert session._pcm_buffer_max == int(AIRPLAY_LATE_JOIN_RING_MIN_SECONDS * PCM_SAMPLE_SIZE)
1834
1835    unanchored = _make_session(0.0, 8.0)
1836    with patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now):
1837        unanchored._observe_write_head_lead()
1838    assert unanchored._peak_lead_seconds == 0.0
1839
1840
1841@pytest.mark.asyncio
1842async def test_late_join_silence_pad_is_bounded_and_reports_the_residual() -> None:
1843    """An implausible ack is padded only up to the ring bound, and says so."""
1844    now = 1_000_000.0
1845    session = _make_session(now - 400.0, 420.0)
1846    session._pcm_total_fed = int(420.0 * PCM_SAMPLE_SIZE)
1847    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 2)
1848    session._pcm_buffer_max = PCM_SAMPLE_SIZE * 2
1849    session._peak_lead_seconds = 1e6
1850    logger = MagicMock()
1851    session.prov.logger = logger
1852    player = _make_late_joiner()
1853
1854    written: list[tuple[str, bytes]] = []
1855
1856    async def capture_write(target: Any, chunk: bytes) -> None:
1857        written.append((target.player_id, chunk))
1858
1859    def setup_with_stale_ack(*_args: Any, **_kwargs: Any) -> None:
1860        _setup_stream(player)()
1861        # The binary reports an instant far behind the commanded one, mapping
1862        # the joiner back near the start of the session. 320s of head is owed.
1863        player.stream.start = AsyncMock(return_value=int((now - 300.0) * 1000))
1864
1865    with (
1866        patch.object(session, "_start_client", side_effect=setup_with_stale_ack),
1867        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1868        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1869    ):
1870        await session.add_client(player)
1871
1872    prime = next(chunk for pid, chunk in written if pid == player.player_id)
1873    # Bounded at one ring of silence plus the ring itself - never the 320s owed.
1874    assert len(prime) / PCM_SAMPLE_SIZE == pytest.approx(4.0, abs=0.01)
1875    assert (session._pcm_total_fed - len(prime)) % session._pcm_frame_size == 0
1876    # The joiner cannot be placed exactly, so the log must not claim sync.
1877    logger.warning.assert_called_once()
1878    tail = logger.warning.call_args.args[-1]
1879    assert "ahead of the group" in tail
1880    assert "in sync" not in tail
1881
1882
1883@pytest.mark.asyncio
1884async def test_start_client_releases_a_foreign_mute_latch() -> None:
1885    """A client joining the session gets its foreign mute latch released on start."""
1886    session = _make_session(start_time=0.0, seconds_streamed=0.0)
1887    player = _make_late_joiner()
1888
1889    with (
1890        patch(
1891            "music_assistant.providers.airplay.stream_session.AirPlayStream",
1892            return_value=MagicMock(connect=AsyncMock()),
1893        ),
1894        patch.object(session, "_start_player_ffmpeg", AsyncMock()),
1895    ):
1896        await session._start_client(player, use_shared_ptp=False)
1897
1898    player.release_foreign_mute_latch.assert_called_once_with()
1899