path: root/app/tests/test_tts.py

"""Tests for the TTS client wrappers (language handling and payloads)."""

import base64
import io
import json
import struct
import tempfile
import time
import urllib.error
import unittest
import wave
from contextlib import redirect_stdout
from pathlib import Path
from unittest.mock import MagicMock, patch

from converter import config
from converter import converter as converter_mod
from converter.clients import (
    AUDIOCPP_DEFAULT_FAMILY_PROFILE,
    AUDIOCPP_FAMILY_PROFILES,
    AUDIOCPP_LANG_OMIT,
    AUDIOCPP_TASK_TTS,
    AUDIOCPP_TASK_VDES,
    AUDIOCPP_VOICE_CLONE,
    AUDIOCPP_VOICE_DESIGN,
    AUDIOCPP_VOICE_NONE,
    AUDIOCPP_VOICE_OPTIONAL,
    AUDIOCPP_VOICE_REQUIRED,
    AUDIOCPP_VOICE_SPEAKER,
    BACKEND_AUDIOCPP,
    BACKEND_FASTER,
    BACKEND_QWEN,
    LANGUAGE_CHOICES,
    LANGUAGE_ISO_CODES,
    MODEL_SIZE,
    SAMPLE_RATE,
    TTS_LANGUAGES,
    VOICE_MODE_CLONE,
    VOICE_MODE_CUSTOM,
    VOICE_MODE_DESIGN,
    VOICE_MODES,
    AudioCppTTSClient,
    FasterTTSClient,
    QWEN3_TTS_SPEAKERS,
    QwenTTSClient,
    audiocpp_entry_voice_capability,
    audiocpp_family_narrates,
    audiocpp_family_voice_policy,
    audiocpp_request_error,
    audiocpp_script_input,
    audiocpp_voice_for_run,
    allocation_log_note,
    nvidia_device_memory_report,
    normalize_language,
    transcribe_reference_audio_detailed,
    whisper_backend_problem,
)
from converter.clients import audiocpp as audiocpp_client
from converter.clients.base import NonRetryableTTSError
from converter.converter import AudiobookConverter

# Chunks folder handed to clients whose tests never write chunk files.
_DUMMY_CHUNKS = Path(tempfile.gettempdir()) / "audiobook_tts_test_chunks"

# A concrete audio.cpp model entry id (no config default anymore): the
# tests request it explicitly, the way --model / the Generate form does.
_AUDIOCPP_MODEL_ID = "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF"


class NormalizeLanguageTests(unittest.TestCase):
    def test_display_names_case_insensitive(self):
        self.assertEqual(normalize_language("english"), "English")
        self.assertEqual(normalize_language("ENGLISH"), "English")
        self.assertEqual(normalize_language(" Japanese "), "Japanese")

    def test_auto_accepted(self):
        self.assertEqual(normalize_language("auto"), "Auto")
        self.assertEqual(normalize_language("Auto"), "Auto")

    def test_iso_aliases(self):
        self.assertEqual(normalize_language("en"), "English")
        self.assertEqual(normalize_language("ja"), "Japanese")
        self.assertEqual(normalize_language("zh"), "Chinese")
        self.assertEqual(normalize_language("ko"), "Korean")
        self.assertEqual(normalize_language("de"), "German")
        self.assertEqual(normalize_language("fr"), "French")
        self.assertEqual(normalize_language("ru"), "Russian")
        self.assertEqual(normalize_language("pt"), "Portuguese")
        self.assertEqual(normalize_language("es"), "Spanish")
        self.assertEqual(normalize_language("it"), "Italian")
        self.assertEqual(normalize_language("ar"), "Arabic")
        self.assertEqual(normalize_language("hi"), "Hindi")
        self.assertEqual(normalize_language("vi"), "Vietnamese")

    def test_audio_cpp_menu_languages_accepted(self):
        # audio.cpp's WebUI menus add Arabic, Hindi and Vietnamese; the
        # MagpieTTS Arabic regional variants collapse to plain Arabic.
        self.assertEqual(normalize_language("arabic"), "Arabic")
        self.assertEqual(normalize_language("Hindi"), "Hindi")
        self.assertEqual(normalize_language("vietnamese"), "Vietnamese")
        for variant in ("ar-AE", "ar-MSA", "ar-SA"):
            self.assertEqual(normalize_language(variant), "Arabic")

    def test_language_choices_are_valid_display_names(self):
        # The TUI's static picker lists a permutation of TTS_LANGUAGES
        # (common languages first), so every entry normalizes.
        self.assertEqual(sorted(LANGUAGE_CHOICES), sorted(TTS_LANGUAGES))
        for name in LANGUAGE_CHOICES:
            self.assertEqual(normalize_language(name), name)

    def test_all_supported_languages_round_trip(self):
        for name in TTS_LANGUAGES:
            self.assertEqual(normalize_language(name.lower()), name)

    def test_unknown_language_rejected_with_guidance(self):
        with self.assertRaises(ValueError) as ctx:
            normalize_language("klingon")
        message = str(ctx.exception)
        self.assertIn("klingon", message)
        self.assertIn("English", message)

    def test_none_and_empty_rejected(self):
        with self.assertRaises(ValueError):
            normalize_language(None)
        with self.assertRaises(ValueError):
            normalize_language("   ")


class QwenTTSClientLanguageTests(unittest.TestCase):
    """Language validation and defaults, without touching the network."""

    def _make_client(self, **kwargs):
        with patch.object(QwenTTSClient, "_connect"):
            return QwenTTSClient(_DUMMY_CHUNKS, **kwargs)

    def test_default_follows_config_for_each_mode(self):
        custom = self._make_client(voice_mode=VOICE_MODE_CUSTOM,
                                   voice="Vivian")
        self.assertEqual(custom.language, config.LANGUAGE)
        clone = self._make_client(voice_mode=VOICE_MODE_CLONE,
                                  voice_clone_ref_audio="ref.wav")
        self.assertEqual(clone.language, config.LANGUAGE)

    def test_explicit_language_normalized(self):
        client = self._make_client(voice_mode=VOICE_MODE_CUSTOM,
                                   language="ja", voice="Vivian")
        self.assertEqual(client.language, "Japanese")

    def test_invalid_language_fails_before_connect(self):
        with patch.object(QwenTTSClient, "_connect") as mock_connect:
            with self.assertRaises(ValueError):
                QwenTTSClient(_DUMMY_CHUNKS, language="klingon")
        mock_connect.assert_not_called()

    def test_api_url_override_stored(self):
        client = self._make_client(voice_mode=VOICE_MODE_CUSTOM,
                                   api_url="http://10.0.0.5:7860",
                                   voice="Vivian")
        self.assertEqual(client.api_url, "http://10.0.0.5:7860")

    def test_api_url_override_used_by_connect(self):
        with patch.object(QwenTTSClient, "_init_client") as mk_init:
            client = QwenTTSClient.__new__(QwenTTSClient)
            client.voice_mode = VOICE_MODE_CUSTOM
            client.api_url = "http://10.0.0.5:7860"
            client._connect()
        mk_init.assert_called_once_with("http://10.0.0.5:7860", clone=False)


class SeedResolutionTests(unittest.TestCase):
    """CONSTANT_SEED: one seed per run, reused for every request, so the
    voice stays consistent across chunk boundaries (the servers
    re-sample the voice when the seed changes between generations)."""

    def _make_client(self, **kwargs):
        with patch.object(QwenTTSClient, "_connect"):
            return QwenTTSClient(_DUMMY_CHUNKS, **kwargs)

    def test_constant_seed_draws_one_nonnegative_seed(self):
        with patch.object(config, "CONSTANT_SEED", True), \
                patch.object(config, "SEED", -1):
            client = self._make_client(voice_mode=VOICE_MODE_CUSTOM,
                                       voice="Vivian")
        self.assertGreaterEqual(client._seed, 0)

    def test_explicit_seed_wins_over_constant_seed(self):
        with patch.object(config, "CONSTANT_SEED", True), \
                patch.object(config, "SEED", 42):
            client = self._make_client(voice_mode=VOICE_MODE_CUSTOM,
                                       voice="Vivian")
        self.assertEqual(client._seed, 42)

    def test_without_constant_seed_minus_one_is_forwarded(self):
        with patch.object(config, "CONSTANT_SEED", False), \
                patch.object(config, "SEED", -1):
            client = self._make_client(voice_mode=VOICE_MODE_CUSTOM,
                                       voice="Vivian")
        self.assertEqual(client._seed, -1)

    def test_resolved_seed_is_reused_across_requests(self):
        api_info = {
            "named_endpoints": {
                "/run_custom_voice": {
                    "parameters": [{"parameter_name": "seed"}]
                }
            }
        }
        client = QwenTTSClient.__new__(QwenTTSClient)
        client.voice_mode = VOICE_MODE_CUSTOM
        client.speaker = "Vivian"
        client.language = "English"
        client._seed = 1234
        client.api_info = api_info
        client.client = MagicMock()
        client._generate_custom_voice("first text")
        client._generate_custom_voice("second text")
        seeds = [call.kwargs["seed"]
                 for call in client.client.predict.call_args_list]
        self.assertEqual(seeds, [1234, 1234])


class PayloadLanguageTests(unittest.TestCase):
    """The language must reach the API payload in every endpoint variant."""

    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()
        self.ref_audio = Path(self._tmp.name) / "reference.wav"
        self.ref_audio.write_bytes(b"x")

    def tearDown(self):
        self._tmp.cleanup()

    def _custom_client(self, language, endpoint, api_info=None):
        client = QwenTTSClient.__new__(QwenTTSClient)
        client.voice_mode = VOICE_MODE_CUSTOM
        client.speaker = "Vivian"
        client.language = language
        client._seed = config.SEED
        client.api_info = api_info if api_info is not None else {
            "named_endpoints": {endpoint: {}}
        }
        client.client = MagicMock()
        return client

    def _clone_client(self, language, endpoint, api_info=None, ref_text="hello"):
        client = QwenTTSClient.__new__(QwenTTSClient)
        client.voice_mode = VOICE_MODE_CLONE
        client.language = language
        client._seed = config.SEED
        client.voice_clone_ref_audio = str(self.ref_audio)
        client.voice_clone_ref_text = ref_text
        client.clone_api_info = api_info if api_info is not None else {
            "named_endpoints": {endpoint: {}}
        }
        client.clone_client = MagicMock()
        client._ref_audio_filedata = {"dummy": "payload"}
        return client

    def test_custom_voice_run_instruct_uses_language(self):
        client = self._custom_client("Japanese", "/run_instruct")
        client._generate_custom_voice("text")
        kwargs = client.client.predict.call_args.kwargs
        self.assertEqual(kwargs["lang_disp"], "Japanese")

    def test_custom_voice_alt_endpoint_uses_language(self):
        client = self._custom_client("French", "/run_custom_voice")
        client._generate_custom_voice("text")
        kwargs = client.client.predict.call_args.kwargs
        self.assertEqual(kwargs["language"], "French")

    def test_voice_clone_run_voice_clone_uses_language(self):
        client = self._clone_client("Japanese", "/run_voice_clone")
        client._generate_voice_clone("text")
        kwargs = client.clone_client.predict.call_args.kwargs
        self.assertEqual(kwargs["lang_disp"], "Japanese")

    def test_voice_clone_alt_endpoint_uses_language(self):
        client = self._clone_client("Korean", "/generate_voice_clone")
        client._generate_voice_clone("text")
        kwargs = client.clone_client.predict.call_args.kwargs
        self.assertEqual(kwargs["language"], "Korean")

    def test_voice_clone_alt_endpoint_includes_optional_params(self):
        api_info = {
            "named_endpoints": {
                "/generate_voice_clone": {
                    "parameters": [
                        {"parameter_name": "model_size"},
                        {"parameter_name": "seed"},
                    ]
                }
            }
        }
        client = self._clone_client("English", "/generate_voice_clone", api_info=api_info)
        client._generate_voice_clone("text")
        kwargs = client.clone_client.predict.call_args.kwargs
        self.assertEqual(kwargs["model_size"], MODEL_SIZE)
        self.assertEqual(kwargs["seed"], config.SEED)


class FasterTTSClientHealthTests(unittest.TestCase):
    """Connection behavior of the faster-qwen3-tts client."""

    def _health_response(self, model_loaded=True):
        response = MagicMock()
        response.__enter__.return_value = response
        response.read.return_value = json.dumps(
            {"status": "ok", "model_loaded": model_loaded}).encode("utf-8")
        return response

    def test_unreachable_server_raises_with_readme_pointer(self):
        import urllib.error
        with patch("converter.clients.faster.urllib.request.urlopen",
                   side_effect=urllib.error.URLError("Connection refused")):
            with self.assertRaises(RuntimeError) as ctx:
                FasterTTSClient(_DUMMY_CHUNKS, voice="narrator")
        message = str(ctx.exception)
        self.assertIn("not reachable", message)
        self.assertIn("README", message)

    def test_model_not_loaded_raises(self):
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=self._health_response(model_loaded=False)):
            with self.assertRaises(RuntimeError) as ctx:
                FasterTTSClient(_DUMMY_CHUNKS, voice="narrator")
        self.assertIn("not loaded", str(ctx.exception))

    def test_missing_voice_raises_before_connecting(self):
        # There is no configured default voice: a faster run names its
        # voice per run (the server silently falls back when the key is
        # not in its voices.json).
        with patch("converter.clients.faster.urllib.request.urlopen") \
                as mock_urlopen:
            with self.assertRaises(RuntimeError) as ctx:
                FasterTTSClient(_DUMMY_CHUNKS)
        self.assertIn("requires a voice", str(ctx.exception))
        mock_urlopen.assert_not_called()

    def test_healthy_server_uses_the_requested_voice(self):
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=self._health_response()):
            client = FasterTTSClient(_DUMMY_CHUNKS, voice="narrator")
        self.assertEqual(client.voice, "narrator")
        self.assertEqual(client.api_url, config.FASTER_API_URL.rstrip("/"))

    def test_explicit_voice_and_url_override_config(self):
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=self._health_response()):
            client = FasterTTSClient(_DUMMY_CHUNKS,
                                     voice="narrator", api_url="http://10.0.0.5:9000/")
        self.assertEqual(client.voice, "narrator")
        self.assertEqual(client.api_url, "http://10.0.0.5:9000")


