fix: model-specific instructions support and set cfg-scale 4 for breeze automaticallyHEADmain

1 files changed, 419 insertions, 0 deletions

diff --git a/app/tests/test_instruction_capabilities.py b/app/tests/test_instruction_capabilities.py
new file mode 100644
index 0000000..440ee08
--- /dev/null
+++ b/app/tests/test_instruction_capabilities.py
@@ -0,0 +1,419 @@
+"""Instruction-support and guidance regressions across the TTS backends.
+
+Covers the capabilities the models actually implement (verified against
+each backend's serving code) and what the clients send for them:
+Breeze-TTS 2's recommended guidance strength with instructions, the
+audio.cpp Qwen3-TTS variant split (CustomVoice reads instructions, the
+Base cloner does not), the SGLang models that consume a separate style
+instruction alongside their voice conditioning, and the Qwen demo's
+CustomVoice instruction parameter.
+"""
+
+import io
+import json
+import tempfile
+import unittest
+import wave
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from converter.clients import (
+    BACKEND_QWEN, AudioCppTTSClient, QwenTTSClient,
+    VOICE_MODE_CLONE, VOICE_MODE_CUSTOM, VOICE_MODE_DESIGN,
+)
+from converter.clients.audiocpp import (
+    AUDIOCPP_FAMILY_BREEZE_TTS,
+    AUDIOCPP_FAMILY_PROFILES,
+    AUDIOCPP_VOICE_OPTIONAL,
+    audiocpp_entry_supports_instructions,
+    audiocpp_family_voice_policy,
+)
+
+
+_WAV_BYTES = b"RIFF\x18\x00\x00\x00WAVEfmt \x10\x00\x00\x00"
+
+
+# ---------------------------------------------------------------------------
+# Pure helpers
+# ---------------------------------------------------------------------------
+
+class VoicePolicyKnownFamiliesTests(unittest.TestCase):
+    """Families whose verified policy must survive a stale local spec."""
+
+    def test_breeze_is_tts_plus_clone_even_without_a_local_spec(self):
+        # Remote Breeze entries against older local checkouts carry no
+        # breeze_tts spec at all: the fallback keeps instructions-only
+        # voice direction connectable instead of demanding a reference.
+        self.assertEqual(
+            audiocpp_family_voice_policy(AUDIOCPP_FAMILY_BREEZE_TTS),
+            AUDIOCPP_VOICE_OPTIONAL)
+
+    def test_vibevoice_accepts_reference_audio_despite_its_spec(self):
+        # vibevoice.json declares only "tts", but the implementation
+        # accepts reference audio: a mixed tts+clone family, so a picked
+        # voice must not be silently dropped.
+        self.assertEqual(audiocpp_family_voice_policy("vibevoice"),
+                         AUDIOCPP_VOICE_OPTIONAL)
+
+
+class EntryInstructionSupportTests(unittest.TestCase):
+    """audiocpp_entry_supports_instructions: True/False/None per entry."""
+
+    def test_breeze_supports_instructions(self):
+        self.assertIs(
+            audiocpp_entry_supports_instructions(
+                AUDIOCPP_FAMILY_BREEZE_TTS, "tts", "Breeze-TTS-2-GGUF"),
+            True)
+
+    def test_qwen_customvoice_and_design_support_instructions(self):
+        self.assertIs(
+            audiocpp_entry_supports_instructions(
+                "qwen3_tts", "tts", "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF"),
+            True)
+        self.assertIs(
+            audiocpp_entry_supports_instructions(
+                "qwen3_tts", "vdes", "Qwen3-TTS-12Hz-1.7B-VoiceDesign"),
+            True)
+
+    def test_qwen_base_cloner_provably_does_not(self):
+        self.assertIs(
+            audiocpp_entry_supports_instructions(
+                "qwen3_tts", "tts", "Qwen3-TTS-12Hz-1.7B-Base-GGUF"),
+            False)
+
+    def test_unknown_families_are_unknown_not_unsupported(self):
+        self.assertIsNone(
+            audiocpp_entry_supports_instructions("chatterbox",
+                                                 "tts", "x"))
+
+
+# ---------------------------------------------------------------------------
+# audio.cpp request payloads (instructed Breeze runs)
+# ---------------------------------------------------------------------------
+
+def _breeze_client(instructions=None, request_options=None,
+                   voice="narrator", seed=-1):
+    """A fully-initialized Breeze client (no HTTP machinery touched)."""
+    with patch.object(AudioCppTTSClient, "_connect"):
+        client = AudioCppTTSClient(
+            Path("."), voice=voice, instructions=instructions,
+            request_options=request_options)
+    client.api_url = "http://127.0.0.1:8080"
+    client.model_id = "Breeze-TTS-2-GGUF"
+    client.family = AUDIOCPP_FAMILY_BREEZE_TTS
+    client.task = "tts"
+    client.profile = AUDIOCPP_FAMILY_PROFILES[AUDIOCPP_FAMILY_BREEZE_TTS]
+    client.design_mode = False
+    client.instruction_voice = False
+    client.plain_mode = False
+    client.preset_mode = True
+    client.speaker_mode = False
+    client._seed = seed
+    client._resolve_auto_guidance()
+    return client
+
+
+def _captured_payload(client):
+    """The JSON body _request_wav sends, via a stubbed urlopen."""
+    response = MagicMock()
+    response.read.return_value = _WAV_BYTES
+    response.__enter__ = lambda self: response
+    response.__exit__ = lambda self, *exc: None
+    with patch("converter.clients.audiocpp.urllib.request.urlopen") \
+            as urlopen:
+        urlopen.return_value = response
+        client._request_wav("Hello there.")
+    request = urlopen.call_args[0][0]
+    return json.loads(request.data.decode("utf-8"))
+
+
+class BreezeGuidanceDefaultTests(unittest.TestCase):
+    """Breeze guidance: recommended 4 with instructions, otherwise none."""
+
+    def test_instructed_clone_carries_guidance_4_and_the_instruction(self):
+        client = _breeze_client(instructions="Screaming, crazed, yelling")
+        self.assertEqual(client._auto_guidance_scale, 4.0)
+        payload = _captured_payload(client)
+        self.assertEqual(payload["guidance_scale"], 4.0)
+        self.assertEqual(payload["voice"], "narrator")
+        self.assertEqual(
+            payload["options"],
+            {"instruction": "Screaming, crazed, yelling"})
+        self.assertNotIn("instructions", payload)
+
+    def test_option_instruction_also_gets_the_guidance_default(self):
+        client = _breeze_client(request_options={
+            "instruction": "Read slowly and warmly."})
+        self.assertEqual(client._auto_guidance_scale, 4.0)
+        payload = _captured_payload(client)
+        self.assertEqual(payload["guidance_scale"], 4.0)
+        self.assertEqual(payload["options"]["instruction"],
+                         "Read slowly and warmly.")
+
+    def test_explicit_guidance_option_is_preserved(self):
+        client = _breeze_client(
+            instructions="Screaming, crazed, yelling",
+            request_options={"guidance_scale": "2.5"})
+        self.assertIsNone(client._auto_guidance_scale)
+        payload = _captured_payload(client)
+        self.assertNotIn("guidance_scale", payload)
+        self.assertEqual(payload["options"]["guidance_scale"], "2.5")
+
+    def test_guidance_0_override_still_counts_as_explicit(self):
+        # 0 selects the instruction-free branch: a deliberate setting.
+        client = _breeze_client(
+            instructions="Screaming",
+            request_options={"guidance_scale": "0"})
+        payload = _captured_payload(client)
+        self.assertNotIn("guidance_scale", payload)
+
+    def test_plain_clone_without_instructions_uses_the_backend_default(self):
+        client = _breeze_client()
+        self.assertIsNone(client._auto_guidance_scale)
+        payload = _captured_payload(client)
+        self.assertNotIn("guidance_scale", payload)
+        self.assertNotIn("options", payload)
+        self.assertEqual(payload["voice"], "narrator")
+
+
+class InstructionConflictTests(unittest.TestCase):
+    """Two different instruction sources are refused before connecting."""
+
+    def test_conflicting_instructions_and_option_raise_without_a_server(self):
+        with self.assertRaises(RuntimeError) as ctx:
+            _breeze_client(instructions="calm narration",
+                           request_options={"instruction": "screaming"})
+        self.assertIn("Two conflicting instructions",
+                      str(ctx.exception))
+
+    def test_identical_instructions_from_both_sources_are_accepted(self):
+        client = _breeze_client(
+            instructions="calm narration",
+            request_options={"instruction": "calm narration"})
+        self.assertEqual(client.instructions, "calm narration")
+
+    def test_option_only_instruction_is_folded_into_the_reports(self):
+        client = _breeze_client(
+            request_options={"instruction": "calm narration"})
+        self.assertEqual(client.instructions, "calm narration")
+
+
+class SeedPrecisionTests(unittest.TestCase):
+    """Full-range uint64 seeds travel as decimal strings (audio.cpp docs)."""
+
+    def test_seed_above_2_pow_53_is_sent_as_a_string(self):
+        seed = 2 ** 53 + 3  # beyond the exact JSON-number integer range
+        client = _breeze_client(seed=seed)
+        payload = _captured_payload(client)
+        self.assertEqual(payload["seed"], str(seed))
+
+    def test_ordinary_seeds_stay_numbers(self):
+        client = _breeze_client(seed=42)
+        payload = _captured_payload(client)
+        self.assertEqual(payload["seed"], 42)
+
+
+# ---------------------------------------------------------------------------
+# SGLang-Omni: instructions on supported pipelines
+# ---------------------------------------------------------------------------
+
+class SgOmniInstructionTests(unittest.TestCase):
+    """instructions reach the payload only where the serving code reads it."""
+
+    _tmp_dir = None
+    _REF = None
+
+    @classmethod
+    def setUpClass(cls):
+        buffer = io.BytesIO()
+        with wave.open(buffer, "wb") as wav_file:
+            wav_file.setnchannels(1)
+            wav_file.setsampwidth(2)
+            wav_file.setframerate(24000)
+            wav_file.writeframes(b"\x01\x00" * 16)
+        cls._tmp_dir = tempfile.TemporaryDirectory()
+        cls._REF = Path(cls._tmp_dir.name) / "narrator.wav"
+        cls._REF.write_bytes(buffer.getvalue())
+        cls.addClassCleanup(cls._tmp_dir.cleanup)
+
+    def _client(self, model, **kwargs):
+        from converter.clients import SgOmniTTSClient
+        with patch.object(SgOmniTTSClient, "_connect"):
+            client = SgOmniTTSClient(
+                Path("."), model=model, ref_audio=str(self._REF),
+                ref_text="Hello transcript.", instructions="screaming",
+                **kwargs)
+        entry = client.entry
+        payload = client._request_payload("Hello there.")
+        return entry, payload
+
+    def test_qwen_base_clone_carries_ref_and_instruction(self):
+        entry, payload = self._client("qwen3_tts_1_7b_base")
+        self.assertIn("ref_audio", payload)
+        self.assertNotEqual(payload.get("task_type"), "VoiceDesign")
+        self.assertEqual(payload["instructions"], "screaming")
+
+    def test_moss_clone_carries_ref_and_instruction(self):
+        entry, payload = self._client("moss_tts")
+        self.assertIn("ref_audio", payload)
+        self.assertEqual(payload["instructions"], "screaming")
+
+    def test_customvoice_speaker_with_instruction(self):
+        entry, payload = self._client("qwen3_tts_0_6b_customvoice",
+                                      voice="Vivian")
+        self.assertEqual(payload["voice"], "Vivian")
+        self.assertEqual(payload["instructions"], "screaming")
+        self.assertNotIn("task_type", payload)
+
+    def test_design_remains_voice_design_with_instruction(self):
+        from converter.clients import SgOmniTTSClient
+        with patch.object(SgOmniTTSClient, "_connect"):
+            client = SgOmniTTSClient(
+                Path("."), model="qwen3_tts_1_7b_voicedesign",
+                instructions="a warm narrator")
+        payload = client._request_payload("Hello there.")
+        self.assertEqual(payload["task_type"], "VoiceDesign")
+        self.assertEqual(payload["instructions"], "a warm narrator")
+        self.assertNotIn("ref_audio", payload)
+
+    def test_unsupported_model_refuses_instructions_at_connect(self):
+        with self.assertRaises(RuntimeError) as ctx:
+            self._client("higgs_audio_v3_tts")
+        self.assertIn("does not consume style instructions",
+                      str(ctx.exception))
+
+
+# ---------------------------------------------------------------------------
+# Qwen demo: CustomVoice instruction parameter
+# ---------------------------------------------------------------------------
+
+class QwenCustomVoiceInstructionTests(unittest.TestCase):
+    """The run_instruct endpoint takes an ``instruct`` delivery control."""
+
+    def test_run_instruct_sends_instruct_alongside_the_speaker(self):
+        client = QwenTTSClient.__new__(QwenTTSClient)
+        client.voice_mode = VOICE_MODE_CUSTOM
+        client.speaker = "Vivian"
+        client.language = "Auto"
+        client.instructions = "screaming, crazed"
+        client._seed = -1
+        client.client = MagicMock()
+        client._resolve_api_name = lambda *names: names[0]
+        client._endpoint_accepts_param = MagicMock(return_value=True)
+        client._generate_custom_voice("Hello there.")
+        predict = client.client.predict
+        predict.assert_called_once_with(
+            text="Hello there.", lang_disp="Auto",
+            spk_disp="Vivian", instruct="screaming, crazed",
+            api_name="/run_instruct")
+
+    def test_custom_voice_without_instructions_is_unchanged(self):
+        client = QwenTTSClient.__new__(QwenTTSClient)
+        client.voice_mode = VOICE_MODE_CUSTOM
+        client.speaker = "Vivian"
+        client.language = "Auto"
+        client.instructions = ""
+        client._seed = -1
+        client.client = MagicMock()
+        client._resolve_api_name = lambda *names: names[0]
+        client._endpoint_accepts_param = MagicMock(return_value=True)
+        client._generate_custom_voice("Hello there.")
+        _, kwargs = client.client.predict.call_args
+        self.assertNotIn("instruct", kwargs)
+
+
+# ---------------------------------------------------------------------------
+# audiobook voice-mode routing (qwen)
+# ---------------------------------------------------------------------------
+
+class CatalogInstructionFlagsTests(unittest.TestCase):
+    """Only the verified pipelines carry supports_instructions."""
+
+    def test_catalog_marks_only_the_verified_pipelines(self):
+        from backends.sglomni.catalog import ENTRIES
+        supported = {"qwen3_tts_0_6b_customvoice", "qwen3_tts_0_6b_base",
+                     "qwen3_tts_1_7b_base", "qwen3_tts_1_7b_voicedesign",
+                     "moss_tts", "moss_tts_local"}
+        for entry in ENTRIES:
+            with self.subTest(entry=entry.key):
+                self.assertEqual(entry.supports_instructions,
+                                 entry.key in supported)
+
+
+class GradioPrefixProbeTests(unittest.TestCase):
+    """Qwen demos under modern Gradio sit behind /gradio_api."""
+
+    def _identify(self, modern_payload, legacy_payload=None):
+        import backends.probe as probe
+        seen = []
+
+        def fake_get_json(url, timeout):
+            seen.append(url)
+            if url == "http://x/gradio_api/info":
+                return modern_payload
+            if url == "http://x/info":
+                return legacy_payload
+            return None
+
+        with patch.object(probe, "_get_json", side_effect=fake_get_json):
+            with patch.object(probe.common, "server_running",
+                              return_value=True):
+                identity = probe._identify_gradio("http://x", 1.0)
+        return identity, seen
+
+    def test_modern_prefix_is_probed_first_and_identifies(self):
+        payload = {"named_endpoints": {"/run_instruct": {}}}
+        identity, seen = self._identify(payload)
+        self.assertEqual(identity, probe_identity("qwen-custom"))
+        self.assertEqual(seen, ["http://x/gradio_api/info"])
+
+    def test_legacy_info_still_identifies_older_gradio(self):
+        payload = {"named_endpoints": {"/run_voice_clone": {}}}
+        identity, seen = self._identify(None, payload)
+        self.assertEqual(identity, probe_identity("qwen-clone"))
+        self.assertEqual(seen, ["http://x/gradio_api/info",
+                                "http://x/info"])
+
+    def test_neither_prefix_answers_none(self):
+        identity, _ = self._identify(None)
+        self.assertIsNone(identity)
+
+
+def probe_identity(name):
+    """The probe's IDENTITY_* constant for a backend NAME (local import)."""
+    import backends.probe as probe
+    return {"qwen-custom": probe.IDENTITY_QWEN_CUSTOM,
+            "qwen-clone": probe.IDENTITY_QWEN_CLONE,
+            }[name]
+
+
+# ---------------------------------------------------------------------------
+# audiobook voice-mode routing (qwen)
+# ---------------------------------------------------------------------------
+
+class QwenVoiceModeRoutingTests(unittest.TestCase):
+    """speaker + instructions is a directed CustomVoice run, not Design."""
+
+    def test_voice_mode_for_qwen_combinations(self):
+        from converter.converter import voice_mode_for
+        cases = [
+            (dict(voice=None, clone=None, instructions=None),
+             VOICE_MODE_CUSTOM),
+            (dict(voice=None, clone=None, instructions="screaming"),
+             VOICE_MODE_DESIGN),
+            (dict(voice="Vivian", clone=None, instructions="screaming"),
+             VOICE_MODE_CUSTOM),
+            (dict(voice=None, clone="ref.wav", instructions=None),
+             VOICE_MODE_CLONE),
+        ]
+        for kwargs, expected in cases:
+            with self.subTest(**kwargs):
+                self.assertEqual(
+                    voice_mode_for(BACKEND_QWEN, voice=kwargs["voice"],
+                                   clone=kwargs["clone"],
+                                   instructions=kwargs["instructions"]),
+                    expected)
+
+
+if __name__ == "__main__":
+    unittest.main()