fix: spec santizer for glm, outetts, miotts, minimax.

1 files changed, 454 insertions, 1 deletions

diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py
index 7edb3c9..18c38c4 100644
--- a/app/tests/test_backends_audiocpp.py
+++ b/app/tests/test_backends_audiocpp.py
@@ -710,7 +710,7 @@ class BuildEntriesHostingTests(unittest.TestCase):
                 "design": False, "recommended": True}
 
     def _entries(self, catalog_entry):
-        entries, _, _, _, _ = make_server.wizard._build_entries(
+        entries, _, _, _, _, _ = make_server.wizard._build_entries(
             [catalog_entry["family"]],
             {catalog_entry["family"]: [self._option(catalog_entry["family"])]},
             {catalog_entry["family"]: catalog_entry},
@@ -4242,3 +4242,456 @@ class PrebuiltFallbackTests(unittest.TestCase):
         self.assertEqual(rc, 0)
         mk_i.assert_not_called()
         mk_build.assert_called_once()
+
+
+class MissingStripPrefixSanitizeTests(unittest.TestCase):
+    """The missing-strip_prefix repair for nested single-GGUF packages."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.checkout_holder = Path(self._tmp.name) / "audio.cpp"
+        self.checkout_holder.mkdir()
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    _GGUF_SPEC = {
+        "family": "glm_tts",
+        "sources": [{"format": "gguf",
+                     "roots": {"model": ".", "weights": "$gguf"}}],
+    }
+
+    def _spec(self, files, roots=(("model", "."), ("weights", "$gguf")),
+              fmt="gguf"):
+        spec = json.loads(json.dumps(self._GGUF_SPEC))
+        spec["sources"][0]["roots"] = dict(roots)
+        spec["sources"][0]["format"] = fmt
+        spec["packages"] = [{"id": "pkg", "format": fmt, "files": files}]
+        return spec
+
+    def test_nested_single_gguf_gets_its_prefix(self):
+        spec = self._spec(["Text to audio (TTS)/GLM-TTS_Q8.gguf"])
+        self.assertTrue(make_server.catalog.sanitize_model_spec(spec))
+        self.assertEqual(spec["packages"][0]["strip_prefix"],
+                         "Text to audio (TTS)")
+
+    def test_explicit_gguf_paths_without_gguf_root_are_untouched(self):
+        # minimax_music3-style: the gguf source names tensors by explicit
+        # file paths (no $gguf root), so the nested layout is intentional.
+        spec = self._spec(["config/a.json", "tokenizer/t.json",
+                           "language_model_q4_0.gguf"],
+                          roots=(("model", "."),))
+        self.assertFalse(make_server.catalog.sanitize_model_spec(spec))
+
+    def test_mixed_prefixes_are_untouched(self):
+        spec = self._spec(["config/a.json", "model.safetensors"])
+        self.assertFalse(make_server.catalog.sanitize_model_spec(spec))
+
+    def test_flat_package_is_untouched(self):
+        spec = self._spec(["model.gguf"])
+        self.assertFalse(make_server.catalog.sanitize_model_spec(spec))
+
+    def test_safetensors_packages_are_untouched(self):
+        spec = self._spec(["Some-Dir/model.safetensors"], fmt="safetensors")
+        self.assertFalse(make_server.catalog.sanitize_model_spec(spec))
+
+    def test_catalog_carries_the_sanitized_prefix(self):
+        _write_spec(self.checkout_holder, "glm_like",
+                    packages=[{
+                        "id": "glm_like_q8_0", "default": True,
+                        "format": "gguf",
+                        "target_directory": "GLM-Like-Q8",
+                        "files": ["Text to audio (TTS)/GLM-Like_Q8.gguf"],
+                    }])
+        specs_dir = self.checkout_holder / "model_specs"
+        spec = json.loads(
+            (specs_dir / "glm_like.json").read_text(encoding="utf-8"))
+        spec["sources"] = [{"format": "gguf",
+                            "roots": {"model": ".", "weights": "$gguf"}}]
+        (specs_dir / "glm_like.json").write_text(
+            json.dumps(spec), encoding="utf-8")
+        catalog = make_server.catalog.load_model_catalog(self.checkout_holder)
+        glm_like = next(e for e in catalog if e["family"] == "glm_like")
+        self.assertEqual(glm_like["packages"][0]["strip_prefix"],
+                         "Text to audio (TTS)")
+        self.assertEqual(
+            make_server.catalog.entry_model_path(glm_like),
+            "models/GLM-Like-Q8")
+
+
+class EntryModelPathTests(unittest.TestCase):
+    """entry_model_path: directory hosting vs. the multi-GGUF file rule."""
+
+    def _entry(self, packages, default_directory=None):
+        if default_directory is None and packages:
+            default_directory = str(
+                packages[0].get("target_directory") or "Bundle-GGUF")
+        return {"family": "minimax_h3", "packages": packages,
+                "default_path": f"models/{default_directory or ''}"}
+
+    def test_multi_gguf_package_is_hosted_from_its_first_gguf(self):
+        entry = self._entry([{
+            "id": "minimax_h3_q4_k", "default": True, "format": "gguf",
+            "target_directory": "MiniMax-H3-Q4-GGUF",
+            "strip_prefix": "MiniMax-H3-Q4-GGUF",
+            "files": [
+                "MiniMax-H3-Q4-GGUF/configuration.json",
+                "MiniMax-H3-Q4-GGUF/text_encoder_q4_k.gguf",
+                "MiniMax-H3-Q4-GGUF/dit.gguf",
+                "MiniMax-H3-Q4-GGUF/audio_vae_folded_f16.gguf",
+                "MiniMax-H3-Q4-GGUF/video_vae.gguf",
+            ],
+        }])
+        self.assertEqual(
+            make_server.catalog.entry_model_path(entry),
+            "models/MiniMax-H3-Q4-GGUF/text_encoder_q4_k.gguf")
+
+    def test_single_gguf_package_hosts_the_directory(self):
+        entry = self._entry([{
+            "id": "voxcpm2_q8_0", "default": True, "format": "gguf",
+            "target_directory": "VoxCPM2-GGUF",
+            "strip_prefix": "VoxCPM2-GGUF",
+            "files": ["VoxCPM2-GGUF/voxcpm2-q8_0.gguf"],
+        }], default_directory="VoxCPM2-GGUF")
+        self.assertEqual(make_server.catalog.entry_model_path(entry),
+                         "models/VoxCPM2-GGUF")
+
+    def test_alternate_directory_uses_that_packages_files(self):
+        entry = self._entry([
+            {"id": "a_q8", "default": True, "format": "gguf",
+             "target_directory": "A-GGUF",
+             "files": ["A-GGUF/a.gguf", "A-GGUF/b.gguf"]},
+            {"id": "b_q8", "format": "gguf", "target_directory": "B-GGUF",
+             "files": ["B-GGUF/b.gguf"]},
+        ])
+        self.assertEqual(make_server.catalog.entry_model_path(entry, "B-GGUF"),
+                         "models/B-GGUF")
+
+    def test_safetensors_directory_hosts_the_directory(self):
+        entry = self._entry([{
+            "id": "voxcpm2_safetensors", "format": "safetensors",
+            "target_directory": "VoxCPM2",
+            "files": ["config.json", "model.safetensors"],
+        }], default_directory="VoxCPM2")
+        self.assertEqual(make_server.catalog.entry_model_path(entry),
+                         "models/VoxCPM2")
+
+    def test_entry_without_packages_falls_back_to_the_directory(self):
+        entry = self._entry([], default_directory="Fallback-GGUF")
+        self.assertEqual(make_server.catalog.entry_model_path(entry),
+                         "models/Fallback-GGUF")
+
+
+class BuildModelEntrySessionOptionsTests(unittest.TestCase):
+    """build_model_entry carries per-entry session options when given."""
+
+    def test_session_options_added_when_given(self):
+        entry = make_server.catalog.build_model_entry(
+            "miotts", "MioTTS-1.7B-GGUF", "models/MioTTS-1.7B-GGUF",
+            session_options={"miotts.codec_model_path": "models/MioCodec"})
+        self.assertEqual(entry["session_options"],
+                         {"miotts.codec_model_path": "models/MioCodec"})
+
+    def test_session_options_omitted_when_empty(self):
+        entry = make_server.catalog.build_model_entry(
+            "miotts", "MioTTS-1.7B-GGUF", "models/MioTTS-1.7B-GGUF")
+        self.assertNotIn("session_options", entry)
+
+
+class FilePrecisePresenceTests(unittest.TestCase):
+    """Installed checks are file-precise against the catalog packages."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.checkout = Path(self._tmp.name) / "audio.cpp"
+        self.checkout.mkdir()
+        specs = self.checkout / "model_specs"
+        specs.mkdir()
+        spec = {
+            "family": "glm_like", "category": "tts", "tasks": ["tts", "clone"],
+            "packages": [{
+                "id": "glm_like_q8_0", "default": True, "format": "gguf",
+                "target_directory": "GLM-Like-Q8",
+                "files": ["Text to audio (TTS)/GLM-Like_Q8.gguf"],
+            }],
+            "sources": [{"format": "gguf",
+                         "roots": {"model": ".", "weights": "$gguf"}}],
+        }
+        (specs / "glm_like.json").write_text(json.dumps(spec),
+                                             encoding="utf-8")
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def _entry(self, rel="models/GLM-Like-Q8"):
+        return {"id": "GLM-Like-Q8", "family": "glm_like", "path": rel}
+
+    def test_stale_nested_layout_counts_as_missing(self):
+        stale = self.checkout / "models" / "GLM-Like-Q8" \
+            / "Text to audio (TTS)"
+        stale.mkdir(parents=True)
+        (stale / "GLM-Like_Q8.gguf").write_bytes(b"x")
+        self.assertFalse(
+            make_server.models._all_models_present(self.checkout,
+                                                   [self._entry()]))
+        server_json = self.checkout / "server.json"
+        server_json.write_text(json.dumps({"models": [self._entry()]}),
+                               encoding="utf-8")
+        missing = make_server.models.missing_model_entries(server_json)
+        self.assertEqual([m["id"] for m in missing], ["GLM-Like-Q8"])
+
+    def test_flat_layout_after_the_repair_counts_as_installed(self):
+        target = self.checkout / "models" / "GLM-Like-Q8"
+        target.mkdir(parents=True)
+        (target / "GLM-Like_Q8.gguf").write_bytes(b"x")
+        self.assertTrue(
+            make_server.models._all_models_present(self.checkout,
+                                                   [self._entry()]))
+        server_json = self.checkout / "server.json"
+        server_json.write_text(json.dumps({"models": [self._entry()]}),
+                               encoding="utf-8")
+        self.assertEqual(
+            make_server.models.missing_model_entries(server_json), [])
+
+    def test_file_style_entry_of_a_multi_gguf_package(self):
+        specs = self.checkout / "model_specs"
+        spec = {
+            "family": "multi", "category": "tts", "tasks": ["tts"],
+            "packages": [{
+                "id": "multi_q4", "default": True, "format": "gguf",
+                "target_directory": "Multi-Q4-GGUF",
+                "strip_prefix": "Multi-Q4-GGUF",
+                "files": ["Multi-Q4-GGUF/dit.gguf",
+                          "Multi-Q4-GGUF/vae.gguf"],
+            }],
+        }
+        (specs / "multi.json").write_text(json.dumps(spec), encoding="utf-8")
+        target = self.checkout / "models" / "Multi-Q4-GGUF"
+        target.mkdir(parents=True)
+        (target / "dit.gguf").write_bytes(b"x")
+        (target / "vae.gguf").write_bytes(b"x")
+        self.assertTrue(make_server.models._all_models_present(
+            self.checkout,
+            [{"id": "Multi", "path": "models/Multi-Q4-GGUF/dit.gguf"}]))
+
+    def test_unmatched_entry_keeps_the_plain_path_check(self):
+        target = self.checkout / "elsewhere"
+        target.mkdir()
+        (target / "m.gguf").write_bytes(b"x")
+        self.assertTrue(make_server.models._all_models_present(
+            self.checkout, [{"id": "x", "path": str(target)}]))
+
+
+class CompanionInstallTests(unittest.TestCase):
+    """Companion packages (MioCodec for MioTTS) join the install list."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.checkout = Path(self._tmp.name) / "audio.cpp"
+        self.checkout.mkdir()
+        specs = self.checkout / "model_specs"
+        specs.mkdir()
+        manager = self.checkout / "tools" / "model_manager_v2.py"
+        manager.parent.mkdir()
+        manager.write_text("#!/usr/bin/env python3\n", encoding="utf-8")
+        (specs / "miocodec.json").write_text(json.dumps({
+            "family": "miocodec", "category": "audio_tools",
+            "tasks": ["codec"],
+            "packages": [{
+                "id": "miocodec_q8_0", "default": True, "format": "gguf",
+                "target_directory": "MioCodec-25Hz-44.1kHz-v2-GGUF",
+                "strip_prefix": "MioCodec-25Hz-44.1kHz-v2-GGUF",
+                "files": ["MioCodec-25Hz-44.1kHz-v2-GGUF/codec.gguf"],
+            }],
+        }), encoding="utf-8")
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_missing_companion_is_merged_into_pending(self):
+        merged = make_server.models._merge_companions(
+            self.checkout, [], [("MioCodec", "miocodec_q8_0")])
+        self.assertEqual(merged, [("MioCodec", "miocodec_q8_0")])
+
+    def test_installed_companion_is_reported_and_skipped(self):
+        target = self.checkout / "models" / "MioCodec-25Hz-44.1kHz-v2-GGUF"
+        target.mkdir(parents=True)
+        (target / "codec.gguf").write_bytes(b"x")
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            merged = make_server.models._merge_companions(
+                self.checkout, [], [("MioCodec", "miocodec_q8_0")])
+        self.assertEqual(merged, [])
+        self.assertIn("MioCodec is already installed", buf.getvalue())
+
+    def test_install_models_prints_the_companion_command(self):
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            rc = make_server.models._install_models(
+                self.checkout, [], download=False,
+                companions=[("MioCodec", "miocodec_q8_0")])
+        self.assertEqual(rc, 0)
+        self.assertIn("install miocodec_q8_0", buf.getvalue())
+
+    def test_install_models_runs_the_companion_download(self):
+        buf = io.StringIO()
+        with redirect_stdout(buf), \
+                patch.object(common, "run_console_subprocess",
+                             return_value=0) as run:
+            rc = make_server.models._install_models(
+                self.checkout, [], download=True,
+                companions=[("MioCodec", "miocodec_q8_0")])
+        self.assertEqual(rc, 0)
+        argv = run.call_args[0][0]
+        self.assertEqual(argv[-2:], ["install", "miocodec_q8_0"])
+
+
+class ApplyEntrySessionOptionsTests(unittest.TestCase):
+    """The wizard bakes companion/session options into server.json entries."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.checkout = Path(self._tmp.name) / "audio.cpp"
+        self.checkout.mkdir()
+        self.wav_dir = Path(self._tmp.name) / "voices"
+        self.wav_dir.mkdir()
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def _write_wav(self, name, seconds, rate=16000):
+        import wave as wave_mod
+        with wave_mod.open(str(self.wav_dir / name), "wb") as handle:
+            handle.setnchannels(1)
+            handle.setsampwidth(2)
+            handle.setframerate(rate)
+            handle.writeframes(b"\x00\x00" * int(rate * seconds))
+
+    def test_miotts_entry_gets_the_codec_model_path(self):
+        entries = [{"id": "MioTTS-1.7B-GGUF", "family": "miotts"}]
+        applied = make_server.catalog.apply_entry_session_options(
+            entries, None, self.checkout)
+        self.assertEqual(applied, ["MioTTS-1.7B-GGUF"])
+        self.assertEqual(
+            entries[0]["session_options"]["miotts.codec_model_path"],
+            "models/MioCodec-25Hz-44.1kHz-v2-GGUF")
+
+    def test_hand_set_codec_path_is_not_overridden(self):
+        entries = [{"id": "MioTTS-1.7B-GGUF", "family": "miotts",
+                    "session_options":
+                        {"miotts.codec_model_path": "/custom/codec"}}]
+        applied = make_server.catalog.apply_entry_session_options(
+            entries, None, self.checkout)
+        self.assertEqual(applied, [])
+        self.assertEqual(
+            entries[0]["session_options"]["miotts.codec_model_path"],
+            "/custom/codec")
+
+    def test_voxcpm_capacity_sized_to_the_longest_voice(self):
+        self._write_wav("short.wav", 5)
+        self._write_wav("long.wav", 40)
+        entries = [{"id": "VoxCPM2-GGUF", "family": "voxcpm2"}]
+        applied = make_server.catalog.apply_entry_session_options(
+            entries, self.wav_dir, self.checkout)
+        self.assertEqual(applied, ["VoxCPM2-GGUF"])
+        self.assertEqual(
+            entries[0]["session_options"]
+            ["voxcpm2.audiovae_encoder_sample_capacity"], "720000")
+
+    def test_short_voices_need_no_capacity(self):
+        self._write_wav("short.wav", 5)
+        entries = [{"id": "VoxCPM2-GGUF", "family": "voxcpm2"}]
+        applied = make_server.catalog.apply_entry_session_options(
+            entries, self.wav_dir, self.checkout)
+        self.assertEqual(applied, [])
+        self.assertNotIn("session_options", entries[0])
+
+    def test_no_voice_directory_means_no_capacity(self):
+        entries = [{"id": "VoxCPM2-GGUF", "family": "voxcpm2"}]
+        make_server.catalog.apply_entry_session_options(entries, None,
+                                                        self.checkout)
+        self.assertNotIn("session_options", entries[0])
+
+    def test_existing_session_options_are_preserved(self):
+        entries = [{"id": "VoxCPM2-GGUF", "family": "voxcpm2",
+                    "session_options": {"voxcpm2.mem_saver": "true"}}]
+        self._write_wav("long.wav", 40)
+        make_server.catalog.apply_entry_session_options(
+            entries, self.wav_dir, self.checkout)
+        options = entries[0]["session_options"]
+        self.assertEqual(options["voxcpm2.mem_saver"], "true")
+        self.assertIn("voxcpm2.audiovae_encoder_sample_capacity", options)
+
+
+class FilePathSelectionTests(unittest.TestCase):
+    """Selections and unused-model matching for file-hosted entries."""
+
+    def test_file_style_entry_maps_to_its_directory(self):
+        config = {"models": [
+            {"id": "MiniMax-H3-Q4-GGUF", "family": "minimax_h3",
+             "path": "models/MiniMax-H3-Q4-GGUF/text_encoder_q4_k.gguf",
+             "task": "tts"},
+        ]}
+        catalog = [{"family": "minimax_h3", "packages": [],
+                    "default_path": "models/MiniMax-H3-Q4-GGUF"}]
+        selected, tasks = make_server.catalog.server_config_selections(
+            config, catalog)
+        self.assertEqual(selected["minimax_h3"], ["MiniMax-H3-Q4-GGUF"])
+        self.assertEqual(tasks[("minimax_h3", "MiniMax-H3-Q4-GGUF")], "tts")
+
+    def test_directory_style_entry_matches_a_file_style_selection(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        server_json = Path(self._tmp.name) / "server.json"
+        server_json.write_text(json.dumps({"models": [{
+            "id": "MiniMax-H3-Q4-GGUF", "family": "minimax_h3",
+            "path": "models/MiniMax-H3-Q4-GGUF",
+        }]}), encoding="utf-8")
+        unused = make_server.models.unused_installed_entries(
+            server_json, {"models/MiniMax-H3-Q4-GGUF/dit.gguf"})
+        self.assertEqual(unused, [])
+
+
+class WizardCompanionGuidanceTests(unittest.TestCase):
+    """_build_entries returns MioCodec guidance for MioTTS selections."""
+
+    _CATALOG_ENTRY = {
+        "family": "miotts", "display_name": "MioTTS", "description": "",
+        "languages": ["en"], "tasks": ["tts", "clone"], "clone_capable": True,
+        "packages": [{
+            "id": "miotts_1_7b_q8_0", "default": True, "format": "gguf",
+            "target_directory": "MioTTS-1.7B-GGUF",
+            "strip_prefix": "MioTTS-1.7B-GGUF",
+            "files": ["MioTTS-1.7B-GGUF/miotts-1.7b-q8_0.gguf"],
+        }],
+        "install_id": "miotts_1_7b_q8_0",
+        "default_path": "models/MioTTS-1.7B-GGUF",
+    }
+
+    def test_miotts_selection_carries_the_mio_codec_companion(self):
+        entries, _ids, guidance, companions, _design, _clone = \
+            make_server.wizard._build_entries(
+                ["miotts"],
+                {"miotts": [{"target_directory": "MioTTS-1.7B-GGUF",
+                             "install_id": "miotts_1_7b_q8_0",
+                             "design": False, "recommended": True}]},
+                {"miotts": self._CATALOG_ENTRY},
+                lambda install_id: "tts")
+        self.assertEqual(companions,
+                         [("MioCodec 25Hz 44.1kHz v2 (required by MioTTS)",
+                           "miocodec_q8_0")])
+        self.assertEqual(entries[0]["path"], "models/MioTTS-1.7B-GGUF")
+        self.assertEqual(len(guidance), 1)
+
+    def test_other_families_carry_no_companions(self):
+        catalog_entry = dict(self._CATALOG_ENTRY, family="voxcpm2",
+                             display_name="VoxCPM2")
+        _entries, _ids, _guidance, companions, _design, _clone = \
+            make_server.wizard._build_entries(
+                ["voxcpm2"],
+                {"voxcpm2": [{"target_directory": "VoxCPM2-GGUF",
+                              "install_id": "voxcpm2_q8_0",
+                              "design": False, "recommended": True}]},
+                {"voxcpm2": catalog_entry},
+                lambda install_id: "tts")
+        self.assertEqual(companions, [])