class FasterTTSClientGenerateTests(unittest.TestCase):
    """Chunk generation: sub-chunking, WAV output, retries, bookkeeping."""

    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()
        self._sleep = patch("converter.clients.base.time.sleep")
        self._sleep.start()

    def tearDown(self):
        self._sleep.stop()
        self._tmp.cleanup()

    def _make_client(self):
        client = FasterTTSClient.__new__(FasterTTSClient)
        client.chunks_dir = Path(self._tmp.name)
        client.voice = "default"
        client.api_url = "http://127.0.0.1:8000"
        return client

    def _read_wav(self, path):
        with wave.open(str(path), "rb") as wav_file:
            return (wav_file.getnchannels(), wav_file.getsampwidth(),
                    wav_file.getframerate(), wav_file.readframes(wav_file.getnframes()))

    def test_generate_chunk_writes_valid_wav(self):
        client = self._make_client()
        pcm = b"\x01\x00" * 100
        with patch.object(client, "_request_pcm", return_value=pcm):
            result = client.generate_chunk("Hello world.", 1)
        self.assertIsNotNone(result)
        path = Path(result)
        self.assertEqual(path.name, "chunk_0001.wav")
        channels, sampwidth, framerate, frames = self._read_wav(path)
        self.assertEqual(channels, 1)
        self.assertEqual(sampwidth, 2)
        self.assertEqual(framerate, SAMPLE_RATE)
        self.assertEqual(frames, pcm)

    def test_long_text_is_subchunked_and_concatenated_in_order(self):
        client = self._make_client()
        sentences = [" ".join(f"word{i}" for i in range(6)) + "." for _ in range(3)]
        text = " ".join(sentences)
        pcm_parts = [b"\x01\x00" * 10, b"\x02\x00" * 20, b"\x03\x00" * 30]
        with patch.object(config, "CHUNK_SIZE", 10), \
                patch.object(client, "_request_pcm", side_effect=pcm_parts) as mock_pcm:
            result = client.generate_chunk(text, 1)
        self.assertEqual(mock_pcm.call_count, 3)
        _, _, _, frames = self._read_wav(Path(result))
        self.assertEqual(frames, b"".join(pcm_parts))

    def test_subchunk_size_follows_config_chunk_size(self):
        client = self._make_client()
        text = " ".join(f"word{i}" for i in range(8))
        pcm = b"\x01\x00" * 10
        with patch.object(config, "CHUNK_SIZE", 4), \
                patch.object(client, "_request_pcm", return_value=pcm) as mock_pcm:
            result = client.generate_chunk(text, 1)
        # The sub-chunk split follows config.CHUNK_SIZE, so the whole
        # (8-word) text needs two 4-word requests here.
        self.assertEqual(mock_pcm.call_count, 2)
        self.assertIsNotNone(result)

    def test_stale_chunk_files_are_removed(self):
        stale = Path(self._tmp.name) / "chunk_0001.mp3"
        stale.write_bytes(b"old")
        client = self._make_client()
        with patch.object(client, "_request_pcm", return_value=b"\x01\x00"):
            client.generate_chunk("Hello.", 1)
        remaining = sorted(path.name for path in Path(self._tmp.name).glob("chunk_0001.*"))
        self.assertEqual(remaining, ["chunk_0001.wav"])

    def test_transient_failure_fails_the_chunk_attempt(self):
        # Retrying is the chunk-level policy's job
        # (process_chunk_with_retry); one generate_chunk call makes one
        # request attempt per sub-chunk.
        client = self._make_client()
        pcm = b"\x01\x00" * 10
        with patch.object(client, "_request_pcm",
                          side_effect=[RuntimeError("boom"), pcm]) as mock_pcm:
            result = client.generate_chunk("Hello.", 1)
        self.assertIsNone(result)
        self.assertEqual(mock_pcm.call_count, 1)

    def test_empty_pcm_response_fails_the_chunk(self):
        client = self._make_client()

        def _response(body):
            response = MagicMock()
            response.__enter__.return_value = response
            response.read.return_value = body
            return response

        with patch("converter.clients.faster.urllib.request.urlopen",
                   side_effect=[_response(b"")]) as mock_urlopen:
            result = client.generate_chunk("Hello.", 1)
        self.assertIsNone(result)
        self.assertEqual(mock_urlopen.call_count, 1)

    def test_subchunk_request_failure_fails_the_chunk(self):
        client = self._make_client()
        with patch.object(client, "_request_pcm",
                          side_effect=RuntimeError("down")) as mock_pcm:
            result = client.generate_chunk("Hello.", 1)
        self.assertIsNone(result)
        self.assertEqual(mock_pcm.call_count, 1)

    def test_empty_text_fails_the_chunk(self):
        client = self._make_client()
        with patch.object(client, "_request_pcm") as mock_pcm:
            result = client.generate_chunk("   ", 1)
        self.assertIsNone(result)
        mock_pcm.assert_not_called()

    def test_request_payload_includes_voice_text_and_format(self):
        client = self._make_client()
        response = MagicMock()
        response.__enter__.return_value = response
        response.read.return_value = b"\x01\x00" * 10
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=response) as mock_urlopen:
            pcm = client._request_pcm("Hello world.")
        self.assertEqual(pcm, b"\x01\x00" * 10)
        request = mock_urlopen.call_args[0][0]
        self.assertEqual(request.full_url, "http://127.0.0.1:8000/v1/audio/speech")
        payload = json.loads(request.data.decode("utf-8"))
        self.assertEqual(payload["input"], "Hello world.")
        self.assertEqual(payload["voice"], "default")
        self.assertEqual(payload["response_format"], "pcm")

    def test_full_length_pcm_passes(self):
        client = self._make_client()
        text = " ".join(f"word{i}" for i in range(12))
        # 12 words -> expected 4.8s, half is 2.4s -> 2.5s of audio passes.
        pcm = b"\x01\x00" * int(2.5 * SAMPLE_RATE)
        with patch.object(client, "_request_pcm", return_value=pcm):
            result = client.generate_chunk(text, 1)
        self.assertIsNotNone(result)


class QwenTTSClientGenerateTests(unittest.TestCase):
    """Qwen chunk generation: sub-request splitting and concatenation."""

    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()

    def tearDown(self):
        self._tmp.cleanup()

    def _make_client(self):
        client = QwenTTSClient.__new__(QwenTTSClient)
        client.chunks_dir = Path(self._tmp.name)
        client.voice_mode = VOICE_MODE_CUSTOM
        return client

    @staticmethod
    def _write_wav(path: Path, frames: bytes) -> Path:
        with wave.open(str(path), "wb") as wav_file:
            wav_file.setnchannels(1)
            wav_file.setsampwidth(2)
            wav_file.setframerate(SAMPLE_RATE)
            wav_file.writeframes(frames)
        return path

    def _read_wav_frames(self, path: Path) -> bytes:
        with wave.open(str(path), "rb") as wav_file:
            return wav_file.readframes(wav_file.getnframes())

    def test_single_request_copies_audio(self):
        client = self._make_client()
        source = self._write_wav(Path(self._tmp.name) / "server.wav", b"\x01\x00" * 50)
        with patch.object(client, "_generate_custom_voice",
                          return_value=(str(source),)) as mock_generate:
            result = client.generate_chunk("Hello world.", 1)
        mock_generate.assert_called_once_with("Hello world.")
        path = Path(result)
        self.assertEqual(path.name, "chunk_0001.wav")
        self.assertEqual(self._read_wav_frames(path), b"\x01\x00" * 50)

    def test_oversized_input_is_split_and_concatenated_in_order(self):
        client = self._make_client()
        first = self._write_wav(Path(self._tmp.name) / "one.wav", b"\x01\x00" * 10)
        second = self._write_wav(Path(self._tmp.name) / "two.wav", b"\x02\x00" * 20)
        text = " ".join(f"word{i}" for i in range(12))
        with patch.object(config, "CHUNK_SIZE", 5), \
                patch.object(client, "_generate_custom_voice",
                             side_effect=[(str(first),), (str(second),),
                                          (str(first),)]) as mock_generate:
            result = client.generate_chunk(text, 1)
        self.assertEqual(mock_generate.call_count, 3)
        path = Path(result)
        self.assertEqual(path.name, "chunk_0001.wav")
        self.assertEqual(self._read_wav_frames(path),
                         b"\x01\x00" * 10 + b"\x02\x00" * 20 + b"\x01\x00" * 10)
        for call in mock_generate.call_args_list:
            self.assertLessEqual(len(call[0][0].split()), 5)

    def test_empty_text_fails_the_chunk(self):
        client = self._make_client()
        with patch.object(client, "_generate_custom_voice") as mock_generate:
            result = client.generate_chunk("   ", 1)
        self.assertIsNone(result)
        mock_generate.assert_not_called()


class QwenTTSClientVoiceDesignTests(unittest.TestCase):
    """Qwen VoiceDesign mode: instructions and the /run_voice_design call."""

    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()

    def tearDown(self):
        self._tmp.cleanup()

    def _client(self, instructions=None):
        client = QwenTTSClient.__new__(QwenTTSClient)
        client.chunks_dir = Path(self._tmp.name)
        client.voice_mode = VOICE_MODE_DESIGN
        client.language = config.LANGUAGE
        client.instructions = (instructions or "").strip()
        client.api_info = {"named_endpoints": {"/run_voice_design": {
            "parameters": [
                {"parameter_name": "text"},
                {"parameter_name": "lang_disp"},
                {"parameter_name": "design"},
            ]}}}
        return client

    def _fake_output(self) -> str:
        out = Path(self._tmp.name) / "server_out.wav"
        out.write_bytes(b"\x01\x00")
        return str(out)

    def test_voice_mode_design_is_valid(self):
        self.assertIn(VOICE_MODE_DESIGN, VOICE_MODES)

    def test_generate_payload_and_return(self):
        client = self._client(instructions="A warm narrator")
        fake = MagicMock(return_value=(self._fake_output(),))
        with patch.object(client, "_generate_voice_design", fake):
            result = client._generate_sub_request(
                "Hello there.", self._tmp.name, 1, 1, 1)
        fake.assert_called_once_with("Hello there.")
        self.assertEqual(Path(result).name, "part_01.wav")

    def test_payload_uses_design_field_language_and_instruction(self):
        client = self._client(instructions="A warm narrator")
        captured = {}

        def fake_predict(**payload):
            captured.update(payload)
            return (self._fake_output(),)

        client.client = MagicMock()
        client.client.predict.side_effect = fake_predict
        result = client._generate_voice_design("Hi.")
        self.assertEqual(captured["text"], "Hi.")
        self.assertEqual(captured["lang_disp"], config.LANGUAGE)
        self.assertEqual(captured["design"], "A warm narrator")
        self.assertNotIn("seed", captured)  # not accepted by this endpoint
        self.assertEqual(result, (self._fake_output(),))

    def test_payload_uses_empty_design_field_when_no_instructions_given(self):
        # There is no configured default instruction: the client sends
        # whatever the run provided (empty when none).
        client = self._client(instructions=None)
        self.assertEqual(client.instructions, "")

    def test_unknown_api_falls_back_to_the_requested_name(self):
        client = self._client()
        client.api_info = {"named_endpoints": {}}
        client.client = MagicMock()
        client.client.predict.return_value = (self._fake_output(),)
        client._generate_voice_design("Hi.")
        _, kwargs = client.client.predict.call_args
        self.assertEqual(kwargs["api_name"], "/run_voice_design")



