fix: warn before using a likely too-big chunk size for sglang-omni models

1 files changed, 190 insertions, 4 deletions

diff --git a/app/tests/test_backends_servers.py b/app/tests/test_backends_servers.py
index 569c7bd..9fd71f2 100644
--- a/app/tests/test_backends_servers.py
+++ b/app/tests/test_backends_servers.py
@@ -168,6 +168,13 @@ class StartTests(unittest.TestCase):
             ['File "...", in resolve_checkpoint',
              "ModuleNotFoundError: No module named 'qwen_tts'"]))
 
+    def test_boot_hint_names_a_taken_port(self):
+        # uvicorn's bind failure (a launcher without a port fallback) is
+        # the exited-path face of the port-conflict problem.
+        self.assertIn("held by another process", servers._boot_hint(
+            ["OSError: [Errno 98] error while attempting to bind on "
+             "address 0.0.0.0:9999: address already in use"]))
+
     def test_console_progress_prints_the_hint(self):
         out = io.StringIO()
         with redirect_stdout(out):
@@ -222,16 +229,20 @@ class StartTests(unittest.TestCase):
         """Readiness needs the server to answer HTTP as its identity.
 
         A TCP-accepting but still-booting server (lazy model load, slow
-        listen-before-serve) must not count as ready.
+        listen-before-serve) must not count as ready. The port opens only
+        once the spawned server binds it: free at the spawn-time probe,
+        answering TCP from the first poll on, with the HTTP identity
+        trailing one iteration behind.
         """
         spec = ServerSpec("test", "http://127.0.0.1:9999",
                           [str(self.exe)], identity="audiocpp")
         proc = self._boot_proc()
         with patch.object(servers, "LOG_DIR", self.dir), \
                 patch("subprocess.Popen", return_value=proc), \
-                patch("backends.common.server_running", return_value=True), \
+                patch("backends.common.server_running",
+                      side_effect=[False, True, True]), \
                 patch.object(servers.probe, "identify_server",
-                             side_effect=[None, None, "audiocpp"]), \
+                             side_effect=[None, "audiocpp"]), \
                 patch("time.sleep"):
             ok = servers.start(spec)
         self.assertTrue(ok)
@@ -243,7 +254,8 @@ class StartTests(unittest.TestCase):
         proc = self._boot_proc()
         with patch.object(servers, "LOG_DIR", self.dir), \
                 patch("subprocess.Popen", return_value=proc), \
