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