class AudioCppTTSClientHealthTests(unittest.TestCase):
    """Connection behavior of the audio.cpp client."""

    @staticmethod
    def _json_response(payload):
        response = MagicMock()
        response.__enter__.return_value = response
        response.read.return_value = json.dumps(payload).encode("utf-8")
        return response

    def _get_responses(self, health=None, models=None, voices=None):
        """Side effect dispatching GET responses by URL."""
        def _dispatch(request, **_kwargs):
            url = request if isinstance(request, str) else request.full_url
            if url.endswith("/health"):
                return self._json_response(health if health is not None
                                           else {"status": "ok"})
            if url.endswith("/v1/models"):
                return self._json_response(models if models is not None else
                                           {"data": [{"id": _AUDIOCPP_MODEL_ID,
                                                      "family": "qwen3_tts"}]})
            if "/v1/audio/voices" in url:
                if voices is Exception:
                    raise Exception("voices endpoint down")
                return self._json_response(voices if voices is not None
                                           else {"voices": ["narrator"]})
            raise AssertionError(f"unexpected URL: {url}")
        return _dispatch

    def _client(self, voice=None, language=None,
                model_id=_AUDIOCPP_MODEL_ID, **kwargs):
        with patch("converter.clients.faster.urllib.request.urlopen",
                   side_effect=self._get_responses(**kwargs)):
            return AudioCppTTSClient(_DUMMY_CHUNKS, voice=voice,
                                     language=language, model_id=model_id)

    def test_unreachable_server_raises_with_readme_pointer(self):
        import urllib.error
        with patch("converter.clients.faster.urllib.request.urlopen",
                   side_effect=urllib.error.URLError("Connection refused")):
            with self.assertRaises(RuntimeError) as ctx:
                AudioCppTTSClient(_DUMMY_CHUNKS)
        message = str(ctx.exception)
        self.assertIn("not reachable", message)
        self.assertIn("README", message)

    def test_unhealthy_status_raises(self):
        with self.assertRaises(RuntimeError) as ctx:
            self._client(health={"status": "starting"})
        self.assertIn("starting", str(ctx.exception))

    def test_unknown_model_id_raises_with_configured_ids(self):
        with self.assertRaises(RuntimeError) as ctx:
            self._client(models={"data": [{"id": "pocket-tts"}, {"id": "other"}]})
        message = str(ctx.exception)
        self.assertIn(_AUDIOCPP_MODEL_ID, message)
        self.assertIn("pocket-tts", message)
        self.assertIn("other", message)

    def test_healthy_server_speaker_mode_defaults(self):
        client = self._client(voice="Vivian")
        self.assertEqual(client.api_url, config.AUDIOCPP_API_URL.rstrip("/"))
        self.assertEqual(client.model_id, _AUDIOCPP_MODEL_ID)
        self.assertEqual(client.language, config.LANGUAGE)
        self.assertEqual(client.voice, "Vivian")
        self.assertFalse(client.preset_mode)
        self.assertTrue(client.speaker_mode)

    def test_speaker_mode_normalizes_the_speaker_name(self):
        client = self._client(voice="uncle_fu")
        self.assertEqual(client.voice, "Uncle Fu")
        self.assertTrue(client.speaker_mode)

    def test_no_voice_on_speaker_entry_raises(self):
        # There is no configured default speaker: a CustomVoice entry
        # without --voice fails fast instead of guessing one.
        with self.assertRaises(RuntimeError) as ctx:
            self._client()
        message = str(ctx.exception)
        self.assertIn("built-in speakers", message)
        self.assertIn("--voice", message)

    def test_voice_speaker_name_selects_speaker_mode(self):
        # --voice naming a built-in CustomVoice speaker selects speaker
        # mode; the name is normalized to its wire (display) form and no
        # preset validation runs.
        client = self._client(voice="Uncle_Fu")
        self.assertEqual(client.voice, "Uncle Fu")
        self.assertFalse(client.preset_mode)
        self.assertTrue(client.speaker_mode)

    def test_voice_speaker_name_on_clone_entry_is_a_preset(self):
        # --voice on a clone-only (Base) entry is a server-side preset,
        # not a built-in speaker, so the name is validated against the
        # server's voice library.
        with self.assertRaises(RuntimeError) as ctx:
            self._client(voice="Ryan", model_id="Qwen3-TTS-12Hz-1.7B-Base-GGUF",
                         models={"data": [
                             {"id": "Qwen3-TTS-12Hz-1.7B-Base-GGUF",
                              "family": "qwen3_tts"}]})
        message = str(ctx.exception)
        self.assertIn("'Ryan'", message)
        self.assertIn("--voice", message)

    def test_speaker_mode_stays_on_the_selected_entry(self):
        # A built-in speaker name selects speaker mode on the entry the
        # run picked; no second-entry rerouting exists anymore.
        client = self._client(
            voice="Ryan", model_id="Qwen3-TTS-CustomVoice",
            models={"data": [{"id": "Qwen3-TTS-CustomVoice",
                              "family": "qwen3_tts"},
                             {"id": "qwen3-tts-clone",
                              "family": "qwen3_tts"}]})
        self.assertEqual(client.model_id, "Qwen3-TTS-CustomVoice")
        self.assertTrue(client.speaker_mode)
        self.assertFalse(client.preset_mode)

    def test_no_voice_on_base_entry_raises_instead_of_silent_speaker(self):
        # The Base model has no built-in speakers: without --voice the run
        # fails fast instead of silently sending a speaker name that the
        # model ignores.
        with self.assertRaises(RuntimeError) as ctx:
            self._client(model_id="Qwen3-TTS-12Hz-1.7B-Base-GGUF",
                         models={"data": [
                             {"id": "Qwen3-TTS-12Hz-1.7B-Base-GGUF",
                              "family": "qwen3_tts"}]})
        message = str(ctx.exception)
        self.assertIn("Base-GGUF", message)
        self.assertIn("--voice", message)

    def test_preset_mode_uses_requested_voice(self):
        client = self._client(voice="narrator")
        self.assertEqual(client.voice, "narrator")
        self.assertTrue(client.preset_mode)

    def test_preset_mode_validates_voice_against_server_list(self):
        with self.assertRaises(RuntimeError) as ctx:
            self._client(voice="ghost", voices={"voices": ["narrator", "obama"]})
        message = str(ctx.exception)
        self.assertIn("ghost", message)
        self.assertIn("narrator", message)
        self.assertIn("obama", message)

    def test_preset_mode_skips_validation_when_voices_endpoint_fails(self):
        client = self._client(voice="narrator", voices=Exception)
        self.assertEqual(client.voice, "narrator")

    def test_invalid_language_fails_before_connect(self):
        with patch("converter.clients.faster.urllib.request.urlopen") as mock_urlopen:
            with self.assertRaises(ValueError):
                AudioCppTTSClient(_DUMMY_CHUNKS, language="klingon")
        mock_urlopen.assert_not_called()

    def test_explicit_language_normalized(self):
        client = self._client(language="ja", voice="Vivian")
        self.assertEqual(client.language, "Japanese")

    def test_seed_resolved_once_per_run(self):
        with patch.object(config, "CONSTANT_SEED", True), \
                patch.object(config, "SEED", -1):
            client = self._client(voice="Vivian")
        self.assertGreaterEqual(client._seed, 0)

    def test_preset_mode_stays_on_the_requested_entry(self):
        # Preset (cloning) requests synthesize with the entry the run
        # selected; pick the Base entry with --model to clone on it.
        client = self._client(
            voice="narrator", model_id="qwen3-tts",
            models={"data": [{"id": "qwen3-tts"}, {"id": "qwen3-tts-clone"}]})
        self.assertEqual(client.model_id, "qwen3-tts")
        self.assertTrue(client.preset_mode)

    def test_empty_model_id_auto_picks_single_server_entry(self):
        # A multi-model server used without editing config.py: an empty
        # --model resolves to the only hosted entry automatically.
        client = self._client(
            voice="narrator", model_id="",
            models={"data": [{"id": "higgs", "family": "higgs_audio_tts"}]},
            voices={"voices": ["narrator"]})
        self.assertEqual(client.model_id, "higgs")

    def test_empty_model_id_with_multiple_entries_requires_explicit_choice(self):
        with self.assertRaises(RuntimeError) as ctx:
            self._client(
                voice="narrator", model_id="",
                models={"data": [{"id": "higgs"}, {"id": "voxcpm2"}]},
                voices={"voices": ["narrator"]})
        message = str(ctx.exception)
        self.assertIn("--model", message)
        self.assertIn("higgs", message)
        self.assertIn("voxcpm2", message)

    def test_model_id_reaches_request(self):
        # The per-run --model value is what the client requests.
        client = self._client(
            voice="narrator", model_id="higgs",
            models={"data": [{"id": "higgs", "family": "higgs_audio_tts"}]},
            voices={"voices": ["narrator"]})
        self.assertEqual(client.model_id, "higgs")

    def test_preset_mode_on_a_single_clone_entry_server(self):
        # A server hosting only the Base (cloning) entry: select it with
        # --model and a preset voice works.
        client = self._client(
            voice="narrator", model_id="qwen3-tts-clone",
            models={"data": [{"id": "qwen3-tts-clone"}]})
        self.assertEqual(client.model_id, "qwen3-tts-clone")
        self.assertTrue(client.preset_mode)

    def test_unknown_model_id_error_suggests_a_model(self):
        # Requesting an id the server does not host fails fast and names
        # both the requested and the hosted ids.
        with self.assertRaises(RuntimeError) as ctx:
            self._client(voice="narrator", model_id="qwen3-tts",
                         models={"data": [{"id": "qwen3-tts-clone"}]})
        message = str(ctx.exception)
        self.assertIn("qwen3-tts", message)
        self.assertIn("qwen3-tts-clone", message)
        self.assertIn("--model", message)

    def test_preset_mode_with_no_matching_model_lists_both_ids(self):
        with self.assertNoLogs("converter.clients.audiocpp", level="WARNING"):
            with self.assertRaises(RuntimeError) as ctx:
                self._client(voice="narrator", model_id="qwen3-tts",
                             models={"data": [{"id": "pocket-tts"}]})
        message = str(ctx.exception)
        self.assertIn("qwen3-tts", message)
        self.assertIn("pocket-tts", message)


class AudioCppTaskDetectionTests(unittest.TestCase):
    """Task auto-detection (tts/clon/vdes) and voice design validation."""

    @staticmethod
    def _json_response(payload):
        response = MagicMock()
        response.__enter__.return_value = response
        response.read.return_value = json.dumps(payload).encode("utf-8")
        return response

    def _client(self, voice=None, instructions=None, request_options=None,
                models=None):
        if models is None:
            models = {"data": [{"id": _AUDIOCPP_MODEL_ID,
                                "family": "qwen3_tts"}]}

        def _dispatch(request, **_kwargs):
            url = request if isinstance(request, str) else request.full_url
            if url.endswith("/health"):
                return self._json_response({"status": "ok"})
            if url.endswith("/v1/models"):
                return self._json_response(models)
            if "/v1/audio/voices" in url:
                return self._json_response({"voices": ["narrator"]})
            raise AssertionError(f"unexpected URL: {url}")

        with patch("converter.clients.faster.urllib.request.urlopen",
                   side_effect=_dispatch):
            return AudioCppTTSClient(_DUMMY_CHUNKS, voice=voice,
                                     instructions=instructions,
                                     request_options=request_options,
                                     model_id=_AUDIOCPP_MODEL_ID)

    def test_missing_task_falls_back_to_tts(self):
        # Servers that predate the task field hosted plain TTS models.
        client = self._client(voice="Vivian", models={"data": [
            {"id": _AUDIOCPP_MODEL_ID, "family": "qwen3_tts"}]})
        self.assertEqual(client.task, AUDIOCPP_TASK_TTS)
        self.assertFalse(client.design_mode)

    def test_task_detected_from_models_endpoint(self):
        client = self._client(models={"data": [
            {"id": _AUDIOCPP_MODEL_ID, "family": "qwen3_tts",
             "task": "vdes"}]},
            instructions="A warm adult narrator")
        self.assertEqual(client.task, AUDIOCPP_TASK_VDES)
        self.assertTrue(client.design_mode)

    def test_clon_task_entry_connects_in_preset_mode(self):
        client = self._client(voice="narrator", models={"data": [
            {"id": _AUDIOCPP_MODEL_ID, "family": "chatterbox",
             "task": "clon"}]})
        self.assertEqual(client.task, "clon")
        self.assertFalse(client.design_mode)
        self.assertTrue(client.preset_mode)

    def test_unsupported_task_rejected_with_available_entries(self):
        with self.assertRaises(RuntimeError) as ctx:
            self._client(models={"data": [
                {"id": _AUDIOCPP_MODEL_ID, "family": "qwen3_asr",
                 "task": "asr"},
                {"id": "tts-1", "family": "qwen3_tts", "task": "tts"}]},
                instructions="unused")
        message = str(ctx.exception)
        self.assertIn("'asr'", message)
        self.assertIn("--model", message)
        self.assertIn("tts-1", message)

    def test_vdes_without_instructions_requires_description(self):
        with self.assertRaises(RuntimeError) as ctx:
            self._client(models={"data": [
                {"id": _AUDIOCPP_MODEL_ID, "family": "qwen3_tts",
                 "task": "vdes"}]})
        message = str(ctx.exception)
        self.assertIn("voice design", message)
        self.assertIn("--instructions", message)

    def test_vdes_with_voice_rejected(self):
        with self.assertRaises(RuntimeError) as ctx:
            self._client(voice="narrator", models={"data": [
                {"id": _AUDIOCPP_MODEL_ID, "family": "qwen3_tts",
                 "task": "vdes"}]},
                instructions="A warm adult narrator")
        self.assertIn("--voice", str(ctx.exception))
        self.assertIn("--instructions", str(ctx.exception))

    def test_vdes_with_instructions_connects_in_design_mode(self):
        buf = io.StringIO()
        with redirect_stdout(buf):
            client = self._client(models={"data": [
                {"id": _AUDIOCPP_MODEL_ID, "family": "qwen3_tts",
                 "task": "vdes"}]},
                instructions="A warm adult narrator")
        self.assertTrue(client.design_mode)
        self.assertEqual(client.instructions, "A warm adult narrator")
        out = buf.getvalue()
        self.assertIn("voice design", out)
        self.assertIn("A warm adult narrator", out)

    def test_instructions_without_voice_on_generic_family_connects(self):
        # Families without built-in speakers can get their voice from the
        # instruction alone (e.g. OmniVoice voice design).
        buf = io.StringIO()
        with redirect_stdout(buf):
            client = self._client(models={"data": [
                {"id": _AUDIOCPP_MODEL_ID, "family": "omnivoice",
                 "task": "tts"}]},
                instructions="female, young adult, moderate pitch")
        self.assertFalse(client.design_mode)
        self.assertTrue(client.instruction_voice)
        self.assertIn("instruction voice", buf.getvalue())

    def test_instructions_with_builtin_speaker_family_stays_speaker_mode(self):
        buf = io.StringIO()
        with redirect_stdout(buf):
            client = self._client(
                voice="Vivian",
                models={"data": [
                    {"id": _AUDIOCPP_MODEL_ID, "family": "qwen3_tts",
                     "task": "tts"}]},
                instructions="Very happy.")
        self.assertFalse(client.design_mode)
        self.assertFalse(client.instruction_voice)
        self.assertIn("speaker 'Vivian'", buf.getvalue())

    def test_instructions_reach_the_client(self):
        client = self._client(
            models={"data": [
                {"id": _AUDIOCPP_MODEL_ID, "family": "qwen3_tts",
                 "task": "vdes"}]},
            instructions="from flag")
        self.assertEqual(client.instructions, "from flag")


