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

7 files changed, 590 insertions, 9 deletions

diff --git a/app/tests/test_audiobook_cli.py b/app/tests/test_audiobook_cli.py
index ab13107..f25ec23 100644
--- a/app/tests/test_audiobook_cli.py
+++ b/app/tests/test_audiobook_cli.py
@@ -334,6 +334,80 @@ class ConvertWiringTests(unittest.TestCase):
             self._convert(output_file=self.tmp / "dune.mp3")
 
 
+class SglomniChunkClampTests(unittest.TestCase):
+    """convert(backend="sglomni"): the chunk-cap popup runs pre-flight
+    for CLI runs and its answer rides chunk_size into the converter."""
+
+    def setUp(self):
+        self.tmp = Path(tempfile.mkdtemp(prefix="audiobook_sglomni_"))
+        self.addCleanup(shutil.rmtree, self.tmp, True)
+        self.book = _make_book(self.tmp)
+        self._old_folders = (converter_mod.BOOKS_FOLDER,
+                             converter_mod.AUDIOBOOKS_FOLDER)
+        self.addCleanup(self._restore_folders)
+
+    def _restore_folders(self):
+        converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER = \
+            self._old_folders
+
+    def _convert(self, *, prompt=None, **kwargs):
+        """Run convert() for a higgs run with the prompt mocked.
+
+        PROMPT replaces converter.prompt_chunk_clamp (default: a MagicMock
+        answering 80). Returns (code, prompt mock, converter ctor kwargs).
+        """
+        if prompt is None:
+            prompt = MagicMock(return_value=80)
+        preflight = MagicMock(
+            return_value=([self.book], [(self.book, "dune")]))
+        fake_instance = MagicMock()
+        fake_instance.run.return_value = True
+        fake_class = MagicMock(return_value=fake_instance)
+        fake_class.preflight_overwrites = preflight
+        stdout = io.StringIO()
+        with patch.object(audiobook, "setup_logging"), \
+                patch.object(audiobook, "setup_directories"), \
+                patch.object(audiobook, "AudiobookConverter", fake_class), \
+                patch.object(converter_mod, "prompt_chunk_clamp", prompt), \
+                contextlib.redirect_stdout(stdout):
+            code = audiobook.convert(
+                backend="sglomni", model_id="higgs_audio_v3_tts",
+                api_url="http://127.0.0.1:8100", **kwargs)
+        # A cancelled run stops before any converter is constructed.
+        ctor = (fake_class.call_args.kwargs
+                if fake_class.call_args is not None else None)
+        return code, prompt, stdout, ctor
+
+    def test_cli_run_asks_and_carries_the_clamp(self):
+        code, prompt, _, ctor = self._convert()
+        self.assertEqual(code, 0)
+        self.assertEqual(prompt.call_args[0][0].key, "higgs_audio_v3_tts")
+        self.assertEqual(ctor["chunk_size"], 80)
+
+    def test_prompt_anyway_sends_no_clamp(self):
+        _, prompt, _, ctor = self._convert(
+            prompt=MagicMock(return_value=None))
+        self.assertIsNone(ctor["chunk_size"])
+
+    def test_prompt_cancel_stops_the_run_unstarted(self):
+        def cancel(entry):
+            raise converter_mod.ChunkClampCancelled("cancelled")
+        code, prompt, stdout, ctor = self._convert(prompt=cancel)
+        self.assertEqual(code, 0)
+        self.assertIn("Conversion cancelled", stdout.getvalue())
+        self.assertFalse(ctor)
+
+    def test_hub_run_with_a_plan_skips_the_prompt(self):
+        # The hub pre-flights inside the TUI (where the popup lives) and
+        # carries the answer as chunk_size; convert() must not re-ask.
+        _, prompt, _, ctor = self._convert(
+            prompt=MagicMock(side_effect=AssertionError("should not ask")),
+            book_files=[self.book],
+            planned=[(self.book, "dune")],
+            chunk_size=80)
+        self.assertEqual(ctor["chunk_size"], 80)
+
+
 class AllModelsConvertTests(unittest.TestCase):
     """convert(model_ids=...): the "All (multiple generation)" loop.
 
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()
diff --git a/app/tests/test_backends_sglomni.py b/app/tests/test_backends_sglomni.py
index ebac500..8cb86cb 100644
--- a/app/tests/test_backends_sglomni.py
+++ b/app/tests/test_backends_sglomni.py
@@ -395,12 +395,24 @@ class HiggsConfigTests(unittest.TestCase):
         self.assertRegex(text, r"gpu_memory_fraction:\s*0\.80")
         # The engine's 2048-frame default (~27 s at 75 fps) silently
         # truncates a full 250-word sub-chunk; per-request values are
-        # clamped to this factory cap server-side.
-        self.assertRegex(text, r"max_new_tokens:\s*12288")
+        # clamped to this factory cap server-side. 3000 frames (~40 s)
+        # is the most the pinned 4095-token admission window allows
+        # after the prompt tokens.
+        self.assertRegex(text, r"max_new_tokens:\s*3000")
 
     def test_entry_sends_the_raised_frame_cap_per_request(self):
         entry = entry_by_key("higgs_audio_v3_tts")
-        self.assertEqual(entry.max_new_tokens, 12288)
+        self.assertEqual(entry.max_new_tokens, 3000)
+
+    def test_entry_caps_sub_requests_for_the_admission_window(self):
+        """The server pins prompt + generation at 4096 tokens for Higgs;
+        80 words (~30-40 s at 75 fps) narrates inside the 3000-frame
+        cap, and the pre-flight popup offers the clamp for a run."""
+        entry = entry_by_key("higgs_audio_v3_tts")
+        self.assertEqual(entry.chunk_words, 80)
+        # A full CHUNK_SIZE sub-chunk does NOT fit one Higgs request.
+        self.assertLess(entry.chunk_words, 250)
+        self.assertIsNone(entry_by_key("zonos2").chunk_words)
 
 
 class Fp8FallbackTests(unittest.TestCase):
diff --git a/app/tests/test_converter.py b/app/tests/test_converter.py
index 0b0eb0d..85ae6ca 100644
--- a/app/tests/test_converter.py
+++ b/app/tests/test_converter.py
@@ -21,7 +21,11 @@ from converter.clients import (
 from converter import converter as converter_mod
 from converter.converter import (
     AudiobookConverter,
+    ChunkClampCancelled,
+    chunk_clamp_message,
+    chunk_clamp_needed,
     find_existing_outputs,
+    prompt_chunk_clamp,
     prompt_overwrite,
     setup_logging,
 )
@@ -554,6 +558,71 @@ class PromptOverwriteTests(unittest.TestCase):
         self.assertIn("overwrite them", prompt_text)
 
 
+class ChunkClampPromptTests(unittest.TestCase):
+    """The chunk-cap popup for models that cannot narrate a full
+    CHUNK_SIZE sub-request (Higgs)."""
+
+    def _entry(self):
+        from backends.sglomni.catalog import entry_by_key
+        return entry_by_key("higgs_audio_v3_tts")
+
+    def test_uncapped_models_need_no_clamp(self):
+        from backends.sglomni.catalog import entry_by_key
+        self.assertFalse(chunk_clamp_needed(None))
+        self.assertFalse(chunk_clamp_needed(entry_by_key("zonos2")))
+
+    def test_higgs_needs_a_clamp_at_the_default_chunk_size(self):
+        self.assertTrue(chunk_clamp_needed(self._entry()))
+
+    def test_message_names_the_model_and_the_cap(self):
+        lines = chunk_clamp_message(self._entry())
+        text = " ".join(lines)
+        self.assertIn("Higgs Audio v3 TTS", text)
+        self.assertIn("80 words", text)
+        self.assertIn("cut off mid-sentence", text)
+
+    def test_clamped_models_at_or_below_the_cap_need_no_clamp(self):
+        with patch.object(config, "CHUNK_SIZE", 80):
+            self.assertFalse(chunk_clamp_needed(self._entry()))
+
+    def test_prompt_answers(self):
+        entry = self._entry()
+        with patch("builtins.input", return_value=""):
+            self.assertEqual(prompt_chunk_clamp(entry), 80)
+        with patch("builtins.input", return_value="s"):
+            self.assertEqual(prompt_chunk_clamp(entry), 80)
+        with patch("builtins.input", return_value="t"):
+            self.assertIsNone(prompt_chunk_clamp(entry))
+        with patch("builtins.input", return_value="cancel"):
+            with self.assertRaises(ChunkClampCancelled):
+                prompt_chunk_clamp(entry)
+
+    def test_prompt_invalid_answer_reasked(self):
+        with patch("builtins.input", side_effect=["maybe", "t"]) as mock_input:
+            self.assertIsNone(prompt_chunk_clamp(self._entry()))
+        self.assertEqual(mock_input.call_count, 2)
+
+    def test_prompt_eof_clamps_for_unattended_runs(self):
+        with patch("builtins.input", side_effect=EOFError):
+            self.assertEqual(prompt_chunk_clamp(self._entry()), 80)
+
+    def test_ask_callback_replaces_the_console(self):
+        calls = []
+
+        def ask(lines, words):
+            calls.append((lines, words))
+            return "clamp"
+
+        self.assertEqual(prompt_chunk_clamp(self._entry(), ask=ask), 80)
+        self.assertEqual(len(calls), 1)
+        self.assertEqual(calls[0][1], 80)
+
+    def test_ask_anyway_returns_no_clamp(self):
+        self.assertIsNone(prompt_chunk_clamp(
+            self._entry(), ask=lambda lines, words: "anyway"))
+
+
+
 class PreflightOverwritesTests(unittest.TestCase):
     """The pre-flight overwrite check runs without a TTS server connection."""
 
diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py
index e38acbe..84796ee 100644
--- a/app/tests/test_hub.py
+++ b/app/tests/test_hub.py
@@ -2661,6 +2661,8 @@ class ConvertFlowTests(unittest.TestCase):
                 self._mock_preflight()
                 self._answer_form(backend="sglomni", model_id=entry.key,
                                   voice=str(ref), instructions="")
+                # Higgs's chunk-cap popup: keep the configured chunk size.
+                self.tui.script.append("anyway")
                 cmd = self._convert(None, [self._ready("sglomni",
                                                        "SGLang-Omni")])
                 fields = self.tui.forms_seen[-1][1]
@@ -2691,6 +2693,8 @@ class ConvertFlowTests(unittest.TestCase):
             self._mock_preflight()
             self._answer_form(backend="sglomni", model_id=entry.key,
                               voice="", instructions="")
+            # Higgs's chunk-cap popup: keep the configured chunk size.
+            self.tui.script.append("anyway")
             cmd = self._convert(None, [self._ready("sglomni",
                                                    "SGLang-Omni")])
         kwargs = cmd[2]
@@ -2738,6 +2742,8 @@ class ConvertFlowTests(unittest.TestCase):
             self._answer_form(backend="sglomni-remote",
                               model_id="higgs_audio_v3_tts",
                               voice="narrator", instructions="")
+            # Higgs's chunk-cap popup: keep the configured chunk size.
+            self.tui.script.append("anyway")
             cmd = self._convert(
                 None, [self._remote("sglomni", "SGLang-Omni",
                                     url="http://sgl.local:8100")])
@@ -2870,6 +2876,10 @@ class ConvertFlowTests(unittest.TestCase):
                         else:  # design
                             overrides["instructions"] = "A warm narrator."
                         self._answer_form(**overrides)
+                        if hub.converter_mod.chunk_clamp_needed(entry):
+                            # Higgs's chunk-cap popup: keep the configured
+                            # chunk size.
+                            self.tui.script.append("anyway")
                         cmd = self._convert(None, [
                             self._ready("sglomni", "SGLang-Omni")])
                         self.assertIsNotNone(cmd)
@@ -3196,6 +3206,54 @@ class PreflightTests(unittest.TestCase):
             with self.assertRaises(hub._BackToForm):
                 confirm("overwrite?", True)
 
+    # -- sglomni: the chunk-cap popup ------------------------------------
+
+    def _sglomni_cmd(self, model_id="higgs_audio_v3_tts"):
+        return ("convert", "sglomni",
+                {"model_id": model_id, "clone": None, "output_format": "mp3"})
+
+    def _run_sglomni_preflight(self, cmd, menu_answer):
+        stdscr = object()
+        with patch.object(hub.AudiobookConverter, "preflight_overwrites",
+                          return_value=(["book.txt"], [("book.txt", "x")])), \
+                patch.object(hub.tui, "menu", return_value=menu_answer) \
+                as mk_menu:
+            outcome = hub._preflight(stdscr, cmd)
+        return outcome, mk_menu
+
+    def test_sglomni_run_asks_the_chunk_popup_and_stashes_the_clamp(self):
+        cmd = self._sglomni_cmd()
+        outcome, mk_menu = self._run_sglomni_preflight(cmd, "clamp")
+        self.assertTrue(outcome)
+        self.assertEqual(cmd[2]["chunk_size"], 80)
+        options = mk_menu.call_args[0][2]
+        self.assertEqual([value for _label, value in options],
+                         ["clamp", "anyway", "cancel"])
+        self.assertIn("Set Chunk to 80", options[0][0])
+
+    def test_sglomni_run_try_anyway_stashes_no_clamp(self):
+        cmd = self._sglomni_cmd()
+        outcome, mk_menu = self._run_sglomni_preflight(cmd, "anyway")
+        self.assertTrue(outcome)
+        self.assertNotIn("chunk_size", cmd[2])
+        mk_menu.assert_called_once()
+
+    def test_sglomni_run_cancel_raises_back_to_form(self):
+        # Cancel backs out to the Generate form (the run never starts).
+        stdscr = object()
+        with patch.object(hub.AudiobookConverter, "preflight_overwrites",
+                          return_value=(["book.txt"], [("book.txt", "x")])), \
+                patch.object(hub.tui, "menu", return_value="cancel"):
+            with self.assertRaises(hub._BackToForm):
+                hub._preflight(stdscr, self._sglomni_cmd())
+
+    def test_sglomni_uncapped_model_skips_the_popup(self):
+        outcome, mk_menu = self._run_sglomni_preflight(
+            self._sglomni_cmd("zonos2"),
+            MagicMock(side_effect=AssertionError("should not ask")))
+        self.assertTrue(outcome)
+        mk_menu.assert_not_called()
+
     # -- "All (multiple generation)": one plan per model ----------------
 
     def _all_cmd(self):
diff --git a/app/tests/test_runview.py b/app/tests/test_runview.py
index 501f1f7..53f0a44 100644
--- a/app/tests/test_runview.py
+++ b/app/tests/test_runview.py
@@ -246,6 +246,44 @@ class StateTransitionTests(_FakeTui, unittest.TestCase):
         self.assertEqual(view.boot_hint,
                          "FP8 needs compute capability 8.9+")
 
+    def test_port_taken_event_is_error_with_the_port_specifics(self):
+        # The launcher moved the server to a random port (configured one
+        # taken) and the boot was killed: the error screen names the ports
+        # instead of sitting on "starting" until the timeout.
+        view, _ = self.make_view()
+        view.handle_event({"kind": "starting", "name": "sglomni",
+                           "log_path": "/tmp/sglomni-server.log"})
+        view.handle_event(
+            {"kind": "port_taken", "name": "sglomni",
+             "taken": 8100, "moved": 37183,
+             "message": "the sglomni launcher moved the server from port "
+                        "8100 (already in use by another process) to port "
+                        "37183; clients poll 8100, so this boot cannot "
+                        "become ready — stop whatever holds port 8100 "
+                        "and start again",
+             "log_tail": ["Using port 37183 instead."]})
+        self.assertEqual(view.phase, "error")
+        self.assertEqual(view.server, "error")
+        self.assertIn("8100", view.server_message)
+        self.assertIn("37183", view.server_message)
+        self.assertEqual(view.log_tail, ["Using port 37183 instead."])
+
+    def test_port_taken_boot_failure_is_recorded_in_the_dated_log(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            log_path = os.path.join(tmp, "audiobook_test.log")
+            view, _ = self.make_view(log_path=log_path)
+            view.handle_event({"kind": "starting", "name": "sglomni",
+                               "log_path": "/tmp/sglomni-server.log"})
+            view.handle_event({"kind": "port_taken", "name": "sglomni",
+                               "taken": 8100, "moved": 37183,
+                               "message": "moved from port 8100",
+                               "log_tail": []})
+            with open(log_path, encoding="utf-8") as logf:
+                text = logf.read()
+        self.assertIn("ERROR - moved from port 8100", text)
+        self.assertIn("the server's own output is in "
+                      "/tmp/sglomni-server.log", text)
+
     def test_boot_failure_is_recorded_in_the_dated_log(self):
         # A failed boot never reaches the converter, so without this the
         # dated log the failure pointers name would stay blank.
diff --git a/app/tests/test_tts_sglomni.py b/app/tests/test_tts_sglomni.py
index 85327e1..0e3a7de 100644
--- a/app/tests/test_tts_sglomni.py
+++ b/app/tests/test_tts_sglomni.py
@@ -191,6 +191,7 @@ class PayloadTests(unittest.TestCase):
         client.instructions = kwargs.get("instructions", "")
         client.language = "English"
         client._seed = None
+        client._kv_fit = None
         return client
 
     def test_speaker_payload_sends_the_preset_name(self):
@@ -256,10 +257,19 @@ class PayloadTests(unittest.TestCase):
 
     def test_higgs_payload_raises_the_generation_cap(self):
         """Higgs's 2048-frame engine default caps a request at ~27 s
