music-assistant-server

1"""
2Tests for crossfade transition timing math.
3
4Covers the ``CrossfadeTimingInfo`` contract that drives lyrics-sync correctness:
5
6  pre_crossfade_duration  + crossfade_duration         = portion attributed to A
7  fadein_trimmed_duration + crossfade_duration         = where B's listener actually is
8                                                          when CF ends (the value the
9                                                          flow loop writes into
10                                                          streamdetails.seek_position)
11
12Pure math tests use a fake ``SmartFade`` subclass that just assigns the test-provided
13timing in its ``build``. Smart/standard end-to-end behavior is exercised through
14``SmartFadesMixer.build``.
15"""
16
17from __future__ import annotations
18
19import logging
20from collections.abc import AsyncGenerator
21from unittest.mock import AsyncMock, MagicMock
22
23import numpy as np
24import pytest
25from music_assistant_models.enums import ContentType, CrossfadeMode, MediaType, StreamType
26from music_assistant_models.media_items import AudioFormat
27from music_assistant_models.streamdetails import StreamDetails
28
29import music_assistant.controllers.streams.smart_fades.mixer as mixer_module
30from music_assistant.constants import VERBOSE_LOG_LEVEL
31from music_assistant.controllers.streams.smart_fades.fades import (
32    CrossfadeTimingInfo,
33    SmartCrossFade,
34    SmartFade,
35    SmartFadeNotApplicable,
36    StandardCrossFade,
37)
38from music_assistant.controllers.streams.smart_fades.filters import (
39    FadeOutTrimFilter,
40    GradualTimeStretchFilter,
41    StreamingCrossfadeFilter,
42)
43from music_assistant.controllers.streams.smart_fades.helpers import SMART_CROSSFADE_DURATION
44from music_assistant.controllers.streams.smart_fades.mixer import SmartFadesMixer
45from music_assistant.models.audio_analysis import AudioAnalysisData
46
47PCM = AudioFormat(
48    content_type=ContentType.PCM_S16LE,
49    sample_rate=44100,
50    bit_depth=16,
51    channels=2,
52)
53SAMPLE_SIZE = PCM.pcm_sample_size  # bytes per second
54
55
56def _seconds(seconds: float) -> int:
57    """Return the number of bytes that represent ``seconds`` of PCM audio."""
58    return int(seconds * SAMPLE_SIZE)
59
60
61def _streamdetails(item_id: str = "test", provider: str = "test") -> StreamDetails:
62    """Return a minimal StreamDetails object for mixer.build() input."""
63    return StreamDetails(
64        provider=provider,
65        item_id=item_id,
66        audio_format=PCM,
67        media_type=MediaType.TRACK,
68        stream_type=StreamType.HTTP,
69    )
70
71
72def _make_mixer(analysis_for: dict[str, AudioAnalysisData] | None = None) -> SmartFadesMixer:
73    """Build a SmartFadesMixer with a stubbed StreamsController."""
74    analysis_for = analysis_for or {}
75    streams = MagicMock()
76    streams.logger = logging.getLogger("test_smartfade_transition_timings")
77    streams.audio_analysis = MagicMock()
78
79    async def _get_analysis(item_id: str, _provider: str, **_kwargs: object) -> object:
80        return analysis_for.get(item_id)
81
82    streams.audio_analysis.get_audio_analysis = AsyncMock(side_effect=_get_analysis)
83    return SmartFadesMixer(streams)
84
85
86def _beats(start: float, count: int, interval: float) -> np.ndarray:
87    """Return ``count`` beat positions starting at ``start`` spaced by ``interval`` seconds."""
88    return np.arange(count, dtype=np.float32) * interval + start
89
90
91def _analysis(
92    bpm: float,
93    beats_start: float = 0.0,
94    beats_count: int = 200,
95    duration: float | None = None,
96    rms_energy: np.ndarray | None = None,
97) -> AudioAnalysisData:
98    """Synthetic analysis data with enough beats for SmartCrossFade.build() to succeed."""
99    interval = 60.0 / bpm  # seconds per beat
100    # When an explicit duration is given, generate beats spanning the full track so the
101    # buffer-local shift (duration - 45s) leaves real beats inside the 45s window.
102    if duration is not None:
103        count = max(beats_count, int(duration / interval) + 1)
104        beats = _beats(beats_start, count, interval)
105    else:
106        beats = _beats(beats_start, beats_count, interval)
107        duration = float(beats[-1] + interval)
108    downbeats = beats[::4]  # 4/4 time signature
109    return AudioAnalysisData(
110        duration=duration,
111        bpm=bpm,
112        beats=beats.tolist(),
113        downbeats=downbeats.tolist(),
114        rms_energy=rms_energy.tolist() if rms_energy is not None else None,
115    )
116
117
118def _with_vocal_activity(
119    analysis: AudioAnalysisData,
120    windows: list[tuple[float, float]],
121) -> AudioAnalysisData:
122    """
123    Add a valid 1800-bin vocal probability timeline to an analysis row.
124
125    :param analysis: Analysis row to update.
126    :param windows: Vocal windows in full-track media seconds.
127    """
128    assert analysis.duration is not None
129    frame_duration = analysis.duration / 1800
130    probabilities = [0.0] * 1800
131    for start, end in windows:
132        for index in range(int(start / frame_duration), int(end / frame_duration)):
133            probabilities[index] = 0.9
134    analysis.extra_data = {"vocal_activity": probabilities}
135    return analysis
136
137
138class _FixedTimingFade(SmartFade):
139    """Test-only SmartFade whose build just assigns a caller-provided timing."""
140
141    def __init__(self, timing: CrossfadeTimingInfo) -> None:
142        super().__init__(logging.getLogger("test_fixed_timing_fade"))
143        self._fixed_timing = timing
144
145    def build(
146        self,
147        fade_out_bytes_len: int,
148        fade_in_bytes_len: int,
149        pcm_format: AudioFormat,
150    ) -> None:
151        """Assign the timing supplied at construction time."""
152        self.filters = []  # non-empty would normally be required, but unused here
153        self.timing_info = self._fixed_timing
154
155
156# ---------------------------------------------------------------------------
157# CrossfadeTimingInfo dataclass
158# ---------------------------------------------------------------------------
159
160
161class TestCrossfadeTimingInfo:
162    """Cover the dataclass surface used by callers."""
163
164    def test_fields_are_set(self) -> None:
165        """Constructor stores every duration on the dataclass."""
166        timing = CrossfadeTimingInfo(
167            pre_crossfade_duration=1.0,
168            crossfade_duration=2.0,
169            fadein_trimmed_duration=3.0,
170            post_crossfade_duration=4.0,
171        )
172        assert timing.pre_crossfade_duration == 1.0
173        assert timing.crossfade_duration == 2.0
174        assert timing.fadein_trimmed_duration == 3.0
175        assert timing.post_crossfade_duration == 4.0
176
177    def test_default_values(self) -> None:
178        """All fields default to 0.0 so build can populate them incrementally."""
179        timing = CrossfadeTimingInfo()
180        assert timing.pre_crossfade_duration == 0.0
181        assert timing.crossfade_duration == 0.0
182        assert timing.fadein_trimmed_duration == 0.0
183        assert timing.post_crossfade_duration == 0.0
184
185
186# ---------------------------------------------------------------------------
187# StandardCrossFade.build — timing math via the real subclass
188# ---------------------------------------------------------------------------
189
190
191class TestStandardCrossFadeBuild:
192    """StandardCrossFade.build must produce the expected timing for given inputs."""
193
194    def _build(
195        self,
196        crossfade_duration: float,
197        fade_out_seconds: float,
198        fade_in_seconds: float,
199    ) -> CrossfadeTimingInfo:
200        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=crossfade_duration)
201        fade.build(_seconds(fade_out_seconds), _seconds(fade_in_seconds), PCM)
202        return fade.timing_info
203
204    def test_symmetric_buffers(self) -> None:
205        """Standard with full symmetric buffers — TRIM stays 0."""
206        timing = self._build(crossfade_duration=10.0, fade_out_seconds=30, fade_in_seconds=30)
207        assert timing.crossfade_duration == pytest.approx(10.0)
208        assert timing.fadein_trimmed_duration == 0.0
209        assert timing.pre_crossfade_duration == pytest.approx(20.0)
210        assert timing.post_crossfade_duration == pytest.approx(20.0)
211
212    def test_buffer_equals_overlap(self) -> None:
213        """Standard with X == CF leaves no PRE or POST — mix output is pure overlap."""
214        timing = self._build(crossfade_duration=10.0, fade_out_seconds=10, fade_in_seconds=10)
215        assert timing.pre_crossfade_duration == pytest.approx(0.0)
216        assert timing.crossfade_duration == pytest.approx(10.0)
217        assert timing.fadein_trimmed_duration == 0.0
218        assert timing.post_crossfade_duration == pytest.approx(0.0)
219
220    def test_short_fadein_clamps_overlap(self) -> None:
221        """Fade-in shorter than the configured CF clamps the effective CF down."""
222        timing = self._build(crossfade_duration=10.0, fade_out_seconds=20, fade_in_seconds=4)
223        assert timing.crossfade_duration == pytest.approx(4.0)
224        assert timing.pre_crossfade_duration == pytest.approx(16.0)
225        assert timing.post_crossfade_duration == pytest.approx(0.0)
226        assert timing.fadein_trimmed_duration == 0.0
227
228    def test_short_fadeout_clamps_overlap(self) -> None:
229        """Fade-out shorter than the configured CF clamps the effective CF down."""
230        timing = self._build(crossfade_duration=10.0, fade_out_seconds=3, fade_in_seconds=20)
231        assert timing.crossfade_duration == pytest.approx(3.0)
232        assert timing.pre_crossfade_duration == pytest.approx(0.0)
233        assert timing.post_crossfade_duration == pytest.approx(17.0)
234
235    def test_filter_duration_matches_clamped_timing(self) -> None:
236        """The crossfade filter must use the clamped duration, not the configured one."""
237        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=10.0)
238        # only 6s of (stripped) fade-out audio available
239        fade.build(_seconds(6), _seconds(45), PCM)
240        assert fade.timing_info.crossfade_duration == pytest.approx(6.0)
241        crossfade_filter = fade.filters[0]
242        assert isinstance(crossfade_filter, StreamingCrossfadeFilter)
243        assert crossfade_filter.crossfade_samples == int(6.0 * PCM.sample_rate)
244
245    def test_fractional_overlap_keeps_filter_aligned_to_buffer(self) -> None:
246        """
247        A non-integer clamped overlap keeps the filter's sample count aligned to the buffer.
248
249        Regression for the silent "FFmpeg produced no output" fallback: a fractional
250        effective crossfade made the byte slice a fraction of a sample shorter than the
251        ``d=`` the filter requested, so ffmpeg's acrossfade emitted nothing.
252        """
253        frame_size = (PCM.bit_depth // 8) * PCM.channels
254        # ~6.3333s of audible fade-out: a real PCM buffer is frame-aligned, yet still not a
255        # whole number of seconds, so the effective crossfade stays fractional
256        fade_out_len = _seconds(6.3333) // frame_size * frame_size
257        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=10.0)
258        fade.build(fade_out_len, _seconds(45), PCM)
259        crossfade_filter = fade.filters[0]
260        assert isinstance(crossfade_filter, StreamingCrossfadeFilter)
261        # the source-of-truth byte size is frame-aligned ...
262        assert fade.crossfade_size % frame_size == 0
263        # ... and the filter's sample count is exactly that buffer, in samples
264        assert crossfade_filter.crossfade_samples == fade.crossfade_size // frame_size
265        # the timing duration round-trips from the same integer, never the other way
266        assert fade.timing_info.crossfade_duration == pytest.approx(
267            fade.crossfade_size / PCM.pcm_sample_size
268        )
269
270
271# ---------------------------------------------------------------------------
272# StandardCrossFade.apply — byte slicing must follow the clamped timing
273# ---------------------------------------------------------------------------
274
275
276class TestStandardCrossFadeApplySlicing:
277    """apply() must slice the fade-out buffer by the clamped duration, not the configured one."""
278
279    @pytest.mark.asyncio
280    async def test_apply_slices_with_clamped_duration(
281        self, monkeypatch: pytest.MonkeyPatch
282    ) -> None:
283        """A 6s fade-out with a 10s configured CF must hand the base mixer all 6s, no more."""
284        captured: dict[str, bytes] = {}
285        crossfade_marker = b"crossfade-output"
286
287        async def fake_base_apply(
288            _self: SmartFade,
289            fade_out_part: bytes,
290            _fade_in_part: bytes | AsyncGenerator[bytes],
291            _pcm_format: AudioFormat,
292        ) -> AsyncGenerator[bytes]:
293            captured["fade_out"] = fade_out_part
294            yield crossfade_marker
295
296        monkeypatch.setattr(SmartFade, "apply", fake_base_apply)
297        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=10.0)
298        fade.build(_seconds(6), _seconds(45), PCM)
299        chunks = [
300            chunk async for chunk in fade.apply(b"\x00" * _seconds(6), b"\x00" * _seconds(45), PCM)
301        ]
302        assert len(captured["fade_out"]) == _seconds(6)
303        # nothing precedes the crossfade — the 6s buffer is consumed entirely by the overlap
304        assert chunks[0] == crossfade_marker
305
306    @pytest.mark.asyncio
307    async def test_apply_feeds_exactly_the_filter_sample_count(
308        self, monkeypatch: pytest.MonkeyPatch
309    ) -> None:
310        """
311        apply() must feed the base mixer exactly the filter's sample count.
312
313        Otherwise ffmpeg receives fewer samples than the filter was built for and the
314        overlap comes out short — the silent crossfade failure this regression guards
315        against.
316        """
317        captured: dict[str, bytes] = {}
318
319        async def fake_base_apply(
320            _self: SmartFade,
321            fade_out_part: bytes,
322            fade_in_part: bytes | AsyncGenerator[bytes],
323            _pcm_format: AudioFormat,
324        ) -> AsyncGenerator[bytes]:
325            captured["fade_out"] = fade_out_part
326            assert isinstance(fade_in_part, bytes)
327            captured["fade_in"] = fade_in_part
328            yield b"crossfade-output"
329
330        monkeypatch.setattr(SmartFade, "apply", fake_base_apply)
331        frame_size = (PCM.bit_depth // 8) * PCM.channels
332        # frame-aligned like a real PCM buffer, but a fractional number of seconds
333        fade_out_len = _seconds(6.3333) // frame_size * frame_size
334        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=10.0)
335        fade.build(fade_out_len, _seconds(45), PCM)
336        crossfade_filter = fade.filters[0]
337        assert isinstance(crossfade_filter, StreamingCrossfadeFilter)
338        assert crossfade_filter.crossfade_samples is not None
339        async for _ in fade.apply(b"\x00" * fade_out_len, b"\x11" * _seconds(45), PCM):
340            pass
341        expected_bytes = crossfade_filter.crossfade_samples * frame_size
342        assert len(captured["fade_out"]) == expected_bytes
343        assert len(captured["fade_in"]) == expected_bytes
344
345    @pytest.mark.asyncio
346    async def test_apply_before_build_fails_fast(self) -> None:
347        """apply() without a prior build() must error, not silently hard-cut."""
348        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=10.0)
349        with pytest.raises(RuntimeError, match="not built"):
350            async for _ in fade.apply(b"\x00" * _seconds(5), b"\x11" * _seconds(5), PCM):
351                pass
352
353    @pytest.mark.asyncio
354    async def test_zero_crossfade_skips_ffmpeg(self, monkeypatch: pytest.MonkeyPatch) -> None:
355        """When crossfade_duration == 0, apply() must concatenate without calling ffmpeg."""
356        base_apply_invoked: list[bool] = []
357
358        async def _base_apply_sentinel(
359            _self: SmartFade,
360            _fade_out_part: bytes,
361            _fade_in_part: bytes | AsyncGenerator[bytes],
362            _pcm_format: AudioFormat,
363        ) -> AsyncGenerator[bytes]:
364            base_apply_invoked.append(True)
365            yield b""
366
367        monkeypatch.setattr(SmartFade, "apply", _base_apply_sentinel)
368
369        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=10.0)
370        # fade_out_bytes_len=0 → effective_cf = min(10, 0, 45) = 0 → crossfade_duration == 0
371        fade.build(0, _seconds(45), PCM)
372        assert fade.timing_info.crossfade_duration == 0.0
373
374        fade_out_data = b"\x00" * _seconds(5)
375        fade_in_data = b"\x11" * _seconds(5)
376        chunks = [chunk async for chunk in fade.apply(fade_out_data, fade_in_data, PCM)]
377        combined = b"".join(chunks)
378        assert len(combined) == _seconds(10), f"Expected {_seconds(10)} bytes, got {len(combined)}"
379        assert combined[0:1] == b"\x00", "fade_out bytes should come first"
380        assert combined[-1:] == b"\x11", "fade_in bytes should come last"
381        assert not base_apply_invoked, (
382            "SmartFade.apply (ffmpeg path) must not be called for zero-length crossfade"
383        )
384
385
386# ---------------------------------------------------------------------------
387# Pure math/invariant tests via _FixedTimingFade (no beat-alignment dependency)
388# ---------------------------------------------------------------------------
389
390
391class TestLyricsSyncInvariants:
392    """Invariants the flow loop and per-track loop rely on."""
393
394    def _continuation_offset(self, t: CrossfadeTimingInfo) -> float:
395        """Return the value the flow loop writes into streamdetails.seek_position."""
396        return t.fadein_trimmed_duration + t.crossfade_duration
397
398    def _fadeout_share(self, t: CrossfadeTimingInfo) -> float:
399        """Return the seconds of mix output attributed to the outgoing track."""
400        return t.pre_crossfade_duration + t.crossfade_duration
401
402    def test_continuation_offset_no_trim(self) -> None:
403        """Without trim the listener is exactly CF seconds into the incoming track."""
404        timing = CrossfadeTimingInfo(
405            pre_crossfade_duration=20.0,
406            crossfade_duration=10.0,
407            fadein_trimmed_duration=0.0,
408            post_crossfade_duration=20.0,
409        )
410        assert self._continuation_offset(timing) == pytest.approx(10.0)
411
412    def test_continuation_offset_with_trim(self) -> None:
413        """With trim the listener is TRIM + CF seconds into the incoming track."""
414        timing = CrossfadeTimingInfo(
415            pre_crossfade_duration=29.0,
416            crossfade_duration=16.0,
417            fadein_trimmed_duration=3.0,
418            post_crossfade_duration=26.0,
419        )
420        assert self._continuation_offset(timing) == pytest.approx(19.0)
421
422    def test_fadeout_share_accounts_for_full_outgoing_input(self) -> None:
423        """PRE + CF equals the full outgoing input — A is fully accounted for."""
424        timing = CrossfadeTimingInfo(
425            pre_crossfade_duration=29.0,
426            crossfade_duration=16.0,
427            fadein_trimmed_duration=3.0,
428            post_crossfade_duration=26.0,
429        )
430        # fade_out_seconds = PRE + CF = 45
431        assert self._fadeout_share(timing) == pytest.approx(45.0)
432
433    def test_fixed_timing_fade_round_trip(self) -> None:
434        """_FixedTimingFade.timing_info returns whatever was passed in."""
435        original = CrossfadeTimingInfo(
436            pre_crossfade_duration=1.0,
437            crossfade_duration=2.0,
438            fadein_trimmed_duration=3.0,
439            post_crossfade_duration=4.0,
440        )
441        fade = _FixedTimingFade(original)
442        fade.build(0, 0, PCM)
443        assert fade.timing_info == original
444
445
446# ---------------------------------------------------------------------------
447# SmartFadesMixer.build() — the entry point the flow loop calls
448# ---------------------------------------------------------------------------
449
450
451class TestMixerBuild:
452    """build() resolves smart vs standard, primes filters, and returns a SmartFade."""
453
454    @pytest.mark.asyncio
455    async def test_standard_mode_returns_standard_crossfade(
456        self, monkeypatch: pytest.MonkeyPatch
457    ) -> None:
458        """Standard mode always builds a StandardCrossFade with the configured duration."""
459
460        async def identity_strip(audio_data: bytes, **_kwargs: object) -> bytes:
461            return audio_data  # no stripping — test pure timing math
462
463        monkeypatch.setattr(mixer_module, "strip_silence", identity_strip)
464        mixer = _make_mixer()
465        fade = await mixer.build(
466            fade_in_streamdetails=_streamdetails("in"),
467            fade_out_streamdetails=_streamdetails("out"),
468            pcm_format=PCM,
469            standard_crossfade_duration=8,
470            mode=CrossfadeMode.STANDARD_CROSSFADE,
471            fade_out_data=b"\x00" * _seconds(20),
472            fade_in_bytes_len=_seconds(20),
473        )
474        assert isinstance(fade, StandardCrossFade)
475        timing = fade.timing_info
476        assert timing.crossfade_duration == pytest.approx(8.0)
477        assert timing.fadein_trimmed_duration == 0.0
478        assert timing.pre_crossfade_duration == pytest.approx(12.0)
479        assert timing.post_crossfade_duration == pytest.approx(12.0)
480        # continuation offset for standard fades is just CF
481        assert timing.fadein_trimmed_duration + timing.crossfade_duration == pytest.approx(8.0)
482
483    @pytest.mark.asyncio
484    async def test_smart_mode_returns_smart_crossfade_when_analysis_available(self) -> None:
485        """With audio analysis on both tracks, build() returns a SmartCrossFade."""
486        analysis_out = _analysis(120.0)
487        analysis_in = _analysis(124.0, beats_start=0.4)
488        mixer = _make_mixer({"out": analysis_out, "in": analysis_in})
489        fade_out_data = b"\x00" * _seconds(SMART_CROSSFADE_DURATION)
490        fade = await mixer.build(
491            fade_in_streamdetails=_streamdetails("in"),
492            fade_out_streamdetails=_streamdetails("out"),
493            pcm_format=PCM,
494            standard_crossfade_duration=10,
495            mode=CrossfadeMode.SMART_CROSSFADE,
496            fade_out_data=fade_out_data,
497            fade_in_bytes_len=_seconds(SMART_CROSSFADE_DURATION),
498        )
499        assert isinstance(fade, SmartCrossFade)
500        timing = fade.timing_info
501        # SmartCrossFade applies a beat-aligned trim, so TRIM is non-zero.
502        assert timing.fadein_trimmed_duration > 0
503        assert timing.crossfade_duration > 0
504        # Invariants the flow loop depends on:
505        #   PRE + CF == rendered_fade_out_seconds  (A's audio fully accounted for)
506        #   TRIM + CF + POST == fade_in_seconds    (B's audio fully accounted for)
507        # When time-stretch is active, rendered_fade_out_seconds < buffer_duration because
508        # rubberband compresses the tail; savings come from the plan's TempoPlan.
509        rendered_fade_out_seconds = fade.effective_end - _savings_until(fade, fade.effective_end)
510        fade_in_seconds = float(SMART_CROSSFADE_DURATION)
511        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(
512            rendered_fade_out_seconds, abs=0.05
513        )
514        assert (
515            timing.fadein_trimmed_duration
516            + timing.crossfade_duration
517            + timing.post_crossfade_duration
518            == pytest.approx(fade_in_seconds, abs=0.01)
519        )
520
521    @pytest.mark.asyncio
522    async def test_smart_mode_falls_back_when_no_analysis(
523        self, monkeypatch: pytest.MonkeyPatch
524    ) -> None:
525        """No audio analysis -> build() falls back to StandardCrossFade."""
526
527        async def identity_strip(audio_data: bytes, **_kwargs: object) -> bytes:
528            return audio_data  # no stripping — test pure timing math
529
530        monkeypatch.setattr(mixer_module, "strip_silence", identity_strip)
531        mixer = _make_mixer()
532        fade = await mixer.build(
533            fade_in_streamdetails=_streamdetails("in"),
534            fade_out_streamdetails=_streamdetails("out"),
535            pcm_format=PCM,
536            standard_crossfade_duration=10,
537            mode=CrossfadeMode.SMART_CROSSFADE,
538            fade_out_data=b"\x00" * _seconds(45),
539            fade_in_bytes_len=_seconds(45),
540        )
541        assert isinstance(fade, StandardCrossFade)
542        timing = fade.timing_info
543        assert timing.crossfade_duration == pytest.approx(10.0)
544        assert timing.fadein_trimmed_duration == 0.0
545        assert timing.pre_crossfade_duration == pytest.approx(35.0)
546        assert timing.post_crossfade_duration == pytest.approx(35.0)
547
548    @pytest.mark.asyncio
549    async def test_smart_mode_falls_back_when_analysis_missing_bpm(
550        self, monkeypatch: pytest.MonkeyPatch
551    ) -> None:
552        """Analysis present but missing bpm/beats -> falls back to standard."""
553
554        async def identity_strip(audio_data: bytes, **_kwargs: object) -> bytes:
555            return audio_data
556
557        monkeypatch.setattr(mixer_module, "strip_silence", identity_strip)
558        incomplete = AudioAnalysisData(duration=180.0, bpm=None, beats=None)
559        mixer = _make_mixer({"out": incomplete, "in": incomplete})
560        fade = await mixer.build(
561            fade_in_streamdetails=_streamdetails("in"),
562            fade_out_streamdetails=_streamdetails("out"),
563            pcm_format=PCM,
564            standard_crossfade_duration=10,
565            mode=CrossfadeMode.SMART_CROSSFADE,
566            fade_out_data=b"\x00" * _seconds(30),
567            fade_in_bytes_len=_seconds(30),
568        )
569        assert isinstance(fade, StandardCrossFade)
570        assert fade.timing_info.fadein_trimmed_duration == 0.0
571
572    @pytest.mark.asyncio
573    async def test_build_returns_fade_with_timing_info_readable(
574        self, monkeypatch: pytest.MonkeyPatch
575    ) -> None:
576        """timing_info is queryable immediately after build() — no apply() needed."""
577
578        async def identity_strip(audio_data: bytes, **_kwargs: object) -> bytes:
579            return audio_data
580
581        monkeypatch.setattr(mixer_module, "strip_silence", identity_strip)
582        mixer = _make_mixer()
583        fade = await mixer.build(
584            fade_in_streamdetails=_streamdetails("in"),
585            fade_out_streamdetails=_streamdetails("out"),
586            pcm_format=PCM,
587            standard_crossfade_duration=10,
588            mode=CrossfadeMode.STANDARD_CROSSFADE,
589            fade_out_data=b"\x00" * _seconds(15),
590            fade_in_bytes_len=_seconds(15),
591        )
592        assert isinstance(fade.timing_info, CrossfadeTimingInfo)
593
594    @pytest.mark.asyncio
595    async def test_standard_mode_strips_trailing_silence_before_timing(
596        self, monkeypatch: pytest.MonkeyPatch
597    ) -> None:
598        """Timing must be computed from the stripped length; the plan is stored on the fade."""
599
600        async def fake_strip(
601            audio_data: bytes, *, reverse: bool = False, **_kwargs: object
602        ) -> bytes:
603            assert reverse is True
604            return audio_data[: -_seconds(3)]  # pretend 3s of trailing silence
605
606        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
607        mixer = _make_mixer()
608        fade_out_data = b"\x00" * _seconds(45)
609        smart_fade = await mixer.build(
610            fade_in_streamdetails=_streamdetails("b"),
611            fade_out_streamdetails=_streamdetails("a"),
612            pcm_format=PCM,
613            standard_crossfade_duration=10,
614            mode=CrossfadeMode.STANDARD_CROSSFADE,
615            fade_out_data=fade_out_data,
616            fade_in_bytes_len=_seconds(45),
617        )
618        assert isinstance(smart_fade, StandardCrossFade)
619        assert smart_fade.trailing_silence_bytes == _seconds(3)
620        timing = smart_fade.timing_info
621        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(42.0)
622
623    @pytest.mark.asyncio
624    async def test_smart_mode_never_measures_silence(self, monkeypatch: pytest.MonkeyPatch) -> None:
625        """The smart path must never call strip_silence — beat coordinates map onto the full buffer."""
626
627        async def fail_strip(*_args: object, **_kwargs: object) -> bytes:
628            raise AssertionError("strip_silence must not be called on the smart path")
629
630        monkeypatch.setattr(mixer_module, "strip_silence", fail_strip)
631        mixer = _make_mixer(analysis_for={"a": _analysis(bpm=120.0), "b": _analysis(bpm=120.0)})
632        fade_out_data = b"\x00" * _seconds(45)
633        smart_fade = await mixer.build(
634            fade_in_streamdetails=_streamdetails("b"),
635            fade_out_streamdetails=_streamdetails("a"),
636            pcm_format=PCM,
637            standard_crossfade_duration=10,
638            mode=CrossfadeMode.SMART_CROSSFADE,
639            fade_out_data=fade_out_data,
640            fade_in_bytes_len=_seconds(45),
641        )
642        assert isinstance(smart_fade, SmartCrossFade)
643
644    @pytest.mark.asyncio
645    async def test_smart_fallback_to_standard_strips(self, monkeypatch: pytest.MonkeyPatch) -> None:
646        """Smart mode without analysis falls back to standard — which must measure silence."""
647
648        async def fake_strip(audio_data: bytes, **_kwargs: object) -> bytes:
649            return audio_data[: -_seconds(5)]
650
651        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
652        mixer = _make_mixer(analysis_for={})  # no analysis available
653        smart_fade = await mixer.build(
654            fade_in_streamdetails=_streamdetails("b"),
655            fade_out_streamdetails=_streamdetails("a"),
656            pcm_format=PCM,
657            standard_crossfade_duration=10,
658            mode=CrossfadeMode.SMART_CROSSFADE,
659            fade_out_data=b"\x00" * _seconds(45),
660            fade_in_bytes_len=_seconds(45),
661        )
662        assert isinstance(smart_fade, StandardCrossFade)
663        assert smart_fade.trailing_silence_bytes == _seconds(5)
664
665    @pytest.mark.asyncio
666    async def test_smart_fallback_retains_an_audible_outgoing_vocal(
667        self, monkeypatch: pytest.MonkeyPatch
668    ) -> None:
669        """Validated FireRed activity extends a fallback trim only within audible RMS energy."""
670
671        async def fake_strip(audio_data: bytes, **_kwargs: object) -> bytes:
672            return audio_data[: _seconds(30)]
673
674        def fail_smart_build(*_args: object, **_kwargs: object) -> None:
675            raise SmartFadeNotApplicable("forced fallback")
676
677        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
678        monkeypatch.setattr(SmartCrossFade, "build", fail_smart_build)
679        outgoing = _with_vocal_activity(
680            _analysis(
681                120.0,
682                duration=240.0,
683                rms_energy=_rms_with_silent_tail(240.0, 5.0),
684            ),
685            [(228.0, 232.0)],
686        )
687        mixer = _make_mixer({"a": outgoing, "b": _analysis(120.0, duration=240.0)})
688
689        fade = await mixer.build(
690            fade_in_streamdetails=_streamdetails("b"),
691            fade_out_streamdetails=_streamdetails("a"),
692            pcm_format=PCM,
693            standard_crossfade_duration=10,
694            mode=CrossfadeMode.SMART_CROSSFADE,
695            fade_out_data=b"\x00" * _seconds(45),
696            fade_in_bytes_len=_seconds(45),
697        )
698
699        assert isinstance(fade, StandardCrossFade)
700        retained_seconds = (_seconds(45) - fade.trailing_silence_bytes) / SAMPLE_SIZE
701        assert retained_seconds == pytest.approx(37.75, abs=1 / PCM.sample_rate)
702
703    @pytest.mark.asyncio
704    async def test_smart_fallback_invalid_vocal_data_matches_missing_data(
705        self, monkeypatch: pytest.MonkeyPatch
706    ) -> None:
707        """Missing and stale vocal metadata keep the exact standard silence trim."""
708
709        async def fake_strip(audio_data: bytes, **_kwargs: object) -> bytes:
710            return audio_data[: _seconds(30)]
711
712        def fail_smart_build(*_args: object, **_kwargs: object) -> None:
713            raise SmartFadeNotApplicable("forced fallback")
714
715        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
716        monkeypatch.setattr(SmartCrossFade, "build", fail_smart_build)
717        missing = _analysis(120.0, duration=240.0, rms_energy=_rms_with_silent_tail(240.0, 5.0))
718        stale = _analysis(120.0, duration=240.0, rms_energy=_rms_with_silent_tail(240.0, 5.0))
719        stale.extra_data = {
720            "vocal_activity": {
721                "model": "firered_aed",
722                "frame_duration": 0.1,
723                "probabilities": [0.9] * 2400,
724            }
725        }
726
727        trims: list[int] = []
728        for outgoing in (missing, stale):
729            mixer = _make_mixer({"a": outgoing, "b": _analysis(120.0, duration=240.0)})
730            fade = await mixer.build(
731                fade_in_streamdetails=_streamdetails("b"),
732                fade_out_streamdetails=_streamdetails("a"),
733                pcm_format=PCM,
734                standard_crossfade_duration=10,
735                mode=CrossfadeMode.SMART_CROSSFADE,
736                fade_out_data=b"\x00" * _seconds(45),
737                fade_in_bytes_len=_seconds(45),
738            )
739            assert isinstance(fade, StandardCrossFade)
740            trims.append(fade.trailing_silence_bytes)
741
742        assert trims == [_seconds(15), _seconds(15)]
743
744    @pytest.mark.asyncio
745    async def test_smart_fallback_caps_vocal_retention_at_the_rms_boundary(
746        self, monkeypatch: pytest.MonkeyPatch
747    ) -> None:
748        """FireRed cannot restore a long low-energy tail beyond the audible RMS boundary."""
749
750        async def fake_strip(audio_data: bytes, **_kwargs: object) -> bytes:
751            return audio_data[: _seconds(20)]
752
753        def fail_smart_build(*_args: object, **_kwargs: object) -> None:
754            raise SmartFadeNotApplicable("forced fallback")
755
756        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
757        monkeypatch.setattr(SmartCrossFade, "build", fail_smart_build)
758        outgoing = _with_vocal_activity(
759            _analysis(
760                120.0,
761                duration=240.0,
762                rms_energy=_rms_with_silent_tail(240.0, 20.0),
763            ),
764            [(225.0, 235.0)],
765        )
766        mixer = _make_mixer({"a": outgoing, "b": _analysis(120.0, duration=240.0)})
767
768        fade = await mixer.build(
769            fade_in_streamdetails=_streamdetails("b"),
770            fade_out_streamdetails=_streamdetails("a"),
771            pcm_format=PCM,
772            standard_crossfade_duration=10,
773            mode=CrossfadeMode.SMART_CROSSFADE,
774            fade_out_data=b"\x00" * _seconds(45),
775            fade_in_bytes_len=_seconds(45),
776        )
777
778        assert isinstance(fade, StandardCrossFade)
779        retained_seconds = (_seconds(45) - fade.trailing_silence_bytes) / SAMPLE_SIZE
780        assert retained_seconds == pytest.approx(20.0)
781
782    @pytest.mark.asyncio
783    async def test_standard_mode_fully_silent_tail(self, monkeypatch: pytest.MonkeyPatch) -> None:
784        """A fully silent tail — timing collapses to zero; trailing_silence_bytes is the full input."""
785
786        async def fake_strip(_audio_data: bytes, **_kwargs: object) -> bytes:
787            return b""
788
789        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
790        mixer = _make_mixer()
791        smart_fade = await mixer.build(
792            fade_in_streamdetails=_streamdetails("b"),
793            fade_out_streamdetails=_streamdetails("a"),
794            pcm_format=PCM,
795            standard_crossfade_duration=10,
796            mode=CrossfadeMode.STANDARD_CROSSFADE,
797            fade_out_data=b"\x00" * _seconds(45),
798            fade_in_bytes_len=_seconds(45),
799        )
800        assert isinstance(smart_fade, StandardCrossFade)
801        assert smart_fade.trailing_silence_bytes == _seconds(45)
802        timing = smart_fade.timing_info
803        assert timing.pre_crossfade_duration == 0.0
804        assert timing.crossfade_duration == 0.0
805
806    @pytest.mark.asyncio
807    async def test_standard_mode_measurement_failure_degrades_gracefully(
808        self, monkeypatch: pytest.MonkeyPatch
809    ) -> None:
810        """A strip_silence failure must not propagate — build degrades with trailing_silence_bytes=0."""
811
812        async def broken_strip(*_args: object, **_kwargs: object) -> bytes:
813            raise OSError("ffmpeg spawn failed")
814
815        monkeypatch.setattr(mixer_module, "strip_silence", broken_strip)
816        mixer = _make_mixer()
817        fade_out_data = b"\x00" * _seconds(45)
818        smart_fade = await mixer.build(
819            fade_in_streamdetails=_streamdetails("b"),
820            fade_out_streamdetails=_streamdetails("a"),
821            pcm_format=PCM,
822            standard_crossfade_duration=10,
823            mode=CrossfadeMode.STANDARD_CROSSFADE,
824            fade_out_data=fade_out_data,
825            fade_in_bytes_len=_seconds(45),
826        )
827        assert isinstance(smart_fade, StandardCrossFade)
828        assert smart_fade.trailing_silence_bytes == 0
829        timing = smart_fade.timing_info
830        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(45.0)
831
832    @pytest.mark.asyncio
833    async def test_apply_executes_silence_trim_plan(self, monkeypatch: pytest.MonkeyPatch) -> None:
834        """apply() must slice out trailing_silence_bytes before crossfading."""
835
836        async def fake_strip(audio_data: bytes, **_kwargs: object) -> bytes:
837            return audio_data[: -_seconds(3)]  # pretend 3s of trailing silence
838
839        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
840
841        captured: dict[str, bytes] = {}
842        crossfade_marker = b"crossfade-output"
843
844        async def fake_base_apply(
845            _self: SmartFade,
846            fade_out_part: bytes,
847            _fade_in_part: bytes | AsyncGenerator[bytes],
848            _pcm_format: AudioFormat,
849        ) -> AsyncGenerator[bytes]:
850            captured["fade_out"] = fade_out_part
851            yield crossfade_marker
852
853        monkeypatch.setattr(SmartFade, "apply", fake_base_apply)
854
855        mixer = _make_mixer()
856        fade_out_data = b"\x00" * _seconds(45)
857        smart_fade = await mixer.build(
858            fade_in_streamdetails=_streamdetails("b"),
859            fade_out_streamdetails=_streamdetails("a"),
860            pcm_format=PCM,
861            standard_crossfade_duration=10,
862            mode=CrossfadeMode.STANDARD_CROSSFADE,
863            fade_out_data=fade_out_data,
864            fade_in_bytes_len=_seconds(45),
865        )
866        assert isinstance(smart_fade, StandardCrossFade)
867
868        chunks = [
869            chunk
870            async for chunk in smart_fade.apply(
871                fade_out_part=fade_out_data,
872                fade_in_part=b"\x00" * _seconds(45),
873                pcm_format=PCM,
874            )
875        ]
876
877        # The yielded pre-crossfade bytes are where the trim actually lands:
878        # 45s input - 3s measured silence - 10s clamped CF = 32s. Without the
879        # trim apply() would yield 35s here.
880        marker_idx = chunks.index(crossfade_marker)
881        pre_crossfade_bytes = sum(len(chunk) for chunk in chunks[:marker_idx])
882        assert pre_crossfade_bytes == _seconds(32)
883        assert len(captured["fade_out"]) == _seconds(10)
884
885
886# ---------------------------------------------------------------------------
887# SmartCrossFade — silence-aware fade-out anchoring
888# ---------------------------------------------------------------------------
889
890LOGGER = logging.getLogger(__name__)
891
892
893def _rms_with_silent_tail(track_duration: float, silent_tail: float) -> np.ndarray:
894    bins = np.full(1800, 0.5, dtype=np.float32)
895    bins[0] = 1.0
896    if silent_tail > 0:
897        bins[-int(silent_tail / track_duration * 1800) :] = 0.001
898    return bins
899
900
901class TestSilenceAwareAnchoring:
902    """SmartCrossFade must anchor the fade where audible content ends."""
903
904    def _build_fade(self, silent_tail: float) -> SmartCrossFade:
905        duration = 240.0
906        fade = SmartCrossFade(
907            logger=LOGGER,
908            fade_out_analysis=_analysis(
909                bpm=120.0,
910                duration=duration,
911                rms_energy=_rms_with_silent_tail(duration, silent_tail),
912            ),
913            fade_in_analysis=_analysis(bpm=120.0, duration=duration),
914        )
915        fade.build(_seconds(45), _seconds(45), PCM)
916        return fade
917
918    def test_silent_tail_moves_the_anchor(self) -> None:
919        """A 10s silent tail shortens the audible anchor to ~35s and inserts FadeOutTrimFilter."""
920        fade = self._build_fade(silent_tail=10.0)
921        assert fade.effective_end == pytest.approx(35.0, abs=0.3)
922        # the rendered fade-out covers only the audible region
923        timing = fade.timing_info
924        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(
925            fade.effective_end, abs=0.05
926        )
927        # tail trim is the FIRST filter so later schedules see the trimmed stream
928        assert isinstance(fade.filters[0], FadeOutTrimFilter)
929        assert fade.filters[0].fadeout_end_pos == pytest.approx(fade.effective_end)
930
931    def test_no_silence_keeps_buffer_end_anchor(self) -> None:
932        """Without silence, effective_end equals the full buffer and no trim filter is added."""
933        fade = self._build_fade(silent_tail=0.0)
934        assert fade.effective_end == pytest.approx(45.0, abs=0.3)
935        assert not any(isinstance(f, FadeOutTrimFilter) for f in fade.filters)
936
937    def test_sub_tolerance_silent_tail_snaps_anchor_to_buffer_end(self) -> None:
938        """A silent tail below the trim tolerance keeps the anchor at the rendered buffer end."""
939        fade = self._build_fade(silent_tail=0.4)
940        assert fade.effective_end == pytest.approx(45.0)
941        assert not any(isinstance(f, FadeOutTrimFilter) for f in fade.filters)
942
943    def test_mostly_silent_tail_raises_for_fallback(self) -> None:
944        """A tail with only ~5s of audible content raises SmartFadeNotApplicable so the caller falls back."""
945        with pytest.raises(SmartFadeNotApplicable, match="silent"):
946            self._build_fade(silent_tail=40.0)
947
948    def test_partial_buffer_keeps_beats_aligned(self) -> None:
949        """
950        The live holdback buffer is rarely exactly 45s.
951
952        Beat coordinates must use the actual buffer length or every downbeat snap
953        is off by the difference.
954        """
955        duration = 240.0
956        fade = SmartCrossFade(
957            logger=LOGGER,
958            fade_out_analysis=_analysis(bpm=120.0, duration=duration),
959            fade_in_analysis=_analysis(bpm=120.0, duration=duration),
960        )
961        # 44.3s buffer: 0.7s short of the constant, like a real partial final chunk
962        partial_bytes = int(PCM.pcm_sample_size * 44.3)
963        frame_size = (PCM.bit_depth // 8) * PCM.channels
964        partial_bytes = (partial_bytes // frame_size) * frame_size
965        fade.build(partial_bytes, _seconds(45), PCM)
966        buffer_duration = partial_bytes / PCM.pcm_sample_size
967        # beats are on a strict 0.5s grid from t=0 in the analysis fixture;
968        # in real buffer coordinates each beat must satisfy
969        # (beat + duration - buffer_duration) % 0.5 == 0
970        offset = duration - buffer_duration
971        for beat in fade.fade_out_beats[:8]:
972            track_pos = beat + offset
973            assert abs(track_pos % 0.5) < 1e-3 or abs(track_pos % 0.5 - 0.5) < 1e-3, (
974                f"beat {beat:.4f} maps to track_pos {track_pos:.4f}, "
975                f"not on 0.5s grid (offset={offset:.4f})"
976            )
977        # the snapped crossfade start must land on a real downbeat (2s grid), not 0.7s off
978        crossfade_start = fade.effective_end - fade.timing_info.crossfade_duration
979        start_track_pos = crossfade_start + offset
980        assert abs(start_track_pos % 2.0) < 0.02 or abs(start_track_pos % 2.0 - 2.0) < 0.02, (
981            f"crossfade start {crossfade_start:.4f} maps to track_pos {start_track_pos:.4f}, "
982            f"not on 2s downbeat grid (offset={offset:.4f})"
983        )
984
985    def test_fadeout_beats_are_masked_to_effective_end(self) -> None:
986        """Beats in the silent tail are dropped so no downbeat sits beyond effective_end."""
987        fade = self._build_fade(silent_tail=10.0)
988        assert fade.fade_out_beats.min() >= 0.0
989        assert fade.fade_out_beats.max() <= fade.effective_end + 0.01
990
991    def test_short_audible_tail_keeps_crossfade_inside_it(self) -> None:
992        """A crossfade longer than the audible tail is capped so no schedule goes negative."""
993        duration = 240.0
994        fade = SmartCrossFade(
995            logger=LOGGER,
996            fade_out_analysis=_analysis(
997                bpm=120.0,
998                duration=duration,
999                rms_energy=_rms_with_silent_tail(duration, 33.0),
1000            ),
1001            # 115 vs 120 BPM is within the stretch threshold, so the tempo ramp is active
1002            fade_in_analysis=_analysis(bpm=115.0, duration=duration),
1003        )
1004        fade.build(_seconds(45), _seconds(45), PCM)
1005        assert fade.timing_info.crossfade_duration <= fade.effective_end + 1e-6
1006        # the capped crossfade consumes the whole audible tail, so the stretch is skipped
1007        assert not any(isinstance(f, GradualTimeStretchFilter) for f in fade.filters)
1008
1009
1010def _rms_with_mastered_fade(
1011    track_duration: float, fade_start: float, fade_end: float
1012) -> np.ndarray:
1013    """Steady energy with the record's own gradual fade-out between the given media times."""
1014    bins = np.full(1800, 0.5, dtype=np.float32)
1015    bin_seconds = track_duration / 1800
1016    t = (np.arange(1800) + 0.5) * bin_seconds
1017    ramp = 0.5 * (1.0 - (t - fade_start) / (fade_end - fade_start)) + 0.0005
1018    return np.where(t < fade_start, bins, np.where(t >= fade_end, 0.0005, ramp)).astype(np.float32)
1019
1020
1021class TestQuickFadeMasteredFadeDeadZone:
1022    """
1023    A mastered fade-out under a quick-fade tier lands in the audible-trim dead zone.
1024
1025    The 70% mix-out floor anchors mid-fade while the 5% audible floor sits
1026    several seconds later; every quick-fade rung is far shorter than that gap,
1027    so AudibleTrimPolicy rejects the entire main candidate set. The rescue
1028    pass (ungated trim-closing ladder plus rescue rungs) then ships a
1029    late-anchored fade.
1030    """
1031
1032    def _build_fade(
1033        self, caplog: pytest.LogCaptureFixture, level: int = logging.DEBUG
1034    ) -> SmartCrossFade:
1035        duration = 240.0
1036        fade = SmartCrossFade(
1037            logger=LOGGER,
1038            fade_out_analysis=_analysis(
1039                bpm=128.0,
1040                duration=duration,
1041                # the record fades itself out over 229s..237s: mix-out (70% floor)
1042                # anchors near 231s while audible content (5% floor) runs to ~237s
1043                rms_energy=_rms_with_mastered_fade(duration, 229.0, 237.0),
1044            ),
1045            # 17.2% BPM gap: QUICK_FADE with the [2, 1] rung ladder (3.75s / 1.88s)
1046            fade_in_analysis=_analysis(bpm=150.0, duration=duration),
1047        )
1048        with caplog.at_level(level):
1049            fade.build(_seconds(45), _seconds(45), PCM)
1050        return fade
1051
1052    def test_main_pass_rejects_every_candidate_on_the_trim_guard_alone(
1053        self, caplog: pytest.LogCaptureFixture
1054    ) -> None:
1055        """Every main-pass candidate dies on the one guard; the rescue pass ships the fade."""
1056        self._build_fade(caplog)
1057        assert (
1058            "all 2 candidates rejected (audible trim exceeds a short fade's own duration x2)"
1059            in caplog.text
1060        )
1061        assert (
1062            "shipping a rescue-pass candidate (source=rescue-anchor) instead of the "
1063            "emergency handoff" in caplog.text
1064        )
1065
1066    def test_rescue_ships_a_late_anchored_chain_within_the_trim_bound(
1067        self, caplog: pytest.LogCaptureFixture
1068    ) -> None:
1069        """The shipped chain anchors near the audible end, honoring the short-fade trim bound."""
1070        fade = self._build_fade(caplog)
1071        # rescue anchor: last protective downbeat at/after audio_end - 2 bars (128 BPM)
1072        assert fade.effective_end == pytest.approx(41.25, abs=0.05)
1073        assert isinstance(fade.filters[0], FadeOutTrimFilter)
1074        assert fade.filters[0].fadeout_end_pos == pytest.approx(fade.effective_end)
1075        # the guard's own invariant holds on the shipped plan: audible material
1076        # dropped past the anchor stays within the overlap length (~41.67s RMS boundary)
1077        audible_trim = 41.67 - fade.effective_end
1078        assert audible_trim <= fade.timing_info.crossfade_duration + 1e-6
1079
1080    def test_rescue_pass_scores_trim_closing_candidates(
1081        self, caplog: pytest.LogCaptureFixture
1082    ) -> None:
1083        """The rescue pass runs the audible-end ladder ungated, so the selector scores it."""
1084        self._build_fade(caplog, level=VERBOSE_LOG_LEVEL)
1085        assert "source=trim-closing-anchor" in caplog.text
1086
1087    def test_trim_closing_wins_when_the_ladder_outgrows_the_rescue_rung(
1088        self, caplog: pytest.LogCaptureFixture
1089    ) -> None:
1090        """A 4-bar dead zone ships the audible-end ladder rung, not the capped rescue rung."""
1091        duration = 240.0
1092        fade = SmartCrossFade(
1093            logger=LOGGER,
1094            fade_out_analysis=_analysis(
1095                bpm=128.0,
1096                duration=duration,
1097                # longer mastered fade: the 7.78s trim gap exceeds even the 4-bar
1098                # rung (7.5s) yet stays under the trim-closing generator's 8s gate
1099                rms_energy=_rms_with_mastered_fade(duration, 228.4, 238.9),
1100            ),
1101            # 9.4% BPM gap: QUICK_FADE with the [4, 2, 1] rung ladder
1102            fade_in_analysis=_analysis(bpm=140.0, duration=duration),
1103        )
1104        with caplog.at_level(logging.DEBUG):
1105            fade.build(_seconds(45), _seconds(45), PCM)
1106        assert "shipping a rescue-pass candidate (source=trim-closing-anchor)" in caplog.text
1107        # the audible-end anchor keeps the full 4-bar overlap and trims nothing audible
1108        assert fade.effective_end == pytest.approx(43.40, abs=0.05)
1109        assert fade.timing_info.crossfade_duration == pytest.approx(7.78, abs=0.05)
1110
1111
1112# ---------------------------------------------------------------------------
1113# SmartCrossFade — rubberband stretch savings compensation
1114# ---------------------------------------------------------------------------
1115
1116
1117def _savings_until(fade: SmartCrossFade, t: float) -> float:
1118    """Rendered-time savings of the built fade's tempo plan up to input time t."""
1119    assert fade.plan is not None
1120    return fade.plan.tempo_plan.savings_until(t)
1121
1122
1123class TestStretchSavings:
1124    """
1125    Rendered-time savings from the stretch must reach the timing bookkeeping.
1126
1127    The savings integration math itself is unit-tested on TempoPlan in
1128    tests/controllers/streams/smart_fades/test_models.py.
1129    """
1130
1131    def _stretched_fade(self) -> SmartCrossFade:
1132        duration = 240.0
1133        # 4% BPM difference with >4 bars available -> stretch is applied
1134        fade = SmartCrossFade(
1135            logger=LOGGER,
1136            fade_out_analysis=_analysis(bpm=120.0, duration=duration),
1137            fade_in_analysis=_analysis(bpm=124.8, duration=duration),
1138        )
1139        fade.build(_seconds(45), _seconds(45), PCM)
1140        return fade
1141
1142    def test_pre_plus_cf_equals_rendered_tail(self) -> None:
1143        """PRE + CF equals the rendered tail duration (buffer minus stretch savings)."""
1144        fade = self._stretched_fade()
1145        assert fade.tempo_steps, "test requires the stretch to be active"
1146        total_savings = _savings_until(fade, fade.effective_end)
1147        assert total_savings > 0.0
1148        timing = fade.timing_info
1149        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(
1150            fade.effective_end - total_savings, abs=0.05
1151        )
1152
1153    def test_pre_plus_cf_equals_rendered_tail_when_slowing_down(self) -> None:
1154        """A slower incoming track lengthens the rendered tail — PRE + CF exceeds effective_end."""
1155        duration = 240.0
1156        # ~3.8% BPM difference downwards -> stretch slows the outgoing track
1157        fade = SmartCrossFade(
1158            logger=LOGGER,
1159            fade_out_analysis=_analysis(bpm=120.0, duration=duration),
1160            fade_in_analysis=_analysis(bpm=115.4, duration=duration),
1161        )
1162        fade.build(_seconds(45), _seconds(45), PCM)
1163        assert fade.tempo_steps, "test requires the stretch to be active"
1164        total_savings = _savings_until(fade, fade.effective_end)
1165        assert total_savings < 0.0
1166        timing = fade.timing_info
1167        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(
1168            fade.effective_end - total_savings, abs=0.05
1169        )
1170
1171    def test_bass_kill_completes_at_the_anchor(self) -> None:
1172        """
1173        The outgoing low-shelf kill reaches full depth at or before the audible end.
1174
1175        A-side shelves render BEFORE the rubberband stretch, so their schedules
1176        live in musical input time — no rendered-time remap needed (unlike the
1177        old post-stretch frequency sweeps).
1178        """
1179        fade = self._stretched_fade()
1180        assert fade.plan is not None
1181        low_out = fade.plan.eq_plan.low_out
1182        assert low_out is not None
1183        assert low_out.steps[-1][1] == pytest.approx(-26.0)
1184        assert low_out.steps[-1][0] <= fade.effective_end + 0.05
1185
1186    def test_trim_and_stretch_combined(self) -> None:
1187        """A trimmed silent tail and an active stretch compose: both anchor on the rendered end."""
1188        duration = 240.0
1189        fade = SmartCrossFade(
1190            logger=LOGGER,
1191            fade_out_analysis=_analysis(
1192                bpm=120.0,
1193                duration=duration,
1194                rms_energy=_rms_with_silent_tail(duration, 10.0),
1195            ),
1196            fade_in_analysis=_analysis(bpm=124.8, duration=duration),
1197        )
1198        fade.build(_seconds(45), _seconds(45), PCM)
1199        # tail trim must come first so every later schedule sees the trimmed stream
1200        assert isinstance(fade.filters[0], FadeOutTrimFilter)
1201        assert fade.tempo_steps, "test requires the stretch to be active"
1202        rendered_end = fade.effective_end - _savings_until(fade, fade.effective_end)
1203        timing = fade.timing_info
1204        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(
1205            rendered_end, abs=0.05
1206        )
1207
1208    def test_unstretched_fade_has_zero_savings(self) -> None:
1209        """Without a stretch, savings are zero and PRE + CF equals effective_end exactly."""
1210        fade = SmartCrossFade(
1211            logger=LOGGER,
1212            fade_out_analysis=_analysis(bpm=120.0, duration=240.0),
1213            fade_in_analysis=_analysis(bpm=120.0, duration=240.0),
1214        )
1215        fade.build(_seconds(45), _seconds(45), PCM)
1216        assert fade.tempo_steps == []
1217        assert _savings_until(fade, fade.effective_end) == 0.0
1218        timing = fade.timing_info
1219        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(
1220            fade.effective_end, abs=0.05
1221        )
1222