class AudioCppFamilyDetectionTests(unittest.TestCase):
    """Family auto-detection and per-family adaptations."""

    @staticmethod
    def _json_response(payload):
        response = MagicMock()
        response.__enter__.return_value = response
        response.read.return_value = json.dumps(payload).encode("utf-8")
        return response

    def _client(self, voice="narrator", models=None):
        def _dispatch(request, **_kwargs):
            url = request if isinstance(request, str) else request.full_url
            if url.endswith("/health"):
                return self._json_response({"status": "ok"})
            if url.endswith("/v1/models"):
                return self._json_response(models)
            if "/v1/audio/voices" in url:
                return self._json_response({"voices": [voice] if voice else []})
            raise AssertionError(f"unexpected URL: {url}")

        with patch("converter.clients.faster.urllib.request.urlopen",
                   side_effect=_dispatch):
            return AudioCppTTSClient(_DUMMY_CHUNKS, voice=voice,
                                     model_id=_AUDIOCPP_MODEL_ID)

    def test_family_detected_from_models_endpoint(self):
        client = self._client(models={"data": [
            {"id": _AUDIOCPP_MODEL_ID, "family": "higgs_audio_tts"}]})
        self.assertEqual(client.family, "higgs_audio_tts")
        self.assertIs(client.profile, AUDIOCPP_DEFAULT_FAMILY_PROFILE)

    def test_missing_family_uses_generic_profile(self):
        # A missing family is unknown (not guessed as qwen3_tts): it falls
        # through to the generic clone-only profile.
        client = self._client(models={"data": [
            {"id": _AUDIOCPP_MODEL_ID}]})
        self.assertEqual(client.family, "")
        self.assertIs(client.profile, AUDIOCPP_DEFAULT_FAMILY_PROFILE)

    def test_unknown_family_uses_generic_profile(self):
        client = self._client(models={"data": [
            {"id": _AUDIOCPP_MODEL_ID, "family": "future_tts"}]})
        self.assertEqual(client.family, "future_tts")
        self.assertIs(client.profile, AUDIOCPP_DEFAULT_FAMILY_PROFILE)
        self.assertEqual(client.profile.language_style, AUDIOCPP_LANG_OMIT)

    def test_speech_to_speech_only_family_rejected_at_connect(self):
        # A family whose model spec has no text-synthesis task
        # (PersonaPlex, s2s-only) cannot narrate regardless of its hosted
        # task: the run fails at connect with a pointer at the other
        # entries instead of a mid-run 500 on every request.
        cache = audiocpp_client._FAMILY_SPECS
        cache.clear()
        cache.update({"personaplex": {"tasks": ["s2s"]}})
        self.addCleanup(cache.clear)
        with self.assertRaises(RuntimeError) as ctx:
            self._client(models={"data": [
                {"id": _AUDIOCPP_MODEL_ID, "family": "personaplex",
                 "task": "tts"},
                {"id": "tts-1", "family": "qwen3_tts", "task": "tts"}]})
        message = str(ctx.exception)
        self.assertIn("personaplex", message)
        self.assertIn("speech-to-speech, not TTS", message)
        self.assertIn("cannot synthesize narration", message)
        self.assertIn("Consider deleting the model", message)
        self.assertIn(_AUDIOCPP_MODEL_ID, message)
        self.assertIn("tts-1", message)

    def test_speaker_mode_rejected_for_clone_only_family(self):
        # An unknown family (no spec, no built-in speakers) keeps the
        # conservative clone-only default: without --voice the run fails
        # fast instead of guessing.
        client = None
        try:
            client = self._client(voice=None, models={"data": [
                {"id": _AUDIOCPP_MODEL_ID, "family": "some_new_family"}]})
        except RuntimeError as exc:
            message = str(exc)
            self.assertIn("some_new_family", message)
            self.assertIn("--voice", message)
            self.assertIn("no built-in speakers", message)
        self.assertIsNone(client)

    def test_speaker_mode_allowed_for_customvoice_entry(self):
        # A Qwen3-TTS entry whose id names CustomVoice is speaker-capable;
        # a built-in speaker name selects speaker mode on it.
        client = self._client(
            voice="Vivian", models={"data": [
                {"id": "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF",
                 "family": "qwen3_tts"}]})
        self.assertEqual(client.family, "qwen3_tts")
        self.assertTrue(client.speaker_mode)

    def test_speaker_mode_rejected_for_qwen_base_entry(self):
        # A Qwen3-TTS entry whose id names Base (not CustomVoice) is
        # clone-only, even though its family has built-in speakers on other
        # entries: without --voice it fails fast.
        client = None
        try:
            client = self._client(voice=None, models={"data": [
                {"id": "Qwen3-TTS-12Hz-1.7B-Base-GGUF",
                 "family": "qwen3_tts"}]})
        except RuntimeError as exc:
            message = str(exc)
            self.assertIn("Base-GGUF", message)
            self.assertIn("--voice", message)
        self.assertIsNone(client)

    def test_iso_language_code_helper(self):
        self.assertEqual(LANGUAGE_ISO_CODES["English"], "en")
        self.assertIsNone(LANGUAGE_ISO_CODES.get("Auto"))

    def test_iso_codes_cover_the_audio_cpp_menu_languages(self):
        self.assertEqual(LANGUAGE_ISO_CODES["Arabic"], "ar")
        self.assertEqual(LANGUAGE_ISO_CODES["Hindi"], "hi")
        self.assertEqual(LANGUAGE_ISO_CODES["Vietnamese"], "vi")


class AudiocppEntryVoiceCapabilityTests(unittest.TestCase):
    """The per-entry voice capability resolver (speaker/clone/design)."""

    def _cap(self, family="", task="tts", model_id=""):
        return audiocpp_entry_voice_capability(family, task, model_id)

    def test_vdes_task_is_design(self):
        self.assertEqual(self._cap("qwen3_tts", "vdes",
                                   "Qwen3-TTS-VoiceDesign-GGUF"),
                         AUDIOCPP_VOICE_DESIGN)

    def test_qwen_customvoice_entry_is_speaker(self):
        self.assertEqual(self._cap("qwen3_tts", "tts",
                                   "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF"),
                         AUDIOCPP_VOICE_SPEAKER)

    def test_qwen_base_entry_is_clone(self):
        self.assertEqual(self._cap("qwen3_tts", "tts",
                                   "Qwen3-TTS-12Hz-1.7B-Base-GGUF"),
                         AUDIOCPP_VOICE_CLONE)

    def test_qwen_unidentified_entry_is_clone(self):
        self.assertEqual(self._cap("qwen3_tts", "tts", "qwen"),
                         AUDIOCPP_VOICE_CLONE)

    def test_other_families_are_clone(self):
        self.assertEqual(self._cap("higgs_audio_tts", "tts", "higgs"),
                         AUDIOCPP_VOICE_CLONE)

    def test_missing_family_is_clone(self):
        self.assertEqual(self._cap("", "tts", "legacy"),
                         AUDIOCPP_VOICE_CLONE)

    def test_customvoice_match_is_case_insensitive(self):
        self.assertEqual(self._cap("qwen3_tts", "tts",
                                   "Qwen3-TTS-12Hz-1.7B-CUSTOMVOICE-GGUF"),
                         AUDIOCPP_VOICE_SPEAKER)

    def test_customvoice_id_in_other_family_is_not_speaker(self):
        # The "customvoice" substring only marks a speaker for the qwen3_tts
        # family; another family with a lookalike id stays clone-only.
        self.assertEqual(self._cap("future_tts", "tts",
                                    "Qwen3-TTS-CustomVoice"),
                         AUDIOCPP_VOICE_CLONE)


class AudioCppFamilyVoicePolicyTests(unittest.TestCase):
    """The per-family voice policy resolver (required/optional/none)."""

    def setUp(self):
        # Seed the spec cache instead of reading the (gitignored, setup-
        # downloaded) checkout's model_specs, so the tests are hermetic.
        cache = audiocpp_client._FAMILY_SPECS
        cache.clear()
        cache.update({
            "higgs_audio_tts": {"tasks": ["tts", "clone"]},
            "supertonic": {"tasks": ["tts"]},
            "confucius4_tts": {"tasks": ["clone"]},
        })
        self.addCleanup(cache.clear)

    def test_pure_tts_family_needs_no_voice(self):
        self.assertEqual(audiocpp_family_voice_policy("supertonic"),
                         AUDIOCPP_VOICE_NONE)

    def test_mixed_family_has_an_optional_voice(self):
        self.assertEqual(audiocpp_family_voice_policy("higgs_audio_tts"),
                         AUDIOCPP_VOICE_OPTIONAL)

    def test_clone_only_spec_is_required(self):
        self.assertEqual(audiocpp_family_voice_policy("confucius4_tts"),
                         AUDIOCPP_VOICE_REQUIRED)

    def test_chatterbox_is_required_despite_its_spec(self):
        # The chatterbox spec wrongly lists "tts": the explicit
        # clone-only set wins so the binary's rejection is mirrored.
        self.assertEqual(audiocpp_family_voice_policy("chatterbox"),
                         AUDIOCPP_VOICE_REQUIRED)

    def test_qwen3_tts_is_entry_typed_and_stays_required(self):
        # Qwen3-TTS is decided per entry (speaker/clone/design), so the
        # family policy never loosens its voice requirement.
        self.assertEqual(audiocpp_family_voice_policy("qwen3_tts"),
                         AUDIOCPP_VOICE_REQUIRED)

    def test_vevo2_is_required_despite_its_spec(self):
        # Vevo2's spec lists tts/vc/svc but no "clone" task, yet its
        # zero-shot TTS route refuses every request without a timbre
        # reference: the explicit required set mirrors the server, so an
        # "All" run sends the picked voice instead of failing every
        # request with no voice at all.
        self.assertEqual(audiocpp_family_voice_policy("vevo2"),
                         AUDIOCPP_VOICE_REQUIRED)

    def test_unknown_family_keeps_the_conservative_default(self):
        self.assertEqual(audiocpp_family_voice_policy("brand_new_family"),
                         AUDIOCPP_VOICE_REQUIRED)


class AudioCppFamilyNarratesTests(unittest.TestCase):
    """The per-family "can this model narrate text at all" resolver."""

    def setUp(self):
        # Seed the spec cache like AudioCppFamilyVoicePolicyTests: the
        # tests stay hermetic without a downloaded checkout.
        cache = audiocpp_client._FAMILY_SPECS
        cache.clear()
        cache.update({
            "personaplex": {"tasks": ["s2s"]},
            "vibevoice": {"tasks": ["tts"]},
            "glm_tts": {"tasks": ["tts", "clone"]},
        })
        self.addCleanup(cache.clear)

    def test_speech_to_speech_only_family_cannot_narrate(self):
        self.assertFalse(audiocpp_family_narrates("personaplex"))

    def test_tts_family_narrates(self):
        self.assertTrue(audiocpp_family_narrates("vibevoice"))

    def test_mixed_family_narrates(self):
        self.assertTrue(audiocpp_family_narrates("glm_tts"))

    def test_unlisted_family_is_never_hidden(self):
        # No local spec for the family: conservatively narrating (None),
        # so an unknown family is never silently dropped from the menu.
        self.assertIsNone(audiocpp_family_narrates("brand_new_family"))


class AudioCppPlainTtsModeTests(unittest.TestCase):
    """Plain-TTS runs: families that synthesize without a reference voice."""

    @staticmethod
    def _json_response(payload):
        response = MagicMock()
        response.__enter__.return_value = response
        response.read.return_value = json.dumps(payload).encode("utf-8")
        return response

    # Minimal WAV: _request_wav only validates the RIFF/WAVE header.
    _WAV = b"RIFF\x04\x00\x00\x00WAVE"

    def _client(self, family, task="tts", voice=None, captured=None,
                instructions=None):
        def _dispatch(request, **_kwargs):
            url = request if isinstance(request, str) else request.full_url
            if url.endswith("/health"):
                return self._json_response({"status": "ok"})
            if url.endswith("/v1/models"):
                return self._json_response(
                    {"data": [{"id": "model", "family": family,
                               "task": task}]})
            if "/v1/audio/voices" in url:
                return self._json_response({"voices": ["narrator"]})
            if url.endswith("/v1/audio/speech"):
                if captured is not None:
                    captured.append(json.loads(request.data.decode("utf-8")))
                response = MagicMock()
                response.__enter__.return_value = response
                response.read.return_value = self._WAV
                return response
            if url.endswith("/unload_all_models"):
                return self._json_response({"unloaded": []})
            raise AssertionError(f"unexpected URL: {url}")

        # The patch stays up for the whole test so _request_wav calls land
        # on the dispatch too (capturing the speech payload).
        patcher = patch("converter.clients.faster.urllib.request.urlopen",
                        side_effect=_dispatch)
        patcher.start()
        self.addCleanup(patcher.stop)
        return AudioCppTTSClient(_DUMMY_CHUNKS, voice=voice,
                                 model_id="model",
                                 instructions=instructions)

    def test_pure_tts_family_connects_in_plain_mode(self):
        client = self._client("supertonic")
        self.assertTrue(client.plain_mode)
        self.assertFalse(client.preset_mode)
        self.assertFalse(client.design_mode)

    def test_mixed_family_without_voice_runs_plain(self):
        client = self._client("higgs_audio_tts")
        self.assertTrue(client.plain_mode)

    def test_plain_mode_omits_the_voice_field(self):
        captured = []
        client = self._client("supertonic", captured=captured)
        client._request_wav("Hello world.")
        self.assertNotIn("voice", captured[0])
        self.assertEqual(captured[0]["input"], "Hello world.")

    def test_clone_only_family_without_voice_still_raises(self):
        # Unknown families keep the conservative clone-only default.
        with self.assertRaises(RuntimeError) as ctx:
            self._client("some_new_family")
        self.assertIn("--voice", str(ctx.exception))

    def test_clone_only_family_hosted_as_tts_fails_fast(self):
        # A Chatterbox entry hosted with task "tts" fails every request at
        # session-creation time: refuse at connect with the re-host hint
        # instead of 500ing each chunk.
        with self.assertRaises(RuntimeError) as ctx:
            self._client("chatterbox")
        message = str(ctx.exception)
        self.assertIn("chatterbox", message)
        self.assertIn('"clon"', message)
        self.assertIn("Configure Backends", message)

    def test_clone_only_family_hosted_as_tts_fails_fast_with_voice(self):
        with self.assertRaises(RuntimeError) as ctx:
            self._client("chatterbox", task="tts", voice="narrator")
        self.assertIn('"clon"', str(ctx.exception))

    def test_clone_only_family_hosted_as_clon_needs_a_voice(self):
        with self.assertRaises(RuntimeError) as ctx:
            self._client("chatterbox", task="clon")
        message = str(ctx.exception)
        self.assertIn("--voice", message)
        self.assertIn("voice_preset", message)

    def test_clone_only_instructions_alone_do_not_define_the_voice(self):
        # The REQUIRED-policy refusal precedes the instruction-voice
        # branch: clone-only (and Vevo2-style) families cannot take their
        # voice from an instruction, so a voice-less run fails fast with
        # the --voice fix instead of 500ing every request server-side.
        with self.assertRaises(RuntimeError) as ctx:
            self._client("chatterbox", task="clon",
                         instructions="Calm and steady.")
        message = str(ctx.exception)
        self.assertIn("--voice", message)
        self.assertNotIn("instruction", message)

    def test_vevo2_without_voice_refuses_at_connect(self):
        with self.assertRaises(RuntimeError) as ctx:
            self._client("vevo2")
        message = str(ctx.exception)
        self.assertIn("--voice", message)
        self.assertIn("vevo2", message)

    def test_vevo2_with_voice_connects_in_preset_mode(self):
        client = self._client("vevo2", voice="narrator")
        self.assertTrue(client.preset_mode)
        self.assertFalse(client.plain_mode)