-        (75 fps), below a full 250-word sub-chunk."""
+        (75 fps); the catalog raises it to the most its admission window
+        allows (~40 s after the prompt tokens)."""
         client = self._make_client("higgs_audio_v3_tts")
         payload = client._request_payload("Hello.")
-        self.assertEqual(payload["max_new_tokens"], 12288)
+        self.assertEqual(payload["max_new_tokens"], 3000)
+
+    def test_payload_keeps_a_learned_kv_fit(self):
+        """A capacity learned from an admission rejection caps later
+        requests below the catalog value."""
+        client = self._make_client("higgs_audio_v3_tts")
+        client._kv_fit = 2500
+        self.assertEqual(client._request_payload("Hello.")["max_new_tokens"],
+                         2500)
 
     def test_models_without_a_cap_send_no_max_new_tokens(self):
         client = self._make_client("moss_tts")
@@ -293,6 +303,126 @@ class RequestErrorTests(unittest.TestCase):
         self.assertIsInstance(error, NonRetryableTTSError)
 
 
+_KV_REJECTION_BODY = json.dumps({"error": {
+    "message": "Request requires more tokens than the thinker KV cache "
+               "can hold (input_tokens=684, max_new_tokens=12288, "
+               "required_tokens=12972, kv_capacity=4095). Current "
+               "mem_fraction_static is 0.800; try setting "
+               "--thinker-mem-fraction-static higher.",
+    "type": "InternalServerError", "code": 500}})
+
+
+def _http_error(code: int, body: str) -> urllib.error.HTTPError:
+    return urllib.error.HTTPError(
+        "http://127.0.0.1:8100/v1/audio/speech", code, "error",
+        hdrs=None, fp=io.BytesIO(body.encode("utf-8")))
+
+
+def _speech_response():
+    response = MagicMock()
+    response.__enter__.return_value = response
+    response.read.return_value = _WAV_BYTES
+    return response
+
+
+class KvAdmissionTests(unittest.TestCase):
+    """The KV-window admission rejection refits max_new_tokens once."""
+
+    def _client(self):
+        client = SgOmniTTSClient.__new__(SgOmniTTSClient)
+        from backends.sglomni.catalog import entry_by_key
+        client.entry = entry_by_key("higgs_audio_v3_tts")
+        client.api_url = "http://127.0.0.1:8100"
+        client.voice = None
+        client.ref_audio = None
+        client.ref_text = ""
+        client.instructions = ""
+        client.language = "English"
+        client._seed = None
+        client.chunk_size = None
+        client._kv_fit = None
+        return client
+
+    def test_fit_is_parsed_from_the_server_message(self):
+        fit = self._client()._kv_admission_fit(_KV_REJECTION_BODY)
+        # kv_capacity 4095 - input 684 - the 64-frame margin.
+        self.assertEqual(fit, 3347)
+
+    def test_fit_is_cached_for_later_requests(self):
+        client = self._client()
+        client._kv_admission_fit(_KV_REJECTION_BODY)
+        client._kv_admission_fit(_KV_REJECTION_BODY)
+        self.assertEqual(client._kv_fit, 3347)
+
+    def test_unrelated_errors_do_not_fit(self):
+        client = self._client()
+        self.assertIsNone(client._kv_admission_fit("CUDA out of memory"))
+        self.assertIsNone(client._kv_fit)
+
+    def test_a_window_below_the_floor_raises_with_guidance(self):
+        body = json.dumps({"error": {"message":
+            "Request requires more tokens than the thinker KV cache can "
+            "hold (input_tokens=4000, max_new_tokens=12288, "
+            "required_tokens=16288, kv_capacity=4095).", "code": 500}})
+        with self.assertRaises(NonRetryableTTSError) as ctx:
+            self._client()._kv_admission_fit(body)
+        self.assertIn("shorter reference clip", str(ctx.exception))
+
+    def test_request_wav_refits_and_resends_once(self):
+        client = self._client()
+        with patch(
+                "converter.clients.sglomni.urllib.request.urlopen",
+                side_effect=[_http_error(500, _KV_REJECTION_BODY),
+                             _speech_response()]) as mock_open:
+            wav = client._request_wav("Hello.")
+        self.assertEqual(wav, _WAV_BYTES)
+        self.assertEqual(mock_open.call_count, 2)
+        refit = json.loads(mock_open.call_args[0][0].data)
+        self.assertEqual(refit["max_new_tokens"], 3347)
+
+    def test_request_wav_surfaces_a_refit_that_fails_again(self):
+        client = self._client()
+        with patch(
+                "converter.clients.sglomni.urllib.request.urlopen",
+                side_effect=[_http_error(500, _KV_REJECTION_BODY),
+                             _http_error(500, _KV_REJECTION_BODY)]):
+            with self.assertRaises(RuntimeError) as ctx:
+                client._request_wav("Hello.")
+        message = str(ctx.exception)
+        self.assertIn("HTTP 500", message)
+        self.assertIn("thinker KV cache", message)
+        self.assertNotIsInstance(ctx.exception, NonRetryableTTSError)
+
+    def test_request_wav_does_not_refit_other_errors(self):
+        client = self._client()
+        with patch(
+                "converter.clients.sglomni.urllib.request.urlopen",
+                side_effect=[_http_error(500, "CUDA out of memory")]):
+            with self.assertRaises(RuntimeError) as ctx:
+                client._request_wav("Hello.")
+        self.assertIn("CUDA out of memory", str(ctx.exception))
+        self.assertIsNone(client._kv_fit)
+
+    def test_request_wav_keeps_the_refit_across_sub_requests(self):
+        # A tight window (a long reference clip): the fit binds below the
+        # 3000-frame catalog cap, and every later request carries it.
+        tight_body = json.dumps({"error": {"message":
+            "Request requires more tokens than the thinker KV cache can "
+            "hold (input_tokens=1500, max_new_tokens=3000, "
+            "required_tokens=4500, kv_capacity=4095).", "code": 500}})
+        client = self._client()
+        with patch(
+                "converter.clients.sglomni.urllib.request.urlopen",
+                side_effect=[_http_error(500, tight_body),
+                             _speech_response(),
+                             _speech_response()]) as mock_open:
+            client._request_wav("Hello.")
+            client._request_wav("Hello again.")
+        self.assertEqual(mock_open.call_count, 3)
+        second = json.loads(mock_open.call_args[0][0].data)
+        self.assertEqual(second["max_new_tokens"], 2531)
+
+
 class GenerateChunkTests(unittest.TestCase):
     """Chunk generation: WAV output, sub-chunking, bookkeeping."""
 
@@ -315,6 +445,8 @@ class GenerateChunkTests(unittest.TestCase):
         client.instructions = ""
         client.language = "English"
         client._seed = None
+        client.chunk_size = None
+        client._kv_fit = None
         return client
 
     def _read_wav(self, path):
@@ -347,6 +479,18 @@ class GenerateChunkTests(unittest.TestCase):
         self.assertEqual(len(args[0]), 3)
         self.assertEqual(args[1], Path(result))
 
+    def test_run_chunk_size_caps_the_sub_requests(self):
+        """The pre-flight clamp (a chunk_words-capped model's popup
+        answer) overrides CHUNK_SIZE for this run."""
+        client = self._make_client()
+        client.chunk_size = 10
+        text = " ".join(f"word{i}" for i in range(24))
+        with patch.object(client, "_request_wav",
+                          return_value=_WAV_BYTES) as mock_wav, \
+                patch("converter.clients.sglomni.concat_audio_files"):
+            client.generate_chunk(text, 1)
+        self.assertEqual(mock_wav.call_count, 3)
+
     def test_single_subchunk_skips_concatenation(self):
         client = self._make_client()
         with patch.object(client, "_request_wav", return_value=_WAV_BYTES), \