fix: crash on bad model_specs from audio.cpp, sanitized

1 files changed, 261 insertions, 7 deletions

diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py
index ada7954..08de777 100644
--- a/app/tests/test_backends_audiocpp.py
+++ b/app/tests/test_backends_audiocpp.py
@@ -834,6 +834,203 @@ class InstallModelsTests(unittest.TestCase):
         self.assertFalse(make_server.models._all_models_present(
             self.checkout, [{"path": "models/higgs"}]))
 
+    def _supporting_manager(self):
+        self.manager.write_text(
+            '#!/usr/bin/env python3\n'
+            'parser.add_argument("--specs-dir", default="")\n'
+            'parser.add_argument("--cancel-file", default="")\n',
+            encoding="utf-8")
+
+    def _write_spec(self, name: str, packages: list) -> Path:
+        specs = self.checkout / "model_specs"
+        specs.mkdir(parents=True, exist_ok=True)
+        path = specs / name
+        path.write_text(json.dumps({"family": name[:-5], "packages": packages}),
+                        encoding="utf-8")
+        return path
+
+    def test_dot_strip_prefix_spec_installs_from_sanitized_copy(self):
+        self._supporting_manager()
+        original = self._write_spec("vietneu_tts.json", [{
+            "id": "vietneu_tts_v3_turbo_q8_0",
+            "files": ["model.gguf"],
+            "strip_prefix": ".",
+        }])
+        self._write_spec("other.json", [{
+            "id": "other_q8_0",
+            "files": ["Other-GGUF/model.gguf"],
+            "strip_prefix": "Other-GGUF",
+        }])
+        seen: dict = {}
+
+        def record(argv, **kwargs):
+            specs_dir = Path(argv[3])
+            seen["fixed"] = json.loads(
+                (specs_dir / "vietneu_tts.json").read_text(encoding="utf-8"))
+            seen["other"] = json.loads(
+                (specs_dir / "other.json").read_text(encoding="utf-8"))
+            return 0
+
+        buf = io.StringIO()
+        with redirect_stdout(buf), \
+                patch.object(common, "run_console_subprocess",
+                             side_effect=record) as run:
+            rc = make_server.models._install_models(
+                self.checkout,
+                [("VieNeu-TTS v3 Turbo GGUF", "vietneu_tts_v3_turbo_q8_0")],
+                download=True)
+        self.assertEqual(rc, 0)
+        argv = run.call_args[0][0]
+        self.assertEqual(argv[:2], [sys.executable, str(self.manager)])
+        self.assertEqual(argv[2], "--specs-dir")
+        self.assertEqual(argv[4:], ["install", "vietneu_tts_v3_turbo_q8_0"])
+        self.assertEqual(seen["fixed"]["packages"][0]["strip_prefix"], "")
+        self.assertEqual(seen["other"]["packages"][0]["strip_prefix"],
+                         "Other-GGUF")
+        self.assertIn("sanitized copy", buf.getvalue())
+        # The checkout's own specs are untouched and the temp copy is gone.
+        self.assertEqual(json.loads(
+            original.read_text(encoding="utf-8"))["packages"][0]
+            ["strip_prefix"], ".")
+        self.assertFalse(Path(argv[3]).exists())
+
+    def test_healthy_specs_do_not_add_specs_dir(self):
+        self._supporting_manager()
+        self._write_spec("ok.json", [{
+            "id": "ok_q8_0",
+            "files": ["Ok-GGUF/model.gguf"],
+            "strip_prefix": "Ok-GGUF",
+        }])
+        with patch.object(common, "run_console_subprocess",
+                          return_value=0) as run:
+            make_server.models._install_models(
+                self.checkout, [("Ok", "ok_q8_0")], download=True)
+        self.assertEqual(
+            run.call_args[0][0],
+            [sys.executable, str(self.manager), "install", "ok_q8_0"])
+
+    def test_unknown_prefix_mismatch_left_for_warning_path(self):
+        # A real-directory prefix that matches no files cannot be repaired
+        # confidently; the install is left to fail with the manager's own
+        # error so the remaining downloads continue (warn-and-continue).
+        self._supporting_manager()
+        self._write_spec("broken.json", [{
+            "id": "broken_q8_0",
+            "files": ["model.gguf"],
+            "strip_prefix": "Some-Dir",
+        }])
+        with patch.object(common, "run_console_subprocess",
+                          return_value=0) as run:
+            make_server.models._install_models(
+                self.checkout, [("Broken", "broken_q8_0")], download=True)
+        self.assertEqual(
+            run.call_args[0][0],
+            [sys.executable, str(self.manager), "install", "broken_q8_0"])
+
+    def test_specs_dir_unsupported_manager_leaves_argv_unchanged(self):
+        self._write_spec("vietneu_tts.json", [{
+            "id": "vietneu_tts_v3_turbo_q8_0",
+            "files": ["model.gguf"],
+            "strip_prefix": ".",
+        }])
+        with patch.object(common, "run_console_subprocess",
+                          return_value=0) as run:
+            make_server.models._install_models(
+                self.checkout,
+                [("VieNeu-TTS v3 Turbo GGUF", "vietneu_tts_v3_turbo_q8_0")],
+                download=True)
+        self.assertEqual(
+            run.call_args[0][0],
+            [sys.executable, str(self.manager), "install",
+             "vietneu_tts_v3_turbo_q8_0"])
+
+
+class SanitizeModelSpecTests(unittest.TestCase):
+    """The dot strip_prefix repair and the --specs-dir staging copy."""
+
+    def test_dot_prefix_dropped_when_file_is_bare(self):
+        spec = {"packages": [{"files": ["model.gguf"], "strip_prefix": "."}]}
+        self.assertTrue(make_server.models._sanitize_model_spec(spec))
+        self.assertEqual(spec["packages"][0]["strip_prefix"], "")
+
+    def test_slash_dot_prefix_normalized_like_dot(self):
+        spec = {"packages": [{"files": ["model.gguf"], "strip_prefix": "./"}]}
+        self.assertTrue(make_server.models._sanitize_model_spec(spec))
+        self.assertEqual(spec["packages"][0]["strip_prefix"], "")
+
+    def test_dot_prefix_kept_when_files_carry_it(self):
+        spec = {"packages": [{"files": ["./model.gguf"],
+                              "strip_prefix": "."}]}
+        self.assertFalse(make_server.models._sanitize_model_spec(spec))
+        self.assertEqual(spec["packages"][0]["strip_prefix"], ".")
+
+    def test_real_directory_prefix_untouched(self):
+        spec = {"packages": [{"files": ["model.gguf"],
+                              "strip_prefix": "Kroko-ASR-GGUF"}]}
+        self.assertFalse(make_server.models._sanitize_model_spec(spec))
+        self.assertEqual(spec["packages"][0]["strip_prefix"],
+                         "Kroko-ASR-GGUF")
+
+    def test_valid_prefix_untouched(self):
+        spec = {"packages": [{"files": ["Kroko-ASR-GGUF/model.gguf"],
+                              "strip_prefix": "Kroko-ASR-GGUF"}]}
+        self.assertFalse(make_server.models._sanitize_model_spec(spec))
+
+    def test_missing_or_empty_files_untouched(self):
+        spec = {"packages": [{"strip_prefix": "."},
+                             {"files": [], "strip_prefix": "."},
+                             {"files": "model.gguf", "strip_prefix": "."}]}
+        self.assertFalse(make_server.models._sanitize_model_spec(spec))
+
+    def test_only_broken_packages_repaired(self):
+        spec = {"packages": [
+            {"files": ["model.gguf"], "strip_prefix": "."},
+            {"files": ["./model.gguf"], "strip_prefix": "."},
+        ]}
+        self.assertTrue(make_server.models._sanitize_model_spec(spec))
+        self.assertEqual([p["strip_prefix"] for p in spec["packages"]],
+                         ["", "."])
+
+    def setUp(self):
+        self._td = tempfile.TemporaryDirectory()
+        self.checkout = Path(self._td.name) / "audio.cpp"
+        self.checkout.mkdir()
+
+    def tearDown(self):
+        self._td.cleanup()
+
+    def test_prepare_specs_dir_none_without_specs(self):
+        self.assertIsNone(
+            make_server.models._prepare_specs_dir(self.checkout))
+
+    def test_prepare_specs_dir_none_when_healthy(self):
+        specs = self.checkout / "model_specs"
+        specs.mkdir()
+        (specs / "ok.json").write_text(json.dumps(
+            {"packages": [{"files": ["Ok-GGUF/m.gguf"],
+                           "strip_prefix": "Ok-GGUF"}]}), encoding="utf-8")
+        self.assertIsNone(
+            make_server.models._prepare_specs_dir(self.checkout))
+
+    def test_prepare_specs_dir_writes_all_specs_and_repairs(self):
+        specs = self.checkout / "model_specs"
+        specs.mkdir()
+        (specs / "broken.json").write_text(json.dumps(
+            {"packages": [{"files": ["model.gguf"],
+                           "strip_prefix": "."}]}), encoding="utf-8")
+        (specs / "plain.json").write_text("not json", encoding="utf-8")
+        staging = make_server.models._prepare_specs_dir(self.checkout)
+        try:
+            self.assertIsNotNone(staging)
+            repaired = json.loads(
+                (staging / "broken.json").read_text(encoding="utf-8"))
+            self.assertEqual(repaired["packages"][0]["strip_prefix"], "")
+            self.assertEqual(
+                (staging / "plain.json").read_text(encoding="utf-8"),
+                "not json")
+        finally:
+            shutil.rmtree(staging, ignore_errors=True)
+
 
 class ConfigFormTranscriptionToggleTests(unittest.TestCase):
     """The combined form's Voice transcripts row: one fixed two-way toggle.
@@ -3467,10 +3664,18 @@ class PrebuiltUpdateRoutingTests(unittest.TestCase):
         marker.write_text(json.dumps({"tag": tag, "asset": "x"}),
                           encoding="utf-8")
 
+    def _marker(self, checkout: Path) -> Path:
+        return make_server.prebuilt.marker_path(checkout, "cpu",
+                                                platform="darwin")
+
     def _run(self, checkout: Path, *, marker: Optional[dict],
-             release: Optional[dict], tag: Optional[str] = None):
+             release: Optional[dict], tag: Optional[str] = None,
+             install_rc: int = 0, build_rc: int = 0):
         # TAG is what the quota-free redirect resolution reports; RELEASE
         # is the API fallback (only consulted when the redirect fails).
+        # INSTALL_RC is what the prebuilt re-download reports: 0 (or a
+        # cancellation) returns directly, a failure falls back to the
+        # source-build route, whose BUILD_RC decides the final exit code.
         with patch("sys.platform", "darwin"), \
                 patch.object(make_server.build, "find_local_checkout",
                              return_value=checkout), \
@@ -3485,16 +3690,18 @@ class PrebuiltUpdateRoutingTests(unittest.TestCase):
                 patch.object(make_server.prebuilt, "fetch_latest_release",
                              return_value=release) as mk_fetch, \
                 patch.object(make_server.prebuilt, "install_prebuilt",
-                             return_value=0) as mk_install, \
+                             return_value=install_rc) as mk_install, \
                 patch.object(common, "git_update",
-                             return_value=0) as mk_git:
+                             return_value=0) as mk_git, \
+                patch.object(make_server.build, "build_audiocpp",
+                             return_value=build_rc) as mk_build:
             rc = make_server.build.update()
-        return rc, mk_install, mk_git, mk_tag, mk_fetch
+        return rc, mk_install, mk_git, mk_tag, mk_fetch, mk_build
 
     def test_newer_release_triggers_a_redownload(self):
         checkout = self._checkout()
         self._mark_prebuilt(checkout, "v0.6.0")
-        rc, mk_install, mk_git, _mk_tag, mk_fetch = self._run(
+        rc, mk_install, mk_git, _mk_tag, mk_fetch, mk_build = self._run(
             checkout, marker={"tag": "v0.6.0", "asset": "x"},
             release=_release([], tag="v0.7.0"), tag="v0.7.0")
         self.assertEqual(rc, 0)
@@ -3502,22 +3709,24 @@ class PrebuiltUpdateRoutingTests(unittest.TestCase):
         mk_install.assert_called_once_with(checkout, "cpu", emit=None,
                                            cancel=None)
         mk_fetch.assert_not_called()
+        mk_build.assert_not_called()
 
     def test_current_release_is_a_noop(self):
         checkout = self._checkout()
-        rc, mk_install, mk_git, _mk_tag, mk_fetch = self._run(
+        rc, mk_install, mk_git, _mk_tag, mk_fetch, mk_build = self._run(
             checkout, marker={"tag": "v0.7.0", "asset": "x"},
             release=None, tag="v0.7.0")
         self.assertEqual(rc, 0)
         mk_install.assert_not_called()
         mk_git.assert_not_called()
+        mk_build.assert_not_called()
         # The redirect already answered: the API (rate-limited easily)
         # must not be touched for an "already current" check.
         mk_fetch.assert_not_called()
 
     def test_unreachable_github_keeps_the_install(self):
         checkout = self._checkout()
-        rc, mk_install, mk_git, mk_tag, mk_fetch = self._run(
+        rc, mk_install, mk_git, mk_tag, mk_fetch, mk_build = self._run(
             checkout, marker={"tag": "v0.7.0", "asset": "x"},
             release=None, tag=None)
         self.assertEqual(rc, 0)
@@ -3525,6 +3734,51 @@ class PrebuiltUpdateRoutingTests(unittest.TestCase):
         mk_fetch.assert_called_once()  # the API fallback tried too
         mk_install.assert_not_called()
         mk_git.assert_not_called()
+        mk_build.assert_not_called()
+
+    def test_failed_redownload_falls_back_to_a_source_build(self):
+        checkout = self._checkout()
+        self._mark_prebuilt(checkout, "v0.6.0")
+        rc, mk_install, mk_git, _mk_tag, _mk_fetch, mk_build = self._run(
+            checkout, marker={"tag": "v0.6.0", "asset": "x"},
+            release=_release([], tag="v0.7.0"), tag="v0.7.0", install_rc=1)
+        self.assertEqual(rc, 0)
+        mk_install.assert_called_once_with(checkout, "cpu", emit=None,
+                                           cancel=None)
+        mk_git.assert_called_once()
+        mk_build.assert_called_once_with(checkout, "cpu", emit=None,
+                                         cancel=None)
+        # The fallback source build replaces the prebuilt install: the
+        # marker must go, or the next update re-downloads over the
+        # freshly built binary.
+        self.assertFalse(self._marker(checkout).exists())
+
+    def test_failed_redownload_keeps_the_marker_when_the_build_fails(self):
+        # A failed fallback leaves the previous prebuilt binary in place
+        # (install_prebuilt downloads before touching it), so the marker
+        # stays truthful and the next update retries the re-download.
+        checkout = self._checkout()
+        self._mark_prebuilt(checkout, "v0.6.0")
+        rc, _mk_install, mk_git, _mk_tag, _mk_fetch, mk_build = self._run(
+            checkout, marker={"tag": "v0.6.0", "asset": "x"},
+            release=_release([], tag="v0.7.0"), tag="v0.7.0", install_rc=1,
+            build_rc=1)
+        self.assertEqual(rc, 1)
+        mk_git.assert_called_once()
+        mk_build.assert_called_once()
+        self.assertTrue(self._marker(checkout).exists())
+
+    def test_cancelled_redownload_does_not_fall_back(self):
+        checkout = self._checkout()
+        self._mark_prebuilt(checkout, "v0.6.0")
+        rc, _mk_install, mk_git, _mk_tag, _mk_fetch, mk_build = self._run(
+            checkout, marker={"tag": "v0.6.0", "asset": "x"},
+            release=_release([], tag="v0.7.0"), tag="v0.7.0",
+            install_rc=130)
+        self.assertEqual(rc, 130)
+        mk_git.assert_not_called()
+        mk_build.assert_not_called()
+        self.assertTrue(self._marker(checkout).exists())
 
     def test_source_builds_still_route_through_git(self):
         checkout = self._checkout()