class AudioCppCloneOnlyErrorTests(unittest.TestCase):
    """The non-retryable classification of clone-only hosting 500s."""

    def _error(self, message):
        return audiocpp_request_error(
            500, json.dumps({"error": {"message": message}}))

    def test_chatterbox_hosting_error_is_not_retryable(self):
        exc = self._error(
            "Chatterbox supports VoiceCloning and VoiceConversion")
        self.assertIsInstance(exc, NonRetryableTTSError)
        self.assertIn("VoiceCloning and VoiceConversion", str(exc))
        self.assertIn('"clon"', str(exc))

    def test_confucius_hosting_error_is_not_retryable(self):
        exc = self._error("Confucius4-TTS supports the VoiceCloning task")
        self.assertIsInstance(exc, NonRetryableTTSError)

    def test_echo_hosting_error_is_not_retryable(self):
        exc = self._error("Echo-TTS only supports offline voice cloning")
        self.assertIsInstance(exc, NonRetryableTTSError)

    def test_unrelated_error_stays_retryable(self):
        exc = self._error("model busy")
        self.assertNotIsInstance(exc, NonRetryableTTSError)
        self.assertIn("model busy", str(exc))


class AudioCppDeterministicErrorTests(unittest.TestCase):
    """Deterministic audio.cpp failures are not retried.

    The fragments come from real 500 bodies (incomplete model packages,
    s2s-only families, unresolvable voices, VRAM exhaustion); retrying
    the identical request cannot succeed, so in an "All" run every broken
    model must be skipped in one attempt with its reason on screen.
    """

    def _error(self, message):
        return audiocpp_request_error(
            500, json.dumps({"error": {"message": message}}))

    def test_missing_model_package_file_is_not_retryable(self):
        exc = self._error(
            "failed to load model resources using builtin model spec for "
            "family 'glm_tts' source 'safetensors': missing model package "
            "file 'tokenizer_merges': /models/GLM-TTS_Q8")
        self.assertIsInstance(exc, NonRetryableTTSError)

    def test_missing_model_root_is_not_retryable(self):
        exc = self._error(
            "failed to select safetensors source from builtin model spec "
            "for family 'outetts': missing model root: dac=/models/OuteTTS")
        self.assertIsInstance(exc, NonRetryableTTSError)

    def test_ambiguous_gguf_directory_is_not_retryable(self):
        exc = self._error(
            "model directory contains 4 GGUF files: /models/MiniMax-H3-Q4-"
            "GGUF; found: audio_vae_folded_f16.gguf, dit.gguf, ...")
        self.assertIsInstance(exc, NonRetryableTTSError)

    def test_missing_companion_model_path_is_not_retryable(self):
        exc = self._error(
            "model path does not exist: /tmp/audiocpp-gguf/"
            "MioCodec-25Hz-44.1kHz-v2")
        self.assertIsInstance(exc, NonRetryableTTSError)

    def test_speech_to_speech_only_family_is_not_retryable_with_a_hint(self):
        exc = self._error("PersonaPlex supports only speech-to-speech sessions")
        self.assertIsInstance(exc, NonRetryableTTSError)
        self.assertIn("speech-to-speech, not TTS", str(exc))
        self.assertIn("Consider deleting the model", str(exc))

    def test_unresolvable_clone_voice_is_not_retryable_with_a_hint(self):
        exc = self._error("Vevo2 requires target_voice or voice speaker audio")
        self.assertIsInstance(exc, NonRetryableTTSError)
        self.assertIn("reference audio", str(exc))

    def test_unscripted_vibevoice_prompt_is_not_retryable(self):
        exc = self._error("VibeVoice prompt has no valid Speaker N: lines")
        self.assertIsInstance(exc, NonRetryableTTSError)

    def test_reference_over_encoder_capacity_is_not_retryable_with_a_hint(self):
        exc = self._error("VoxCPM2 AudioVAE encoder sample capacity exceeded")
        self.assertIsInstance(exc, NonRetryableTTSError)
        self.assertIn("trim", str(exc))

    def test_allocation_failures_are_not_retryable_with_a_hint(self):
        # VRAM does not change between attempts of a sequential run (the
        # "All" loop unloads models between books, not between retries).
        # The hint names the server log (which records the exact attempted
        # allocation size) and the DramaBox mem_saver session option.
        for message in ("DramaBox vocoder backend buffer allocation failed",
                        "failed to allocate MOSS codec encoder forward graph"):
            exc = self._error(message)
            self.assertIsInstance(exc, NonRetryableTTSError)
            self.assertIn("audiocpp-server.log", str(exc))
            self.assertIn("dramabox.mem_saver", str(exc))

    def test_missing_companion_hint_names_the_configure_fix(self):
        exc = self._error(
            "model path does not exist: /tmp/audiocpp-gguf/MioCodec-25Hz"
            "-44.1kHz-v2")
        self.assertIn("companion package", str(exc))
        self.assertIn("Configure Backends", str(exc))

    def test_stale_package_layout_hint_names_the_re_download(self):
        exc = self._error("missing model package file 'tokenizer_merges'")
        self.assertIn("Configure Backends", str(exc))
        self.assertIn("re-downloaded", str(exc))

    def test_multi_gguf_directory_hint_names_the_hosting_fix(self):
        exc = self._error("model directory contains 4 GGUF files: /m")
        self.assertIn("several GGUFs", str(exc))
        self.assertIn("Configure Backends", str(exc))

    def test_sample_capacity_hint_names_the_capacity_override(self):
        exc = self._error("VoxCPM2 AudioVAE encoder sample capacity exceeded")
        self.assertIn("encoder-sample capacity", str(exc))
        self.assertIn("Configure Backends", str(exc))

    def test_allocation_log_note_is_appended_to_the_error(self):
        exc = audiocpp_request_error(
            500, json.dumps({"error": {"message":
                "DramaBox audio VAE backend buffer allocation failed"}}),
            log_note=" The server's log (/x) records the failed allocation "
                     "as: allocating 12.5 MiB on device 0")
        self.assertIn("allocating 12.5 MiB on device 0", str(exc))

    def test_log_note_is_not_appended_to_unrelated_errors(self):
        exc = audiocpp_request_error(
            500, json.dumps({"error": {"message": "model busy"}}),
            log_note=" The server's log (/x) records the failed allocation "
                     "as: allocating 12.5 MiB on device 0")
        self.assertNotIn("allocating 12.5 MiB", str(exc))

    def test_max_tokens_before_eoc_stays_retryable(self):
        # Proven transient: a request that hit it has succeeded on retry.
        exc = self._error("Higgs TTS generation reached max_tokens before EOC")
        self.assertNotIsInstance(exc, NonRetryableTTSError)



