music-assistant-server

1"""Unit tests for AI Radio runtime session flow and logging."""
2
3from __future__ import annotations
4
5import asyncio
6import json
7import logging
8import random
9from collections.abc import Awaitable, Callable
10from contextlib import suppress
11from copy import deepcopy
12from pathlib import Path
13from types import SimpleNamespace
14from typing import Any, cast
15from unittest.mock import AsyncMock, MagicMock
16
17import pytest
18from music_assistant_models.enums import (
19    EventType,
20    MediaType,
21    PlaybackState,
22    ProviderFeature,
23    ProviderType,
24)
25from music_assistant_models.errors import MusicAssistantError
26from music_assistant_models.event import MassEvent
27from music_assistant_models.media_items import ProviderMapping, Track
28
29from music_assistant.helpers.datetime import now as host_now
30from music_assistant.models.plugin import AIEngine, PluginProvider, TTSEngine
31from music_assistant.providers.ai_radio import runtime as runtime_module
32from music_assistant.providers.ai_radio.constants import (
33    ATTR_HOST_ID,
34    ATTR_MAX_CHARS,
35    ATTR_PROMPT,
36    ATTR_SESSION_ID,
37    ATTR_STATION_ID,
38    ATTR_WEB_SEARCH_MODE,
39    CONF_AI_ENGINE,
40    CONF_TTS_ENGINE,
41    CONF_WEATHER_PROVIDER,
42    TTS_PRONUNCIATION_INSTRUCTIONS,
43)
44from music_assistant.providers.ai_radio.models import PlannedSection, SessionState, Slot
45from music_assistant.providers.ai_radio.queue_dj import AIRadioQueueDJMixin
46from music_assistant.providers.ai_radio.runtime import AIRadioRuntimeMixin
47from music_assistant.providers.ai_radio.storage import AIRadioStorageMixin
48
49
50class StubConfig:
51    """Minimal ProviderConfig stand-in exposing get_value."""
52
53    def __init__(self, values: dict[str, Any] | None = None) -> None:
54        """Initialize with an optional map of config key -> value."""
55        self._values = values or {}
56
57    def get_value(self, key: str, default: Any = None) -> Any:
58        """Return the stubbed value for key, or default when absent."""
59        return self._values.get(key, default)
60
61
62class DummyRuntime(AIRadioRuntimeMixin):
63    """Minimal runtime harness for testing mixin behavior."""
64
65    def __init__(self, setup_values: dict[str, Any] | None = None) -> None:
66        """Initialize minimal state for runtime tests."""
67        self.logger = logging.getLogger("tests.ai_radio.runtime")
68        self._sessions: dict[str, SessionState] = {}
69        self._sections: dict[str, dict[str, Any]] = {}
70        self.config = cast("Any", StubConfig())
71        self.instance_id = "ai_radio_test"
72        self.domain = "ai_radio"
73        self._setup_values = setup_values or {}
74
75    def get_setup_value(self, key: str, default: Any = None) -> Any:
76        """Return the stubbed setup flow value for key, or default when absent."""
77        return self._setup_values.get(key, default)
78
79    def _schedule_replan(self, queue_id: str) -> None:
80        """No-op stand-in for the queue DJ mixin's replan scheduling."""
81
82    async def set_queue_dj(self, queue_id: str, host_id: str | None) -> dict[str, str]:
83        """No-op stand-in for the queue DJ mixin's set_queue_dj."""
84        return {}
85
86    def _materialize_sections(
87        self, section_ids: list[str], sections_map: dict[str, dict[str, Any]] | None = None
88    ) -> tuple[list[dict[str, Any]], list[str]]:
89        """Resolve section ids against self._sections, mirroring the storage mixin."""
90        source = self._sections if sections_map is None else sections_map
91        sections: list[dict[str, Any]] = []
92        missing: list[str] = []
93        for section_id in section_ids:
94            section = source.get(section_id)
95            if section is None:
96                missing.append(section_id)
97                continue
98            sections.append(deepcopy(section))
99        return sections, missing
100
101
102class FailingRuntime(DummyRuntime):
103    """Runtime harness that forces show execution failure."""
104
105    async def _run_show(
106        self,
107        session: SessionState,
108        station: dict[str, Any],
109    ) -> dict[str, Any]:
110        """Raise to test failed-session behavior."""
111        raise RuntimeError("boom")
112
113
114def _set_runtime_mass(runtime: AIRadioRuntimeMixin, mass: Any) -> None:
115    """Attach lightweight test mass object while bypassing strict runtime typing."""
116    cast("Any", runtime).mass = mass
117
118
119def _create_ai_plugin(instance_id: str, *engine_ids: str) -> MagicMock:
120    """Create a mock plugin provider exposing the given AI engines."""
121    provider = MagicMock(spec=PluginProvider)
122    provider.instance_id = instance_id
123    provider.get_ai_engines = AsyncMock(
124        return_value=[
125            AIEngine(id=engine_id, name=engine_id, provider=provider) for engine_id in engine_ids
126        ]
127    )
128    return provider
129
130
131def _create_tts_plugin(instance_id: str, *engine_ids: str) -> MagicMock:
132    """Create a mock plugin provider exposing the given TTS engines."""
133    provider = MagicMock(spec=PluginProvider)
134    provider.instance_id = instance_id
135    provider.get_tts_engines = AsyncMock(
136        return_value=[
137            TTSEngine(id=engine_id, name=engine_id, provider=provider) for engine_id in engine_ids
138        ]
139    )
140    return provider
141
142
143def _create_engine_mass(feature: ProviderFeature, *providers: Any, **attrs: Any) -> Any:
144    """Create a lightweight mass stand-in serving the given plugins for one feature."""
145
146    class DummyMass:
147        def get_providers_supporting_feature(
148            self,
149            requested: ProviderFeature,
150            priority: tuple[ProviderType, ...] = (),
151        ) -> list[Any]:
152            return list(providers) if requested == feature else []
153
154    mass = DummyMass()
155    for key, value in attrs.items():
156        setattr(mass, key, value)
157    return mass
158
159
160async def test_run_session_sets_completed_and_logs(caplog: Any) -> None:
161    """Complete a session and emit start/completion logs."""
162
163    class SuccessfulRuntime(DummyRuntime):
164        async def _run_show(
165            self,
166            session: SessionState,
167            station: dict[str, Any],
168        ) -> dict[str, Any]:
169            """Return a successful show run result."""
170            return {"ok": True}
171
172    runtime = SuccessfulRuntime()
173    session = SessionState(session_id="s1", station_id="station_a")
174    runtime._sessions[session.session_id] = session
175
176    with caplog.at_level(logging.INFO):
177        await runtime._run_session(session.session_id, {"id": "station_a"})
178
179    assert session.status == "completed"
180    assert session.result == {"ok": True}
181    assert any("AI Radio run started" in message for message in caplog.messages)
182    assert any("AI Radio run completed" in message for message in caplog.messages)
183
184
185async def test_run_session_sets_failed_state(caplog: Any) -> None:
186    """Fail a session and keep the error message in state."""
187    runtime = FailingRuntime()
188    session = SessionState(session_id="s2", station_id="station_b")
189    runtime._sessions[session.session_id] = session
190
191    with caplog.at_level(logging.ERROR):
192        await runtime._run_session(session.session_id, {"id": "station_b"})
193
194    assert session.status == "failed"
195    assert session.error == "boom"
196
197
198async def test_run_session_sets_failed_state_with_empty_exception_message() -> None:
199    """Store exception class name when failure has no message."""
200
201    class EmptyError(Exception):
202        """Exception with empty default message."""
203
204    class EmptyFailingRuntime(DummyRuntime):
205        async def _run_show(
206            self,
207            session: SessionState,
208            station: dict[str, Any],
209        ) -> dict[str, Any]:
210            raise EmptyError
211
212    runtime = EmptyFailingRuntime()
213    session = SessionState(session_id="s2b", station_id="station_b")
214    runtime._sessions[session.session_id] = session
215
216    await runtime._run_session(session.session_id, {"id": "station_b"})
217
218    assert session.status == "failed"
219    assert session.error == "EmptyError"
220
221
222async def test_run_session_sets_stopped_state_on_cancellation(caplog: Any) -> None:
223    """Mark session as stopped when runtime execution is cancelled."""
224
225    class CancelledRuntime(DummyRuntime):
226        async def _run_show(
227            self,
228            session: SessionState,
229            station: dict[str, Any],
230        ) -> dict[str, Any]:
231            raise asyncio.CancelledError
232
233    runtime = CancelledRuntime()
234    session = SessionState(session_id="s3", station_id="station_c")
235    runtime._sessions[session.session_id] = session
236
237    with caplog.at_level(logging.INFO), suppress(asyncio.CancelledError):
238        await runtime._run_session(session.session_id, {"id": "station_c"})
239
240    assert session.status == "stopped"
241    assert any("AI Radio run cancelled" in message for message in caplog.messages)
242
243
244async def test_run_session_reschedules_a_replan_for_the_session_queue() -> None:
245    """Re-arm the queue DJ for the session's queue once its show session ends."""
246
247    class ReplanTrackingRuntime(FailingRuntime):
248        def __init__(self) -> None:
249            super().__init__()
250            self.replanned_queue_ids: list[str] = []
251
252        def _schedule_replan(self, queue_id: str) -> None:
253            self.replanned_queue_ids.append(queue_id)
254
255    runtime = ReplanTrackingRuntime()
256    session = SessionState(session_id="s4", station_id="station_d")
257    session.queue_id = "queue_1"
258    runtime._sessions[session.session_id] = session
259
260    await runtime._run_session(session.session_id, {"id": "station_d"})
261
262    assert runtime.replanned_queue_ids == ["queue_1"]
263
264
265async def test_prepare_runtime_tokens_logs_unsupported_weather_provider(caplog: Any) -> None:
266    """Warn when weather placeholders are used with unsupported provider."""
267    runtime = DummyRuntime()
268    station = {
269        "sections": [
270            {
271                "id": "Weather_Short",
272                "type": "ai_text",
273                "prompt": "Forecast: <weather_hourly>",
274            }
275        ],
276        "section_order": [],
277    }
278    runtime.config = cast(
279        "Any",
280        StubConfig(
281            {
282                "weather_city": "Berlin",
283                "weather_country": "DE",
284                CONF_WEATHER_PROVIDER: "unsupported_provider",
285            }
286        ),
287    )
288
289    with caplog.at_level(logging.WARNING):
290        tokens = await runtime._prepare_runtime_tokens(station)
291
292    assert tokens == {}
293    assert any("Unsupported weather provider" in message for message in caplog.messages)
294
295
296def _weather_program() -> dict[str, Any]:
297    """Return a program whose only section references the hourly weather token."""
298    return {
299        "sections": [
300            {
301                "id": "Weather_Short",
302                "type": "ai_text",
303                "prompt": "Forecast: <weather_hourly>",
304            }
305        ],
306        "section_order": [],
307    }
308
309
310def _count_weather_fetches(runtime: DummyRuntime) -> list[str]:
311    """Replace the forecast lookup with a stub and return the list it records into."""
312    fetches: list[str] = []
313
314    async def _fetch(city: str, **_kwargs: Any) -> tuple[str, str]:
315        fetches.append(city)
316        return "12 degrees", "mild"
317
318    runtime._fetch_open_meteo_weather = _fetch  # type: ignore[method-assign, assignment]
319    return fetches
320
321
322async def test_prepare_runtime_tokens_reuses_the_cached_weather_within_the_ttl() -> None:
323    """A second pass inside the cache window reuses the tokens instead of refetching."""
324    runtime = DummyRuntime()
325    runtime.config = cast("Any", StubConfig({"weather_city": "Berlin", "weather_country": "DE"}))
326    fetches = _count_weather_fetches(runtime)
327
328    first = await runtime._prepare_runtime_tokens(_weather_program())
329    second = await runtime._prepare_runtime_tokens(_weather_program())
330
331    assert first == {"<weather_hourly>": "12 degrees", "<weather_daily>": "mild"}
332    assert second == first
333    assert fetches == ["Berlin"]
334
335
336async def test_prepare_runtime_tokens_refetches_the_weather_once_the_ttl_expired() -> None:
337    """An expired cache entry is refetched rather than served stale forever."""
338    runtime = DummyRuntime()
339    runtime.config = cast("Any", StubConfig({"weather_city": "Berlin", "weather_country": "DE"}))
340    fetches = _count_weather_fetches(runtime)
341
342    await runtime._prepare_runtime_tokens(_weather_program())
343    assert runtime._weather_tokens_cache is not None
344    fetched_at, tokens = runtime._weather_tokens_cache
345    runtime._weather_tokens_cache = (
346        fetched_at - runtime_module.WEATHER_TOKENS_CACHE_SECONDS - 1,
347        tokens,
348    )
349    await runtime._prepare_runtime_tokens(_weather_program())
350
351    assert fetches == ["Berlin", "Berlin"]
352
353
354def test_weather_strings_are_rounded_to_whole_numbers() -> None:
355    """A host reads the forecast out loud, so it says 19 degrees and never 19.2."""
356    runtime = DummyRuntime()
357    payload = {
358        "current": {
359            "time": "2026-08-10T09:00",
360            "temperature_2m": 19.2,
361            "apparent_temperature": 18.7,
362        },
363        "hourly": {
364            "time": ["2026-08-10T09:00", "2026-08-10T10:00"],
365            "temperature_2m": [19.2, 20.6],
366            "precipitation_probability": [12.4, 0],
367        },
368        "daily": {
369            "time": ["2026-08-10"],
370            "temperature_2m_min": [11.4],
371            "temperature_2m_max": [21.49],
372            "precipitation_probability_max": [30.6],
373        },
374    }
375
376    hourly, daily = runtime._format_weather_strings(payload)
377
378    assert hourly == (
379        "now 19C (feels 19C); 2026-08-10 09:00: 19C, rain 12%; 2026-08-10 10:00: 21C, rain 0%"
380    )
381    assert daily == "2026-08-10: 11-21C, rain 31%"
382
383
384async def test_prepare_runtime_tokens_ignores_missing_location(caplog: Any) -> None:
385    """Skip weather preparation when the configured location is incomplete."""
386    runtime = DummyRuntime()
387    station = {
388        "sections": [
389            {
390                "id": "Weather_Short",
391                "type": "ai_text",
392                "prompt": "Forecast: <weather_hourly>",
393            }
394        ],
395        "section_order": [],
396    }
397    runtime.config = cast("Any", StubConfig({"weather_city": "", "weather_country": "DE"}))
398
399    with caplog.at_level(logging.DEBUG):
400        tokens = await runtime._prepare_runtime_tokens(station)
401
402    assert tokens == {}
403    assert any("no location configured" in message for message in caplog.messages)
404
405
406def test_extract_location_reads_provider_config() -> None:
407    """Weather location comes from the provider config, not the station."""
408    runtime = DummyRuntime()
409    runtime.config = cast("Any", StubConfig({"weather_city": "Berlin", "weather_country": "DE"}))
410
411    assert runtime._extract_location() == ("Berlin", "DE")
412
413
414def test_extract_location_defaults_to_empty_when_unset() -> None:
415    """An unconfigured weather location resolves to empty strings, not an error."""
416    runtime = DummyRuntime()
417
418    assert runtime._extract_location() == ("", "")
419
420
421@pytest.mark.parametrize("timezone_value", ["Asia/Tokyo", "  Asia/Tokyo  "])
422def test_configured_now_uses_valid_configured_timezone(timezone_value: str) -> None:
423    """A valid configured IANA timezone name is honored, surrounding whitespace included."""
424    runtime = DummyRuntime()
425    runtime.config = cast("Any", StubConfig({"timezone": timezone_value}))
426
427    result = runtime._configured_now()
428
429    assert str(result.tzinfo) == "Asia/Tokyo"
430
431
432@pytest.mark.parametrize(
433    "timezone_value",
434    ["", "not-a-real-zone", "CEST", "../../etc/passwd"],
435)
436def test_configured_now_falls_back_when_timezone_blank_or_invalid(timezone_value: str) -> None:
437    """A blank or invalid configured timezone falls back to the host local time."""
438    runtime = DummyRuntime()
439    runtime.config = cast("Any", StubConfig({"timezone": timezone_value}))
440
441    result = runtime._configured_now()
442
443    assert result.utcoffset() == host_now().utcoffset()
444
445
446def test_plan_sections_ignores_invalid_optional_chance() -> None:
447    """Treat non-numeric OPTIONAL chance values as zero during planning."""
448    runtime = DummyRuntime()
449    station = {
450        "sections": [
451            {
452                "id": "Song_Transition",
453                "name": "Song Transition",
454                "type": "ai_text",
455                "prompt": "Transition from <prev_songinfo> to <next_songinfo>",
456            }
457        ],
458        "section_order": [
459            {
460                "when": "between_songs",
461                "flow": [
462                    {
463                        "OPTIONAL": {
464                            "section": "Song_Transition",
465                            "chance": "not-a-number",
466                        }
467                    }
468                ],
469            }
470        ],
471        "general": {"timezone": "UTC"},
472    }
473    tracks = [
474        {"name": "A", "artist": "Artist A", "songinfo": "Artist A - A", "duration": 180},
475        {"name": "B", "artist": "Artist B", "songinfo": "Artist B - B", "duration": 180},
476    ]
477
478    planned, _history = runtime._plan_sections(
479        session_id="sess",
480        tracks=tracks,
481        program=station,
482        track_index_offset=0,
483        minute_offset=0.0,
484        history_state={},
485        allowed_slot_when=["between_songs"],
486        runtime_tokens={},
487    )
488
489    assert planned == []
490
491
492async def test_generate_text_wraps_not_connected_error() -> None:
493    """Raise an actionable MusicAssistantError when the AI engine is disconnected."""
494
495    class NotConnected(Exception):
496        """Match hass_client NotConnected exception name."""
497
498    plugin = _create_ai_plugin("hass_1", "ai_task.default")
499    plugin.ai_query = AsyncMock(side_effect=NotConnected)
500    runtime = DummyRuntime({CONF_AI_ENGINE: "hass_1/ai_task.default"})
501    _set_runtime_mass(
502        runtime,
503        _create_engine_mass(
504            ProviderFeature.AI_QUERY, plugin, metadata=SimpleNamespace(locale="en_US")
505        ),
506    )
507
508    with pytest.raises(MusicAssistantError) as error:
509        await runtime._generate_text(
510            instructions="test",
511            prompt="test prompt",
512            web_mode="disabled",
513        )
514    assert "not connected" in str(error.value).lower()
515    assert "hass_1/ai_task.default" in str(error.value)
516
517
518async def test_generate_text_fails_the_section_when_the_engine_stalls(
519    monkeypatch: pytest.MonkeyPatch,
520) -> None:
521    """A stalled AI engine fails the section instead of hanging the session."""
522    monkeypatch.setattr("music_assistant.providers.ai_radio.runtime.AI_QUERY_TIMEOUT_SECONDS", 0.01)
523
524    async def _answers_too_late(*_args: Any, **_kwargs: Any) -> str:
525        await asyncio.sleep(5)
526        return "section text"
527
528    plugin = _create_ai_plugin("hass_1", "ai_task.default")
529    plugin.ai_query = AsyncMock(side_effect=_answers_too_late)
530    runtime = DummyRuntime({CONF_AI_ENGINE: "hass_1/ai_task.default"})
531    _set_runtime_mass(
532        runtime,
533        _create_engine_mass(
534            ProviderFeature.AI_QUERY, plugin, metadata=SimpleNamespace(locale="en_US")
535        ),
536    )
537
538    with pytest.raises(MusicAssistantError) as error:
539        await runtime._generate_text(
540            instructions="test",
541            prompt="test prompt",
542            web_mode="disabled",
543        )
544    assert "did not respond within" in str(error.value)
545
546
547async def test_generate_text_reports_an_engine_side_timeout_as_a_query_failure() -> None:
548    """A timeout raised by the engine itself is reported as a query failure, not our cap."""
549    plugin = _create_ai_plugin("hass_1", "ai_task.default")
550    plugin.ai_query = AsyncMock(side_effect=TimeoutError)
551    runtime = DummyRuntime({CONF_AI_ENGINE: "hass_1/ai_task.default"})
552    _set_runtime_mass(
553        runtime,
554        _create_engine_mass(
555            ProviderFeature.AI_QUERY, plugin, metadata=SimpleNamespace(locale="en_US")
556        ),
557    )
558
559    with pytest.raises(MusicAssistantError) as error:
560        await runtime._generate_text(
561            instructions="test",
562            prompt="test prompt",
563            web_mode="disabled",
564        )
565    assert "query failed: TimeoutError" in str(error.value)
566
567
568async def test_generate_text_asks_for_the_system_locale_language() -> None:
569    """The AI query states the server locale so sections are written in that language."""
570    plugin = _create_ai_plugin("hass_1", "ai_task.default")
571    plugin.ai_query = AsyncMock(return_value="section text")
572    runtime = DummyRuntime({CONF_AI_ENGINE: "hass_1/ai_task.default"})
573    _set_runtime_mass(
574        runtime,
575        _create_engine_mass(
576            ProviderFeature.AI_QUERY, plugin, metadata=SimpleNamespace(locale="nl_NL")
577        ),
578    )
579
580    await runtime._generate_text(
581        instructions="test",
582        prompt="test prompt",
583        web_mode="disabled",
584    )
585
586    assert "nl_NL" in plugin.ai_query.await_args.args[0]
587    assert plugin.ai_query.await_args.kwargs == {"engine_id": "ai_task.default"}
588
589
590async def test_generate_text_prefers_the_hosts_language_over_the_system_locale() -> None:
591    """An explicit host language wins over the server locale in the AI query."""
592    plugin = _create_ai_plugin("hass_1", "ai_task.default")
593    plugin.ai_query = AsyncMock(return_value="section text")
594    runtime = DummyRuntime({CONF_AI_ENGINE: "hass_1/ai_task.default"})
595    _set_runtime_mass(
596        runtime,
597        _create_engine_mass(
598            ProviderFeature.AI_QUERY, plugin, metadata=SimpleNamespace(locale="nl_NL")
599        ),
600    )
601
602    await runtime._generate_text(
603        instructions="test",
604        prompt="test prompt",
605        web_mode="disabled",
606        language="fr_FR",
607    )
608
609    assert "fr_FR" in plugin.ai_query.await_args.args[0]
610    assert "nl_NL" not in plugin.ai_query.await_args.args[0]
611
612
613async def test_generate_text_falls_back_to_the_system_locale_when_language_is_empty() -> None:
614    """An unset host language keeps asking for the server locale, exactly as before."""
615    plugin = _create_ai_plugin("hass_1", "ai_task.default")
616    plugin.ai_query = AsyncMock(return_value="section text")
617    runtime = DummyRuntime({CONF_AI_ENGINE: "hass_1/ai_task.default"})
618    _set_runtime_mass(
619        runtime,
620        _create_engine_mass(
621            ProviderFeature.AI_QUERY, plugin, metadata=SimpleNamespace(locale="nl_NL")
622        ),
623    )
624
625    await runtime._generate_text(
626        instructions="test",
627        prompt="test prompt",
628        web_mode="disabled",
629        language="",
630    )
631
632    assert "nl_NL" in plugin.ai_query.await_args.args[0]
633
634
635@pytest.mark.parametrize("general", [{"instructions": "Host personality: minimal DJ."}, {}])
636async def test_generate_text_always_states_the_pronunciation_rules(
637    general: dict[str, Any],
638) -> None:
639    """Every query carries the TTS pronunciation rules, with or without station instructions."""
640    plugin = _create_ai_plugin("hass_1", "ai_task.default")
641    plugin.ai_query = AsyncMock(return_value="section text")
642    runtime = DummyRuntime({CONF_AI_ENGINE: "hass_1/ai_task.default"})
643    _set_runtime_mass(
644        runtime,
645        _create_engine_mass(
646            ProviderFeature.AI_QUERY, plugin, metadata=SimpleNamespace(locale="en_US")
647        ),
648    )
649
650    await runtime._generate_text(
651        instructions=str(general.get("instructions", "")), prompt="test prompt", web_mode="allow"
652    )
653
654    assert TTS_PRONUNCIATION_INSTRUCTIONS in plugin.ai_query.await_args.args[0]
655
656
657def test_resolve_placeholders_keeps_time_and_weather_deferred() -> None:
658    """Static track placeholders resolve at plan time; time and weather stay deferred."""
659    runtime = DummyRuntime()
660    _set_runtime_mass(runtime, SimpleNamespace(metadata=SimpleNamespace(locale="en_US")))
661    tracks = [
662        {"index": 0, "songinfo": "A - One", "duration": 200},
663        {"index": 1, "songinfo": "B - Two", "duration": 200},
664    ]
665    slot = Slot(
666        when="between_songs",
667        at_index=1,
668        prev_index=0,
669        next_index=1,
670        very_next_index=None,
671        minute_mark=3.3,
672    )
673
674    static, deferred = runtime._resolve_placeholders(
675        program={},
676        tracks=tracks,
677        slot=slot,
678        runtime_tokens={"<weather_hourly>": "12 degrees"},
679    )
680
681    assert static["<prev_songinfo>"] == "A - One"
682    assert static["<next_songinfo>"] == "B - Two"
683    assert "<timestamp>" not in static
684    assert "<weather_hourly>" not in static
685    assert deferred["<weather_hourly>"] == "12 degrees"
686    assert "<timestamp>" in deferred
687
688
689def test_plan_sections_leaves_deferred_tokens_in_the_prompt() -> None:
690    """A planned section's prompt keeps its deferred tokens verbatim."""
691    runtime = DummyRuntime()
692    _set_runtime_mass(runtime, SimpleNamespace(metadata=SimpleNamespace(locale="en_US")))
693    station = {
694        "sections": [
695            {
696                "id": "Weather",
697                "name": "Weather",
698                "prompt": "It is <timestamp>. Weather: <weather_hourly>. Next: <next_songinfo>.",
699                "constraints": {"max_chars": 300},
700                "web_search": "disabled",
701            }
702        ],
703        "section_order": [{"when": "between_songs", "flow": [{"MUST": "Weather"}]}],
704    }
705    tracks = [
706        {"index": 0, "songinfo": "A - One", "duration": 200},
707        {"index": 1, "songinfo": "B - Two", "duration": 200},
708    ]
709
710    planned, _history = runtime._plan_sections(
711        session_id="sess",
712        tracks=tracks,
713        program=station,
714        track_index_offset=0,
715        minute_offset=0.0,
716        history_state={},
717        allowed_slot_when=["between_songs"],
718        runtime_tokens={"<weather_hourly>": "12 degrees"},
719    )
720
721    assert planned
722    prompt = planned[0].prompt
723    assert "<timestamp>" in prompt
724    assert "<weather_hourly>" in prompt
725    assert "B - Two" in prompt
726
727
728def _weather_guarded_station() -> dict[str, Any]:
729    """Return a station whose only section requires the weather-hourly token to be present."""
730    return {
731        "sections": [
732            {
733                "id": "Weather",
734                "name": "Weather",
735                "type": "ai_text",
736                "web_search": "disabled",
737                "prompt": "Current weather: <weather_hourly>.",
738                "constraints": {"max_chars": 200},
739            }
740        ],
741        "section_order": [
742            {
743                "when": "between_songs",
744                "flow": [
745                    {
746                        "OPTIONAL": {
747                            "section": "Weather",
748                            "chance": 100,
749                            "guards": {"require_placeholders_present": ["<weather_hourly>"]},
750                        }
751                    }
752                ],
753            }
754        ],
755    }
756
757
758def test_plan_sections_suppresses_section_when_required_placeholder_is_missing() -> None:
759    """A guarded section plans zero entries when its required placeholder never resolved."""
760    runtime = DummyRuntime()
761    _set_runtime_mass(runtime, SimpleNamespace(metadata=SimpleNamespace(locale="en_US")))
762    tracks = [
763        {"index": 0, "songinfo": "A - One", "duration": 200},
764        {"index": 1, "songinfo": "B - Two", "duration": 200},
765    ]
766
767    planned, _history = runtime._plan_sections(
768        session_id="sess",
769        tracks=tracks,
770        program=_weather_guarded_station(),
771        track_index_offset=0,
772        minute_offset=0.0,
773        history_state={},
774        allowed_slot_when=["between_songs"],
775        runtime_tokens={},
776    )
777
778    assert planned == []
779
780
781def test_plan_sections_includes_section_when_required_placeholder_is_present() -> None:
782    """The same guarded section plans normally once its required placeholder resolved."""
783    runtime = DummyRuntime()
784    _set_runtime_mass(runtime, SimpleNamespace(metadata=SimpleNamespace(locale="en_US")))
785    tracks = [
786        {"index": 0, "songinfo": "A - One", "duration": 200},
787        {"index": 1, "songinfo": "B - Two", "duration": 200},
788    ]
789
790    planned, _history = runtime._plan_sections(
791        session_id="sess",
792        tracks=tracks,
793        program=_weather_guarded_station(),
794        track_index_offset=0,
795        minute_offset=0.0,
796        history_state={},
797        allowed_slot_when=["between_songs"],
798        runtime_tokens={"<weather_hourly>": "12 degrees"},
799    )
800
801    assert len(planned) == 1
802    assert planned[0].section_id == "Weather"
803
804
805def _stub_track(item_id: str) -> Track:
806    """Build a minimal Track with one available ProviderMapping, for build_queue_item."""
807    return Track(
808        item_id=item_id,
809        provider="library",
810        name=f"Track {item_id}",
811        provider_mappings={
812            ProviderMapping(
813                item_id=item_id,
814                provider_domain="library",
815                provider_instance="library",
816            )
817        },
818    )
819
820
821def test_compose_queue_items_places_clips_at_planned_indices() -> None:
822    """Clips are interleaved at their planned indices, carrying their render state."""
823    runtime = DummyRuntime()
824    _set_runtime_mass(runtime, SimpleNamespace())
825    tracks = [
826        {"index": 0, "uri": "library://track/1", "media_item": _stub_track("1")},
827        {"index": 1, "uri": "library://track/2", "media_item": _stub_track("2")},
828    ]
829    sections = [
830        PlannedSection(
831            order=0,
832            clip_id="sess_000",
833            section_id="Intro",
834            section_name="Intro",
835            when="start_of_playlist",
836            insert_at_index=0,
837            prompt="hello <timestamp>",
838            max_chars=200,
839            web_search_mode="disabled",
840        ),
841        PlannedSection(
842            order=1,
843            clip_id="sess_001",
844            section_id="Between",
845            section_name="Between",
846            when="between_songs",
847            insert_at_index=1,
848            prompt="middle <weather_hourly>",
849            max_chars=200,
850            web_search_mode="allow",
851        ),
852    ]
853
854    items = runtime._compose_queue_items(
855        queue_id="player_a",
856        session=SessionState(session_id="sess", station_id="st"),
857        program={"id": "st"},
858        tracks=tracks,
859        sections=sections,
860    )
861
862    assert [item.media_item.item_id for item in items if item.media_item is not None] == [
863        "sess_000",
864        "1",
865        "sess_001",
866        "2",
867    ]
868    intro = items[0]
869    assert intro.media_item is not None
870    assert intro.media_item.media_type == MediaType.SOUND_EFFECT
871    assert intro.extra_attributes[ATTR_PROMPT] == "hello <timestamp>"
872    assert intro.extra_attributes[ATTR_SESSION_ID] == "sess"
873    assert intro.extra_attributes[ATTR_STATION_ID] == "st"
874    assert intro.extra_attributes[ATTR_MAX_CHARS] == 200
875    assert items[2].extra_attributes[ATTR_WEB_SEARCH_MODE] == "allow"
876    # the section name travels as the item's own name, not as an attribute
877    assert intro.name == "Intro"
878    assert "ai_radio_section_name" not in intro.extra_attributes
879    # track items carry no AI Radio state
880    assert items[1].extra_attributes == {}
881
882
883def test_build_program_merges_host_into_station() -> None:
884    """The merged program carries the host's persona, sections and section_order."""
885    runtime = DummyRuntime()
886    runtime._sections = {
887        "Song_Transition": {
888            "id": "Song_Transition",
889            "name": "Song Transition",
890            "type": "ai_text",
891            "prompt": "Prompt",
892            "web_search": "disabled",
893        }
894    }
895    host = {
896        "id": "rick",
897        "name": "Rick",
898        "instructions": "Persona.",
899        "tts_engine": "engine-1",
900        "language": "fr_FR",
901        "section_ids": ["Song_Transition"],
902        "section_order": [{"when": "between_songs", "flow": [{"MUST": "Song_Transition"}]}],
903        "merge_section_id": "",
904    }
905    station = {
906        "id": "station_a",
907        "name": "Station A",
908        "source_playlist_id": "p1",
909        "source_playlist_provider": "library",
910        "default_player_id": "",
911        "max_duration_minutes": 0.0,
912        "shuffle_source_tracks": True,
913        "host_id": "rick",
914    }
915
916    program = runtime._build_program(station, host)
917
918    assert program["instructions"] == "Persona."
919    assert program["tts_engine"] == "engine-1"
920    assert program["language"] == "fr_FR"
921    assert [s["id"] for s in program["sections"]] == ["Song_Transition"]
922    assert program["section_order"] == host["section_order"]
923    assert program["source_playlist_id"] == "p1"
924
925
926def test_clip_item_carries_host_id() -> None:
927    """A planned clip's queue item stamps both the station id and the host id."""
928    runtime = DummyRuntime()
929    section = PlannedSection(
930        order=0,
931        clip_id="sess_000",
932        section_id="Song_Transition",
933        section_name="Song Transition",
934        when="between_songs",
935        insert_at_index=1,
936        prompt="p",
937        max_chars=0,
938        web_search_mode="disabled",
939    )
940    program = {"id": "station_a", "host_id": "rick"}
941
942    item = runtime._section_to_clip_item("queue-1", "sess", program, section)
943
944    assert item.extra_attributes[ATTR_HOST_ID] == "rick"
945    assert item.extra_attributes[ATTR_SESSION_ID] == "sess"
946
947
948async def test_get_ai_engine_requires_a_configured_selection() -> None:
949    """Without a stored selection no engine is picked, so the run fails with a clear error."""
950    runtime = DummyRuntime()
951    _set_runtime_mass(
952        runtime, _create_engine_mass(ProviderFeature.AI_QUERY, _create_ai_plugin("hass_1", "one"))
953    )
954
955    with pytest.raises(MusicAssistantError, match="No AI engine available"):
956        await runtime._get_ai_engine()
957
958
959async def test_get_ai_engine_uses_the_configured_selection() -> None:
960    """A configured engine uid wins over the first available engine."""
961    high_priority = _create_ai_plugin("zz_high", "engine")
962    low_priority = _create_ai_plugin("aa_low", "engine")
963    runtime = DummyRuntime({CONF_AI_ENGINE: "aa_low/engine"})
964    _set_runtime_mass(
965        runtime, _create_engine_mass(ProviderFeature.AI_QUERY, high_priority, low_priority)
966    )
967
968    assert (await runtime._get_ai_engine()).uid == "aa_low/engine"
969
970
971async def test_get_ai_engine_refuses_a_configured_engine_that_disappeared() -> None:
972    """A concrete AI selection is never silently replaced by another available engine."""
973    runtime = DummyRuntime({CONF_AI_ENGINE: "gone/engine"})
974    _set_runtime_mass(
975        runtime, _create_engine_mass(ProviderFeature.AI_QUERY, _create_ai_plugin("hass_1", "one"))
976    )
977
978    with pytest.raises(MusicAssistantError, match="No AI engine available"):
979        await runtime._get_ai_engine()
980
981
982async def test_get_tts_engine_uses_the_configured_selection() -> None:
983    """The stored TTS uid selects its engine, whatever order the plugins are served in."""
984    high_priority = _create_tts_plugin("zz_high", "engine")
985    low_priority = _create_tts_plugin("aa_low", "engine")
986    runtime = DummyRuntime({CONF_TTS_ENGINE: "aa_low/engine"})
987    _set_runtime_mass(
988        runtime, _create_engine_mass(ProviderFeature.TTS, high_priority, low_priority)
989    )
990
991    assert (await runtime._get_tts_engine()).uid == "aa_low/engine"
992
993
994async def test_get_tts_engine_refuses_a_configured_engine_that_disappeared() -> None:
995    """A concrete TTS selection is never silently replaced by another available engine."""
996    runtime = DummyRuntime({CONF_TTS_ENGINE: "gone/engine"})
997    _set_runtime_mass(
998        runtime, _create_engine_mass(ProviderFeature.TTS, _create_tts_plugin("hass_1", "one"))
999    )
1000
1001    with pytest.raises(MusicAssistantError, match="No text-to-speech engine available"):
1002        await runtime._get_tts_engine()
1003
1004
1005async def test_get_tts_engine_falls_back_to_provider_selection_when_host_uid_is_unresolvable(
1006    caplog: Any,
1007) -> None:
1008    """A host engine_uid that no longer resolves falls back to the provider's TTS selection."""
1009    runtime = DummyRuntime({CONF_TTS_ENGINE: "aa_low/engine"})
1010    _set_runtime_mass(
1011        runtime, _create_engine_mass(ProviderFeature.TTS, _create_tts_plugin("aa_low", "engine"))
1012    )
1013
1014    with caplog.at_level(logging.WARNING):
1015        engine = await runtime._get_tts_engine("gone/engine")
1016
1017    assert engine.uid == "aa_low/engine"
1018    assert any("unavailable" in message for message in caplog.messages)
1019
1020
1021def _show_mass_stub(**handlers: Any) -> SimpleNamespace:
1022    """
1023    Build a minimal mass stub for exercising _run_show.
1024
1025    Any player_queues/players/music/metadata handler not passed gets a no-op default.
1026    """
1027
1028    async def _noop_async(*_args: Any, **_kwargs: Any) -> None:
1029        return None
1030
1031    def _noop_sync(*_args: Any, **_kwargs: Any) -> None:
1032        return None
1033
1034    def _noop_get_active_queue(_player_id: str) -> Any:
1035        return None
1036
1037    def _noop_get_player(_player_id: str) -> Any:
1038        return object()
1039
1040    def _noop_items(_queue_id: str, limit: int = 500, offset: int = 0) -> list[Any]:  # noqa: ARG001
1041        return []
1042
1043    subscribers: list[Callable[[Any], None]] = []
1044
1045    def _recording_subscribe(
1046        cb_func: Callable[[Any], None],
1047        event_filter: Any = None,  # noqa: ARG001
1048        id_filter: Any = None,  # noqa: ARG001
1049    ) -> Callable[[], None]:
1050        subscribers.append(cb_func)
1051
1052        def _unsubscribe() -> None:
1053            subscribers.remove(cb_func)
1054
1055        return _unsubscribe
1056
1057    def _emit_queue_updated(queue_id: str) -> None:
1058        event = MassEvent(event=EventType.QUEUE_UPDATED, object_id=queue_id)
1059        for cb_func in subscribers:
1060            cb_func(event)
1061
1062    def _emit_player_removed(player_id: str) -> None:
1063        event = MassEvent(event=EventType.PLAYER_REMOVED, object_id=player_id)
1064        for cb_func in subscribers:
1065            cb_func(event)
1066
1067    player_queues = SimpleNamespace(
1068        clear=handlers.get("clear", _noop_sync),
1069        get=handlers.get("get", lambda _queue_id: None),
1070        get_active_queue=handlers.get("get_active_queue", _noop_get_active_queue),
1071        set_shuffle=handlers.get("set_shuffle", _noop_async),
1072        load=handlers.get("load", _noop_async),
1073        play_index=handlers.get("play_index", _noop_async),
1074        items=handlers.get("items", _noop_items),
1075        signal_update=handlers.get("signal_update", _noop_sync),
1076        stop=handlers.get("stop", _noop_async),
1077    )
1078    return SimpleNamespace(
1079        player_queues=player_queues,
1080        players=SimpleNamespace(get_player=handlers.get("get_player", _noop_get_player)),
1081        music=SimpleNamespace(playlists=handlers.get("playlists", SimpleNamespace())),
1082        metadata=SimpleNamespace(locale=handlers.get("locale", "en_US")),
1083        create_task=handlers.get("create_task", _noop_sync),
1084        subscribe=handlers.get("subscribe", _recording_subscribe),
1085        emit_queue_updated=_emit_queue_updated,
1086        emit_player_removed=_emit_player_removed,
1087    )
1088
1089
1090def _stub_queue(state: PlaybackState, current_index: int | None) -> SimpleNamespace:
1091    """Build a mutable queue stand-in exposing the fields _await_show_end reads."""
1092    return SimpleNamespace(state=state, current_index=current_index)
1093
1094
1095def _stub_clip_queue_item(clip_id: str, session_id: str) -> SimpleNamespace:
1096    """Build a queue item stand-in whose extra_attributes carry a session id."""
1097    return SimpleNamespace(extra_attributes={ATTR_SESSION_ID: session_id}, item_id=clip_id)
1098
1099
1100def _recording_set_shuffle(log: list[str]) -> Callable[[str, bool], Awaitable[None]]:
1101    """Return an async set_shuffle stub that appends "set_shuffle" to the given call-order log."""
1102
1103    async def _set_shuffle(_queue_id: str, _shuffle_enabled: bool) -> None:
1104        log.append("set_shuffle")
1105
1106    return _set_shuffle
1107
1108
1109def _show_station() -> dict[str, Any]:
1110    """Return a station config for _run_show tests, whose section_order yields clips."""
1111    return {
1112        "id": "st",
1113        "name": "Show Station",
1114        "default_player_id": "living_room",
1115        "source_playlist_id": "playlist-1",
1116        "source_playlist_provider": "library",
1117        "shuffle_source_tracks": False,
1118        "general": {"timezone": "UTC"},
1119        "sections": [
1120            {
1121                "id": "Song_Introduction_Start",
1122                "name": "Intro",
1123                "type": "ai_text",
1124                "web_search": "disabled",
1125                "prompt": "Welcome, next up is <next_songinfo>.",
1126                "constraints": {"max_chars": 200},
1127            },
1128            {
1129                "id": "Song_Transition",
1130                "name": "Transition",
1131                "type": "ai_text",
1132                "web_search": "disabled",
1133                "prompt": "From <prev_songinfo> to <next_songinfo>.",
1134                "constraints": {"max_chars": 200},
1135            },
1136        ],
1137        "section_order": [
1138            {"when": "start_of_playlist", "flow": [{"MUST": "Song_Introduction_Start"}]},
1139            {"when": "between_songs", "flow": [{"MUST": "Song_Transition"}]},
1140        ],
1141    }
1142
1143
1144class ShowRuntime(DummyRuntime):
1145    """Runtime harness exercising the real _run_show with stubbed track sourcing."""
1146
1147    async def _fetch_source_tracks(
1148        self, station: dict[str, Any]
1149    ) -> tuple[list[dict[str, Any]], str]:
1150        """Return two fixed tracks, each carrying its resolved media item."""
1151        return [
1152            {"index": 0, "songinfo": "A - One", "duration": 200, "media_item": _stub_track("1")},
1153            {"index": 1, "songinfo": "B - Two", "duration": 200, "media_item": _stub_track("2")},
1154        ], "Source Playlist"
1155
1156    async def _prepare_runtime_tokens(self, station: dict[str, Any]) -> dict[str, str]:
1157        """Skip the weather lookup; runtime tokens are irrelevant to these tests."""
1158        return {}
1159
1160
1161class ShowRuntimeWithDJ(AIRadioQueueDJMixin, AIRadioStorageMixin, ShowRuntime):
1162    """ShowRuntime harness that also carries sticky queue DJ state."""
1163
1164    def __init__(self, tmp_path: Path) -> None:
1165        """Initialize show runtime state plus queue DJ bookkeeping."""
1166        super().__init__()
1167        self._hosts: dict[str, dict[str, Any]] = {
1168            "rick": {"id": "rick", "name": "Rick", "instructions": "x", "tts_engine": ""},
1169        }
1170        self._dj_queues: dict[str, Any] = {}
1171        self._dj_file = tmp_path / "queue_dj.json"
1172        self._dj_lock = asyncio.Lock()
1173        self._unloading = False
1174
1175
1176def _recording_create_task(scheduled: list[str]) -> Callable[..., None]:
1177    """Return a create_task stub that records the task id and discards the coroutine."""
1178
1179    def _create_task(coro: Any, task_id: str | None = None, **_kwargs: Any) -> None:
1180        if task_id:
1181            scheduled.append(task_id)
1182        coro.close()
1183
1184    return _create_task
1185
1186
1187async def test_run_show_loads_the_whole_show_then_plays_index_zero() -> None:
1188    """The show is loaded in one call, fully stamped, before playback is started."""
1189    runtime = ShowRuntime()
1190    call_order: list[str] = []
1191    loaded: list[tuple[Any, dict[str, Any]]] = []
1192
1193    async def _load(_queue_id: str, queue_items: list[Any], **kwargs: Any) -> None:
1194        call_order.append("load")
1195        # snapshot media_item + extra_attributes now: asserting on the live queue_item
1196        # objects after _run_show returns would pass even if stamping happened later
1197        loaded.extend((item.media_item, dict(item.extra_attributes)) for item in queue_items)
1198        assert kwargs["shuffle"] is False
1199        assert kwargs["keep_remaining"] is False
1200        assert kwargs["keep_played"] is False
1201
1202    async def _play_index(_queue_id: str, index: int) -> None:
1203        call_order.append(f"play_index:{index}")
1204
1205    _set_runtime_mass(
1206        runtime,
1207        _show_mass_stub(
1208            load=_load,
1209            play_index=_play_index,
1210            clear=lambda _queue_id: call_order.append("clear"),
1211            set_shuffle=_recording_set_shuffle(call_order),
1212        ),
1213    )
1214
1215    await runtime._run_show(SessionState(session_id="sess", station_id="st"), _show_station())
1216
1217    assert call_order == ["clear", "set_shuffle", "load", "play_index:0"]
1218    clips = [
1219        (media_item, attrs)
1220        for media_item, attrs in loaded
1221        if media_item.media_type == MediaType.SOUND_EFFECT
1222    ]
1223    assert clips
1224    # every clip was already fully stamped at the moment load() was called
1225    assert all(attrs[ATTR_PROMPT] for _media_item, attrs in clips)
1226    assert all(attrs[ATTR_SESSION_ID] == "sess" for _media_item, attrs in clips)
1227
1228
1229async def test_run_show_targets_active_group_queue() -> None:
1230    """Queue and start the show on the active (group) queue when the player is grouped."""
1231    runtime = ShowRuntime()
1232    clear_calls: list[str] = []
1233    load_queue_ids: list[str] = []
1234    play_index_queue_ids: list[str] = []
1235
1236    async def _load(queue_id: str, **_kwargs: Any) -> None:
1237        load_queue_ids.append(queue_id)
1238
1239    async def _play_index(queue_id: str, _index: int) -> None:
1240        play_index_queue_ids.append(queue_id)
1241
1242    _set_runtime_mass(
1243        runtime,
1244        _show_mass_stub(
1245            get_active_queue=lambda _player_id: SimpleNamespace(queue_id="group_1"),
1246            clear=clear_calls.append,
1247            load=_load,
1248            play_index=_play_index,
1249        ),
1250    )
1251    session = SessionState(session_id="s1", station_id="st")
1252
1253    result = await runtime._run_show(session, _show_station())
1254
1255    assert result["queue_id"] == "group_1"
1256    assert clear_calls == ["group_1"]
1257    assert load_queue_ids == ["group_1"]
1258    assert play_index_queue_ids == ["group_1"]
1259    assert session.queue_id == "group_1"
1260
1261
1262async def test_run_show_clears_the_queues_sticky_dj(tmp_path: Path) -> None:
1263    """Starting a show drops that queue's existing sticky DJ assignment."""
1264    runtime = ShowRuntimeWithDJ(tmp_path)
1265    _set_runtime_mass(runtime, _show_mass_stub())
1266    await runtime.set_queue_dj("living_room", "rick")
1267    assert "living_room" in runtime._dj_queues
1268
1269    await runtime._run_show(SessionState(session_id="sess", station_id="st"), _show_station())
1270
1271    assert "living_room" not in runtime._dj_queues
1272    persisted = json.loads(runtime._dj_file.read_text())
1273    assert persisted["queues"] == {}
1274
1275
1276async def test_run_show_clears_the_dj_on_the_resolved_group_queue(tmp_path: Path) -> None:
1277    """A grouped player's DJ is cleared on the active (group) queue, not the raw player id."""
1278    runtime = ShowRuntimeWithDJ(tmp_path)
1279    _set_runtime_mass(
1280        runtime,
1281        _show_mass_stub(get_active_queue=lambda _player_id: SimpleNamespace(queue_id="group_1")),
1282    )
1283    # a stale assignment on the raw player id must survive untouched: the show never
1284    # played there, only on the resolved group queue
1285    await runtime.set_queue_dj("living_room", "rick")
1286    await runtime.set_queue_dj("group_1", "rick")
1287
1288    await runtime._run_show(SessionState(session_id="s1", station_id="st"), _show_station())
1289
1290    assert "group_1" not in runtime._dj_queues
1291    assert "living_room" in runtime._dj_queues
1292
1293
1294async def test_run_session_finally_replans_a_dj_armed_mid_show(tmp_path: Path) -> None:
1295    """A DJ armed via the menu while a show plays still gets scheduled once the show ends."""
1296    runtime = ShowRuntimeWithDJ(tmp_path)
1297    scheduled: list[str] = []
1298    _set_runtime_mass(runtime, _show_mass_stub(create_task=_recording_create_task(scheduled)))
1299    session = SessionState(session_id="sess", station_id="st", queue_id="living_room")
1300    runtime._sessions[session.session_id] = session
1301
1302    # arming mid-show already requested a pass; that pass would drain against the running-show
1303    # guard in _replan_queue and clear replan_pending without planning anything, so reset it
1304    # here to isolate the finally block's own request instead of piggybacking on this one
1305    await runtime.set_queue_dj("living_room", "rick")
1306    runtime._dj_queues["living_room"].replan_pending = False
1307    scheduled.clear()
1308
1309    async def _run_show_stub(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
1310        raise RuntimeError("show over")
1311
1312    runtime._run_show = _run_show_stub  # type: ignore[method-assign]
1313
1314    await runtime._run_session(session.session_id, {"id": "st"})
1315
1316    assert scheduled == ["ai_radio_dj_replan_living_room"]
1317    assert runtime._dj_queues["living_room"].ready is True
1318
1319
1320async def test_run_show_ends_as_stopped_when_the_user_stops_the_queue() -> None:
1321    """A queue stopped part-way through the show ends the run as a user stop."""
1322    runtime = DummyRuntime()
1323    session = SessionState(session_id="sess", station_id="st")
1324    queue = _stub_queue(state=PlaybackState.PLAYING, current_index=0)
1325    mass = _show_mass_stub(
1326        get=lambda _queue_id: queue,
1327        items=lambda _queue_id, limit=500, offset=0: [  # noqa: ARG005
1328            _stub_clip_queue_item("sess_000", session_id="sess")
1329        ],
1330    )
1331    _set_runtime_mass(runtime, mass)
1332
1333    task = asyncio.create_task(
1334        runtime._await_show_end(session, "player_a", last_index=5, has_clips=True)
1335    )
1336    await asyncio.sleep(0)
1337    assert not task.done()
1338
1339    queue.state = PlaybackState.IDLE
1340    mass.emit_queue_updated("player_a")
1341
1342    assert await asyncio.wait_for(task, timeout=1) == "queue_stopped"
1343
1344
1345async def test_run_show_ends_as_exhausted_when_the_show_plays_out() -> None:
1346    """Reaching the last enqueued entry ends the run as a normal completion."""
1347    runtime = DummyRuntime()
1348    session = SessionState(session_id="sess", station_id="st")
1349    queue = _stub_queue(state=PlaybackState.PLAYING, current_index=5)
1350    mass = _show_mass_stub(
1351        get=lambda _queue_id: queue,
1352        items=lambda _queue_id, limit=500, offset=0: [  # noqa: ARG005
1353            _stub_clip_queue_item("sess_000", session_id="sess")
1354        ],
1355    )
1356    _set_runtime_mass(runtime, mass)
1357
1358    task = asyncio.create_task(
1359        runtime._await_show_end(session, "player_a", last_index=5, has_clips=True)
1360    )
1361    await asyncio.sleep(0)
1362
1363    queue.state = PlaybackState.IDLE
1364    mass.emit_queue_updated("player_a")
1365
1366    assert await asyncio.wait_for(task, timeout=1) == "source_exhausted"
1367
1368
1369async def test_run_show_ignores_an_idle_queue_before_playback_starts() -> None:
1370    """A queue that has not started yet is not mistaken for a stopped one."""
1371    runtime = DummyRuntime()
1372    session = SessionState(session_id="sess", station_id="st")
1373    queue = _stub_queue(state=PlaybackState.IDLE, current_index=None)
1374    mass = _show_mass_stub(
1375        get=lambda _queue_id: queue,
1376        items=lambda _queue_id, limit=500, offset=0: [  # noqa: ARG005
1377            _stub_clip_queue_item("sess_000", session_id="sess")
1378        ],
1379    )
1380    _set_runtime_mass(runtime, mass)
1381
1382    task = asyncio.create_task(
1383        runtime._await_show_end(session, "player_a", last_index=5, has_clips=True)
1384    )
1385    mass.emit_queue_updated("player_a")
1386    await asyncio.sleep(0)
1387
1388    assert not task.done()
1389    task.cancel()
1390
1391
1392async def test_run_show_keeps_a_paused_queue_on_air() -> None:
1393    """A paused queue keeps the show running."""
1394    runtime = DummyRuntime()
1395    session = SessionState(session_id="sess", station_id="st")
1396    queue = _stub_queue(state=PlaybackState.PLAYING, current_index=1)
1397    mass = _show_mass_stub(
1398        get=lambda _queue_id: queue,
1399        items=lambda _queue_id, limit=500, offset=0: [  # noqa: ARG005
1400            _stub_clip_queue_item("sess_000", session_id="sess")
1401        ],
1402    )
1403    _set_runtime_mass(runtime, mass)
1404
1405    task = asyncio.create_task(
1406        runtime._await_show_end(session, "player_a", last_index=5, has_clips=True)
1407    )
1408    await asyncio.sleep(0)
1409
1410    queue.state = PlaybackState.PAUSED
1411    mass.emit_queue_updated("player_a")
1412    await asyncio.sleep(0)
1413
1414    assert not task.done()
1415    task.cancel()
1416
1417
1418async def test_run_show_ends_when_the_queue_no_longer_holds_its_clips() -> None:
1419    """A queue cleared or taken over by other playback ends the run."""
1420    runtime = DummyRuntime()
1421    session = SessionState(session_id="sess", station_id="st")
1422    queue = _stub_queue(state=PlaybackState.PLAYING, current_index=1)
1423    queue_items = [_stub_clip_queue_item("sess_000", session_id="sess")]
1424    mass = _show_mass_stub(
1425        get=lambda _queue_id: queue,
1426        items=lambda _queue_id, limit=500, offset=0: queue_items[offset : offset + limit],
1427    )
1428    _set_runtime_mass(runtime, mass)
1429
1430    task = asyncio.create_task(
1431        runtime._await_show_end(session, "player_a", last_index=5, has_clips=True)
1432    )
1433    await asyncio.sleep(0)
1434
1435    queue_items.clear()
1436    mass.emit_queue_updated("player_a")
1437
1438    assert await asyncio.wait_for(task, timeout=1) == "queue_stopped"
1439
1440
1441async def test_run_show_ends_when_its_player_is_removed() -> None:
1442    """Removing the target player must not pin the session's slot forever."""
1443    runtime = DummyRuntime()
1444    session = SessionState(session_id="sess", station_id="st")
1445    queue = _stub_queue(state=PlaybackState.PLAYING, current_index=1)
1446    queue_holder: list[Any] = [queue]
1447    mass = _show_mass_stub(
1448        get=lambda _queue_id: queue_holder[0],
1449        items=lambda _queue_id, limit=500, offset=0: [  # noqa: ARG005
1450            _stub_clip_queue_item("sess_000", session_id="sess")
1451        ],
1452    )
1453    _set_runtime_mass(runtime, mass)
1454
1455    task = asyncio.create_task(
1456        runtime._await_show_end(session, "player_a", last_index=5, has_clips=True)
1457    )
1458    await asyncio.sleep(0)
1459    assert not task.done()
1460
1461    # on_player_remove pops the queue data before PLAYER_REMOVED is signaled
1462    queue_holder[0] = None
1463    mass.emit_player_removed("player_a")
1464
1465    assert await asyncio.wait_for(task, timeout=1) == "queue_stopped"
1466
1467
1468async def test_run_show_keeps_waiting_when_the_show_has_no_clips_to_lose() -> None:
1469    """A clip-free show is not mistaken for one whose clips got cleared out."""
1470    runtime = DummyRuntime()
1471    session = SessionState(session_id="sess", station_id="st")
1472    queue = _stub_queue(state=PlaybackState.PLAYING, current_index=0)
1473    # track-only queue: no item carries ATTR_SESSION_ID, exactly like a show whose
1474    # section rules never selected anything to insert
1475    mass = _show_mass_stub(
1476        get=lambda _queue_id: queue,
1477        items=lambda _queue_id, limit=500, offset=0: [  # noqa: ARG005
1478            SimpleNamespace(extra_attributes={})
1479        ],
1480    )
1481    _set_runtime_mass(runtime, mass)
1482
1483    task = asyncio.create_task(
1484        runtime._await_show_end(session, "player_a", last_index=5, has_clips=False)
1485    )
1486    await asyncio.sleep(0)
1487    assert not task.done()
1488
1489    queue.current_index = 5
1490    queue.state = PlaybackState.IDLE
1491    mass.emit_queue_updated("player_a")
1492
1493    assert await asyncio.wait_for(task, timeout=1) == "source_exhausted"
1494
1495
1496async def test_await_show_end_fails_when_playback_never_starts(
1497    monkeypatch: pytest.MonkeyPatch,
1498) -> None:
1499    """A show whose playback never starts is declared failed instead of waiting forever."""
1500    monkeypatch.setattr(
1501        "music_assistant.providers.ai_radio.runtime.SHOW_START_TIMEOUT_SECONDS", 0.05
1502    )
1503    runtime = DummyRuntime()
1504    session = SessionState(session_id="sess", station_id="st")
1505    queue = _stub_queue(state=PlaybackState.IDLE, current_index=None)
1506    mass = _show_mass_stub(
1507        get=lambda _queue_id: queue,
1508        items=lambda _queue_id, limit=500, offset=0: [  # noqa: ARG005
1509            _stub_clip_queue_item("sess_000", session_id="sess")
1510        ],
1511    )
1512    _set_runtime_mass(runtime, mass)
1513
1514    with pytest.raises(MusicAssistantError, match="did not start"):
1515        await runtime._await_show_end(session, "player_a", last_index=5, has_clips=True)
1516
1517
1518async def test_await_show_end_does_not_time_out_once_playback_starts(
1519    monkeypatch: pytest.MonkeyPatch,
1520) -> None:
1521    """Playback starting before the start-timeout elapses lets the show proceed as normal."""
1522    monkeypatch.setattr(
1523        "music_assistant.providers.ai_radio.runtime.SHOW_START_TIMEOUT_SECONDS", 0.2
1524    )
1525    runtime = DummyRuntime()
1526    session = SessionState(session_id="sess", station_id="st")
1527    queue = _stub_queue(state=PlaybackState.IDLE, current_index=None)
1528    mass = _show_mass_stub(
1529        get=lambda _queue_id: queue,
1530        items=lambda _queue_id, limit=500, offset=0: [  # noqa: ARG005
1531            _stub_clip_queue_item("sess_000", session_id="sess")
1532        ],
1533    )
1534    _set_runtime_mass(runtime, mass)
1535
1536    task = asyncio.create_task(
1537        runtime._await_show_end(session, "player_a", last_index=5, has_clips=True)
1538    )
1539    await asyncio.sleep(0)
1540    assert not task.done()
1541
1542    queue.state = PlaybackState.PLAYING
1543    queue.current_index = 0
1544    mass.emit_queue_updated("player_a")
1545    await asyncio.sleep(0)
1546    assert not task.done()
1547
1548    queue.current_index = 5
1549    queue.state = PlaybackState.IDLE
1550    mass.emit_queue_updated("player_a")
1551
1552    assert await asyncio.wait_for(task, timeout=1) == "source_exhausted"
1553
1554
1555def test_passes_optional_guards_handles_non_numeric_guard_values() -> None:
1556    """Treat non-numeric guard values as disabled instead of raising ValueError."""
1557    runtime = DummyRuntime()
1558    slot = Slot(
1559        when="between_songs",
1560        at_index=1,
1561        prev_index=0,
1562        next_index=1,
1563        very_next_index=2,
1564        minute_mark=5.0,
1565    )
1566
1567    result = runtime._passes_optional_guards(
1568        section_id="Weather_Short",
1569        guards={"min_gap_songs": "abc", "max_per_60min": "xyz"},
1570        history={},
1571        slot=slot,
1572        tracks=[{}, {}, {}],
1573        placeholders={},
1574        track_index_offset=0,
1575        minute_offset=0.0,
1576    )
1577
1578    assert result is True
1579
1580
1581async def test_fetch_source_tracks_skips_tracks_with_no_resolvable_uri(caplog: Any) -> None:
1582    """Skip and warn about source tracks with no resolvable uri instead of queuing a dead entry."""
1583
1584    class DummyPlaylist:
1585        name = "Source Playlist"
1586
1587    class DummyPlaylistsController:
1588        def __init__(self, tracks: list[Any]) -> None:
1589            self._tracks = tracks
1590
1591        async def get(self, playlist_id: str, provider: str) -> Any:
1592            return DummyPlaylist()
1593
1594        async def tracks(self, playlist_id: str, provider: str) -> Any:
1595            for track in self._tracks:
1596                yield track
1597
1598    class DummyTrack:
1599        def __init__(self, item_id: str, name: str, uri: str = "") -> None:
1600            self.item_id = item_id
1601            self.name = name
1602            self.artists: list[Any] = []
1603            self.duration = 180
1604            self.uri = uri
1605            self.provider_mappings: list[Any] = []
1606
1607    good_track_1 = DummyTrack("1", "Track One", uri="library://track/1")
1608    unresolvable_track = DummyTrack("2", "Track Two")
1609    good_track_2 = DummyTrack("3", "Track Three", uri="library://track/3")
1610
1611    class DummyMusic:
1612        playlists = DummyPlaylistsController([good_track_1, unresolvable_track, good_track_2])
1613
1614    class DummyMass:
1615        music = DummyMusic()
1616
1617    runtime = DummyRuntime()
1618    _set_runtime_mass(runtime, DummyMass())
1619    station = {"source_playlist_id": "playlist-1", "source_playlist_provider": "library"}
1620
1621    with caplog.at_level(logging.WARNING):
1622        tracks, playlist_name = await runtime._fetch_source_tracks(station)
1623
1624    assert playlist_name == "Source Playlist"
1625    assert [track["item_id"] for track in tracks] == ["1", "3"]
1626    assert [track["index"] for track in tracks] == [0, 1]
1627    # the resolved media item travels on the normalized dict, unchanged
1628    assert [track["media_item"] for track in tracks] == [good_track_1, good_track_2]
1629    assert any("Track Two" in record.message for record in caplog.records)
1630
1631
1632def test_apply_source_shuffle_returns_unchanged_when_disabled() -> None:
1633    """Leave the source list untouched when the station does not request shuffling."""
1634    runtime = DummyRuntime()
1635    tracks = [{"index": 0, "uri": "a"}, {"index": 1, "uri": "b"}]
1636    station = {"shuffle_source_tracks": False}
1637
1638    result = runtime._apply_source_shuffle(tracks, station)
1639
1640    assert result is tracks
1641
1642
1643def test_apply_source_shuffle_reorders_and_records_source_index(
1644    monkeypatch: pytest.MonkeyPatch,
1645) -> None:
1646    """Shuffle every track into a new order while keeping all of them and their origin."""
1647    # capture the real class before patching it away: runtime.random is the same shared
1648    # stdlib module object, so the lambda below would otherwise re-look-up itself
1649    original_random_cls = random.Random
1650    monkeypatch.setattr(
1651        "music_assistant.providers.ai_radio.runtime.random.Random",
1652        lambda: original_random_cls(1234),
1653    )
1654    runtime = DummyRuntime()
1655    tracks = [{"uri": f"track/{i}"} for i in range(5)]
1656    station = {"shuffle_source_tracks": True}
1657
1658    result = runtime._apply_source_shuffle(tracks, station)
1659
1660    assert [track["index"] for track in result] == list(range(len(tracks)))
1661    assert {track["uri"] for track in result} == {track["uri"] for track in tracks}
1662    for track in result:
1663        assert tracks[track["source_index"]]["uri"] == track["uri"]
1664    # a seeded shuffle must actually reorder, not silently pass through in place
1665    assert [track["source_index"] for track in result] != list(range(len(tracks)))
1666
1667
1668def test_apply_track_duration_limit_keeps_prefix_of_given_order() -> None:
1669    """Truncate to the playtime cap by walking the given order, no shuffling."""
1670    runtime = DummyRuntime()
1671    tracks = [
1672        {"uri": "a", "duration": 120},
1673        {"uri": "b", "duration": 120},
1674        {"uri": "c", "duration": 120},
1675        {"uri": "d", "duration": 120},
1676    ]
1677    station = {"max_duration_minutes": 3}
1678
1679    result = runtime._apply_track_duration_limit(tracks, station)
1680
1681    assert [track["uri"] for track in result] == ["a", "b"]
1682    assert [track["index"] for track in result] == [0, 1]
1683    assert [track["source_index"] for track in result] == [0, 1]
1684
1685
1686def test_apply_track_duration_limit_zero_cap_is_noop() -> None:
1687    """A cap of 0 disables truncation entirely."""
1688    runtime = DummyRuntime()
1689    tracks = [{"uri": "a", "duration": 120}, {"uri": "b", "duration": 120}]
1690    station = {"max_duration_minutes": 0}
1691
1692    result = runtime._apply_track_duration_limit(tracks, station)
1693
1694    assert result is tracks
1695
1696
1697async def test_run_show_disables_shuffle_before_load() -> None:
1698    """Disable queue shuffle before the items are loaded, so sections keep their planned order."""
1699    runtime = ShowRuntime()
1700    call_order: list[str] = []
1701    set_shuffle_calls: list[tuple[str, bool]] = []
1702
1703    async def _set_shuffle(queue_id: str, shuffle_enabled: bool) -> None:
1704        set_shuffle_calls.append((queue_id, shuffle_enabled))
1705        call_order.append("set_shuffle")
1706
1707    async def _load(_queue_id: str, **_kwargs: Any) -> None:
1708        call_order.append("load")
1709
1710    _set_runtime_mass(runtime, _show_mass_stub(set_shuffle=_set_shuffle, load=_load))
1711    session = SessionState(session_id="s1", station_id="st")
1712
1713    await runtime._run_show(session, _show_station())
1714
1715    assert set_shuffle_calls == [("living_room", False)]
1716    assert call_order.index("set_shuffle") < call_order.index("load")
1717
1718
1719async def test_run_show_stays_running_while_the_queue_plays_and_stop_cancels_it() -> None:
1720    """The session stays 'running' for as long as the show plays; a stop cancels it mid-show."""
1721    runtime = ShowRuntime()
1722    queue = _stub_queue(state=PlaybackState.PLAYING, current_index=0)
1723    unsubscribed = False
1724
1725    def _subscribe(
1726        cb_func: Callable[[Any], None],  # noqa: ARG001
1727        event_filter: Any = None,  # noqa: ARG001
1728        id_filter: Any = None,  # noqa: ARG001
1729    ) -> Callable[[], None]:
1730        def _unsubscribe() -> None:
1731            nonlocal unsubscribed
1732            unsubscribed = True
1733
1734        return _unsubscribe
1735
1736    mass = _show_mass_stub(
1737        get=lambda _queue_id: queue,
1738        items=lambda _queue_id, limit=500, offset=0: [  # noqa: ARG005
1739            _stub_clip_queue_item("sess_000", session_id="s1")
1740        ],
1741        subscribe=_subscribe,
1742    )
1743    _set_runtime_mass(runtime, mass)
1744    session = SessionState(session_id="s1", station_id="st")
1745    runtime._sessions[session.session_id] = session
1746
1747    task = asyncio.create_task(runtime._run_session(session.session_id, _show_station()))
1748    await asyncio.sleep(0)
1749    await asyncio.sleep(0)
1750
1751    # the show is on air: the session must stay running, exactly like start_run's
1752    # max-concurrent-runs and station-already-active guards require
1753    assert session.status == "running"
1754    assert not task.done()
1755
1756    # this is what stop_run does to end a run mid-show
1757    task.cancel()
1758    with pytest.raises(asyncio.CancelledError):
1759        await task
1760
1761    assert session.status == "stopped"
1762    assert unsubscribed
1763
1764
1765async def test_run_show_binds_the_session_to_the_target_queue() -> None:
1766    """Record the queue a show plays on so stopping the show can stop it."""
1767    runtime = ShowRuntime()
1768    _set_runtime_mass(runtime, _show_mass_stub())
1769    session = SessionState(session_id="s1", station_id="st")
1770
1771    await runtime._run_show(session, _show_station())
1772
1773    assert session.queue_id == "living_room"
1774
1775
1776async def test_run_session_reports_a_queue_stop_as_stopped() -> None:
1777    """Report a run that ended because the queue was stopped as stopped, not completed."""
1778
1779    class QueueStoppedRuntime(AIRadioRuntimeMixin):
1780        def __init__(self) -> None:
1781            self.logger = logging.getLogger("tests.ai_radio.runtime.queue_stopped")
1782            self._sessions: dict[str, SessionState] = {}
1783
1784        async def _run_show(self, session: SessionState, station: dict[str, Any]) -> dict[str, Any]:
1785            return {"ended_reason": "queue_stopped"}
1786
1787    runtime = QueueStoppedRuntime()
1788    session = SessionState(session_id="s1", station_id="st")
1789    runtime._sessions[session.session_id] = session
1790
1791    await runtime._run_session(session.session_id, {"id": "st"})
1792
1793    assert session.status == "stopped"
1794    assert session.ended_at is not None
1795