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    CrossfadeFilter,
40    FadeOutTrimFilter,
41    GradualTimeStretchFilter,
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 acrossfade 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, CrossfadeFilter)
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 acrossfade ``ns=`` 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, CrossfadeFilter)
261        # the source-of-truth byte size is frame-aligned ...
262        assert fade.crossfade_size % frame_size == 0
263        # ... and the acrossfade 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 acrossfade ``ns=`` sample count.
312
313        Otherwise ffmpeg's acrossfade receives fewer samples than requested and emits
314        nothing — the silent crossfade failure this regression guards against.
315        """
316        captured: dict[str, bytes] = {}
317
318        async def fake_base_apply(
319            _self: SmartFade,
320            fade_out_part: bytes,
321            fade_in_part: bytes | AsyncGenerator[bytes],
322            _pcm_format: AudioFormat,
323        ) -> AsyncGenerator[bytes]:
324            captured["fade_out"] = fade_out_part
325            assert isinstance(fade_in_part, bytes)
326            captured["fade_in"] = fade_in_part
327            yield b"crossfade-output"
328
329        monkeypatch.setattr(SmartFade, "apply", fake_base_apply)
330        frame_size = (PCM.bit_depth // 8) * PCM.channels
331        # frame-aligned like a real PCM buffer, but a fractional number of seconds
332        fade_out_len = _seconds(6.3333) // frame_size * frame_size
333        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=10.0)
334        fade.build(fade_out_len, _seconds(45), PCM)
335        crossfade_filter = fade.filters[0]
336        assert isinstance(crossfade_filter, CrossfadeFilter)
337        assert crossfade_filter.crossfade_samples is not None
338        async for _ in fade.apply(b"\x00" * fade_out_len, b"\x11" * _seconds(45), PCM):
339            pass
340        expected_bytes = crossfade_filter.crossfade_samples * frame_size
341        assert len(captured["fade_out"]) == expected_bytes
342        assert len(captured["fade_in"]) == expected_bytes
343
344    @pytest.mark.asyncio
345    async def test_apply_before_build_fails_fast(self) -> None:
346        """apply() without a prior build() must error, not silently hard-cut."""
347        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=10.0)
348        with pytest.raises(RuntimeError, match="not built"):
349            async for _ in fade.apply(b"\x00" * _seconds(5), b"\x11" * _seconds(5), PCM):
350                pass
351
352    @pytest.mark.asyncio
353    async def test_zero_crossfade_skips_ffmpeg(self, monkeypatch: pytest.MonkeyPatch) -> None:
354        """When crossfade_duration == 0, apply() must concatenate without calling ffmpeg."""
355        base_apply_invoked: list[bool] = []
356
357        async def _base_apply_sentinel(
358            _self: SmartFade,
359            _fade_out_part: bytes,
360            _fade_in_part: bytes | AsyncGenerator[bytes],
361            _pcm_format: AudioFormat,
362        ) -> AsyncGenerator[bytes]:
363            base_apply_invoked.append(True)
364            yield b""
365
366        monkeypatch.setattr(SmartFade, "apply", _base_apply_sentinel)
367
368        fade = StandardCrossFade(logger=logging.getLogger(), crossfade_duration=10.0)
369        # fade_out_bytes_len=0 → effective_cf = min(10, 0, 45) = 0 → crossfade_duration == 0
370        fade.build(0, _seconds(45), PCM)
371        assert fade.timing_info.crossfade_duration == 0.0
372
373        fade_out_data = b"\x00" * _seconds(5)
374        fade_in_data = b"\x11" * _seconds(5)
375        chunks = [chunk async for chunk in fade.apply(fade_out_data, fade_in_data, PCM)]
376        combined = b"".join(chunks)
377        assert len(combined) == _seconds(10), f"Expected {_seconds(10)} bytes, got {len(combined)}"
378        assert combined[0:1] == b"\x00", "fade_out bytes should come first"
379        assert combined[-1:] == b"\x11", "fade_in bytes should come last"
380        assert not base_apply_invoked, (
381            "SmartFade.apply (ffmpeg path) must not be called for zero-length crossfade"
382        )
383
384
385# ---------------------------------------------------------------------------
386# Pure math/invariant tests via _FixedTimingFade (no beat-alignment dependency)
387# ---------------------------------------------------------------------------
388
389
390class TestLyricsSyncInvariants:
391    """Invariants the flow loop and per-track loop rely on."""
392
393    def _continuation_offset(self, t: CrossfadeTimingInfo) -> float:
394        """Return the value the flow loop writes into streamdetails.seek_position."""
395        return t.fadein_trimmed_duration + t.crossfade_duration
396
397    def _fadeout_share(self, t: CrossfadeTimingInfo) -> float:
398        """Return the seconds of mix output attributed to the outgoing track."""
399        return t.pre_crossfade_duration + t.crossfade_duration
400
401    def test_continuation_offset_no_trim(self) -> None:
402        """Without trim the listener is exactly CF seconds into the incoming track."""
403        timing = CrossfadeTimingInfo(
404            pre_crossfade_duration=20.0,
405            crossfade_duration=10.0,
406            fadein_trimmed_duration=0.0,
407            post_crossfade_duration=20.0,
408        )
409        assert self._continuation_offset(timing) == pytest.approx(10.0)
410
411    def test_continuation_offset_with_trim(self) -> None:
412        """With trim the listener is TRIM + CF seconds into the incoming track."""
413        timing = CrossfadeTimingInfo(
414            pre_crossfade_duration=29.0,
415            crossfade_duration=16.0,
416            fadein_trimmed_duration=3.0,
417            post_crossfade_duration=26.0,
418        )
419        assert self._continuation_offset(timing) == pytest.approx(19.0)
420
421    def test_fadeout_share_accounts_for_full_outgoing_input(self) -> None:
422        """PRE + CF equals the full outgoing input — A is fully accounted for."""
423        timing = CrossfadeTimingInfo(
424            pre_crossfade_duration=29.0,
425            crossfade_duration=16.0,
426            fadein_trimmed_duration=3.0,
427            post_crossfade_duration=26.0,
428        )
429        # fade_out_seconds = PRE + CF = 45
430        assert self._fadeout_share(timing) == pytest.approx(45.0)
431
432    def test_fixed_timing_fade_round_trip(self) -> None:
433        """_FixedTimingFade.timing_info returns whatever was passed in."""
434        original = CrossfadeTimingInfo(
435            pre_crossfade_duration=1.0,
436            crossfade_duration=2.0,
437            fadein_trimmed_duration=3.0,
438            post_crossfade_duration=4.0,
439        )
440        fade = _FixedTimingFade(original)
441        fade.build(0, 0, PCM)
442        assert fade.timing_info == original
443
444
445# ---------------------------------------------------------------------------
446# SmartFadesMixer.build() — the entry point the flow loop calls
447# ---------------------------------------------------------------------------
448
449
450class TestMixerBuild:
451    """build() resolves smart vs standard, primes filters, and returns a SmartFade."""
452
453    @pytest.mark.asyncio
454    async def test_standard_mode_returns_standard_crossfade(
455        self, monkeypatch: pytest.MonkeyPatch
456    ) -> None:
457        """Standard mode always builds a StandardCrossFade with the configured duration."""
458
459        async def identity_strip(audio_data: bytes, **_kwargs: object) -> bytes:
460            return audio_data  # no stripping — test pure timing math
461
462        monkeypatch.setattr(mixer_module, "strip_silence", identity_strip)
463        mixer = _make_mixer()
464        fade = await mixer.build(
465            fade_in_streamdetails=_streamdetails("in"),
466            fade_out_streamdetails=_streamdetails("out"),
467            pcm_format=PCM,
468            standard_crossfade_duration=8,
469            mode=CrossfadeMode.STANDARD_CROSSFADE,
470            fade_out_data=b"\x00" * _seconds(20),
471            fade_in_bytes_len=_seconds(20),
472        )
473        assert isinstance(fade, StandardCrossFade)
474        timing = fade.timing_info
475        assert timing.crossfade_duration == pytest.approx(8.0)
476        assert timing.fadein_trimmed_duration == 0.0
477        assert timing.pre_crossfade_duration == pytest.approx(12.0)
478        assert timing.post_crossfade_duration == pytest.approx(12.0)
479        # continuation offset for standard fades is just CF
480        assert timing.fadein_trimmed_duration + timing.crossfade_duration == pytest.approx(8.0)
481
482    @pytest.mark.asyncio
483    async def test_smart_mode_returns_smart_crossfade_when_analysis_available(self) -> None:
484        """With audio analysis on both tracks, build() returns a SmartCrossFade."""
485        analysis_out = _analysis(120.0)
486        analysis_in = _analysis(124.0, beats_start=0.4)
487        mixer = _make_mixer({"out": analysis_out, "in": analysis_in})
488        fade_out_data = b"\x00" * _seconds(SMART_CROSSFADE_DURATION)
489        fade = await mixer.build(
490            fade_in_streamdetails=_streamdetails("in"),
491            fade_out_streamdetails=_streamdetails("out"),
492            pcm_format=PCM,
493            standard_crossfade_duration=10,
494            mode=CrossfadeMode.SMART_CROSSFADE,
495            fade_out_data=fade_out_data,
496            fade_in_bytes_len=_seconds(SMART_CROSSFADE_DURATION),
497        )
498        assert isinstance(fade, SmartCrossFade)
499        timing = fade.timing_info
500        # SmartCrossFade applies a beat-aligned trim, so TRIM is non-zero.
501        assert timing.fadein_trimmed_duration > 0
502        assert timing.crossfade_duration > 0
503        # Invariants the flow loop depends on:
504        #   PRE + CF == rendered_fade_out_seconds  (A's audio fully accounted for)
505        #   TRIM + CF + POST == fade_in_seconds    (B's audio fully accounted for)
506        # When time-stretch is active, rendered_fade_out_seconds < buffer_duration because
507        # rubberband compresses the tail; savings come from the plan's TempoPlan.
508        rendered_fade_out_seconds = fade.effective_end - _savings_until(fade, fade.effective_end)
509        fade_in_seconds = float(SMART_CROSSFADE_DURATION)
510        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(
511            rendered_fade_out_seconds, abs=0.05
512        )
513        assert (
514            timing.fadein_trimmed_duration
515            + timing.crossfade_duration
516            + timing.post_crossfade_duration
517            == pytest.approx(fade_in_seconds, abs=0.01)
518        )
519
520    @pytest.mark.asyncio
521    async def test_smart_mode_falls_back_when_no_analysis(
522        self, monkeypatch: pytest.MonkeyPatch
523    ) -> None:
524        """No audio analysis -> build() falls back to StandardCrossFade."""
525
526        async def identity_strip(audio_data: bytes, **_kwargs: object) -> bytes:
527            return audio_data  # no stripping — test pure timing math
528
529        monkeypatch.setattr(mixer_module, "strip_silence", identity_strip)
530        mixer = _make_mixer()
531        fade = await mixer.build(
532            fade_in_streamdetails=_streamdetails("in"),
533            fade_out_streamdetails=_streamdetails("out"),
534            pcm_format=PCM,
535            standard_crossfade_duration=10,
536            mode=CrossfadeMode.SMART_CROSSFADE,
537            fade_out_data=b"\x00" * _seconds(45),
538            fade_in_bytes_len=_seconds(45),
539        )
540        assert isinstance(fade, StandardCrossFade)
541        timing = fade.timing_info
542        assert timing.crossfade_duration == pytest.approx(10.0)
543        assert timing.fadein_trimmed_duration == 0.0
544        assert timing.pre_crossfade_duration == pytest.approx(35.0)
545        assert timing.post_crossfade_duration == pytest.approx(35.0)
546
547    @pytest.mark.asyncio
548    async def test_smart_mode_falls_back_when_analysis_missing_bpm(
549        self, monkeypatch: pytest.MonkeyPatch
550    ) -> None:
551        """Analysis present but missing bpm/beats -> falls back to standard."""
552
553        async def identity_strip(audio_data: bytes, **_kwargs: object) -> bytes:
554            return audio_data
555
556        monkeypatch.setattr(mixer_module, "strip_silence", identity_strip)
557        incomplete = AudioAnalysisData(duration=180.0, bpm=None, beats=None)
558        mixer = _make_mixer({"out": incomplete, "in": incomplete})
559        fade = await mixer.build(
560            fade_in_streamdetails=_streamdetails("in"),
561            fade_out_streamdetails=_streamdetails("out"),
562            pcm_format=PCM,
563            standard_crossfade_duration=10,
564            mode=CrossfadeMode.SMART_CROSSFADE,
565            fade_out_data=b"\x00" * _seconds(30),
566            fade_in_bytes_len=_seconds(30),
567        )
568        assert isinstance(fade, StandardCrossFade)
569        assert fade.timing_info.fadein_trimmed_duration == 0.0
570
571    @pytest.mark.asyncio
572    async def test_build_returns_fade_with_timing_info_readable(
573        self, monkeypatch: pytest.MonkeyPatch
574    ) -> None:
575        """timing_info is queryable immediately after build() — no apply() needed."""
576
577        async def identity_strip(audio_data: bytes, **_kwargs: object) -> bytes:
578            return audio_data
579
580        monkeypatch.setattr(mixer_module, "strip_silence", identity_strip)
581        mixer = _make_mixer()
582        fade = await mixer.build(
583            fade_in_streamdetails=_streamdetails("in"),
584            fade_out_streamdetails=_streamdetails("out"),
585            pcm_format=PCM,
586            standard_crossfade_duration=10,
587            mode=CrossfadeMode.STANDARD_CROSSFADE,
588            fade_out_data=b"\x00" * _seconds(15),
589            fade_in_bytes_len=_seconds(15),
590        )
591        assert isinstance(fade.timing_info, CrossfadeTimingInfo)
592
593    @pytest.mark.asyncio
594    async def test_standard_mode_strips_trailing_silence_before_timing(
595        self, monkeypatch: pytest.MonkeyPatch
596    ) -> None:
597        """Timing must be computed from the stripped length; the plan is stored on the fade."""
598
599        async def fake_strip(
600            audio_data: bytes, *, reverse: bool = False, **_kwargs: object
601        ) -> bytes:
602            assert reverse is True
603            return audio_data[: -_seconds(3)]  # pretend 3s of trailing silence
604
605        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
606        mixer = _make_mixer()
607        fade_out_data = b"\x00" * _seconds(45)
608        smart_fade = await mixer.build(
609            fade_in_streamdetails=_streamdetails("b"),
610            fade_out_streamdetails=_streamdetails("a"),
611            pcm_format=PCM,
612            standard_crossfade_duration=10,
613            mode=CrossfadeMode.STANDARD_CROSSFADE,
614            fade_out_data=fade_out_data,
615            fade_in_bytes_len=_seconds(45),
616        )
617        assert isinstance(smart_fade, StandardCrossFade)
618        assert smart_fade.trailing_silence_bytes == _seconds(3)
619        timing = smart_fade.timing_info
620        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(42.0)
621
622    @pytest.mark.asyncio
623    async def test_smart_mode_never_measures_silence(self, monkeypatch: pytest.MonkeyPatch) -> None:
624        """The smart path must never call strip_silence — beat coordinates map onto the full buffer."""
625
626        async def fail_strip(*_args: object, **_kwargs: object) -> bytes:
627            raise AssertionError("strip_silence must not be called on the smart path")
628
629        monkeypatch.setattr(mixer_module, "strip_silence", fail_strip)
630        mixer = _make_mixer(analysis_for={"a": _analysis(bpm=120.0), "b": _analysis(bpm=120.0)})
631        fade_out_data = b"\x00" * _seconds(45)
632        smart_fade = await mixer.build(
633            fade_in_streamdetails=_streamdetails("b"),
634            fade_out_streamdetails=_streamdetails("a"),
635            pcm_format=PCM,
636            standard_crossfade_duration=10,
637            mode=CrossfadeMode.SMART_CROSSFADE,
638            fade_out_data=fade_out_data,
639            fade_in_bytes_len=_seconds(45),
640        )
641        assert isinstance(smart_fade, SmartCrossFade)
642
643    @pytest.mark.asyncio
644    async def test_smart_fallback_to_standard_strips(self, monkeypatch: pytest.MonkeyPatch) -> None:
645        """Smart mode without analysis falls back to standard — which must measure silence."""
646
647        async def fake_strip(audio_data: bytes, **_kwargs: object) -> bytes:
648            return audio_data[: -_seconds(5)]
649
650        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
651        mixer = _make_mixer(analysis_for={})  # no analysis available
652        smart_fade = await mixer.build(
653            fade_in_streamdetails=_streamdetails("b"),
654            fade_out_streamdetails=_streamdetails("a"),
655            pcm_format=PCM,
656            standard_crossfade_duration=10,
657            mode=CrossfadeMode.SMART_CROSSFADE,
658            fade_out_data=b"\x00" * _seconds(45),
659            fade_in_bytes_len=_seconds(45),
660        )
661        assert isinstance(smart_fade, StandardCrossFade)
662        assert smart_fade.trailing_silence_bytes == _seconds(5)
663
664    @pytest.mark.asyncio
665    async def test_smart_fallback_retains_an_audible_outgoing_vocal(
666        self, monkeypatch: pytest.MonkeyPatch
667    ) -> None:
668        """Validated FireRed activity extends a fallback trim only within audible RMS energy."""
669
670        async def fake_strip(audio_data: bytes, **_kwargs: object) -> bytes:
671            return audio_data[: _seconds(30)]
672
673        def fail_smart_build(*_args: object, **_kwargs: object) -> None:
674            raise SmartFadeNotApplicable("forced fallback")
675
676        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
677        monkeypatch.setattr(SmartCrossFade, "build", fail_smart_build)
678        outgoing = _with_vocal_activity(
679            _analysis(
680                120.0,
681                duration=240.0,
682                rms_energy=_rms_with_silent_tail(240.0, 5.0),
683            ),
684            [(228.0, 232.0)],
685        )
686        mixer = _make_mixer({"a": outgoing, "b": _analysis(120.0, duration=240.0)})
687
688        fade = await mixer.build(
689            fade_in_streamdetails=_streamdetails("b"),
690            fade_out_streamdetails=_streamdetails("a"),
691            pcm_format=PCM,
692            standard_crossfade_duration=10,
693            mode=CrossfadeMode.SMART_CROSSFADE,
694            fade_out_data=b"\x00" * _seconds(45),
695            fade_in_bytes_len=_seconds(45),
696        )
697
698        assert isinstance(fade, StandardCrossFade)
699        retained_seconds = (_seconds(45) - fade.trailing_silence_bytes) / SAMPLE_SIZE
700        assert retained_seconds == pytest.approx(37.75, abs=1 / PCM.sample_rate)
701
702    @pytest.mark.asyncio
703    async def test_smart_fallback_invalid_vocal_data_matches_missing_data(
704        self, monkeypatch: pytest.MonkeyPatch
705    ) -> None:
706        """Missing and stale vocal metadata keep the exact standard silence trim."""
707
708        async def fake_strip(audio_data: bytes, **_kwargs: object) -> bytes:
709            return audio_data[: _seconds(30)]
710
711        def fail_smart_build(*_args: object, **_kwargs: object) -> None:
712            raise SmartFadeNotApplicable("forced fallback")
713
714        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
715        monkeypatch.setattr(SmartCrossFade, "build", fail_smart_build)
716        missing = _analysis(120.0, duration=240.0, rms_energy=_rms_with_silent_tail(240.0, 5.0))
717        stale = _analysis(120.0, duration=240.0, rms_energy=_rms_with_silent_tail(240.0, 5.0))
718        stale.extra_data = {
719            "vocal_activity": {
720                "model": "firered_aed",
721                "frame_duration": 0.1,
722                "probabilities": [0.9] * 2400,
723            }
724        }
725
726        trims: list[int] = []
727        for outgoing in (missing, stale):
728            mixer = _make_mixer({"a": outgoing, "b": _analysis(120.0, duration=240.0)})
729            fade = await mixer.build(
730                fade_in_streamdetails=_streamdetails("b"),
731                fade_out_streamdetails=_streamdetails("a"),
732                pcm_format=PCM,
733                standard_crossfade_duration=10,
734                mode=CrossfadeMode.SMART_CROSSFADE,
735                fade_out_data=b"\x00" * _seconds(45),
736                fade_in_bytes_len=_seconds(45),
737            )
738            assert isinstance(fade, StandardCrossFade)
739            trims.append(fade.trailing_silence_bytes)
740
741        assert trims == [_seconds(15), _seconds(15)]
742
743    @pytest.mark.asyncio
744    async def test_smart_fallback_caps_vocal_retention_at_the_rms_boundary(
745        self, monkeypatch: pytest.MonkeyPatch
746    ) -> None:
747        """FireRed cannot restore a long low-energy tail beyond the audible RMS boundary."""
748
749        async def fake_strip(audio_data: bytes, **_kwargs: object) -> bytes:
750            return audio_data[: _seconds(20)]
751
752        def fail_smart_build(*_args: object, **_kwargs: object) -> None:
753            raise SmartFadeNotApplicable("forced fallback")
754
755        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
756        monkeypatch.setattr(SmartCrossFade, "build", fail_smart_build)
757        outgoing = _with_vocal_activity(
758            _analysis(
759                120.0,
760                duration=240.0,
761                rms_energy=_rms_with_silent_tail(240.0, 20.0),
762            ),
763            [(225.0, 235.0)],
764        )
765        mixer = _make_mixer({"a": outgoing, "b": _analysis(120.0, duration=240.0)})
766
767        fade = await mixer.build(
768            fade_in_streamdetails=_streamdetails("b"),
769            fade_out_streamdetails=_streamdetails("a"),
770            pcm_format=PCM,
771            standard_crossfade_duration=10,
772            mode=CrossfadeMode.SMART_CROSSFADE,
773            fade_out_data=b"\x00" * _seconds(45),
774            fade_in_bytes_len=_seconds(45),
775        )
776
777        assert isinstance(fade, StandardCrossFade)
778        retained_seconds = (_seconds(45) - fade.trailing_silence_bytes) / SAMPLE_SIZE
779        assert retained_seconds == pytest.approx(20.0)
780
781    @pytest.mark.asyncio
782    async def test_standard_mode_fully_silent_tail(self, monkeypatch: pytest.MonkeyPatch) -> None:
783        """A fully silent tail — timing collapses to zero; trailing_silence_bytes is the full input."""
784
785        async def fake_strip(_audio_data: bytes, **_kwargs: object) -> bytes:
786            return b""
787
788        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
789        mixer = _make_mixer()
790        smart_fade = await mixer.build(
791            fade_in_streamdetails=_streamdetails("b"),
792            fade_out_streamdetails=_streamdetails("a"),
793            pcm_format=PCM,
794            standard_crossfade_duration=10,
795            mode=CrossfadeMode.STANDARD_CROSSFADE,
796            fade_out_data=b"\x00" * _seconds(45),
797            fade_in_bytes_len=_seconds(45),
798        )
799        assert isinstance(smart_fade, StandardCrossFade)
800        assert smart_fade.trailing_silence_bytes == _seconds(45)
801        timing = smart_fade.timing_info
802        assert timing.pre_crossfade_duration == 0.0
803        assert timing.crossfade_duration == 0.0
804
805    @pytest.mark.asyncio
806    async def test_standard_mode_measurement_failure_degrades_gracefully(
807        self, monkeypatch: pytest.MonkeyPatch
808    ) -> None:
809        """A strip_silence failure must not propagate — build degrades with trailing_silence_bytes=0."""
810
811        async def broken_strip(*_args: object, **_kwargs: object) -> bytes:
812            raise OSError("ffmpeg spawn failed")
813
814        monkeypatch.setattr(mixer_module, "strip_silence", broken_strip)
815        mixer = _make_mixer()
816        fade_out_data = b"\x00" * _seconds(45)
817        smart_fade = await mixer.build(
818            fade_in_streamdetails=_streamdetails("b"),
819            fade_out_streamdetails=_streamdetails("a"),
820            pcm_format=PCM,
821            standard_crossfade_duration=10,
822            mode=CrossfadeMode.STANDARD_CROSSFADE,
823            fade_out_data=fade_out_data,
824            fade_in_bytes_len=_seconds(45),
825        )
826        assert isinstance(smart_fade, StandardCrossFade)
827        assert smart_fade.trailing_silence_bytes == 0
828        timing = smart_fade.timing_info
829        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(45.0)
830
831    @pytest.mark.asyncio
832    async def test_apply_executes_silence_trim_plan(self, monkeypatch: pytest.MonkeyPatch) -> None:
833        """apply() must slice out trailing_silence_bytes before crossfading."""
834
835        async def fake_strip(audio_data: bytes, **_kwargs: object) -> bytes:
836            return audio_data[: -_seconds(3)]  # pretend 3s of trailing silence
837
838        monkeypatch.setattr(mixer_module, "strip_silence", fake_strip)
839
840        captured: dict[str, bytes] = {}
841        crossfade_marker = b"crossfade-output"
842
843        async def fake_base_apply(
844            _self: SmartFade,
845            fade_out_part: bytes,
846            _fade_in_part: bytes | AsyncGenerator[bytes],
847            _pcm_format: AudioFormat,
848        ) -> AsyncGenerator[bytes]:
849            captured["fade_out"] = fade_out_part
850            yield crossfade_marker
851
852        monkeypatch.setattr(SmartFade, "apply", fake_base_apply)
853
854        mixer = _make_mixer()
855        fade_out_data = b"\x00" * _seconds(45)
856        smart_fade = await mixer.build(
857            fade_in_streamdetails=_streamdetails("b"),
858            fade_out_streamdetails=_streamdetails("a"),
859            pcm_format=PCM,
860            standard_crossfade_duration=10,
861            mode=CrossfadeMode.STANDARD_CROSSFADE,
862            fade_out_data=fade_out_data,
863            fade_in_bytes_len=_seconds(45),
864        )
865        assert isinstance(smart_fade, StandardCrossFade)
866
867        chunks = [
868            chunk
869            async for chunk in smart_fade.apply(
870                fade_out_part=fade_out_data,
871                fade_in_part=b"\x00" * _seconds(45),
872                pcm_format=PCM,
873            )
874        ]
875
876        # The yielded pre-crossfade bytes are where the trim actually lands:
877        # 45s input - 3s measured silence - 10s clamped CF = 32s. Without the
878        # trim apply() would yield 35s here.
879        marker_idx = chunks.index(crossfade_marker)
880        pre_crossfade_bytes = sum(len(chunk) for chunk in chunks[:marker_idx])
881        assert pre_crossfade_bytes == _seconds(32)
882        assert len(captured["fade_out"]) == _seconds(10)
883
884
885# ---------------------------------------------------------------------------
886# SmartCrossFade — silence-aware fade-out anchoring
887# ---------------------------------------------------------------------------
888
889LOGGER = logging.getLogger(__name__)
890
891
892def _rms_with_silent_tail(track_duration: float, silent_tail: float) -> np.ndarray:
893    bins = np.full(1800, 0.5, dtype=np.float32)
894    bins[0] = 1.0
895    if silent_tail > 0:
896        bins[-int(silent_tail / track_duration * 1800) :] = 0.001
897    return bins
898
899
900class TestSilenceAwareAnchoring:
901    """SmartCrossFade must anchor the fade where audible content ends."""
902
903    def _build_fade(self, silent_tail: float) -> SmartCrossFade:
904        duration = 240.0
905        fade = SmartCrossFade(
906            logger=LOGGER,
907            fade_out_analysis=_analysis(
908                bpm=120.0,
909                duration=duration,
910                rms_energy=_rms_with_silent_tail(duration, silent_tail),
911            ),
912            fade_in_analysis=_analysis(bpm=120.0, duration=duration),
913        )
914        fade.build(_seconds(45), _seconds(45), PCM)
915        return fade
916
917    def test_silent_tail_moves_the_anchor(self) -> None:
918        """A 10s silent tail shortens the audible anchor to ~35s and inserts FadeOutTrimFilter."""
919        fade = self._build_fade(silent_tail=10.0)
920        assert fade.effective_end == pytest.approx(35.0, abs=0.3)
921        # the rendered fade-out covers only the audible region
922        timing = fade.timing_info
923        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(
924            fade.effective_end, abs=0.05
925        )
926        # tail trim is the FIRST filter so later schedules see the trimmed stream
927        assert isinstance(fade.filters[0], FadeOutTrimFilter)
928        assert fade.filters[0].fadeout_end_pos == pytest.approx(fade.effective_end)
929
930    def test_no_silence_keeps_buffer_end_anchor(self) -> None:
931        """Without silence, effective_end equals the full buffer and no trim filter is added."""
932        fade = self._build_fade(silent_tail=0.0)
933        assert fade.effective_end == pytest.approx(45.0, abs=0.3)
934        assert not any(isinstance(f, FadeOutTrimFilter) for f in fade.filters)
935
936    def test_sub_tolerance_silent_tail_snaps_anchor_to_buffer_end(self) -> None:
937        """A silent tail below the trim tolerance keeps the anchor at the rendered buffer end."""
938        fade = self._build_fade(silent_tail=0.4)
939        assert fade.effective_end == pytest.approx(45.0)
940        assert not any(isinstance(f, FadeOutTrimFilter) for f in fade.filters)
941
942    def test_mostly_silent_tail_raises_for_fallback(self) -> None:
943        """A tail with only ~5s of audible content raises SmartFadeNotApplicable so the caller falls back."""
944        with pytest.raises(SmartFadeNotApplicable, match="silent"):
945            self._build_fade(silent_tail=40.0)
946
947    def test_partial_buffer_keeps_beats_aligned(self) -> None:
948        """
949        The live holdback buffer is rarely exactly 45s.
950
951        Beat coordinates must use the actual buffer length or every downbeat snap
952        is off by the difference.
953        """
954        duration = 240.0
955        fade = SmartCrossFade(
956            logger=LOGGER,
957            fade_out_analysis=_analysis(bpm=120.0, duration=duration),
958            fade_in_analysis=_analysis(bpm=120.0, duration=duration),
959        )
960        # 44.3s buffer: 0.7s short of the constant, like a real partial final chunk
961        partial_bytes = int(PCM.pcm_sample_size * 44.3)
962        frame_size = (PCM.bit_depth // 8) * PCM.channels
963        partial_bytes = (partial_bytes // frame_size) * frame_size
964        fade.build(partial_bytes, _seconds(45), PCM)
965        buffer_duration = partial_bytes / PCM.pcm_sample_size
966        # beats are on a strict 0.5s grid from t=0 in the analysis fixture;
967        # in real buffer coordinates each beat must satisfy
968        # (beat + duration - buffer_duration) % 0.5 == 0
969        offset = duration - buffer_duration
970        for beat in fade.fade_out_beats[:8]:
971            track_pos = beat + offset
972            assert abs(track_pos % 0.5) < 1e-3 or abs(track_pos % 0.5 - 0.5) < 1e-3, (
973                f"beat {beat:.4f} maps to track_pos {track_pos:.4f}, "
974                f"not on 0.5s grid (offset={offset:.4f})"
975            )
976        # the snapped crossfade start must land on a real downbeat (2s grid), not 0.7s off
977        crossfade_start = fade.effective_end - fade.timing_info.crossfade_duration
978        start_track_pos = crossfade_start + offset
979        assert abs(start_track_pos % 2.0) < 0.02 or abs(start_track_pos % 2.0 - 2.0) < 0.02, (
980            f"crossfade start {crossfade_start:.4f} maps to track_pos {start_track_pos:.4f}, "
981            f"not on 2s downbeat grid (offset={offset:.4f})"
982        )
983
984    def test_fadeout_beats_are_masked_to_effective_end(self) -> None:
985        """Beats in the silent tail are dropped so no downbeat sits beyond effective_end."""
986        fade = self._build_fade(silent_tail=10.0)
987        assert fade.fade_out_beats.min() >= 0.0
988        assert fade.fade_out_beats.max() <= fade.effective_end + 0.01
989
990    def test_short_audible_tail_keeps_crossfade_inside_it(self) -> None:
991        """A crossfade longer than the audible tail is capped so no schedule goes negative."""
992        duration = 240.0
993        fade = SmartCrossFade(
994            logger=LOGGER,
995            fade_out_analysis=_analysis(
996                bpm=120.0,
997                duration=duration,
998                rms_energy=_rms_with_silent_tail(duration, 33.0),
999            ),
1000            # 115 vs 120 BPM is within the stretch threshold, so the tempo ramp is active
1001            fade_in_analysis=_analysis(bpm=115.0, duration=duration),
1002        )
1003        fade.build(_seconds(45), _seconds(45), PCM)
1004        assert fade.timing_info.crossfade_duration <= fade.effective_end + 1e-6
1005        # the capped crossfade consumes the whole audible tail, so the stretch is skipped
1006        assert not any(isinstance(f, GradualTimeStretchFilter) for f in fade.filters)
1007
1008
1009def _rms_with_mastered_fade(
1010    track_duration: float, fade_start: float, fade_end: float
1011) -> np.ndarray:
1012    """Steady energy with the record's own gradual fade-out between the given media times."""
1013    bins = np.full(1800, 0.5, dtype=np.float32)
1014    bin_seconds = track_duration / 1800
1015    t = (np.arange(1800) + 0.5) * bin_seconds
1016    ramp = 0.5 * (1.0 - (t - fade_start) / (fade_end - fade_start)) + 0.0005
1017    return np.where(t < fade_start, bins, np.where(t >= fade_end, 0.0005, ramp)).astype(np.float32)
1018
1019
1020class TestQuickFadeMasteredFadeDeadZone:
1021    """
1022    A mastered fade-out under a quick-fade tier lands in the audible-trim dead zone.
1023
1024    The 70% mix-out floor anchors mid-fade while the 5% audible floor sits
1025    several seconds later; every quick-fade rung is far shorter than that gap,
1026    so AudibleTrimPolicy rejects the entire main candidate set. The rescue
1027    pass (ungated trim-closing ladder plus rescue rungs) then ships a
1028    late-anchored fade.
1029    """
1030
1031    def _build_fade(
1032        self, caplog: pytest.LogCaptureFixture, level: int = logging.DEBUG
1033    ) -> SmartCrossFade:
1034        duration = 240.0
1035        fade = SmartCrossFade(
1036            logger=LOGGER,
1037            fade_out_analysis=_analysis(
1038                bpm=128.0,
1039                duration=duration,
1040                # the record fades itself out over 229s..237s: mix-out (70% floor)
1041                # anchors near 231s while audible content (5% floor) runs to ~237s
1042                rms_energy=_rms_with_mastered_fade(duration, 229.0, 237.0),
1043            ),
1044            # 17.2% BPM gap: QUICK_FADE with the [2, 1] rung ladder (3.75s / 1.88s)
1045            fade_in_analysis=_analysis(bpm=150.0, duration=duration),
1046        )
1047        with caplog.at_level(level):
1048            fade.build(_seconds(45), _seconds(45), PCM)
1049        return fade
1050
1051    def test_main_pass_rejects_every_candidate_on_the_trim_guard_alone(
1052        self, caplog: pytest.LogCaptureFixture
1053    ) -> None:
1054        """Every main-pass candidate dies on the one guard; the rescue pass ships the fade."""
1055        self._build_fade(caplog)
1056        assert (
1057            "all 2 candidates rejected (audible trim exceeds a short fade's own duration x2)"
1058            in caplog.text
1059        )
1060        assert (
1061            "shipping a rescue-pass candidate (source=rescue-anchor) instead of the "
1062            "emergency handoff" in caplog.text
1063        )
1064
1065    def test_rescue_ships_a_late_anchored_chain_within_the_trim_bound(
1066        self, caplog: pytest.LogCaptureFixture
1067    ) -> None:
1068        """The shipped chain anchors near the audible end, honoring the short-fade trim bound."""
1069        fade = self._build_fade(caplog)
1070        # rescue anchor: last protective downbeat at/after audio_end - 2 bars (128 BPM)
1071        assert fade.effective_end == pytest.approx(41.25, abs=0.05)
1072        assert isinstance(fade.filters[0], FadeOutTrimFilter)
1073        assert fade.filters[0].fadeout_end_pos == pytest.approx(fade.effective_end)
1074        # the guard's own invariant holds on the shipped plan: audible material
1075        # dropped past the anchor stays within the overlap length (~41.67s RMS boundary)
1076        audible_trim = 41.67 - fade.effective_end
1077        assert audible_trim <= fade.timing_info.crossfade_duration + 1e-6
1078
1079    def test_rescue_pass_scores_trim_closing_candidates(
1080        self, caplog: pytest.LogCaptureFixture
1081    ) -> None:
1082        """The rescue pass runs the audible-end ladder ungated, so the selector scores it."""
1083        self._build_fade(caplog, level=VERBOSE_LOG_LEVEL)
1084        assert "source=trim-closing-anchor" in caplog.text
1085
1086    def test_trim_closing_wins_when_the_ladder_outgrows_the_rescue_rung(
1087        self, caplog: pytest.LogCaptureFixture
1088    ) -> None:
1089        """A 4-bar dead zone ships the audible-end ladder rung, not the capped rescue rung."""
1090        duration = 240.0
1091        fade = SmartCrossFade(
1092            logger=LOGGER,
1093            fade_out_analysis=_analysis(
1094                bpm=128.0,
1095                duration=duration,
1096                # longer mastered fade: the 7.78s trim gap exceeds even the 4-bar
1097                # rung (7.5s) yet stays under the trim-closing generator's 8s gate
1098                rms_energy=_rms_with_mastered_fade(duration, 228.4, 238.9),
1099            ),
1100            # 9.4% BPM gap: QUICK_FADE with the [4, 2, 1] rung ladder
1101            fade_in_analysis=_analysis(bpm=140.0, duration=duration),
1102        )
1103        with caplog.at_level(logging.DEBUG):
1104            fade.build(_seconds(45), _seconds(45), PCM)
1105        assert "shipping a rescue-pass candidate (source=trim-closing-anchor)" in caplog.text
1106        # the audible-end anchor keeps the full 4-bar overlap and trims nothing audible
1107        assert fade.effective_end == pytest.approx(43.40, abs=0.05)
1108        assert fade.timing_info.crossfade_duration == pytest.approx(7.78, abs=0.05)
1109
1110
1111# ---------------------------------------------------------------------------
1112# SmartCrossFade — rubberband stretch savings compensation
1113# ---------------------------------------------------------------------------
1114
1115
1116def _savings_until(fade: SmartCrossFade, t: float) -> float:
1117    """Rendered-time savings of the built fade's tempo plan up to input time t."""
1118    assert fade.plan is not None
1119    return fade.plan.tempo_plan.savings_until(t)
1120
1121
1122class TestStretchSavings:
1123    """
1124    Rendered-time savings from the stretch must reach the timing bookkeeping.
1125
1126    The savings integration math itself is unit-tested on TempoPlan in
1127    tests/controllers/streams/smart_fades/test_models.py.
1128    """
1129
1130    def _stretched_fade(self) -> SmartCrossFade:
1131        duration = 240.0
1132        # 4% BPM difference with >4 bars available -> stretch is applied
1133        fade = SmartCrossFade(
1134            logger=LOGGER,
1135            fade_out_analysis=_analysis(bpm=120.0, duration=duration),
1136            fade_in_analysis=_analysis(bpm=124.8, duration=duration),
1137        )
1138        fade.build(_seconds(45), _seconds(45), PCM)
1139        return fade
1140
1141    def test_pre_plus_cf_equals_rendered_tail(self) -> None:
1142        """PRE + CF equals the rendered tail duration (buffer minus stretch savings)."""
1143        fade = self._stretched_fade()
1144        assert fade.tempo_steps, "test requires the stretch to be active"
1145        total_savings = _savings_until(fade, fade.effective_end)
1146        assert total_savings > 0.0
1147        timing = fade.timing_info
1148        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(
1149            fade.effective_end - total_savings, abs=0.05
1150        )
1151
1152    def test_pre_plus_cf_equals_rendered_tail_when_slowing_down(self) -> None:
1153        """A slower incoming track lengthens the rendered tail — PRE + CF exceeds effective_end."""
1154        duration = 240.0
1155        # ~3.8% BPM difference downwards -> stretch slows the outgoing track
1156        fade = SmartCrossFade(
1157            logger=LOGGER,
1158            fade_out_analysis=_analysis(bpm=120.0, duration=duration),
1159            fade_in_analysis=_analysis(bpm=115.4, duration=duration),
1160        )
1161        fade.build(_seconds(45), _seconds(45), PCM)
1162        assert fade.tempo_steps, "test requires the stretch to be active"
1163        total_savings = _savings_until(fade, fade.effective_end)
1164        assert total_savings < 0.0
1165        timing = fade.timing_info
1166        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(
1167            fade.effective_end - total_savings, abs=0.05
1168        )
1169
1170    def test_bass_kill_completes_at_the_anchor(self) -> None:
1171        """
1172        The outgoing low-shelf kill reaches full depth at or before the audible end.
1173
1174        A-side shelves render BEFORE the rubberband stretch, so their schedules
1175        live in musical input time — no rendered-time remap needed (unlike the
1176        old post-stretch frequency sweeps).
1177        """
1178        fade = self._stretched_fade()
1179        assert fade.plan is not None
1180        low_out = fade.plan.eq_plan.low_out
1181        assert low_out is not None
1182        assert low_out.steps[-1][1] == pytest.approx(-26.0)
1183        assert low_out.steps[-1][0] <= fade.effective_end + 0.05
1184
1185    def test_trim_and_stretch_combined(self) -> None:
1186        """A trimmed silent tail and an active stretch compose: both anchor on the rendered end."""
1187        duration = 240.0
1188        fade = SmartCrossFade(
1189            logger=LOGGER,
1190            fade_out_analysis=_analysis(
1191                bpm=120.0,
1192                duration=duration,
1193                rms_energy=_rms_with_silent_tail(duration, 10.0),
1194            ),
1195            fade_in_analysis=_analysis(bpm=124.8, duration=duration),
1196        )
1197        fade.build(_seconds(45), _seconds(45), PCM)
1198        # tail trim must come first so every later schedule sees the trimmed stream
1199        assert isinstance(fade.filters[0], FadeOutTrimFilter)
1200        assert fade.tempo_steps, "test requires the stretch to be active"
1201        rendered_end = fade.effective_end - _savings_until(fade, fade.effective_end)
1202        timing = fade.timing_info
1203        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(
1204            rendered_end, abs=0.05
1205        )
1206
1207    def test_unstretched_fade_has_zero_savings(self) -> None:
1208        """Without a stretch, savings are zero and PRE + CF equals effective_end exactly."""
1209        fade = SmartCrossFade(
1210            logger=LOGGER,
1211            fade_out_analysis=_analysis(bpm=120.0, duration=240.0),
1212            fade_in_analysis=_analysis(bpm=120.0, duration=240.0),
1213        )
1214        fade.build(_seconds(45), _seconds(45), PCM)
1215        assert fade.tempo_steps == []
1216        assert _savings_until(fade, fade.effective_end) == 0.0
1217        timing = fade.timing_info
1218        assert timing.pre_crossfade_duration + timing.crossfade_duration == pytest.approx(
1219            fade.effective_end, abs=0.05
1220        )
1221