class AudioCppTTSClientRequestTests(unittest.TestCase):
    """The /v1/audio/speech payload and response validation."""

    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()
        self._sleep = patch("converter.clients.base.time.sleep")
        self._sleep.start()

    def tearDown(self):
        self._sleep.stop()
        self._tmp.cleanup()

    def _make_client(self, preset_mode=False, voice="Vivian", language="English", seed=-1,
                     family="qwen3_tts", task="tts",
                     instructions=None, request_options=None):
        client = AudioCppTTSClient.__new__(AudioCppTTSClient)
        client.chunks_dir = Path(self._tmp.name)
        client.api_url = "http://127.0.0.1:8080"
        client.model_id = _AUDIOCPP_MODEL_ID
        client.preset_mode = preset_mode
        client.voice = voice
        client.language = language
        client._seed = seed
        client.family = family
        client.task = task
        client.profile = AUDIOCPP_FAMILY_PROFILES.get(
            family, AUDIOCPP_DEFAULT_FAMILY_PROFILE)
        client.instructions = instructions or ""
        client.request_options = dict(request_options or {})
        client.design_mode = task == AUDIOCPP_TASK_VDES
        client.plain_mode = False
        # Mirrors the connect-time rule: an instruction-defined voice on a
        # clone-capable entry with no --voice (design mode takes precedence).
        capability = audiocpp_entry_voice_capability(
            family, task, client.model_id)
        client.instruction_voice = (
            not preset_mode and not client.design_mode
            and capability == AUDIOCPP_VOICE_CLONE
            and bool(client.instructions))
        return client

    @staticmethod
    def _wav_bytes(frames=b"\x01\x00" * 10, rate=SAMPLE_RATE):
        buffer = io.BytesIO()
        with wave.open(buffer, "wb") as wav_file:
            wav_file.setnchannels(1)
            wav_file.setsampwidth(2)
            wav_file.setframerate(rate)
            wav_file.writeframes(frames)
        return buffer.getvalue()

    def _post_response(self, body):
        response = MagicMock()
        response.__enter__.return_value = response
        response.read.return_value = body
        return response

    def test_payload_includes_model_input_voice_language_and_seed(self):
        client = self._make_client(preset_mode=True, voice="narrator",
                                   language="Japanese", seed=1234)
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=self._post_response(self._wav_bytes())) as mock_urlopen:
            client._request_wav("Hello world.")
        request = mock_urlopen.call_args[0][0]
        self.assertEqual(request.full_url,
                         "http://127.0.0.1:8080/v1/audio/speech")
        payload = json.loads(request.data.decode("utf-8"))
        self.assertEqual(payload["model"], _AUDIOCPP_MODEL_ID)
        self.assertEqual(payload["input"], "Hello world.")
        self.assertEqual(payload["voice"], "narrator")
        self.assertEqual(payload["language"], "Japanese")
        self.assertEqual(payload["seed"], 1234)
        self.assertNotIn("instructions", payload)

    def test_negative_seed_omitted_from_payload(self):
        client = self._make_client(preset_mode=True, voice="narrator", seed=-1)
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=self._post_response(self._wav_bytes())) as mock_urlopen:
            client._request_wav("Hello world.")
        payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8"))
        self.assertNotIn("seed", payload)

    def test_request_timeout_is_the_configured_api_timeout(self):
        client = self._make_client()
        long_text = " ".join(f"word{i}" for i in range(1500))
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=self._post_response(self._wav_bytes())) as mock_urlopen:
            client._request_wav(long_text)
        timeout = mock_urlopen.call_args[1]["timeout"]
        self.assertEqual(timeout, config.API_TIMEOUT)

    def test_speaker_mode_without_instructions_omits_the_field(self):
        # There is no configured style instruction: speaker mode sends no
        # instructions field unless the run provides one.
        client = self._make_client(preset_mode=False)
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=self._post_response(self._wav_bytes())) as mock_urlopen:
            client._request_wav("Hello.")
        payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8"))
        self.assertNotIn("instructions", payload)

    def test_explicit_instructions_reach_the_payload(self):
        client = self._make_client(preset_mode=False,
                                   instructions="Read whisper quiet.")
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=self._post_response(self._wav_bytes())) as mock_urlopen:
            client._request_wav("Hello.")
        payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8"))
        self.assertEqual(payload["instructions"], "Read whisper quiet.")

    def test_preset_mode_sends_instructions_alongside_voice(self):
        # Clone + style control: both the server-side voice and the
        # instruction reach the model.
        client = self._make_client(preset_mode=True, voice="narrator",
                                   instructions="Calm and steady.")
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=self._post_response(self._wav_bytes())) as mock_urlopen:
            client._request_wav("Hello.")
        payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8"))
        self.assertEqual(payload["voice"], "narrator")
        self.assertEqual(payload["instructions"], "Calm and steady.")

    def test_design_mode_payload_omits_voice_and_sends_instructions(self):
        client = self._make_client(task="vdes",
                                   instructions="A warm adult narrator")
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=self._post_response(self._wav_bytes())) as mock_urlopen:
            client._request_wav("Hello.")
        payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8"))
        self.assertNotIn("voice", payload)
        self.assertEqual(payload["instructions"], "A warm adult narrator")

    def test_design_mode_language_follows_family_profile(self):
        # The VoiceDesign package is family qwen3_tts, whose language field
        # takes Qwen display names like the other variants.
        client = self._make_client(task="vdes", language="Japanese",
                                   instructions="A warm adult narrator")
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=self._post_response(self._wav_bytes())) as mock_urlopen:
            client._request_wav("Hello.")
        payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8"))
        self.assertEqual(payload["language"], "Japanese")

    def test_instruction_voice_payload_omits_voice(self):
        # Instruction-defined voice on a family without built-in speakers:
        # no speaker name is invented, the instruction carries the voice.
        client = self._make_client(family="omnivoice",
                                   instructions="female, young adult")
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=self._post_response(self._wav_bytes())) as mock_urlopen:
            client._request_wav("Hello.")
        payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8"))
        self.assertNotIn("voice", payload)
        self.assertNotIn("language", payload)  # generic profile: omitted
        self.assertEqual(payload["instructions"], "female, young adult")

    def test_request_options_forwarded_in_payload(self):
        client = self._make_client(preset_mode=True, voice="narrator",
                                   request_options={"emotion": "neutral",
                                                    "speed": "1.1"})
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=self._post_response(self._wav_bytes())) as mock_urlopen:
            client._request_wav("Hello.")
        payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8"))
        self.assertEqual(payload["options"], {"emotion": "neutral",
                                              "speed": "1.1"})

    def test_empty_request_options_omit_options_field(self):
        client = self._make_client(preset_mode=True, voice="narrator")
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=self._post_response(self._wav_bytes())) as mock_urlopen:
            client._request_wav("Hello.")
        payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8"))
        self.assertNotIn("options", payload)

    def test_generic_family_omits_language_and_instructions(self):
        # Clone-only families (higgs_audio_tts, voxcpm2, ...) detect the
        # language themselves and take no style instruction.
        client = self._make_client(preset_mode=False, family="higgs_audio_tts")
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=self._post_response(self._wav_bytes())) as mock_urlopen:
            client._request_wav("Hello.")
        payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8"))
        self.assertNotIn("language", payload)
        self.assertNotIn("instructions", payload)

    def test_iso_family_sends_language_code(self):
        client = self._make_client(preset_mode=True, voice="narrator",
                                   language="Japanese", family="index_tts2")
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=self._post_response(self._wav_bytes())) as mock_urlopen:
            client._request_wav("Hello.")
        payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8"))
        self.assertEqual(payload["language"], "ja")

    def test_iso_family_auto_omits_language(self):
        client = self._make_client(preset_mode=True, voice="narrator",
                                   language="Auto", family="index_tts2")
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=self._post_response(self._wav_bytes())) as mock_urlopen:
            client._request_wav("Hello.")
        payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8"))
        self.assertNotIn("language", payload)

    def test_qwen_language_display_name_still_sent(self):
        client = self._make_client(preset_mode=True, voice="narrator",
                                   language="Japanese", family="qwen3_tts")
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=self._post_response(self._wav_bytes())) as mock_urlopen:
            client._request_wav("Hello.")
        payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8"))
        self.assertEqual(payload["language"], "Japanese")

    def test_non_wav_response_rejected(self):
        client = self._make_client()
        for body in (b"", b"RIFFxxxx", b"MP3DATA-MP3DATA", b"RIFF\x00\x00\x00\x00mpeg"):
            with patch("converter.clients.faster.urllib.request.urlopen",
                       return_value=self._post_response(body)):
                with self.assertRaises(RuntimeError):
                    client._request_wav("Hello.")

    def test_vibevoice_prompt_is_flattened_into_one_script_line(self):
        # VibeVoice parses the prompt line by line and silently drops
        # every line without a "Speaker N:" prefix, so the client formats
        # each sub-request as one Speaker-1 line (the server maps the
        # lowest speaker to the cloned reference voice).
        client = self._make_client(preset_mode=True, voice="narrator",
                                   family="vibevoice")
        with patch("converter.clients.faster.urllib.request.urlopen",
                   return_value=self._post_response(self._wav_bytes())) as mock_urlopen:
            client._request_wav("Hello world.\n\nSecond paragraph  here.")
        payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8"))
        self.assertEqual(payload["input"],
                         "Speaker 1: Hello world. Second paragraph here.")

    def test_other_families_keep_the_raw_text(self):
        # The script formatting is the vibevoice profile's alone: every
        # other family sends the text untouched.
        self.assertIsNone(
            AUDIOCPP_FAMILY_PROFILES.get("higgs_audio_tts",
                                         AUDIOCPP_DEFAULT_FAMILY_PROFILE)
            .script_prefix)
        self.assertEqual(audiocpp_script_input("Speaker 1", "a\nb"),
                         "Speaker 1: a b")

    def test_http_error_body_surfaced(self):
        import urllib.error
        client = self._make_client()
        error = urllib.error.HTTPError(
            "http://127.0.0.1:8080/v1/audio/speech", 500,
            "Server Error", {}, io.BytesIO(b'{"error":"bad voice"}'))
        with patch("converter.clients.faster.urllib.request.urlopen", side_effect=error):
            with self.assertRaises(RuntimeError) as ctx:
                client._request_wav("Hello.")
        self.assertIn("500", str(ctx.exception))
        self.assertIn("bad voice", str(ctx.exception))

    def test_reference_text_error_is_not_retryable(self):
        # Qwen3-TTS Base cloning without a server-side transcript fails
        # identically on every attempt: the error must carry the fix
        # (prompt_text / x_vector_only_mode) and skip the retry budget.
        client = self._make_client(preset_mode=True, voice="narrator")
        error = urllib.error.HTTPError(
            "http://127.0.0.1:8080/v1/audio/speech", 500,
            "Server Error", {},
            io.BytesIO(b'{"error":{"message":"Qwen3 voice clone ICL mode '
                      b'requires reference text","type":"server_error"}}'))
        with patch("converter.clients.faster.urllib.request.urlopen",
                   side_effect=error):
            with self.assertRaises(NonRetryableTTSError) as ctx:
                client._request_wav("Hello.")
        message = str(ctx.exception)
        self.assertIn("requires reference text", message)
        self.assertIn("'narrator'", message)
        self.assertIn("prompt_text", message)
        self.assertIn("x_vector_only_mode", message)

    def test_model_contract_error_is_not_retryable(self):
        client = self._make_client(preset_mode=True, voice="narrator")
        error = urllib.error.HTTPError(
            "http://127.0.0.1:8080/v1/audio/speech", 500,
            "Server Error", {},
            io.BytesIO(b'{"error":{"message":"model contract spec not found '
                      b"for family 'qwen3_tts' (provide --model-spec-override)\"}}"))
        with patch("converter.clients.faster.urllib.request.urlopen",
                   side_effect=error):
            with self.assertRaises(NonRetryableTTSError) as ctx:
                client._request_wav("Hello.")
        message = str(ctx.exception)
        self.assertIn("not retryable", message)
        self.assertIn("model contract spec not found for family 'qwen3_tts'",
                      message)
        self.assertIn("--model-spec-override", message)

    def test_unknown_model_id_error_is_not_retryable(self):
        client = self._make_client(preset_mode=True, voice="narrator")
        error = urllib.error.HTTPError(
            "http://127.0.0.1:8080/v1/audio/speech", 500,
            "Server Error", {},
            io.BytesIO(b'{"error":{"message":"unknown model id: nope"}}'))
        with patch("converter.clients.faster.urllib.request.urlopen",
                   side_effect=error):
            with self.assertRaises(NonRetryableTTSError) as ctx:
                client._request_wav("Hello.")
        message = str(ctx.exception)
        self.assertIn("not retryable", message)
        self.assertIn("unknown model id: nope", message)

    def test_unmatched_server_error_stays_retryable(self):
        # Only known-deterministic fragments skip the retry budget; device
        # hiccups, OOM, and anything unrecognized keep the plain error the
        # retry loop has always retried.
        client = self._make_client()
        error = urllib.error.HTTPError(
            "http://127.0.0.1:8080/v1/audio/speech", 500,
            "Server Error", {},
            io.BytesIO(b'{"error":{"message":"CUDA error at ggml-cuda.cu"}}'))
        with patch("converter.clients.faster.urllib.request.urlopen",
                   side_effect=error):
            with self.assertRaises(RuntimeError) as ctx:
                client._request_wav("Hello.")
        self.assertNotIsInstance(ctx.exception, NonRetryableTTSError)
        self.assertIn("CUDA error", str(ctx.exception))

    def test_non_retryable_error_skips_remaining_attempts(self):
        client = self._make_client()
        with patch.object(client, "generate_chunk",
                          side_effect=NonRetryableTTSError("nope")) as mock_gen, \
                patch("converter.clients.base.time.sleep") as mock_sleep:
            with self.assertRaises(NonRetryableTTSError):
                client.process_chunk_with_retry(1, "Hello.")
        self.assertEqual(mock_gen.call_count, 1)
        mock_sleep.assert_not_called()

    def test_transient_failure_fails_the_chunk_attempt(self):
        # Retrying is the chunk-level policy's job
        # (process_chunk_with_retry); one generate_chunk call makes one
        # request attempt per sub-chunk.
        client = self._make_client()
        wav = self._wav_bytes()
        with patch.object(client, "_request_wav",
                          side_effect=[RuntimeError("boom"), wav]) as mock_request:
            result = client.generate_chunk("Hello.", 1)
        self.assertIsNone(result)
        self.assertEqual(mock_request.call_count, 1)

    def test_request_failure_fails_the_chunk(self):
        client = self._make_client()
        with patch.object(client, "_request_wav",
                          side_effect=RuntimeError("down")) as mock_request:
            result = client.generate_chunk("Hello.", 1)
        self.assertIsNone(result)
        self.assertEqual(mock_request.call_count, 1)

    def test_empty_text_fails_the_chunk(self):
        client = self._make_client()
        with patch.object(client, "_request_wav") as mock_request:
            result = client.generate_chunk("   ", 1)
        self.assertIsNone(result)
        mock_request.assert_not_called()

    def test_generate_chunk_writes_valid_wav(self):
        client = self._make_client()
        frames = b"\x01\x00" * 100
        with patch.object(client, "_request_wav", return_value=self._wav_bytes(frames)):
            result = client.generate_chunk("Hello world.", 1)
        self.assertIsNotNone(result)
        path = Path(result)
        self.assertEqual(path.name, "chunk_0001.wav")
        with wave.open(str(path), "rb") as wav_file:
            self.assertEqual(wav_file.getnchannels(), 1)
            self.assertEqual(wav_file.getsampwidth(), 2)
            self.assertEqual(wav_file.getframerate(), SAMPLE_RATE)
            self.assertEqual(wav_file.readframes(wav_file.getnframes()), frames)

    def test_long_text_is_subchunked_and_concatenated_in_order(self):
        client = self._make_client()
        sentences = [" ".join(f"word{i}" for i in range(6)) + "." for _ in range(3)]
        text = " ".join(sentences)
        parts = [self._wav_bytes(b"\x01\x00" * 10),
                 self._wav_bytes(b"\x02\x00" * 20),
                 self._wav_bytes(b"\x03\x00" * 30)]
        with patch.object(config, "CHUNK_SIZE", 10), \
                patch.object(client, "_request_wav", side_effect=parts) as mock_request:
            result = client.generate_chunk(text, 1)
        self.assertEqual(mock_request.call_count, 3)
        with wave.open(str(Path(result)), "rb") as wav_file:
            self.assertEqual(wav_file.readframes(wav_file.getnframes()),
                             b"\x01\x00" * 10 + b"\x02\x00" * 20 + b"\x03\x00" * 30)

    def test_stale_chunk_files_are_removed(self):
        stale = Path(self._tmp.name) / "chunk_0001.mp3"
        stale.write_bytes(b"old")
        client = self._make_client()
        with patch.object(client, "_request_wav", return_value=self._wav_bytes()):
            client.generate_chunk("Hello.", 1)
        remaining = sorted(path.name for path in Path(self._tmp.name).glob("chunk_0001.*"))
        self.assertEqual(remaining, ["chunk_0001.wav"])


class TranscribeReasonTests(unittest.TestCase):
    """transcribe_reference_audio_detailed: a reason for every empty result.

    The audio.cpp setup prints the reason per voice, so each failure class
    must be distinguishable: missing package vs broken import vs transcribe
    error vs a silent no-speech result.
    """

    def _transcribe_with_models(self, models, spec_present=True):
        """Run one detailed transcription with _cached_model stubbed.

        MODELS maps backend name -> model object (or exception instance to
        raise in its place). The whisper fallback sees its own entry or a
        ModuleNotFoundError so no real package import ever happens;
        importlib.util.find_spec is pinned so the not-installed vs
        installed-but-broken distinction is deterministic in any env.
        """
        def fake_cached(key, loader):
            backend = key[0]
            entry = models.get(backend)
            if isinstance(entry, Exception):
                raise entry
            return entry
        with patch("converter.clients.transcribe._cached_model",
                   side_effect=fake_cached), \
                patch("importlib.util.find_spec",
                      return_value=MagicMock() if spec_present else None):
            return transcribe_reference_audio_detailed("clip.wav")

    def test_success_returns_text_and_ok(self):
        model = MagicMock()
        model.transcribe.return_value = (iter([MagicMock(text=" Hello. ")]),
                                         MagicMock())
        text, reason = self._transcribe_with_models(
            {"faster_whisper": model, "whisper": ModuleNotFoundError()})
        self.assertEqual(text, "Hello.")
        self.assertEqual(reason, "ok")

    def test_missing_backend_is_not_called_broken(self):
        text, reason = self._transcribe_with_models({
            "faster_whisper": ModuleNotFoundError(
                "No module named 'faster_whisper'"),
            "whisper": ModuleNotFoundError("No module named 'whisper'"),
        }, spec_present=False)
        self.assertIsNone(text)
        self.assertIn("faster_whisper is not installed", reason)
        self.assertIn("whisper is not installed", reason)

    def test_broken_import_is_distinguished_from_missing(self):
        text, reason = self._transcribe_with_models({
            "faster_whisper": ImportError(
                "Error loading shared library ld-linux-x86-64.so.2"),
            "whisper": ModuleNotFoundError("No module named 'whisper'",
                                           name="whisper"),
        })
        self.assertIsNone(text)
        self.assertIn("faster_whisper is installed but failed to import",
                      reason)
        self.assertIn("ld-linux-x86-64.so.2", reason)
        self.assertIn("whisper is not installed", reason)

    def test_transcribe_error_carries_the_exception(self):
        model = MagicMock()
        model.transcribe.side_effect = RuntimeError("decode failed")
        text, reason = self._transcribe_with_models(
            {"faster_whisper": model,
             "whisper": ModuleNotFoundError("No module named 'whisper'")})
        self.assertIsNone(text)
        self.assertIn("faster_whisper transcription failed: decode failed",
                      reason)

    def test_empty_result_reports_no_speech(self):
        model = MagicMock()
        model.transcribe.return_value = (iter([]), MagicMock())
        text, reason = self._transcribe_with_models(
            {"faster_whisper": model,
             "whisper": ModuleNotFoundError("No module named 'whisper'")})
        self.assertIsNone(text)
        self.assertIn("faster_whisper heard no speech", reason)

    def test_whisper_fallback_used_when_faster_whisper_fails(self):
        failing = MagicMock()
        failing.transcribe.side_effect = RuntimeError("boom")
        good = MagicMock()
        # The openai-whisper interface returns a dict with "text".
        good.transcribe.return_value = {"text": " Hi. "}
        text, reason = self._transcribe_with_models(
            {"faster_whisper": failing, "whisper": good})
        self.assertEqual(text, "Hi.")
        self.assertEqual(reason, "ok")

    def test_backend_problem_reports_broken_import(self):
        def fake_import(name, *args, **kwargs):
            raise ImportError("lib load failure")
        with patch("builtins.__import__", side_effect=fake_import), \
                patch("importlib.util.find_spec", return_value=MagicMock()):
            problem = whisper_backend_problem()
        self.assertIn("faster_whisper is installed but failed to import",
                      problem)
        self.assertIn("whisper is installed but failed to import", problem)

    def test_backend_problem_none_when_a_backend_imports(self):
        def fake_import(name, *args, **kwargs):
            if name == "faster_whisper":
                return MagicMock()
            raise ImportError("should not be probed")
        with patch("builtins.__import__", side_effect=fake_import):
            self.assertIsNone(whisper_backend_problem())


