feat: sglang backend support

1 files changed, 93 insertions, 2 deletions

diff --git a/app/tests/test_backends_managed.py b/app/tests/test_backends_managed.py
index cd1f03a..e0bb074 100644
--- a/app/tests/test_backends_managed.py
+++ b/app/tests/test_backends_managed.py
@@ -18,8 +18,8 @@ from backends.managed import ManagedServer, ensure_running
 from backends.probe import (IDENTITY_AUDIOCPP, IDENTITY_QWEN_CLONE,
                             IDENTITY_QWEN_CUSTOM, IDENTITY_QWEN_DESIGN)
 from converter.clients import (BACKEND_AUDIOCPP, BACKEND_QWEN,
-                               VOICE_MODE_CLONE, VOICE_MODE_CUSTOM,
-                               VOICE_MODE_DESIGN)
+                               BACKEND_SGLOMNI, VOICE_MODE_CLONE,
+                               VOICE_MODE_CUSTOM, VOICE_MODE_DESIGN)
 
 
 def _spec(name="audiocpp", url="http://127.0.0.1:8080", identity=None):
@@ -269,6 +269,97 @@ class QwenEnsureRunningTests(unittest.TestCase):
         mk_start.assert_not_called()
 
 
+class SglomniEnsureRunningTests(unittest.TestCase):
+    """sglomni hosts one model per server: the running-model check is
+    keyed on the served HuggingFace repo id (the qwen rules again)."""
+
+    def _detect_sglomni(self):
+        from backends.sglomni import status as sg_status
+        from backends.sglomni.catalog import entry_by_key
+        with patch("backends.sglomni.gpu.compute_capability",
+                   return_value=None):
+            spec = sg_status.build_spec(entry_by_key("zonos2"))
+        return _status([spec], installed=True, label="SGLang-Omni")
+
+    def _run(self, model):
+        out = io.StringIO()
+        with redirect_stdout(out):
+            result = ensure_running(BACKEND_SGLOMNI, VOICE_MODE_CLONE,
+                                    model=model)
+        return result, out.getvalue()
+
+    def test_spec_aims_at_the_model_the_run_selected(self):
+        with patch("backends.detect", return_value=self._detect_sglomni()), \
+                patch("backends.sglomni.models.model_installed",
+                      return_value=True), \
+                patch("backends.sglomni.gpu.compute_capability",
+                      return_value=None), \
+                patch("backends.common.server_running",
+                      return_value=False), \
+                patch.object(servers, "start",
+                             return_value=True) as mk_start:
+            result, _ = self._run("zonos2")
+        spec = mk_start.call_args.args[0]
+        self.assertIn("Zyphra/zonos2", spec.argv)
+        self.assertNotIn("--config", spec.argv)
+        self.assertTrue(result.started)
+
+    def test_fp8_fallback_prints_a_note_and_boots_the_bf16_config(self):
+        with patch("backends.detect", return_value=self._detect_sglomni()), \
+                patch("backends.sglomni.models.model_installed",
+                      return_value=True), \
+                patch("backends.sglomni.gpu.compute_capability",
+                      return_value=(8, 6)), \
+                patch("backends.common.server_running",
+                      return_value=False), \
+                patch.object(servers, "start",
+                             return_value=True) as mk_start:
+            result, output = self._run("zonos2")
+        self.assertIn("bf16", output)
+        self.assertIn("--config", mk_start.call_args.args[0].argv)
+        self.assertTrue(result.started)
+
+    def test_managed_server_hosting_another_model_is_rebooted(self):
+        with patch("backends.detect", return_value=self._detect_sglomni()), \
+                patch("backends.sglomni.models.model_installed",
+                      return_value=True), \
+                patch("backends.sglomni.gpu.compute_capability",
+                      return_value=None), \
+                patch("backends.common.server_running",
+                      return_value=True), \
+                patch("backends.probe.sglomni_served_model",
+                      return_value="Qwen/Qwen3-TTS-12Hz-1.7B-Base"), \
+                patch.object(servers, "alive", return_value=True), \
+                patch.object(servers, "start",
+                             return_value=True) as mk_start, \
+                patch.object(servers, "stop") as mk_stop:
+            result, output = self._run("zonos2")
+        self.assertIn("restarting", output)
+        mk_stop.assert_called_once_with("sglomni")
+        self.assertIn("Zyphra/zonos2",
+                      mk_start.call_args.args[0].argv)
+        self.assertTrue(result.started)
+
+    def test_foreign_server_hosting_another_model_refuses_the_run(self):
+        with patch("backends.detect", return_value=self._detect_sglomni()), \
+                patch("backends.sglomni.models.model_installed",
+                      return_value=True), \
+                patch("backends.sglomni.gpu.compute_capability",
+                      return_value=None), \
+                patch("backends.common.server_running",
+                      return_value=True), \
+                patch("backends.probe.sglomni_served_model",
+                      return_value="Qwen/Qwen3-TTS-12Hz-1.7B-Base"), \
+                patch.object(servers, "alive", return_value=False), \
+                patch.object(servers, "start") as mk_start, \
+                patch.object(servers, "stop") as mk_stop:
+            result, output = self._run("zonos2")
+        self.assertFalse(result.ok)
+        self.assertIn("this run needs Zyphra/zonos2", output)
+        mk_start.assert_not_called()
+        mk_stop.assert_not_called()
+
+
 class ManagedModuleSmokeTests(unittest.TestCase):
     """Import-surface sanity for the module the CLI wires in."""