-                patch("backends.common.server_running", return_value=True), \
+                patch("backends.common.server_running",
+                      side_effect=[False, True, True]), \
                 patch.object(servers.probe, "identify_server",
                              return_value="faster"), \
                 patch.object(servers.probe, "faster_model_loaded",
@@ -467,5 +479,179 @@ class PidForTests(unittest.TestCase):
             self.assertEqual(servers.pid_for("test"), 555)
 
 
+class ReadNewLogTests(unittest.TestCase):
+    """``_read_new_log``: incremental boot-log scanning by byte offset."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.dir = Path(self._tmp.name)
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_reads_only_bytes_appended_since_offset(self):
+        log = self.dir / "log"
+        log.write_text("one\n", encoding="utf-8")
+        offset, text = servers._read_new_log(log, 0)
+        self.assertEqual(text, "one\n")
+        self.assertEqual(offset, 4)
+        self.assertEqual(servers._read_new_log(log, offset), (4, ""))
+        with log.open("a", encoding="utf-8") as fh:
+            fh.write("two\n")
+        offset, text = servers._read_new_log(log, offset)
+        self.assertEqual(text, "two\n")
+        self.assertEqual(offset, 8)
+
+    def test_truncated_log_restarts_from_zero(self):
+        log = self.dir / "log"
+        log.write_text("x" * 100, encoding="utf-8")
+        offset, _text = servers._read_new_log(log, 0)
+        log.write_text("new", encoding="utf-8")
+        offset, text = servers._read_new_log(log, offset)
+        self.assertEqual(text, "new")
+        self.assertEqual(offset, 3)
+
+    def test_missing_file_yields_empty(self):
+        self.assertEqual(servers._read_new_log(self.dir / "nope", 0),
+                         (0, ""))
+
+    def test_undecodable_bytes_are_replaced_not_raised(self):
+        log = self.dir / "log"
+        log.write_bytes(b"ok \xff done\n")
+        _offset, text = servers._read_new_log(log, 0)
+        self.assertIn("done", text)
+
+
+class PortConflictTests(unittest.TestCase):
+    """Doomed boots fail fast instead of polling the wrong port.
+
+    A foreign process on the configured port refuses the spawn outright,
+    and a launcher that logs a silent port fallback (sglang-omni's "Using
+    port N instead") aborts the boot the moment the line appears — the
+    server would keep booting healthily where no client ever polls.
+    """
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.dir = Path(self._tmp.name)
+        self.exe = self.dir / "fake_server"
+        self.exe.write_bytes(b"#!/bin/sh\n")
+        self.spec = ServerSpec("test", "http://127.0.0.1:9999",
+                               [str(self.exe), "--port", "9999"],
+                               identity="sglomni")
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_refuses_to_spawn_when_port_held_by_foreign_process(self):
+        # TCP-up but identity-down at the spec's URL: the listener is not
+        # a usable instance of this server, so a fresh spawn would either
+        # die on the bind or move to a random port. Refuse and name it.
+        with patch.object(servers, "LOG_DIR", self.dir), \
+                patch("subprocess.Popen") as mk, \
+                patch("backends.common.server_running", return_value=True), \
+                patch("backends.probe.identify_server", return_value=None):
+            events = []
+            ok = servers.start(self.spec, progress=events.append)
+        self.assertFalse(ok)
+        mk.assert_not_called()
+        self.assertEqual([e["kind"] for e in events], ["error"])
+        self.assertIn("listening at http://127.0.0.1:9999",
+                      events[0]["message"])
+        self.assertIn("stop that process", events[0]["message"])
+
+    def test_healthy_server_on_the_port_is_reused_not_refused(self):
+        # The pre-flight must not turn "already running" into a conflict:
+        # an endpoint answering as the backend takes the running path.
+        with patch.object(servers, "LOG_DIR", self.dir), \
+                patch("subprocess.Popen") as mk, \
+                patch("backends.common.server_running", return_value=True), \
+                patch("backends.probe.identify_server",
+                      return_value="sglomni"):
+            events = []
+            ok = servers.start(self.spec, progress=events.append)
+        self.assertTrue(ok)
+        mk.assert_not_called()
+        self.assertEqual([e["kind"] for e in events], ["running"])
+
+    def _boot_proc(self):
+        proc = MagicMock()
+        proc.pid = 4242
+        proc.poll.return_value = None
+        return proc
+
+    def test_boot_aborts_when_the_launcher_moves_to_another_port(self):
+        proc = self._boot_proc()
+        fallback = ("[WARNING] Port 9999 is already in use on 0.0.0.0.\n"
+                    "[WARNING] Using port 37183 instead.\n")
+        size = len(fallback.encode("utf-8"))
+        reads = iter([(0, ""), (size, fallback)])
+        with patch.object(servers, "LOG_DIR", self.dir), \
+                patch("subprocess.Popen", return_value=proc) as mk, \
+                patch("backends.common.server_running", return_value=False), \
+                patch.object(servers, "_read_new_log",
+                             side_effect=lambda path, off: next(reads)), \
+                patch.object(servers, "_kill_pid") as mk_kill, \
+                patch("time.sleep"):
+            events = []
+            ok = servers.start(self.spec, progress=events.append)
+        self.assertFalse(ok)
+        mk.assert_called_once()
+        # The misdirected server is killed and unrecorded: it would serve
+        # on a port no client ever polls while holding GPU memory.
+        mk_kill.assert_called_once_with(4242)
+        self.assertFalse((self.dir / "test-server.pid").exists())
+        self.assertEqual([e["kind"] for e in events],
+                         ["starting", "port_taken"])
+        event = events[-1]
+        self.assertEqual(event["taken"], 9999)
+        self.assertEqual(event["moved"], 37183)
+        self.assertIn("moved the server from port 9999", event["message"])
+        self.assertIn("stop whatever holds port 9999", event["message"])
+
+    def test_fallback_split_across_log_reads_still_matches(self):
+        # The launcher prints its two lines back to back, but a 1 s poll
+        # boundary can fall between them — the carried tail re-scans them
+        # together.
+        proc = self._boot_proc()
+        part_a = "WARNING: Port 9999 is already in use on 0.0.0.0.\n"
+        part_b = "WARNING: Using port 37183 instead.\n"
+        off_a = len(part_a.encode("utf-8"))
+        off_b = off_a + len(part_b.encode("utf-8"))
+        reads = iter([(0, ""), (off_a, part_a), (off_b, part_b)])
+        with patch.object(servers, "LOG_DIR", self.dir), \
+                patch("subprocess.Popen", return_value=proc), \
+                patch("backends.common.server_running", return_value=False), \
+                patch.object(servers, "_read_new_log",
+                             side_effect=lambda path, off: next(reads)), \
+                patch.object(servers, "_kill_pid") as mk_kill, \
+                patch("time.sleep"):
+            events = []
+            ok = servers.start(self.spec, progress=events.append)
+        self.assertFalse(ok)
+        mk_kill.assert_called_once_with(4242)
+        self.assertEqual(events[-1]["kind"], "port_taken")
+
+    def test_console_progress_prints_the_port_taken_message(self):
+        proc = self._boot_proc()
+        fallback = ("Port 9999 is already in use on 0.0.0.0.\n"
+                    "Using port 37183 instead.\n")
+        size = len(fallback.encode("utf-8"))
+        reads = iter([(0, ""), (size, fallback)])
+        with patch.object(servers, "LOG_DIR", self.dir), \
+                patch("subprocess.Popen", return_value=proc), \
+                patch("backends.common.server_running", return_value=False), \
+                patch.object(servers, "_read_new_log",
+                             side_effect=lambda path, off: next(reads)), \
+                patch.object(servers, "_kill_pid"), \
+                patch("time.sleep"):
+            out = io.StringIO()
+            with redirect_stdout(out):
+                servers.start(self.spec)
+        self.assertIn("moved the server from port 9999", out.getvalue())
+        self.assertIn("(already in use by another process) to port 37183",
+                      out.getvalue())
+
+
 if __name__ == "__main__":
     unittest.main()