class AudioCppHeartbeatTests(unittest.TestCase):
    """The heartbeat reports chunk progress while a request generates."""

    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()

    def tearDown(self):
        self._tmp.cleanup()

    def _client(self):
        client = AudioCppTTSClient.__new__(AudioCppTTSClient)
        client.chunks_dir = Path(self._tmp.name)
        client.api_url = "http://127.0.0.1:8080"
        client.model_id = _AUDIOCPP_MODEL_ID
        client.preset_mode = False
        client.voice = "Vivian"
        client.language = "English"
        client._seed = -1
        client.family = "qwen3_tts"
        client.profile = AUDIOCPP_DEFAULT_FAMILY_PROFILE
        return client

    @staticmethod
    def _wav_bytes():
        buffer = io.BytesIO()
        with wave.open(buffer, "wb") as wav_file:
            wav_file.setnchannels(1)
            wav_file.setsampwidth(2)
            wav_file.setframerate(SAMPLE_RATE)
            wav_file.writeframes(b"\x01\x00" * 10)
        return buffer.getvalue()

    def _run(self):
        client = self._client()

        def slow_request(*_args, **_kwargs):
            time.sleep(0.12)
            return self._wav_bytes()

        buf = io.StringIO()
        with patch.object(config, "HEARTBEAT_INTERVAL_SECONDS", 0.03), \
                patch.object(client, "_request_wav",
                             side_effect=slow_request), \
                redirect_stdout(buf):
            result = client.generate_chunk("Hello.", 1)
        self.assertTrue(result)
        return buf.getvalue()

    def test_heartbeat_reports_chunk_progress(self):
        out = self._run()
        self.assertIn("Chunk 1 still generating", out)


class AudioCppTTSClientTruncationTests(unittest.TestCase):
    """Audio far shorter than its text implies fails the request."""

    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()

    def tearDown(self):
        self._tmp.cleanup()

    def _make_client(self):
        client = AudioCppTTSClient.__new__(AudioCppTTSClient)
        client.chunks_dir = Path(self._tmp.name)
        client.api_url = "http://127.0.0.1:8080"
        client.model_id = _AUDIOCPP_MODEL_ID
        client.preset_mode = True
        client.voice = "narrator"
        client.language = "English"
        client._seed = -1
        client.family = "qwen3_tts"
        client.profile = AUDIOCPP_FAMILY_PROFILES["qwen3_tts"]
        return client

    @staticmethod
    def _wav_bytes(frames):
        buffer = io.BytesIO()
        with wave.open(buffer, "wb") as wav_file:
            wav_file.setnchannels(1)
            wav_file.setsampwidth(2)
            wav_file.setframerate(SAMPLE_RATE)
            wav_file.writeframes(frames)
        return buffer.getvalue()

    def test_full_length_wav_passes(self):
        client = self._make_client()
        text = " ".join(f"word{i}" for i in range(12))
        # 12 words -> expected 4.8s, half is 2.4s -> 2.5s of audio passes.
        wav = self._wav_bytes(b"\x01\x00" * int(2.5 * SAMPLE_RATE))
        with patch.object(client, "_request_wav", return_value=wav):
            result = client.generate_chunk(text, 1)
        self.assertIsNotNone(result)


class AudioCppUnloadModelsTests(unittest.TestCase):
    """Before generating, the client asks the server to drop loaded models."""

    @staticmethod
    def _client():
        client = AudioCppTTSClient.__new__(AudioCppTTSClient)
        client.api_url = "http://127.0.0.1:8080"
        return client

    @staticmethod
    def _response(body):
        response = MagicMock()
        response.__enter__.return_value = response
        response.read.return_value = body
        return response

    def test_posts_to_unload_all_models(self):
        client = self._client()
        with patch("converter.clients.audiocpp.urllib.request.urlopen",
                   return_value=self._response(b'{"unloaded": ["qwen"]}')) as mock_urlopen:
            client._unload_server_models()
        request = mock_urlopen.call_args[0][0]
        self.assertEqual(request.full_url,
                         "http://127.0.0.1:8080/v1/tasks/unload_all_models")
        self.assertEqual(request.method, "POST")
        self.assertEqual(request.data, b"")

    def test_reports_unloaded_ids(self):
        client = self._client()
        buf = io.StringIO()
        with patch("converter.clients.audiocpp.urllib.request.urlopen",
                   return_value=self._response(b'{"unloaded": ["a", "b"]}')), \
                redirect_stdout(buf):
            client._unload_server_models()
        self.assertIn("Unloaded 2 model(s)", buf.getvalue())
        self.assertIn("a, b", buf.getvalue())

    def test_no_loaded_models_is_silent(self):
        client = self._client()
        buf = io.StringIO()
        with patch("converter.clients.audiocpp.urllib.request.urlopen",
                   return_value=self._response(b'{"unloaded": []}')), \
                redirect_stdout(buf):
            client._unload_server_models()
        self.assertEqual(buf.getvalue(), "")

    def test_http_error_warns_and_continues(self):
        client = self._client()
        buf = io.StringIO()
        with patch("converter.clients.audiocpp.urllib.request.urlopen",
                   side_effect=urllib.error.HTTPError(
                       "http://127.0.0.1:8080/v1/tasks/unload_all_models",
                       404, "Not Found", None, io.BytesIO())), \
                redirect_stdout(buf):
            client._unload_server_models()
        out = buf.getvalue()
        self.assertIn("[WARNING]", out)
        self.assertIn("404", out)

    def test_connection_error_warns_and_continues(self):
        client = self._client()
        buf = io.StringIO()
        with patch("converter.clients.audiocpp.urllib.request.urlopen",
                   side_effect=urllib.error.URLError("refused")), \
                redirect_stdout(buf):
            client._unload_server_models()
        self.assertIn("[WARNING]", buf.getvalue())

    def test_connect_unloads_before_returning(self):
        client = AudioCppTTSClient.__new__(AudioCppTTSClient)
        client.api_url = "http://127.0.0.1:8080"
        client.model_id = _AUDIOCPP_MODEL_ID
        client.preset_mode = True
        client.voice = "narrator"
        client.language = "English"
        client._seed = -1
        client.family = "qwen3_tts"
        client.task = AUDIOCPP_TASK_TTS
        client.profile = AUDIOCPP_FAMILY_PROFILES["qwen3_tts"]
        client.design_mode = False
        client.instruction_voice = False
        client.speaker_mode = False
        client.instructions = ""
        client._unload_models_override = None
        with patch.object(client, "_check_health"), \
                patch.object(client, "_list_models",
                             return_value=[{"id": client.model_id,
                                            "family": "qwen3_tts",
                                            "task": "tts"}]), \
                patch.object(client, "_auto_pick_model_id"), \
                patch.object(client, "_require_model_id"), \
                patch.object(client, "_resolve_family"), \
                patch.object(client, "_resolve_task"), \
                patch.object(client, "_check_voice"), \
                patch.object(client, "_unload_server_models") as mock_unload:
            client._connect()
        mock_unload.assert_called_once()

    def test_connect_unload_override_forces_unload(self):
        # "All (multiple generation)" runs force the unload regardless of
        # the AUDIOCPP_UNLOAD_MODELS setting, so each model starts clean.
        client = AudioCppTTSClient.__new__(AudioCppTTSClient)
        client.api_url = "http://127.0.0.1:8080"
        client.model_id = _AUDIOCPP_MODEL_ID
        client.preset_mode = True
        client.voice = "narrator"
        client.language = "English"
        client._seed = -1
        client.family = "qwen3_tts"
        client.task = AUDIOCPP_TASK_TTS
        client.profile = AUDIOCPP_FAMILY_PROFILES["qwen3_tts"]
        client.design_mode = False
        client.instruction_voice = False
        client.speaker_mode = False
        client.instructions = ""
        client._unload_models_override = True
        with patch.object(client, "_check_health"), \
                patch.object(client, "_list_models",
                             return_value=[{"id": client.model_id,
                                            "family": "qwen3_tts",
                                            "task": "tts"}]), \
                patch.object(client, "_auto_pick_model_id"), \
                patch.object(client, "_require_model_id"), \
                patch.object(client, "_resolve_family"), \
                patch.object(client, "_resolve_task"), \
                patch.object(client, "_check_voice"), \
                patch.object(config, "AUDIOCPP_UNLOAD_MODELS", False), \
                patch.object(client, "_unload_server_models") as mock_unload:
            client._connect()
        mock_unload.assert_called_once()

    def test_connect_unload_override_false_skips_unload(self):
        client = AudioCppTTSClient.__new__(AudioCppTTSClient)
        client.api_url = "http://127.0.0.1:8080"
        client.model_id = _AUDIOCPP_MODEL_ID
        client.preset_mode = True
        client.voice = "narrator"
        client.language = "English"
        client._seed = -1
        client.family = "qwen3_tts"
        client.task = AUDIOCPP_TASK_TTS
        client.profile = AUDIOCPP_FAMILY_PROFILES["qwen3_tts"]
        client.design_mode = False
        client.instruction_voice = False
        client.speaker_mode = False
        client.instructions = ""
        client._unload_models_override = False
        with patch.object(client, "_check_health"), \
                patch.object(client, "_list_models",
                             return_value=[{"id": client.model_id,
                                            "family": "qwen3_tts",
                                            "task": "tts"}]), \
                patch.object(client, "_auto_pick_model_id"), \
                patch.object(client, "_require_model_id"), \
                patch.object(client, "_resolve_family"), \
                patch.object(client, "_resolve_task"), \
                patch.object(client, "_check_voice"), \
                patch.object(config, "AUDIOCPP_UNLOAD_MODELS", True), \
                patch.object(client, "_unload_server_models") as mock_unload:
            client._connect()
        mock_unload.assert_not_called()

    def test_connect_skips_unload_when_disabled(self):
        client = AudioCppTTSClient.__new__(AudioCppTTSClient)
        client.api_url = "http://127.0.0.1:8080"
        client.model_id = _AUDIOCPP_MODEL_ID
        client.preset_mode = True
        client.voice = "narrator"
        client.language = "English"
        client._seed = -1
        client.family = "qwen3_tts"
        client.task = AUDIOCPP_TASK_TTS
        client.profile = AUDIOCPP_FAMILY_PROFILES["qwen3_tts"]
        client.design_mode = False
        client.instruction_voice = False
        client.speaker_mode = False
        client.instructions = ""
        client._unload_models_override = None
        with patch.object(client, "_check_health"), \
                patch.object(client, "_list_models",
                             return_value=[{"id": client.model_id,
                                            "family": "qwen3_tts",
                                            "task": "tts"}]), \
                patch.object(client, "_auto_pick_model_id"), \
                patch.object(client, "_require_model_id"), \
                patch.object(client, "_resolve_family"), \
                patch.object(client, "_resolve_task"), \
                patch.object(client, "_check_voice"), \
                patch.object(config, "AUDIOCPP_UNLOAD_MODELS", False), \
                patch.object(client, "_unload_server_models") as mock_unload:
            client._connect()
        mock_unload.assert_not_called()


