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