"""Ticket #20: memory-aware model ladder, conditional backend order, retry. All behavior is exercised with mocked probes and loaders: no GPU, no network, no hardware. The threshold-derivation test pins the rung thresholds to ticket #19's measured transcribe peaks (1132 MiB for q8_0, 846 MiB for q5_0) plus the 512 MiB protocol headroom, so any re-measure forces a threshold review. """ from unittest.mock import MagicMock, patch from vocalinux.speech_recognition import model_ladder from vocalinux.speech_recognition.model_ladder import ( DEFAULT_RUNG, FLOOR_RUNG_ENGLISH, FLOOR_RUNG_MULTILINGUAL, HEADROOM_MIB, LOW_MEMORY_RUNG, PREFER_Q5_0_ABOVE_MIB, PREFER_Q8_0_ABOVE_MIB, cuda_toolkit_ready_from_probe, next_rung_down, preferred_backend_order, resolve_startup_rung, select_ladder_rung, ) from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager from vocalinux.utils.whispercpp_model_info import WHISPERCPP_MODEL_INFO def _size_mb(name: str) -> float: return float(WHISPERCPP_MODEL_INFO[name]["size_mb"]) class TestThresholdDerivation: """Thresholds match the baseline ticket's recorded recommendation.""" #: Ticket #19 measured transcribe peaks (MiB) per rung. Q8_0_MEASURED_MIB: float = 1132.0 Q5_0_MEASURED_MIB: float = 846.0 def test_q8_threshold_is_measured_footprint_plus_headroom(self) -> None: assert PREFER_Q8_0_ABOVE_MIB == self.Q8_0_MEASURED_MIB + HEADROOM_MIB assert PREFER_Q8_0_ABOVE_MIB == 1644.0 def test_q5_threshold_is_measured_footprint_plus_headroom(self) -> None: assert PREFER_Q5_0_ABOVE_MIB == self.Q5_0_MEASURED_MIB + HEADROOM_MIB assert PREFER_Q5_0_ABOVE_MIB == 1358.0 def test_rungs_are_ordered(self) -> None: assert PREFER_Q8_0_ABOVE_MIB > PREFER_Q5_0_ABOVE_MIB assert _size_mb(DEFAULT_RUNG) > _size_mb(LOW_MEMORY_RUNG) class TestSelectLadderRung: """Free memory picks the rung; the floor pairs with the language.""" def test_plenty_of_memory_keeps_default(self) -> None: assert select_ladder_rung(True, PREFER_Q8_0_ABOVE_MIB) == DEFAULT_RUNG assert select_ladder_rung(False, 99999.0) == DEFAULT_RUNG def test_between_thresholds_steps_to_q5(self) -> None: assert select_ladder_rung(True, PREFER_Q8_0_ABOVE_MIB - 1) == LOW_MEMORY_RUNG assert select_ladder_rung(True, PREFER_Q5_0_ABOVE_MIB) == LOW_MEMORY_RUNG def test_below_q5_fits_the_floor_by_language(self) -> None: assert select_ladder_rung(True, PREFER_Q5_0_ABOVE_MIB - 1) == FLOOR_RUNG_ENGLISH assert select_ladder_rung(False, PREFER_Q5_0_ABOVE_MIB - 1) == FLOOR_RUNG_MULTILINGUAL assert select_ladder_rung(True, 0.0) == FLOOR_RUNG_ENGLISH assert select_ladder_rung(False, 0.0) == FLOOR_RUNG_MULTILINGUAL def test_recorded_boundaries_select_recorded_rungs(self) -> None: assert select_ladder_rung(True, 1644.0) == DEFAULT_RUNG assert select_ladder_rung(True, 1500.0) == LOW_MEMORY_RUNG assert select_ladder_rung(True, 1358.0) == LOW_MEMORY_RUNG assert select_ladder_rung(True, 1200.0) == FLOOR_RUNG_ENGLISH assert select_ladder_rung(False, 1200.0) == FLOOR_RUNG_MULTILINGUAL class TestResolveStartupRung: """A manual pin wins verbatim; only unpinned loads use the ladder.""" def test_pinned_variant_never_overridden(self) -> None: assert resolve_startup_rung("tiny", True, 99999.0) == "tiny" assert resolve_startup_rung(LOW_MEMORY_RUNG, False, 0.0) == LOW_MEMORY_RUNG assert resolve_startup_rung(FLOOR_RUNG_ENGLISH, True, 99999.0) == FLOOR_RUNG_ENGLISH def test_unpinned_uses_ladder(self) -> None: assert resolve_startup_rung("", True, 99999.0) == DEFAULT_RUNG assert resolve_startup_rung("", False, 0.0) == FLOOR_RUNG_MULTILINGUAL class TestNextRungDown: """One step down per failure; the floor fails loudly.""" def test_default_steps_to_low_memory(self) -> None: assert next_rung_down(DEFAULT_RUNG, True) == LOW_MEMORY_RUNG assert next_rung_down(DEFAULT_RUNG, False) == LOW_MEMORY_RUNG def test_low_memory_steps_to_floor_by_language(self) -> None: assert next_rung_down(LOW_MEMORY_RUNG, True) == FLOOR_RUNG_ENGLISH assert next_rung_down(LOW_MEMORY_RUNG, False) == FLOOR_RUNG_MULTILINGUAL def test_floor_has_no_step_down(self) -> None: assert next_rung_down(FLOOR_RUNG_ENGLISH, True) is None assert next_rung_down(FLOOR_RUNG_MULTILINGUAL, False) is None def test_non_ladder_rungs_have_no_step_down(self) -> None: assert next_rung_down("tiny", True) is None assert next_rung_down("medium", False) is None class TestCudaToolkitReadiness: """Driver alone is not a toolkit; arch must match too.""" def test_complete_matching_toolkit_is_ready(self) -> None: assert cuda_toolkit_ready_from_probe(True, True, (8, 6)) is True assert cuda_toolkit_ready_from_probe(True, True, (7, 0)) is True def test_driver_without_toolkit_is_not_ready(self) -> None: assert cuda_toolkit_ready_from_probe(True, False, (8, 6)) is False def test_no_gpu_is_not_ready(self) -> None: assert cuda_toolkit_ready_from_probe(False, True, (8, 6)) is False assert cuda_toolkit_ready_from_probe(False, False, None) is False def test_old_architecture_is_not_ready(self) -> None: assert cuda_toolkit_ready_from_probe(True, True, (6, 1)) is False assert cuda_toolkit_ready_from_probe(True, True, None) is False class TestPreferredBackendOrder: """CUDA ranks first only with a complete toolkit.""" def test_cuda_first_with_toolkit(self) -> None: with patch.object(model_ladder, "cuda_toolkit_ready", return_value=True): assert preferred_backend_order() == ("cuda", "vulkan", "cpu") def test_vulkan_then_cpu_without_toolkit(self) -> None: with patch.object(model_ladder, "cuda_toolkit_ready", return_value=False): assert preferred_backend_order() == ("vulkan", "cpu") def _make_manager(**kwargs): """Create a whisper.cpp manager without running any initializer.""" with ( patch.object(SpeechRecognitionManager, "_init_vosk"), patch.object(SpeechRecognitionManager, "_init_whisper"), patch.object(SpeechRecognitionManager, "_init_whispercpp"), ): return SpeechRecognitionManager( engine="whisper_cpp", model_size="small", language="en-us", defer_download=True, **kwargs, ) class TestStepDownRetry: """The loader steps one rung down and retries once, then surfaces.""" def test_first_failure_steps_down_and_retries_once(self) -> None: manager = _make_manager() manager.model_size = DEFAULT_RUNG calls = [] def fail_once() -> None: calls.append(manager.model_size) if len(calls) == 1: raise RuntimeError("CUDA out of memory") with patch.object(SpeechRecognitionManager, "_init_whispercpp_once", side_effect=fail_once): manager._init_whispercpp() assert calls == [DEFAULT_RUNG, LOW_MEMORY_RUNG] assert manager.model_size == LOW_MEMORY_RUNG def test_second_failure_surfaces(self) -> None: manager = _make_manager() manager.model_size = DEFAULT_RUNG with patch.object( SpeechRecognitionManager, "_init_whispercpp_once", side_effect=RuntimeError("boom") ): try: manager._init_whispercpp() except RuntimeError: pass else: raise AssertionError("second failure must surface") assert manager.model_size == LOW_MEMORY_RUNG def test_manual_pin_never_steps_down(self) -> None: manager = _make_manager(whispercpp_variant_pinned=True) manager.model_size = DEFAULT_RUNG loader = MagicMock(side_effect=RuntimeError("CUDA out of memory")) with patch.object(SpeechRecognitionManager, "_init_whispercpp_once", loader): try: manager._init_whispercpp() except RuntimeError: pass else: raise AssertionError("manual pick must fail loudly") assert loader.call_count == 1 assert manager.model_size == DEFAULT_RUNG def test_floor_rung_fails_loudly(self) -> None: manager = _make_manager() manager.model_size = FLOOR_RUNG_ENGLISH loader = MagicMock(side_effect=OSError("no space")) with patch.object(SpeechRecognitionManager, "_init_whispercpp_once", loader): try: manager._init_whispercpp() except OSError: pass else: raise AssertionError("floor failure must surface") assert loader.call_count == 1 def test_non_ladder_model_does_not_step_down(self) -> None: manager = _make_manager() manager.model_size = "medium" loader = MagicMock(side_effect=RuntimeError("boom")) with patch.object(SpeechRecognitionManager, "_init_whispercpp_once", loader): try: manager._init_whispercpp() except RuntimeError: pass else: raise AssertionError("non-ladder failure must surface") assert loader.call_count == 1 class TestStartupRungSelection: """Unpinned defaults resolve through the ladder; explicit picks do not.""" def _manager_with(self, saved_size, pinned_variant, language="en-us"): from vocalinux.ui.config_manager import ConfigManager manager = ConfigManager.__new__(ConfigManager) manager.config = { "speech_recognition": { "whisper_cpp_model_size": saved_size, "whisper_cpp_model_variant": pinned_variant, "language": language, } } return manager def test_unpinned_default_selects_rung_by_memory(self) -> None: manager = self._manager_with("tiny", "", "en-us") with patch( "vocalinux.speech_recognition.model_ladder.free_memory_mib", return_value=99999.0 ): with patch( "vocalinux.utils.whispercpp_model_info.is_model_downloaded", return_value=False ): assert manager.get_model_size_for_engine("whisper_cpp") == DEFAULT_RUNG def test_unpinned_default_low_memory_selects_q5(self) -> None: manager = self._manager_with("tiny", "", "en-us") with patch( "vocalinux.speech_recognition.model_ladder.free_memory_mib", return_value=PREFER_Q5_0_ABOVE_MIB, ): with patch( "vocalinux.utils.whispercpp_model_info.is_model_downloaded", return_value=False ): assert manager.get_model_size_for_engine("whisper_cpp") == LOW_MEMORY_RUNG def test_unpinned_default_tight_memory_selects_floor(self) -> None: manager = self._manager_with("tiny", "", "pl") with patch("vocalinux.speech_recognition.model_ladder.free_memory_mib", return_value=100.0): with patch( "vocalinux.utils.whispercpp_model_info.is_model_downloaded", return_value=False ): assert manager.get_model_size_for_engine("whisper_cpp") == FLOOR_RUNG_MULTILINGUAL def test_pinned_variant_never_uses_ladder(self) -> None: manager = self._manager_with("tiny", LOW_MEMORY_RUNG, "en-us") with patch( "vocalinux.speech_recognition.model_ladder.free_memory_mib", return_value=99999.0 ): assert manager.get_model_size_for_engine("whisper_cpp") == LOW_MEMORY_RUNG def test_explicit_size_keeps_legacy_derivation(self) -> None: manager = self._manager_with("medium", "", "en-us") with patch( "vocalinux.speech_recognition.model_ladder.free_memory_mib", return_value=99999.0 ): with patch( "vocalinux.utils.whispercpp_model_info.is_model_downloaded", return_value=False ): assert manager.get_model_size_for_engine("whisper_cpp") == "medium.en" class TestFreeMemoryForBackend: """The memory reading follows the installed build (spec #16, ticket #20).""" def test_vulkan_build_reads_discrete_vram_when_present(self) -> None: with ( patch.object(model_ladder, "free_vram_mib", return_value=2600.0), patch.object(model_ladder, "free_system_ram_mib", return_value=9000.0), ): assert model_ladder.free_memory_for_backend("vulkan") == (2600.0, "VRAM") def test_cuda_build_reads_discrete_vram_when_present(self) -> None: with patch.object(model_ladder, "free_vram_mib", return_value=1500.0): assert model_ladder.free_memory_for_backend("cuda") == (1500.0, "VRAM") def test_gpu_build_without_an_nvidia_reading_reads_ram(self) -> None: """Integrated graphics: no discrete VRAM reading, so system RAM decides.""" with ( patch.object(model_ladder, "free_vram_mib", return_value=None), patch.object(model_ladder, "free_system_ram_mib", return_value=6000.0), ): assert model_ladder.free_memory_for_backend("vulkan") == (6000.0, "RAM") def test_cpu_build_never_reads_vram(self) -> None: vram = MagicMock(return_value=2600.0) with ( patch.object(model_ladder, "free_vram_mib", vram), patch.object(model_ladder, "free_system_ram_mib", return_value=1200.0), ): assert model_ladder.free_memory_for_backend("cpu") == (1200.0, "RAM") vram.assert_not_called()