class BackendWiringTests(unittest.TestCase):
    """AudiobookConverter wiring for the --backend selector."""

    def test_faster_backend_uses_faster_client_without_reference(self):
        with patch("converter.converter.FasterTTSClient") as mock_faster, \
                patch("converter.converter.QwenTTSClient") as mock_qwen, \
                patch("converter.converter.AudioCppTTSClient") as mock_audiocpp:
            AudiobookConverter(voice_mode=VOICE_MODE_CLONE,
                               backend=BACKEND_FASTER, voice="narrator")
        mock_faster.assert_called_once_with(chunks_dir=converter_mod.CHUNKS_FOLDER,
                                            voice="narrator", api_url=None,
                                            quiet=False, cancel=None)
        mock_qwen.assert_not_called()
        mock_audiocpp.assert_not_called()

    def test_audiocpp_backend_with_voice_uses_audiocpp_client(self):
        with patch("converter.converter.FasterTTSClient") as mock_faster, \
                patch("converter.converter.QwenTTSClient") as mock_qwen, \
                patch("converter.converter.AudioCppTTSClient") as mock_audiocpp:
            AudiobookConverter(voice_mode=VOICE_MODE_CLONE,
                               backend=BACKEND_AUDIOCPP, voice="narrator",
                               language="ja")
        mock_audiocpp.assert_called_once_with(chunks_dir=converter_mod.CHUNKS_FOLDER,
                                              voice="narrator", language="Japanese",
                                              model_id=None,
                                              instructions=None,
                                              request_options={},
                                              api_url=None, quiet=False,
                                              unload_models=None, cancel=None)
        mock_faster.assert_not_called()
        mock_qwen.assert_not_called()

    def test_audiocpp_backend_without_voice_uses_audiocpp_client(self):
        with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp:
            AudiobookConverter(voice_mode=VOICE_MODE_CUSTOM,
                               backend=BACKEND_AUDIOCPP)
        mock_audiocpp.assert_called_once_with(chunks_dir=converter_mod.CHUNKS_FOLDER,
                                              voice=None, language=config.LANGUAGE,
                                              model_id=None,
                                              instructions=None,
                                              request_options={},
                                              api_url=None, quiet=False,
                                              unload_models=None, cancel=None)

    def test_audiocpp_backend_model_id_is_wired_through(self):
        with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp:
            AudiobookConverter(voice_mode=VOICE_MODE_CLONE,
                               backend=BACKEND_AUDIOCPP, voice="narrator",
                               model_id="higgs")
        mock_audiocpp.assert_called_once_with(
            chunks_dir=converter_mod.CHUNKS_FOLDER,
            voice="narrator", language=config.LANGUAGE,
            model_id="higgs", instructions=None,
            request_options={}, api_url=None, quiet=False,
            unload_models=None, cancel=None)

    def test_audiocpp_backend_instructions_and_options_are_wired_through(self):
        with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp:
            AudiobookConverter(voice_mode=VOICE_MODE_CUSTOM,
                               backend=BACKEND_AUDIOCPP,
                               instructions="A warm adult narrator",
                               request_options={"emotion": "neutral",
                                                "speed": "1.1"})
        mock_audiocpp.assert_called_once_with(
            chunks_dir=converter_mod.CHUNKS_FOLDER,
            voice=None, language=config.LANGUAGE,
            model_id=None,
            instructions="A warm adult narrator",
            request_options={"emotion": "neutral", "speed": "1.1"},
            api_url=None, quiet=False, unload_models=None,
            cancel=None)

    def test_qwen_backend_uses_qwen_client(self):
        with patch("converter.converter.FasterTTSClient") as mock_faster, \
                patch("converter.converter.QwenTTSClient") as mock_qwen, \
                patch("converter.converter.AudioCppTTSClient") as mock_audiocpp:
            AudiobookConverter(voice_mode=VOICE_MODE_CUSTOM,
                               backend=BACKEND_QWEN, voice="Vivian")
        _, kwargs = mock_qwen.call_args
        self.assertEqual(kwargs["voice"], "Vivian")
        mock_faster.assert_not_called()
        mock_audiocpp.assert_not_called()

    def test_qwen_clone_mode_still_requires_reference(self):
        with patch("converter.converter.QwenTTSClient"):
            with self.assertRaises(ValueError):
                AudiobookConverter(voice_mode=VOICE_MODE_CLONE,
                                   backend=BACKEND_QWEN)

    def test_qwen_design_mode_without_instructions_rejected(self):
        # A VoiceDesign run needs a description; an empty instructions
        # value (not even the config default) is refused up front.
        with patch("converter.converter.QwenTTSClient"):
            with self.assertRaises(ValueError):
                AudiobookConverter(voice_mode=VOICE_MODE_DESIGN,
                                   backend=BACKEND_QWEN, instructions="   ")

    def test_qwen_design_mode_threads_instructions_to_the_client(self):
        with patch("converter.converter.QwenTTSClient") as mock_qwen:
            AudiobookConverter(voice_mode=VOICE_MODE_DESIGN,
                               backend=BACKEND_QWEN,
                               instructions="A warm adult female narrator")
        _, kwargs = mock_qwen.call_args
        self.assertEqual(kwargs["instructions"],
                         "A warm adult female narrator")

    def test_qwen_design_narrator_tag_uses_designed(self):
        self.assertEqual(AudiobookConverter.compute_narrator_tag(
            BACKEND_QWEN, None, VOICE_MODE_DESIGN, None,
            "A warm adult female narrator"), "designed")

    def test_api_url_override_reaches_each_client(self):
        # A remote conversion threads api_url through to the selected client.
        with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp:
            AudiobookConverter(voice_mode=VOICE_MODE_CLONE,
                               backend=BACKEND_AUDIOCPP, voice="narrator",
                               api_url="http://10.0.0.5:8080")
        mock_audiocpp.assert_called_once_with(
            chunks_dir=converter_mod.CHUNKS_FOLDER,
            voice="narrator", language=config.LANGUAGE, model_id=None,
            instructions=None, request_options={},
            api_url="http://10.0.0.5:8080", quiet=False,
            unload_models=None, cancel=None)
        with patch("converter.converter.FasterTTSClient") as mock_faster:
            AudiobookConverter(voice_mode=VOICE_MODE_CLONE,
                               backend=BACKEND_FASTER, voice="narrator",
                               api_url="http://10.0.0.5:8000")
        mock_faster.assert_called_once_with(chunks_dir=converter_mod.CHUNKS_FOLDER,
                                            voice="narrator",
                                            api_url="http://10.0.0.5:8000",
                                            quiet=False, cancel=None)
        with patch("converter.converter.QwenTTSClient") as mock_qwen:
            AudiobookConverter(voice_mode=VOICE_MODE_CUSTOM,
                               backend=BACKEND_QWEN, voice="Vivian",
                               api_url="http://10.0.0.5:7860")
        mock_qwen.assert_called_once_with(
            chunks_dir=converter_mod.CHUNKS_FOLDER,
            voice_mode=VOICE_MODE_CUSTOM, voice_clone_ref_audio=None,
            voice_clone_ref_text=None, skip_transcription=False,
            language=config.LANGUAGE, instructions=None,
            api_url="http://10.0.0.5:7860", quiet=False, voice="Vivian",
            cancel=None)

    def test_audiocpp_clone_mode_does_not_require_reference(self):
        # Cloning is server-side for the audiocpp backend, so the
        # clone-mode voice can be selected without local reference audio.
        with patch("converter.converter.AudioCppTTSClient"):
            converter = AudiobookConverter(voice_mode=VOICE_MODE_CLONE,
                                           backend=BACKEND_AUDIOCPP,
                                           voice="narrator")
        self.assertIsNone(converter.voice_clone_ref_audio)

    def test_chapter_chunks_audiocpp_splits(self):
        converter = self._audiocpp_converter(voice="narrator")
        text = " ".join(f"word{i}" for i in range(50))
        with patch.object(config, "CHUNK_SIZE", 10):
            chunks = converter._chapter_chunks(text)
        self.assertGreater(len(chunks), 1)
        self.assertTrue(all(len(chunk.split()) <= 10 for chunk in chunks))

    def test_chapter_chunks_qwen_always_splits(self):
        with patch("converter.converter.QwenTTSClient"):
            converter = AudiobookConverter(voice_mode=VOICE_MODE_CUSTOM,
                                           backend=BACKEND_QWEN,
                                           voice="Vivian")
        text = " ".join(f"word{i}" for i in range(50))
        with patch.object(config, "CHUNK_SIZE", 10):
            chunks = converter._chapter_chunks(text)
        self.assertGreater(len(chunks), 1)

    def test_faster_backend_still_validates_other_settings(self):
        with patch("converter.converter.FasterTTSClient"):
            with self.assertRaises(ValueError):
                AudiobookConverter(backend=BACKEND_FASTER, speed=0)
            with self.assertRaises(ValueError):
                AudiobookConverter(backend=BACKEND_FASTER, language="klingon")

    def test_audiocpp_backend_still_validates_other_settings(self):
        with patch("converter.converter.AudioCppTTSClient"):
            with self.assertRaises(ValueError):
                AudiobookConverter(backend=BACKEND_AUDIOCPP, speed=0)
            with self.assertRaises(ValueError):
                AudiobookConverter(backend=BACKEND_AUDIOCPP, language="klingon")

    def _faster_converter(self, voice=None):
        with patch("converter.converter.FasterTTSClient"):
            return AudiobookConverter(voice_mode=VOICE_MODE_CLONE,
                                      backend=BACKEND_FASTER, voice=voice)

    def _audiocpp_converter(self, voice=None, instructions=None):
        with patch("converter.converter.AudioCppTTSClient"):
            return AudiobookConverter(
                voice_mode=VOICE_MODE_CLONE if voice else VOICE_MODE_CUSTOM,
                backend=BACKEND_AUDIOCPP, voice=voice,
                instructions=instructions)

    def test_narrator_tag_uses_faster_voice_name(self):
        converter = self._faster_converter(voice="male_richard_poe")
        self.assertEqual(converter._narrator_tag(), "male_richard_poe")

    def test_narrator_tag_faster_without_voice_uses_default_key(self):
        # Unreachable in a valid run (--voice is required); the tag stays
        # stable for pre-flights of runs that will fail client-side.
        converter = self._faster_converter()
        self.assertEqual(converter._narrator_tag(), "default")

    def test_narrator_tag_audiocpp_uses_voice_name(self):
        converter = self._audiocpp_converter(voice="female_narrator")
        self.assertEqual(converter._narrator_tag(), "female_narrator")

    def test_narrator_tag_audiocpp_without_voice_uses_fallback(self):
        # Unreachable in a valid run (the client refuses a speaker-capable
        # entry without --voice); the tag stays stable for pre-flights.
        converter = self._audiocpp_converter()
        self.assertEqual(converter._narrator_tag(), "narrator")

    def test_banner_and_narrator_work_without_reference_audio(self):
        converter = self._faster_converter(voice="male_richard_poe")
        converter._print_banner()  # must not raise (regression: Path(None))
        self.assertIsNone(converter.voice_clone_ref_audio)

    def test_audiocpp_banner_prints_without_reference_audio(self):
        converter = self._audiocpp_converter(voice="narrator")
        converter._print_banner()  # must not raise
        converter = self._audiocpp_converter(instructions="Calm and warm.")
        converter._print_banner()

    def test_audiocpp_banner_prints_model_family(self):
        from contextlib import redirect_stdout
        converter = self._audiocpp_converter(voice="narrator")
        converter.tts.family = "higgs_audio_tts"
        buffer = io.StringIO()
        with redirect_stdout(buffer):
            converter._print_banner()
        self.assertIn("higgs_audio_tts", buffer.getvalue())

    def test_non_faster_narrator_tag_unchanged(self):
        with tempfile.TemporaryDirectory() as tmp:
            ref = Path(tmp) / "ref.wav"
            ref.write_bytes(b"x")
            with patch("converter.converter.QwenTTSClient"):
                converter = AudiobookConverter(voice_mode=VOICE_MODE_CLONE,
                                               voice_clone_ref_audio=str(ref),
                                               backend=BACKEND_QWEN)
        self.assertEqual(converter._narrator_tag(), "ref")


if __name__ == "__main__":
    unittest.main()


class AllocationLogNoteTests(unittest.TestCase):
    """allocation_log_note: the server log's exact allocation numbers."""

    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()
        self.log = Path(self._tmp.name) / "audiocpp-server.log"

    def tearDown(self):
        self._tmp.cleanup()

    def _note(self, message):
        with patch("backends.servers.server_log_path",
                   return_value=self.log):
            return allocation_log_note(message)

    def test_cuda_malloc_failure_line_is_surfaced(self):
        self.log.write_text(
            "I ... engine loaded\n"
            "ggml_backend_cuda_buffer_type_alloc_buffer: allocating "
            "1240.5 MiB on device 0: cudaMalloc failed: out of memory\n"
            "server: request failed\n", encoding="utf-8")
        note = self._note("DramaBox audio VAE backend buffer allocation "
                          "failed")
        self.assertIn("1240.5 MiB", note)
        self.assertIn("device 0", note)
        self.assertIn(str(self.log), note)

    def test_non_allocation_message_gets_no_note(self):
        self.log.write_text("allocating 1.0 MiB on device 0: cudaMalloc "
                            "failed: out of memory\n", encoding="utf-8")
        self.assertEqual(self._note("model busy"), "")

    def test_missing_log_yields_no_note(self):
        self.assertEqual(
            self._note("failed to allocate MOSS codec encoder forward "
                       "graph"), "")

    def test_log_without_allocation_lines_yields_no_note(self):
        self.log.write_text("unrelated\n", encoding="utf-8")
        self.assertEqual(self._note("MOSS codec encoder forward graph "
                                    "allocation failed"), "")


class DeviceMemoryWarningTests(unittest.TestCase):
    """The one-time low-free-VRAM warning before the first request."""

    def _warn(self, report, url="http://127.0.0.1:8080"):
        client = AudioCppTTSClient.__new__(AudioCppTTSClient)
        client.api_url = url
        client.quiet = False
        buf = io.StringIO()
        with redirect_stdout(buf), \
                patch.object(audiocpp_client, "nvidia_device_memory_report",
                             return_value=report):
            client._warn_low_device_memory()
        return buf.getvalue()

    def test_low_free_memory_warns(self):
        out = self._warn("0, 24576, 1024\n1, 24576, 23000")
        self.assertIn("GPU 0", out)
        self.assertIn("1024 MiB free of 24576 MiB", out)
        self.assertNotIn("GPU 1", out)

    def test_healthy_memory_warns_nothing(self):
        self.assertEqual(self._warn("0, 24576, 23000"), "")

    def test_missing_nvidia_smi_warns_nothing(self):
        self.assertEqual(self._warn(None), "")

    def test_remote_host_skips_the_check(self):
        with patch.object(audiocpp_client, "nvidia_device_memory_report",
                          side_effect=AssertionError("should not run")):
            self.assertEqual(self._warn("0, 24576, 1024",
                                        url="http://10.0.0.5:8080"), "")


class ErrorBodyClassificationTests(unittest.TestCase):
    """Deterministic-error detection runs on the FULL HTTP error body."""

    def test_fragment_beyond_200_chars_is_still_classified(self):
        # The deterministic fragment sits deep in a long server message;
        # truncating before matching would misclassify it as retryable.
        filler = "x" * 300
        body = json.dumps({"error": {"message":
            f"{filler} model contract spec not found for family"}})
        error = audiocpp_request_error(500, body)
        self.assertIsInstance(error, NonRetryableTTSError)
        self.assertIn("not retryable", str(error))

    def test_quoted_message_is_truncated_for_display(self):
        body = json.dumps({"error": {"message": "y" * 1000}})
        error = audiocpp_request_error(500, body)
        self.assertNotIn("y" * 300, str(error))
        self.assertLess(len(str(error)), 1000)

    def test_http_error_body_helper_reads_whole_body(self):
        body = b"z" * 500
        exc = urllib.error.HTTPError("http://x", 500, "ISE", {},
                                     io.BytesIO(body))
        self.assertEqual(audiocpp_client._http_error_body(exc), body.decode())


class VoiceForRunTests(unittest.TestCase):
    """audiocpp_voice_for_run: the "All"-run per-model voice resolution."""

    def test_design_takes_no_voice(self):
        self.assertIsNone(audiocpp_client.audiocpp_voice_for_run(
            "voxcpm2", "vdes", "Vox", "narrator", ["narrator"]))

    def test_speaker_entry_takes_the_speaker_pick_or_first_speaker(self):
        model = "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF"
        resolve = audiocpp_client.audiocpp_voice_for_run
        self.assertEqual(resolve("qwen3_tts", "tts", model, "Vivian", []),
                         "Vivian")
        self.assertEqual(resolve("qwen3_tts", "tts", model, "narrator", []),
                         QWEN3_TTS_SPEAKERS[0])

    def test_clone_entry_takes_the_preset_pick_or_first_server_voice(self):
        self.assertEqual(audiocpp_client.audiocpp_voice_for_run(
            "higgs_audio_tts", "tts", "higgs", "narrator",
            ["narrator", "other"]), "narrator")
        self.assertEqual(audiocpp_client.audiocpp_voice_for_run(
            "higgs_audio_tts", "tts", "higgs", "unknown", ["first", "x"]),
            "first")
        self.assertIsNone(audiocpp_client.audiocpp_voice_for_run(
            "higgs_audio_tts", "tts", "higgs", "unknown", []))

    def test_pure_tts_family_takes_no_voice(self):
        self.assertIsNone(audiocpp_client.audiocpp_voice_for_run(
            "supertonic", "tts", "supertonic", "narrator", ["narrator"]))