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() -> None:
1278    """A projected readiness instant anchors the join, just past the receiver's clock."""
1279    # Freeze time so both the test and the code under test agree on `now`.
1280    now = 1_000_000.0
1281    session = _make_session(now - 5.0, 5.0)
1282    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1283    player = _make_late_joiner()
1284    # A cold receiver: its clock is projected usable 3.0s out, well past the floor.
1285    ready_at_unix_ms = int((now + 3.0) * 1000)
1286
1287    def setup_with_projection(*_args: Any, **_kwargs: Any) -> None:
1288        _setup_stream(player)()
1289        player.stream.wait_clock_ready = AsyncMock(
1290            return_value=(ClockReadiness.PROJECTED, ready_at_unix_ms)
1291        )
1292
1293    with (
1294        patch.object(session, "_start_client", side_effect=setup_with_projection),
1295        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1296        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1297    ):
1298        await session.add_client(player)
1299
1300    expected_lead = AIRPLAY_CLOCK_READY_LEAD_MS / 1000
1301    assert _captured_start_at(player) - now == pytest.approx(3.0 + expected_lead, abs=0.01)
1302    player.stream.wait_clock_ready.assert_awaited_once_with(
1303        timeout=AIRPLAY_CLOCK_READY_TIMEOUT_MS / 1000
1304    )
1305
1306
1307@pytest.mark.asyncio
1308async def test_late_join_floor_wins_over_a_clock_that_is_already_ready() -> None:
1309    """A receiver whose clock is already locked still gets the join floor as its anchor."""
1310    # Freeze time so both the test and the code under test agree on `now`.
1311    now = 1_000_000.0
1312    session = _make_session(now - 5.0, 5.0)
1313    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1314    player = _make_late_joiner()
1315    # A warm receiver reports a readiness instant that has already passed.
1316    ready_at_unix_ms = int((now - 1.0) * 1000)
1317
1318    def setup_with_projection(*_args: Any, **_kwargs: Any) -> None:
1319        _setup_stream(player)()
1320        player.stream.wait_clock_ready = AsyncMock(
1321            return_value=(ClockReadiness.PROJECTED, ready_at_unix_ms)
1322        )
1323
1324    with (
1325        patch.object(session, "_start_client", side_effect=setup_with_projection),
1326        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1327        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1328    ):
1329        await session.add_client(player)
1330
1331    expected_headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1332    assert _captured_start_at(player) - now == pytest.approx(expected_headroom, abs=0.01)
1333
1334
1335@pytest.mark.asyncio
1336async def test_late_join_falls_back_to_the_floor_without_a_clock_projection() -> None:
1337    """No projection (NTP timing or a silent receiver) anchors on the floor."""
1338    # Freeze time so both the test and the code under test agree on `now`.
1339    now = 1_000_000.0
1340    session = _make_session(now - 5.0, 5.0)
1341    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1342    player = _make_late_joiner()
1343
1344    with (
1345        patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start,
1346        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1347        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1348    ):
1349        mock_start.side_effect = _setup_stream(player)
1350        await session.add_client(player)
1351
1352    # every fallback shape surfaces as "no projection" to the session
1353    player.stream.wait_clock_ready.assert_awaited_once_with(
1354        timeout=AIRPLAY_CLOCK_READY_TIMEOUT_MS / 1000
1355    )
1356    expected_headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1357    assert _captured_start_at(player) - now == pytest.approx(expected_headroom, abs=0.01)
1358    assert player in session.sync_clients
1359
1360
1361@pytest.mark.asyncio
1362@pytest.mark.parametrize(
1363    "readiness",
1364    [ClockReadiness.UNREPORTED, ClockReadiness.NOT_APPLICABLE],
1365    ids=["unreported", "ntp"],
1366)
1367async def test_late_join_without_a_projection_still_joins(readiness: ClockReadiness) -> None:
1368    """A device with no clock to wait for is a fallback, not a reason to refuse it."""
1369    now = 1_000_000.0
1370    session = _make_session(now - 5.0, 5.0)
1371    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1372    player = _make_late_joiner()
1373
1374    def setup_without_projection(*_args: Any, **_kwargs: Any) -> None:
1375        _setup_stream(player)()
1376        player.stream.wait_clock_ready = AsyncMock(return_value=(readiness, 0))
1377
1378    with (
1379        patch.object(session, "_start_client", side_effect=setup_without_projection),
1380        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1381        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1382    ):
1383        await session.add_client(player)
1384
1385    assert player in session.sync_clients
1386
1387
1388@pytest.mark.asyncio
1389async def test_late_joiner_with_a_stalled_clock_is_not_added() -> None:
1390    """A receiver that never answered our clock renders silence, so keep it out of the group."""
1391    now = 1_000_000.0
1392    session = _make_session(now - 5.0, 5.0)
1393    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1394    player = _make_late_joiner()
1395
1396    def setup_stalled(*_args: Any, **_kwargs: Any) -> None:
1397        _setup_stream(player)()
1398        player.stream.wait_clock_ready = AsyncMock(return_value=(ClockReadiness.STALLED, 0))
1399
1400    with (
1401        patch.object(session, "_start_client", side_effect=setup_stalled),
1402        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1403        patch.object(session, "stop_client", new_callable=AsyncMock) as stop_client,
1404        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1405    ):
1406        await session.add_client(player)
1407
1408    player.stream.start.assert_not_awaited()
1409    assert player not in session.sync_clients
1410    stop_client.assert_awaited_once_with(player, reason="receiver clock stalled")
1411
1412
1413@pytest.mark.asyncio
1414async def test_late_join_feed_keeps_flowing_while_waiting_for_clock_readiness() -> None:
1415    """The group keeps being fed while a joiner's receiver clock projection is pending."""
1416    session = _make_session(time.time() - 5, 5.0)
1417    player = _make_late_joiner()
1418    readiness_pending = asyncio.Event()
1419    readiness_released = asyncio.Event()
1420
1421    def setup_pending_projection(*_args: Any, **_kwargs: Any) -> None:
1422        _setup_stream(player)()
1423
1424        async def wait_clock_ready(*_args: Any, **_kwargs: Any) -> tuple[ClockReadiness, int]:
1425            readiness_pending.set()
1426            await readiness_released.wait()
1427            return (ClockReadiness.UNREPORTED, 0)
1428
1429        player.stream.wait_clock_ready = AsyncMock(side_effect=wait_clock_ready)
1430
1431    with (
1432        patch.object(session, "_start_client", side_effect=setup_pending_projection),
1433        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1434    ):
1435        join = asyncio.create_task(session.add_client(player))
1436        await asyncio.wait_for(readiness_pending.wait(), timeout=5)
1437        assert await asyncio.wait_for(
1438            session._write_chunk_to_all_players(b"\x02" * PCM_SAMPLE_SIZE), timeout=5
1439        )
1440        readiness_released.set()
1441        await asyncio.wait_for(join, timeout=5)
1442
1443    assert session.seconds_streamed == pytest.approx(6.0)
1444    assert player in session.sync_clients
1445
1446
1447@pytest.mark.asyncio
1448async def test_late_join_feed_keeps_flowing_while_start_ack_is_outstanding() -> None:
1449    """The group keeps being fed while a join's START ack is outstanding."""
1450    # Freeze time so both the test and the code under test agree on `now`.
1451    now = 1_000_000.0
1452    start_time = now - 5.0
1453    session = _make_session(start_time, 5.0)
1454    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1455    player = _make_late_joiner()
1456    ack_outstanding = asyncio.Event()
1457    ack_released = asyncio.Event()
1458
1459    def setup_deferred_ack(*_args: Any, **_kwargs: Any) -> None:
1460        _setup_stream(player)()
1461
1462        async def start(start_unix_ms: int, *_args: Any, **_kwargs: Any) -> int:
1463            # the binary holds its ack until the receiver clock is verified
1464            ack_outstanding.set()
1465            await ack_released.wait()
1466            return start_unix_ms
1467
1468        player.stream.start = AsyncMock(side_effect=start)
1469
1470    writes: list[tuple[str, int]] = []
1471
1472    async def capture_write(target: Any, chunk: bytes) -> None:
1473        writes.append((target.player_id, len(chunk)))
1474
1475    with (
1476        patch.object(session, "_start_client", side_effect=setup_deferred_ack),
1477        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1478        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1479    ):
1480        join = asyncio.create_task(session.add_client(player))
1481        await asyncio.wait_for(ack_outstanding.wait(), timeout=5)
1482        assert await asyncio.wait_for(
1483            session._write_chunk_to_all_players(b"\x02" * PCM_SAMPLE_SIZE), timeout=5
1484        )
1485        ack_released.set()
1486        await asyncio.wait_for(join, timeout=5)
1487
1488    # that second of feed reached the leader and moved the write head to 6.0s
1489    assert ("leader", PCM_SAMPLE_SIZE) in writes
1490    assert session.seconds_streamed == pytest.approx(6.0)
1491    # the anchor (now + the 2.5s floor) is due at 7.5s of feed, so the joiner
1492    # skips only the 1.5s still to come, not the 2.5s due at the commanded
1493    # mapping: the content is mapped against the head the feed actually reached
1494    skip_seconds = session._client_skip_bytes[player.player_id] / PCM_SAMPLE_SIZE
1495    assert skip_seconds == pytest.approx(1.5, abs=0.01)
1496    assert player in session.sync_clients
1497
1498
1499@pytest.mark.asyncio
1500async def test_late_join_cancelled_while_ack_outstanding_stops_the_client() -> None:
1501    """A join cancelled while its START ack is outstanding never half-joins the session."""
1502    session = _make_session(time.time() - 5, 5.0)
1503    player = _make_late_joiner()
1504    ack_outstanding = asyncio.Event()
1505
1506    def setup_pending_ack(*_args: Any, **_kwargs: Any) -> None:
1507        _setup_stream(player)()
1508
1509        async def start(*_args: Any, **_kwargs: Any) -> None:
1510            ack_outstanding.set()
1511            await asyncio.Event().wait()
1512
1513        player.stream.start = AsyncMock(side_effect=start)
1514
1515    with (
1516        patch.object(session, "_start_client", side_effect=setup_pending_ack),
1517        patch.object(session, "stop_client", new_callable=AsyncMock) as stop_client,
1518    ):
1519        join = asyncio.create_task(session.add_client(player))
1520        await asyncio.wait_for(ack_outstanding.wait(), timeout=5)
1521        join.cancel()
1522        with pytest.raises(asyncio.CancelledError):
1523            await join
1524
1525    assert player not in session.sync_clients
1526    stop_client.assert_awaited_once_with(player, reason="late joiner start cancelled")
1527
1528
1529@pytest.mark.asyncio
1530async def test_late_join_maps_content_from_the_acked_instant() -> None:
1531    """A binary that acks later than commanded gets its content mapped to the acked instant."""
1532    # Freeze time so both the test and the code under test agree on `now`.
1533    now = 1_000_000.0
1534    start_time = now - 5.0
1535    session = _make_session(start_time, 5.0)
1536    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1537    player = _make_late_joiner()
1538    # A sync_adjust rides on top of the commanded instant, so it has to be taken
1539    # back out of the ack before the content is mapped onto it.
1540    adjust_ms = 200
1541    player.config.get_value = MagicMock(return_value=adjust_ms)
1542    deferral_ms = 1500
1543
1544    def setup_deferred_ack(*_args: Any, **_kwargs: Any) -> None:
1545        _setup_stream(player)()
1546
1547        async def start(start_unix_ms: int, _position_ms: int, *, join: bool = False) -> int:
1548            assert join is True
1549            return start_unix_ms + deferral_ms
1550
1551        player.stream.start = AsyncMock(side_effect=start)
1552
1553    written_chunks: list[bytes] = []
1554
1555    async def capture_write(_player: Any, chunk: bytes) -> None:
1556        written_chunks.append(chunk)
1557
1558    with (
1559        patch.object(session, "_start_client", side_effect=setup_deferred_ack),
1560        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1561        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1562    ):
1563        await session.add_client(player)
1564
1565    commanded_ms = int((now + AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000) * 1000) + adjust_ms
1566    assert player.stream.start.await_args.args[0] == commanded_ms
1567    # The ack lands 1.5s later than commanded, so the due position is 9.0s of
1568    # feed instead of 7.5s: the joiner skips 4.0s of the live feed.
1569    assert written_chunks == []
1570    skip_seconds = session._client_skip_bytes[player.player_id] / PCM_SAMPLE_SIZE
1571    assert skip_seconds == pytest.approx(4.0, abs=0.01)
1572    # Progress is reported against the sample that lands on the acked instant,
1573    # not the one that would have landed on the commanded instant.
1574    player.stream.rebase_position.assert_called_once_with(9000)
1575
1576
1577@pytest.mark.asyncio
1578async def test_write_chunk_drains_skip_counter_across_chunks() -> None:
1579    """A per-client skip is consumed across chunks, slicing the partial one."""
1580    session = _make_session(0, 0)
1581    player: Any = session.sync_clients[0]
1582    ffmpeg = MagicMock(closed=False)
1583    ffmpeg.write = AsyncMock()
1584    session._player_ffmpeg[player.player_id] = ffmpeg
1585    # skip 1.5s of a 1s-per-chunk feed
1586    session._client_skip_bytes[player.player_id] = PCM_SAMPLE_SIZE * 3 // 2
1587
1588    chunk_one = b"\x01" * PCM_SAMPLE_SIZE
1589    chunk_two = b"\x02" * PCM_SAMPLE_SIZE
1590    chunk_three = b"\x03" * PCM_SAMPLE_SIZE
1591    for chunk in (chunk_one, chunk_two, chunk_three):
1592        await session._write_chunk_to_player(player, chunk)
1593
1594    written = [call_args.args[0] for call_args in ffmpeg.write.await_args_list]
1595    # first chunk fully consumed, second chunk sliced in half, third whole
1596    assert written == [b"\x02" * (PCM_SAMPLE_SIZE // 2), chunk_three]
1597    assert session._client_skip_bytes[player.player_id] == 0
1598
1599
1600def test_effective_start_time_adds_reference_member_shift() -> None:
1601    """The effective anchor adds the first sync client's accumulated shift."""
1602    session = _make_session(100.0, 0)
1603    reference: Any = session.sync_clients[0]
1604    reference.stream.cumulative_shift_seconds = 1.539
1605    assert session.effective_start_time == pytest.approx(101.539)
1606
1607    # the reference transfers to whichever client is first
1608    other = MagicMock()
1609    other.stream.cumulative_shift_seconds = 0.5
1610    session.sync_clients.insert(0, other)
1611    assert session.effective_start_time == pytest.approx(100.5)
1612
1613    # a missing stream falls back to the raw anchor
1614    other.stream = None
1615    assert session.effective_start_time == pytest.approx(100.0)
1616
1617
1618@pytest.mark.asyncio
1619async def test_stop_client_clears_skip_and_shift_state() -> None:
1620    """Tearing a client down drops its skip counter and resets its playout shift."""
1621    session = _make_session(0, 0)
1622    player = _make_late_joiner()
1623    player.stream = _stream_defaults(MagicMock())
1624    player.stream.session = session
1625    player.stream.stop = AsyncMock()
1626    session._client_skip_bytes[player.player_id] = 12_345
1627
1628    await session.stop_client(player)
1629
1630    assert player.player_id not in session._client_skip_bytes
1631    player.stream.reset_reanchor_shift.assert_called_once_with()
1632    player.stream.stop.assert_awaited_once_with(force=True)
1633
1634
1635@pytest.mark.asyncio
1636async def test_replace_clears_skip_and_shift_state() -> None:
1637    """A warm replace re-anchors everyone, so skip counters and shifts reset."""
1638    session = _make_session(0, 0)
1639    player: Any = session.sync_clients[0]
1640    player.config.get_value = MagicMock(return_value=0)
1641    stream = player.stream
1642    stream.running = True
1643    stream.connected = True
1644    stream.flush = AsyncMock(return_value=True)
1645    stream.wait_audio_present = AsyncMock(return_value=True)
1646    session._client_skip_bytes[player.player_id] = 999
1647
1648    with (
1649        patch.object(session, "_start_player_ffmpeg", new_callable=AsyncMock),
1650        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
1651        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=100.0),
1652    ):
1653        assert await session.replace(MagicMock(), MagicMock(elapsed_time=0))
1654
1655    assert session._client_skip_bytes == {}
1656    stream.reset_reanchor_shift.assert_called()
1657
1658
1659@pytest.mark.asyncio
1660async def test_cleanup_after_removal_skips_idle_when_player_has_new_session_stream() -> None:
1661    """Cleanup must not idle a player that was already re-added to another session."""
1662    now = time.time()
1663    session = _make_session(now - 10, 12.5)
1664    player = _make_late_joiner()
1665    other_session = object()
1666    player.set_state_from_stream = MagicMock()
1667    player.stream = _stream_defaults(MagicMock())
1668    player.stream.session = other_session
1669    session.sync_clients.clear()
1670
1671    with (
1672        patch.object(session, "stop_client", new_callable=AsyncMock),
1673        patch.object(session, "stop", new_callable=AsyncMock),
1674    ):
1675        await session._cleanup_after_removal(player)
1676
1677    player.set_state_from_stream.assert_not_called()
1678
1679
1680@pytest.mark.asyncio
1681async def test_late_join_pads_with_silence_when_the_ring_ran_out_under_a_committed_anchor() -> None:
1682    """A due position lost to the ring after the START is covered with silence, not a moved anchor."""
1683    # Freeze time so both the test and the code under test agree on `now`.
1684    now = 1_000_000.0
1685    start_time = now - 100.0
1686    session = _make_session(start_time, 110.0)
1687    session._pcm_total_fed = int(110.0 * PCM_SAMPLE_SIZE)
1688    # An 8s ring against a 10s write-head lead: wide enough when the anchor is
1689    # planned, too narrow by the time the binary owns the instant.
1690    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 8)
1691    session._pcm_buffer_max = PCM_SAMPLE_SIZE * 8
1692    # Freeze the adaptive sizing so this test covers the cap, not the growth.
1693    session._peak_lead_seconds = 1e6
1694    logger = MagicMock()
1695    session.prov.logger = logger
1696    player = _make_late_joiner()
1697
1698    written: list[tuple[str, bytes]] = []
1699
1700    async def capture_write(target: Any, chunk: bytes) -> None:
1701        written.append((target.player_id, chunk))
1702
1703    def setup_with_feed(*_args: Any, **_kwargs: Any) -> None:
1704        _setup_stream(player)()
1705
1706        async def start(start_unix_ms: int, _position_ms: int, *, join: bool = False) -> int:
1707            assert join is True
1708            # The group keeps being fed while the START ack is outstanding: this
1709            # is what pushes the joiner's due position off the back of the ring.
1710            await session._write_chunk_to_all_players(b"\x02" * int(2.0 * PCM_SAMPLE_SIZE))
1711            return start_unix_ms
1712
1713        player.stream.start = AsyncMock(side_effect=start)
1714
1715    with (
1716        patch.object(session, "_start_client", side_effect=setup_with_feed),
1717        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1718        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1719    ):
1720        await session.add_client(player)
1721
1722    # The anchor is the one that was commanded: an acked instant is never moved.
1723    headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1724    assert _captured_start_at(player) == pytest.approx(now + headroom, abs=0.001)
1725    prime = next(chunk for pid, chunk in written if pid == player.player_id)
1726    # due is 102.5s of feed and the write head reached 112.0s, so 9.5s is owed
1727    # while only 8s survives in the ring: the missing 1.5s opens as silence.
1728    assert len(prime) / PCM_SAMPLE_SIZE == pytest.approx(9.5, abs=0.001)
1729    pad = int(1.5 * PCM_SAMPLE_SIZE)
1730    assert prime[:pad] == bytes(pad), "the missing head must be silence"
1731    assert set(prime[pad:]) == {1, 2}, "the rest must be the buffered feed"
1732    assert pad % session._pcm_frame_size == 0
1733    assert session._client_skip_bytes[player.player_id] == 0
1734    # Position still reports where the GROUP is at that instant, because the
1735    # real content lands exactly where it would have without the shortfall.
1736    player.stream.rebase_position.assert_not_called()
1737    logger.warning.assert_called_once()
1738    assert "silence" in logger.warning.call_args.args[0]
1739
1740
1741@pytest.mark.asyncio
1742async def test_late_join_ring_shortfall_keeps_a_misaligned_ring_head_frame_aligned() -> None:
1743    """A silence pad over a mid-frame ring head still lands the feed on frame boundaries."""
1744    now = 1_000_000.0
1745    session = _make_session(now - 100.0, 110.0)
1746    # Both the write head and the ring head sit mid-frame.
1747    session._pcm_total_fed = int(110.0 * PCM_SAMPLE_SIZE) + 3
1748    session._pcm_buffer = bytearray(b"\x01" * (PCM_SAMPLE_SIZE * 8 + 2))
1749    session._pcm_buffer_max = len(session._pcm_buffer)
1750    session._peak_lead_seconds = 1e6
1751    player = _make_late_joiner()
1752
1753    written: list[tuple[str, bytes]] = []
1754
1755    async def capture_write(target: Any, chunk: bytes) -> None:
1756        written.append((target.player_id, chunk))
1757
1758    def setup_with_feed(*_args: Any, **_kwargs: Any) -> None:
1759        _setup_stream(player)()
1760
1761        async def start(start_unix_ms: int, _position_ms: int, *, join: bool = False) -> int:
1762            assert join is True
1763            await session._write_chunk_to_all_players(b"\x02" * (int(2.0 * PCM_SAMPLE_SIZE) + 1))
1764            return start_unix_ms
1765
1766        player.stream.start = AsyncMock(side_effect=start)
1767
1768    with (
1769        patch.object(session, "_start_client", side_effect=setup_with_feed),
1770        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1771        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1772    ):
1773        await session.add_client(player)
1774
1775    prime = next(chunk for pid, chunk in written if pid == player.player_id)
1776    frame_size = session._pcm_frame_size
1777    # The prime ends exactly at the write head, so its start - and therefore the
1778    # whole padded prime - has to sit on an absolute frame boundary.
1779    assert (session._pcm_total_fed - len(prime)) % frame_size == 0
1780    pad_len = len(prime) - len(prime.lstrip(b"\x00"))
1781    assert pad_len % frame_size == 0
1782
1783
1784def test_ring_grows_to_the_observed_write_head_lead() -> None:
1785    """The ring tracks the largest lead a session shows, above a floor and under a byte cap."""
1786    now = 1_000_000.0
1787    session = _make_session(now - 100.0, 100.0)
1788    assert session._pcm_buffer_max == int(AIRPLAY_LATE_JOIN_RING_MIN_SECONDS * PCM_SAMPLE_SIZE)
1789
1790    with patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now):
1791        # A 4s lead stays under the floor, which keeps the ring where it was.
1792        session.seconds_streamed = 104.0
1793        session._observe_write_head_lead()
1794        assert session._pcm_buffer_max == int(AIRPLAY_LATE_JOIN_RING_MIN_SECONDS * PCM_SAMPLE_SIZE)
1795
1796        # A 15s lead carries the ring past the floor, with the margin on top.
1797        session.seconds_streamed = 115.0
1798        session._observe_write_head_lead()
1799        assert session._peak_lead_seconds == pytest.approx(15.0, abs=0.001)
1800        expected = int((15.0 + AIRPLAY_LATE_JOIN_RING_MARGIN_SECONDS) * PCM_SAMPLE_SIZE)
1801        assert session._pcm_buffer_max == expected
1802
1803        # A lead that falls back never shrinks the ring: the history a joiner
1804        # still needs is already in it.
1805        session.seconds_streamed = 108.0
1806        session._observe_write_head_lead()
1807        assert session._pcm_buffer_max == expected
1808
1809        # Growth is bounded in bytes, so a hi-res rate cannot multiply it out.
1810        session.seconds_streamed = 100.0 + 3600.0
1811        session._observe_write_head_lead()
1812        assert session._pcm_buffer_max == AIRPLAY_LATE_JOIN_RING_MAX_BYTES
1813
1814
1815def test_write_head_lead_is_not_measured_before_the_anchor_arrives() -> None:
1816    """Audio fed inside the start lead is not counted as pipeline depth."""
1817    now = 1_000_000.0
1818    # Anchored 2.5s into the future: nothing is audible yet.
1819    session = _make_session(now + 2.5, 8.0)
1820    with patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now):
1821        session._observe_write_head_lead()
1822    assert session._peak_lead_seconds == 0.0
1823    assert session._pcm_buffer_max == int(AIRPLAY_LATE_JOIN_RING_MIN_SECONDS * PCM_SAMPLE_SIZE)
1824
1825    unanchored = _make_session(0.0, 8.0)
1826    with patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now):
1827        unanchored._observe_write_head_lead()
1828    assert unanchored._peak_lead_seconds == 0.0
1829
1830
1831@pytest.mark.asyncio
1832async def test_late_join_silence_pad_is_bounded_and_reports_the_residual() -> None:
1833    """An implausible ack is padded only up to the ring bound, and says so."""
1834    now = 1_000_000.0
1835    session = _make_session(now - 400.0, 420.0)
1836    session._pcm_total_fed = int(420.0 * PCM_SAMPLE_SIZE)
1837    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 2)
1838    session._pcm_buffer_max = PCM_SAMPLE_SIZE * 2
1839    session._peak_lead_seconds = 1e6
1840    logger = MagicMock()
1841    session.prov.logger = logger
1842    player = _make_late_joiner()
1843
1844    written: list[tuple[str, bytes]] = []
1845
1846    async def capture_write(target: Any, chunk: bytes) -> None:
1847        written.append((target.player_id, chunk))
1848
1849    def setup_with_stale_ack(*_args: Any, **_kwargs: Any) -> None:
1850        _setup_stream(player)()
1851        # The binary reports an instant far behind the commanded one, mapping
1852        # the joiner back near the start of the session. 320s of head is owed.
1853        player.stream.start = AsyncMock(return_value=int((now - 300.0) * 1000))
1854
1855    with (
1856        patch.object(session, "_start_client", side_effect=setup_with_stale_ack),
1857        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1858        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1859    ):
1860        await session.add_client(player)
1861
1862    prime = next(chunk for pid, chunk in written if pid == player.player_id)
1863    # Bounded at one ring of silence plus the ring itself - never the 320s owed.
1864    assert len(prime) / PCM_SAMPLE_SIZE == pytest.approx(4.0, abs=0.01)
1865    assert (session._pcm_total_fed - len(prime)) % session._pcm_frame_size == 0
1866    # The joiner cannot be placed exactly, so the log must not claim sync.
1867    logger.warning.assert_called_once()
1868    tail = logger.warning.call_args.args[-1]
1869    assert "ahead of the group" in tail
1870    assert "in sync" not in tail
1871