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

16 files changed, 1036 insertions, 45 deletions

diff --git a/app/backends/servers.py b/app/backends/servers.py
index d352717..fd28866 100644
--- a/app/backends/servers.py
+++ b/app/backends/servers.py
@@ -23,6 +23,7 @@ boot screen, which renders the same events. Pid/log files live under
 """
 
 import os
+import re
 import signal
 import subprocess
 import sys
@@ -62,6 +63,12 @@ _BOOT_HINTS = (
      "re-install the model via Configure Backends (checking the model "
      "installs its companion packages), or pip-install it into the "
      "backend's venv manually"),
+    # A launcher that could not take its configured port — uvicorn's bind
+    # failure dies outright, sglang-omni's silently moves to a random one
+    # (the live detection for that is _PORT_FALLBACK_RE below).
+    ("already in use",
+     "the configured port is held by another process — stop that process "
+     "or move this server to a free port, then start it again"),
 )
 
 # Progress callback: called with an event dict. KIND is one of:
@@ -70,6 +77,10 @@ _BOOT_HINTS = (
 #   "ready"     {name, url}                       server is up and answering
 #   "exited"    {name, returncode, log_tail, hint}  process died while booting
 #   "timeout"   {name, seconds, log_tail, hint}   readiness deadline elapsed
+#   "port_taken" {name, taken, moved, message, log_tail}
+#                                                 launcher moved the boot
+#                                                 to a random port (the
+#                                                 configured one was taken)
 #   "running"   {name, url}                       already up (no spawn)
 #   "cancelled" {name}                            boot aborted via cancel
 #   "error"     {message}                         could not spawn the executable
@@ -104,6 +115,9 @@ def _console_progress(event: dict) -> None:
         _print_tail(event.get("log_tail"))
         if event.get("hint"):
             print(f"[WARNING] hint: {event['hint']}")
+    elif kind == "port_taken":
+        print(f"[ERROR] {event['message']}")
+        _print_tail(event.get("log_tail"))
     elif kind == "error":
         print(f"[ERROR] {event['message']}")
 
@@ -154,6 +168,62 @@ def _print_tail(tail: List[str]) -> None:
         print("---")
 
 
+# sglang-omni's launcher, finding the requested port taken, silently binds
+# a random one instead — the boot then stays healthy but unreachable at
+# the configured URL (every client keeps polling the taken port until the
+# start timeout). Scanned from the boot log so the boot fails in seconds
+# with both port numbers named instead.
+_PORT_FALLBACK_RE = re.compile(
+    r"Port (?P<taken>\d+) is already in use.*?"
+    r"Using port (?P<moved>\d+) instead", re.DOTALL)
+
+# How much of the previous log read is rescanned with the next chunk, so a
+# fallback message split across two reads still matches (the launcher
+# prints its two lines back to back; half a kilobyte is generous).
+_LOG_CARRY_BYTES = 512
+
+
+def _read_new_log(path: Path, offset: int) -> tuple:
+    """Read the bytes appended to the server log since OFFSET (best effort).
+
+    Returns ``(new_offset, text)`` — where to resume and what was read
+    (undecodable bytes replaced). An unreadable or unchanged file yields
+    the offset unchanged and ""; a shrunken file (truncated or rotated)
+    restarts from zero so nothing new is skipped.
+    """
+    try:
+        size = path.stat().st_size
+    except OSError:
+        return offset, ""
+    if size < offset:
+        offset = 0
+    if size == offset:
+        return offset, ""
+    try:
+        with path.open("rb") as fh:
+            fh.seek(offset)
+            raw = fh.read()
+    except OSError:
+        return offset, ""
+    return offset + len(raw), raw.decode("utf-8", errors="replace")
+
+
+def _port_taken_message(spec) -> str:
+    """The refusal message for spawning while SPEC's URL already answers."""
+    return (f"another process is listening at {spec.url} but it is not a "
+            f"usable {spec.name} server — stop that process (or move this "
+            f"server to a free port) and start again")
+
+
+def _port_fallback_message(spec, taken: str, moved: str) -> str:
+    """The boot-failure message when the launcher moved ports on us."""
+    return (f"the {spec.name} launcher moved the server from port "
+            f"{taken} (already in use by another process) to port {moved}; "
+            f"clients poll {taken}, so this boot cannot become ready — "
+            f"stop whatever holds port {taken} (or move this server to a "
+            f"free port) and start again")
+
+
 def _pid_alive(pid: int) -> bool:
     """True when a process with PID is still running (POSIX signal-0 probe)."""
     if sys.platform == "win32":
@@ -286,15 +356,19 @@ def _kill_pid(pid: int) -> bool:
     return True
 
 
-def _server_ready(spec) -> bool:
+def _server_ready(spec, listening: Optional[bool] = None) -> bool:
     """True when the server described by SPEC is usable, not just listening.
 
     Without an IDENTITY this is the plain TCP-connect check. With one, the
     server must also answer HTTP as that backend (``probe.identify_server``);
     for the faster backend (whose model loads after the port opens) the
-    ``/health`` model_loaded flag must additionally be true.
+    ``/health`` model_loaded flag must additionally be true. LISTENING, when
+    given, is a caller's fresh ``common.server_running`` result — reused so
+    one start pass probes the port only once.
     """
-    if not common.server_running(spec.url):
+    if listening is None:
+        listening = common.server_running(spec.url)
+    if not listening:
         return False
     identity = getattr(spec, "identity", None)
     if identity is None:
@@ -320,6 +394,13 @@ def start(spec, progress: ProgressCallback = None,
     exit reports the log tail and returns False. A no-op (True) when the
     server is already running.
 
+    Two port-conflict guards fail fast instead of letting a doomed boot
+    run out the clock: a foreign process already listening on the spec's
+    URL (but not answering as the backend) refuses the spawn outright, and
+    a launcher that logs a port fallback mid-boot ("Port N is already in
+    use ... Using port M instead" — sglang-omni's) aborts the boot with a
+    "port_taken" event and kills the misdirected server.
+
     PROGRESS, when given, receives each boot event (see ProgressCallback);
     the default ``_console_progress`` prints them, preserving the old console
     output. CANCEL (a threading.Event) aborts the boot: the spawned process
@@ -333,7 +414,11 @@ def start(spec, progress: ProgressCallback = None,
                 "message": f"server executable not found: {exe}. Run 'Set "
                            "up a backend' first."})
         return False
-    if _server_ready(spec):
+    # One TCP probe feeds both checks: an already-usable server takes the
+    # "running" path, and a listener that is NOT usable (identity probe
+    # failed) is exactly the foreign-holder conflict refused below.
+    listening = common.server_running(spec.url)
+    if _server_ready(spec, listening):
         report({"kind": "running", "name": spec.name, "url": spec.url})
         return True
 
@@ -350,6 +435,15 @@ def start(spec, progress: ProgressCallback = None,
                            f"{pid_for(spec.name)}) is already starting or "
                            "running; stop it first"})
         return False
+    # Refuse to spawn onto a port a foreign process already holds: TCP-up
+    # but probe-down means the listener is not a usable instance of this
+    # server. A fresh spawn would then either die on the bind or (launchers
+    # that fall back silently, like sglang-omni) move to a random port and
+    # leave every client polling the taken one — the boot watchdog below
+    # catches that late, so name the conflict here.
+    if listening:
+        report({"kind": "error", "message": _port_taken_message(spec)})
+        return False
     pid_file = _pid_path(spec.name)
     if pid_file.exists():
         try:
@@ -419,6 +513,13 @@ def start(spec, progress: ProgressCallback = None,
     started = time.time()
     next_heartbeat = started + 15
     deadline = started + start_timeout
+    # Boot-log watchdog state: scan only what this boot appends (the log
+    # file accumulates across boots, so a previous boot's fallback lines
+    # must not re-fire here), carrying a tail between reads so a fallback
+    # message split across two polls still matches.
+    log_path = _log_path(spec.name)
+    log_offset = log_path.stat().st_size if log_path.exists() else 0
+    log_carry = ""
     while time.time() < deadline:
         if cancel is not None and cancel.is_set():
             # User cancelled while booting: kill what we spawned (the
@@ -441,6 +542,31 @@ def start(spec, progress: ProgressCallback = None,
             except OSError:
                 pass
             return False
+        # Watchdog: a launcher that silently moved to another port (the
+        # configured one was taken) keeps booting healthily where no
+        # client will ever call it — abort now instead of polling the
+        # taken port until the timeout.
+        log_offset, new_text = _read_new_log(log_path, log_offset)
+        if new_text:
+            scan = log_carry + new_text
+            match = _PORT_FALLBACK_RE.search(scan)
+            if match is not None:
+                # Kill the misdirected server: it would serve on a port
+                # no client will call while holding GPU memory.
+                _kill_pid(proc.pid)
+                try:
+                    pid_file.unlink()
+                except OSError:
+                    pass
+                report({"kind": "port_taken", "name": spec.name,
+                        "taken": int(match.group("taken")),
+                        "moved": int(match.group("moved")),
+                        "message": _port_fallback_message(
+                            spec, match.group("taken"),
+                            match.group("moved")),
+                        "log_tail": _read_log_tail(spec.name)})
+                return False
+            log_carry = scan[-_LOG_CARRY_BYTES:]
         if _server_ready(spec):
             report({"kind": "ready", "name": spec.name, "url": spec.url})
             return True
diff --git a/app/backends/sglomni/catalog.py b/app/backends/sglomni/catalog.py
index 1e4ca2d..160eb8b 100644
--- a/app/backends/sglomni/catalog.py
+++ b/app/backends/sglomni/catalog.py
@@ -69,11 +69,20 @@ class ModelEntry:
     # 86.13 fps, ~12 s) and Higgs's to 2048 frames (75 fps, ~27 s, and
     # per-request values are clamped to the engine cap, so its vendored
     # config raises the cap too) — and silently truncate longer text.
-    # 12288 frames covers a full 250-word CHUNK_SIZE sub-chunk on both and
-    # stays under the KV-pool admission check on every GPU that can host
-    # the model (the scheduler rejects prompt + max_new_tokens above it,
-    # and these requests are not auto-clamped).
+    # 12288 frames covers a full 250-word CHUNK_SIZE sub-chunk on ZONOS2
+    # (whose KV pools are >= 28083 tokens on every GPU that can host it)
+    # but NOT on Higgs: its thinker engine pins context_length at 4096,
+    # and the scheduler rejects any request whose prompt tokens plus
+    # max_new_tokens exceed that window (kv_capacity=4095, on every GPU —
+    # the pool side is never the binding constraint there).
     max_new_tokens: Optional[int] = None
+    # The largest sub-request (words) the model can narrate within its
+    # admission window, for models whose engine cannot cover a full
+    # CHUNK_SIZE sub-chunk (None = no cap; config.CHUNK_SIZE stands).
+    # Higgs: 3000 frames ≈ 40 s at 75 fps ≈ 80 words of narration, and
+    # the run pre-flight offers to clamp CHUNK_SIZE for the run (the
+    # client's adaptive retry still rescues requests the window rejects).
+    chunk_words: Optional[int] = None
 
 
 # The Qwen3-TTS CustomVoice speaker table — the same built-in speakers the
@@ -177,9 +186,18 @@ ENTRIES: Tuple[ModelEntry, ...] = (
         requires_reference=False,
         # The engine's 2048-frame default is ~27 s of speech at the codec's
         # 75 fps; requests are clamped to the engine cap server-side, so the
-        # yaml raises the cap and every request carries 12288 frames
-        # (~164 s) — enough for a full CHUNK_SIZE sub-chunk.
-        max_new_tokens=12288,
+        # yaml raises the cap and every request carries 3000 frames (~40 s)
+        # — the most the admission window allows: upstream pins the thinker
+        # engine's context_length at 4096, and the scheduler rejects any
+        # request whose prompt (including the reference-audio tokens) plus
+        # max_new_tokens exceeds it. 3000 frames leaves ~1095 tokens of
+        # prompt headroom (an 80-word chunk with a 20.5 s reference
+        # measured 684). Sub-requests cap at 80 words so the text fits the
+        # window too — the pre-flight offers to clamp CHUNK_SIZE for the
+        # run, and the client refits rejected requests to whatever the
+        # server reports as its capacity.
+        max_new_tokens=3000,
+        chunk_words=80,
         notes="zero-shot narration, cloning from a reference clip",
     ),
     ModelEntry(
diff --git a/app/backends/sglomni/configs/higgs_audio_v3_tts.yaml b/app/backends/sglomni/configs/higgs_audio_v3_tts.yaml
index 74736fa..8a42771 100644
--- a/app/backends/sglomni/configs/higgs_audio_v3_tts.yaml
+++ b/app/backends/sglomni/configs/higgs_audio_v3_tts.yaml
@@ -9,22 +9,27 @@
 # "CUDA out of memory. Tried to allocate 14.00 MiB".
 #
 # 0.80 trims the engine's static pool by ~1.2 GB per 24 GB of VRAM while
-# leaving a KV cache pool (~10 GB on a 24 GB card) far larger than any
-# narration request needs. Cards with heavy other-GPU-process usage can go
-# lower (e.g. 0.75).
+# leaving a KV cache pool far larger than any narration request needs.
+# Cards with heavy other-GPU-process usage can go lower (e.g. 0.75).
 #
 # The tts_engine factory also caps every request at max_new_tokens=2048
 # audio frames, and per-request values are clamped to that cap server-side
 # (make_higgs_scheduler_adapters) — the Higgs codec runs 75 frames per
 # second (24 kHz / 320 downsample), so the default is ~27 s of speech, which
 # silently truncates this tool's full 250-word sub-chunks (~100 s). Raising
-# the factory cap is the only way past it; the catalog also sends
-# max_new_tokens=12288 per request (the same value ZONOS2 uses) so a request
-# may use the room: 12288 frames ≈ 164 s.
+# the factory cap is the only way past it, but the ceiling is hard: upstream
+# pins the thinker engine's context_length at 4096 (HiggsTtsEngineBuilder —
+# not overridable), and the scheduler rejects any request whose prompt
+# tokens (including the reference-audio tokens) plus max_new_tokens exceed
+# that window ("Request requires more tokens than the thinker KV cache can
+# hold", kv_capacity=4095, on every GPU). The cap therefore lands at 3000
+# frames ≈ 40 s — the most the window allows with prompt headroom (an
+# 80-word chunk with a 20.5 s reference measured 684 prompt tokens) — and
+# the catalog caps sub-requests at 80 words to match (chunk_words).
 config_cls: HiggsTtsPipelineConfig
 model_path: bosonai/higgs-audio-v3-tts-4b
 stages:
   tts_engine:
     gpu_memory_fraction: 0.80
     factory:
-      max_new_tokens: 12288
+      max_new_tokens: 3000
diff --git a/app/converter/clients/sglomni.py b/app/converter/clients/sglomni.py
index d4b756f..14c9990 100644
--- a/app/converter/clients/sglomni.py
+++ b/app/converter/clients/sglomni.py
@@ -26,6 +26,7 @@ mandatory (Qwen3-TTS Base, dots.tts, ZONOS2 — those refuse at connect).
 import base64
 import json
 import logging
+import re
 import tempfile
 import urllib.error
 import urllib.parse
@@ -64,6 +65,21 @@ _MIME_BY_SUFFIX = {
 _NON_RETRYABLE_TYPES = ("BadRequestError", "InvalidRequestError",
                         "NotFoundError", "PermissionDeniedError")
 
+# The scheduler's KV-window admission error ("Request requires more tokens
+# than the thinker KV cache can hold (input_tokens=684, max_new_tokens=
+# 12288, required_tokens=12972, kv_capacity=4095)..."): the server names
+# the numbers a refit needs, and upstream classifies the message as a
+# deterministic bad request — the identical request fails on every retry,
+# so the only useful response is to send a smaller one.
+_KV_ADMISSION_MARKER = "thinker KV cache can hold"
+
+# Frames kept below the capacity the server reported, and the smallest
+# refitted cap worth generating with (~14 s of speech at 75 fps): below
+# the floor the request would truncate almost immediately, so the run
+# surfaces guidance instead of near-empty audio.
+_KV_FIT_MARGIN = 64
+_KV_FIT_FLOOR = 1024
+
 
 def _is_loopback(url: str) -> bool:
     """True when URL's host is this machine (the server can read local
@@ -82,6 +98,20 @@ def _data_url(path: Path) -> str:
     return f"data:{mime};base64,{encoded}"
 
 
+def _http_error_detail(exc: urllib.error.HTTPError) -> str:
+    """The error response body as text (empty when it cannot be read)."""
+    try:
+        return exc.read().decode("utf-8", errors="replace")
+    except Exception:
+        return ""
+
+
+def _kv_error_number(detail: str, name: str) -> Optional[int]:
+    """The integer NAME=... reports in a KV-window admission message."""
+    match = re.search(rf"\b{name}=(\d+)", detail)
+    return int(match.group(1)) if match else None
+
+
 class SgOmniTTSClient(BaseTTSClient):
     """Generates audio chunks through an SGLang-Omni server."""
 
@@ -94,6 +124,7 @@ class SgOmniTTSClient(BaseTTSClient):
                  instructions: Optional[str] = None,
                  language: Optional[str] = None,
                  api_url: Optional[str] = None,
+                 chunk_size: Optional[int] = None,
                  quiet: bool = False, cancel=None):
         super().__init__(chunks_dir, quiet=quiet, cancel=cancel)
         # The catalog entry this run targets (the backend package validates
@@ -114,6 +145,13 @@ class SgOmniTTSClient(BaseTTSClient):
         self.ref_text = (ref_text or "").strip()
         self.skip_transcription = skip_transcription
         self.instructions = (instructions or "").strip()
+        # Per-run sub-request word cap (the pre-flight chunk popup's "set
+        # chunk" answer); None follows config.CHUNK_SIZE.
+        self.chunk_size = chunk_size
+        # A KV-window capacity the server taught us via an admission
+        # rejection (None = none learned): later requests keep their
+        # max_new_tokens under it. See _kv_admission_fit.
+        self._kv_fit = None
         # Seed sent with every request: config.SEED as-is, or (with
         # CONSTANT_SEED and SEED < 0) one random value drawn per run and
         # reused for every chunk so the voice stays consistent across
@@ -278,8 +316,13 @@ class SgOmniTTSClient(BaseTTSClient):
             # Models whose engine caps a request below what a full
             # sub-chunk can narrate (Zonos2's 1024-frame default is ~12 s):
             # raise the ceiling per request. Generation still stops at
-            # natural EOS, so an unused margin costs nothing.
-            payload["max_new_tokens"] = entry.max_new_tokens
+            # natural EOS, so an unused margin costs nothing. A capacity
+            # learned from an admission rejection (Higgs pins the window)
+            # keeps later requests under it too.
+            cap = entry.max_new_tokens
+            if self._kv_fit is not None:
+                cap = min(cap, self._kv_fit)
+            payload["max_new_tokens"] = cap
         if entry.capability == "design":
             payload["task_type"] = "VoiceDesign"
             payload["instructions"] = self.instructions
@@ -291,22 +334,42 @@ class SgOmniTTSClient(BaseTTSClient):
 
     def _request_wav(self, text: str) -> bytes:
         """POST one sub-chunk and return the complete WAV bytes."""
+        payload = self._request_payload(text)
+        try:
+            return self._post_speech(payload)
+        except urllib.error.HTTPError as exc:
+            # A KV-window rejection is deterministic (upstream maps it to a
+            # bad request): refit the generation cap to the capacity the
+            # server reported and resend once before surfacing anything.
+            detail = _http_error_detail(exc)
+            fitted = (self._kv_admission_fit(detail)
+                      if "max_new_tokens" in payload else None)
+            if fitted is not None:
+                try:
+                    return self._post_speech(
+                        dict(payload, max_new_tokens=fitted))
+                except urllib.error.HTTPError as retry_exc:
+                    exc = retry_exc
+                    detail = _http_error_detail(exc)
+            raise self._request_error(exc.code, detail) from exc
+        except urllib.error.URLError as exc:
+            raise RuntimeError(
+                f"SGLang-Omni request failed: {exc.reason}") from exc
+
+    def _post_speech(self, payload: dict) -> bytes:
+        """POST PAYLOAD to /v1/audio/speech; HTTPErrors propagate raw."""
         url = f"{self.api_url}/v1/audio/speech"
-        payload = json.dumps(self._request_payload(text)).encode("utf-8")
         request = urllib.request.Request(
-            url, data=payload,
+            url, data=json.dumps(payload).encode("utf-8"),
             headers={"Content-Type": "application/json"}, method="POST")
         try:
             with urllib.request.urlopen(request,
                                         timeout=config.API_TIMEOUT) as response:
                 wav = response.read()
-        except urllib.error.HTTPError as exc:
-            detail = ""
-            try:
-                detail = exc.read().decode("utf-8", errors="replace")
-            except Exception:
-                pass
-            raise self._request_error(exc.code, detail) from exc
+        except urllib.error.HTTPError:
+            # Re-raise raw (HTTPError subclasses URLError): the caller maps
+            # it — and refits KV-window rejections — from the status code.
+            raise
         except urllib.error.URLError as exc:
             raise RuntimeError(
                 f"SGLang-Omni request failed: {exc.reason}") from exc
@@ -314,6 +377,37 @@ class SgOmniTTSClient(BaseTTSClient):
             raise RuntimeError("SGLang-Omni server returned empty audio")
         return wav
 
+    def _kv_admission_fit(self, detail: str) -> Optional[int]:
+        """A refitted max_new_tokens for a KV-window rejection, or None.
+
+        DETAIL is the error response body. The server's message names the
+        request's prompt length and the KV window it must fit; the refit
+        keeps a small margin below the window, is remembered for this
+        client's remaining sub-requests, and the refitted request carries
+        it. When the window leaves less than a useful minimum after the
+        prompt (a very long reference clip), the run fails with guidance
+        instead of near-empty audio.
+        """
+        if _KV_ADMISSION_MARKER not in detail:
+            return None
+        input_tokens = _kv_error_number(detail, "input_tokens")
+        kv_capacity = _kv_error_number(detail, "kv_capacity")
+        if input_tokens is None or kv_capacity is None:
+            return None
+        fitted = kv_capacity - input_tokens - _KV_FIT_MARGIN
+        if fitted < _KV_FIT_FLOOR:
+            raise NonRetryableTTSError(
+                f"SGLang-Omni rejected the request: the model's KV window "
+                f"({kv_capacity} tokens) leaves {fitted} frames after this "
+                f"request's prompt ({input_tokens} tokens) — too little to "
+                "narrate anything useful. Use a shorter reference clip or "
+                "a smaller Chunk Size setting; the server caps prompt plus "
+                "generation at that window for every request.")
+        if self._kv_fit is not None:
+            fitted = min(fitted, self._kv_fit)
+        self._kv_fit = fitted
+        return fitted
+
     def _request_error(self, status: int, detail: str) -> Exception:
         """Map the OpenAI-style error envelope to the retry decision.
 
@@ -347,14 +441,17 @@ class SgOmniTTSClient(BaseTTSClient):
     def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]:
         """Generate one audio chunk; returns its path in the chunks folder.
 
-        The text is split into sub-requests of at most
-        ``config.CHUNK_SIZE`` words each (the book-level chunker normally
+        The text is split into sub-requests of at most ``chunk_size`` words
+        each — the per-run cap the pre-flight chunk popup sets for models
+        whose engine cannot narrate a full CHUNK_SIZE sub-chunk (Higgs),
+        else ``config.CHUNK_SIZE`` (the book-level chunker normally
         guarantees this already; the split is defense in depth against
-        pathological input such as a punctuation-free run of text), and
+        pathological input such as a punctuation-free run of text) — and
         the returned WAV files are concatenated into one chunk file.
         """
         try:
-            sub_chunks = split_into_chunks(text, max_words=config.CHUNK_SIZE)
+            sub_chunks = split_into_chunks(
+                text, max_words=self.chunk_size or config.CHUNK_SIZE)
             if not sub_chunks:
                 raise RuntimeError("No text to synthesize")
 
diff --git a/app/converter/converter.py b/app/converter/converter.py
index 0769258..3b878c8 100644
--- a/app/converter/converter.py
+++ b/app/converter/converter.py
@@ -220,6 +220,73 @@ def prompt_overwrite(existing: List[Path], output_name: str,
         print("Please answer 'y' or 'n' (or press Enter for yes).")
 
 
+class ChunkClampCancelled(Exception):
+    """The chunk-cap popup's Cancel answer: the run stops unstarted."""
+
+
+def chunk_clamp_needed(entry) -> bool:
+    """True when ENTRY cannot narrate a full CHUNK_SIZE sub-request.
+
+    A catalog entry declares ``chunk_words`` when its engine's admission
+    window caps one request below what the configured CHUNK_SIZE can
+    narrate (Higgs: the server pins each request's prompt plus
+    generation at 4096 tokens).
+    """
+    return (entry is not None and entry.chunk_words is not None
+            and entry.chunk_words < config.CHUNK_SIZE)
+
+
+def chunk_clamp_message(entry) -> List[str]:
+    """The popup text for a chunk_words-capped ENTRY (short lines)."""
+    return [
+        f"{entry.label} can narrate at most ~{entry.chunk_words} words "
+        "per request: the server caps",
+        "each request's prompt plus generation at a fixed window. "
+        "Longer sub-chunks",
+        "may cut off mid-sentence.",
+    ]
+
+
+def prompt_chunk_clamp(entry, ask=None) -> Optional[int]:
+    """The per-run sub-request word cap for ENTRY (None = no clamp).
+
+    Models whose catalog entry declares ``chunk_words`` below CHUNK_SIZE
+    (Higgs) cannot narrate a full sub-chunk in one request; the popup
+    offers to clamp CHUNK_SIZE for this run, keep it ("try anyway",
+    risking mid-chunk truncation), or cancel the run. ASK, when given,
+    replaces the console prompt: it is called with (message lines, words)
+    and returns "clamp" or "anyway" (Cancel raises inside the callback —
+    the hub maps it to returning to the Generate form). A closed stdin
+    (non-interactive run) clamps: unattended runs keep working and never
+    produce silently truncated audio.
+    """
+    if not chunk_clamp_needed(entry):
+        return None
+    lines = chunk_clamp_message(entry)
+    if ask is not None:
+        answer = ask(lines, entry.chunk_words)
+        return entry.chunk_words if answer == "clamp" else None
+    for line in lines:
+        print(line)
+    while True:
+        try:
+            answer = input(
+                f"Set Chunk to {entry.chunk_words} for this run, try "
+                "anyway, or cancel? [S/t/c]: ").strip().lower()
+        except EOFError:
+            print(f"\n[WARNING] No interactive input available; clamping "
+                  f"sub-requests to {entry.chunk_words} words for this run")
+            return entry.chunk_words
+        if answer in ("", "s", "set"):
+            return entry.chunk_words
+        if answer in ("t", "try"):
+            return None
+        if answer in ("c", "cancel"):
+            raise ChunkClampCancelled(
+                "Conversion cancelled at the chunk-size prompt")
+        print("Please answer 's', 't', or 'c'.")
+
+
 class AudiobookConverter:
     """Audiobook converter using a local TTS API."""
 
@@ -236,6 +303,7 @@ class AudiobookConverter:
                  instructions: Optional[str] = None,
                  request_options: Optional[Dict[str, str]] = None,
                  api_url: Optional[str] = None,
+                 chunk_size: Optional[int] = None,
                  unload_models: Optional[bool] = None,
                  progress: Optional[Callable[[dict], None]] = None,
                  cancel=None):
@@ -324,13 +392,18 @@ class AudiobookConverter:
                         "catalog key for --model (see the backend docs).")
             model_id = entry.key
             self.model_id = model_id
+            # CHUNK_SIZE carries a per-run override (the pre-flight chunk
+            # popup's clamp for models whose engine caps one request below
+            # a full sub-chunk); None follows the config.CHUNK_SIZE setting.
+            self.chunk_size = chunk_size
             self.tts = SgOmniTTSClient(
                 chunks_dir=CHUNKS_FOLDER, model=model_id, voice=voice,
                 ref_audio=voice_clone_ref_audio,
                 ref_text=voice_clone_ref_text,
                 skip_transcription=skip_transcription,
                 instructions=instructions, language=self.language,
-                api_url=api_url, quiet=quiet, cancel=cancel)
+                api_url=api_url, chunk_size=self.chunk_size,
+                quiet=quiet, cancel=cancel)
         else:
             # Qwen: the voice mode picks the request shape (built-in
             # speaker, clone from a reference .wav, or a designed voice);
diff --git a/app/docs/backend-sglomni.md b/app/docs/backend-sglomni.md
index 4f553f9..58e5ce5 100644
--- a/app/docs/backend-sglomni.md
+++ b/app/docs/backend-sglomni.md
@@ -60,7 +60,7 @@ inside the start timeout.
 | `qwen3_tts_0_6b_base` | Qwen3-TTS 0.6B Base | clone (reference required) | |
 | `qwen3_tts_1_7b_base` | Qwen3-TTS 1.7B Base | clone (reference required) | higher quality |
 | `qwen3_tts_1_7b_voicedesign` | Qwen3-TTS 1.7B VoiceDesign | `--instructions` | |
-| `higgs_audio_v3_tts` | Higgs Audio v3 TTS | default voice or clone | launches with a vendored config: VRAM headroom for 24 GB cards + raised generation cap (requests carry `max_new_tokens=12288`; the engine's 2048-frame default is ~27 s and clamps per-request values) |
+| `higgs_audio_v3_tts` | Higgs Audio v3 TTS | default voice or clone | launches with a vendored config: VRAM headroom for 24 GB cards + a raised generation cap (requests carry `max_new_tokens=3000` — upstream pins the thinker engine's context length at 4096, so prompt + generation must fit 4095 tokens, ~40 s of speech — and sub-requests cap at 80 words, with a pre-run popup offering the CHUNK_SIZE clamp) |
 | `moss_tts` | MOSS-TTS v1.5 | clone (reference required) | |
 | `moss_tts_local` | MOSS-TTS Local v1.5 | default voice or clone | 48 kHz |
 | `voxtral_tts` | Voxtral TTS 4B | preset named voices | e.g. `default`, `casual_male` |
@@ -108,6 +108,28 @@ Unlike `audio.cpp` (server-side voice presets) the reference clip travels
 Reference clips live in the project's `voices/` directory (10-20 seconds of
 clean speech recommended).
 
+### Higgs Audio v3's request window
+
+The Higgs server pins its thinker engine's context length at 4096 tokens
+(not configurable), and the scheduler rejects any request whose prompt —
+including the reference-audio tokens — plus `max_new_tokens` exceeds that
+window ("Request requires more tokens than the thinker KV cache can hold",
+on every GPU: the KV pool side is never the binding constraint). One
+request can therefore narrate at most ~40 s of speech, so:
+
+- Requests carry `max_new_tokens=3000` (the vendored config raises the
+  engine's 2048-frame default to match), and sub-requests cap at 80 words.
+  Before a run with Higgs, a popup offers to clamp CHUNK_SIZE for the run
+  ("Set Chunk to 80"), keep the configured size (audio may cut off
+  mid-chunk), or cancel; CLI runs answer on the console (`[S/t/c]`, and a
+  closed stdin clamps so unattended runs never truncate silently).
+- When a request is still rejected (a long reference clip, for example),
+  the client refits `max_new_tokens` to the capacity the server reports
+  and resends once — the same request never retries as-is.
+- Voice cloning with a *long* reference clip eats into the same window:
+  keep references around 10-20 seconds (as recommended above) for the
+  most usable generation headroom.
+
 ## Server lifecycle
 
 The hub and the CLI start and stop the managed instance around each run
@@ -136,6 +158,18 @@ CLI's `--api-url`) at an `sgl-omni` instance. The hub discovers it via
 model from `GET /v1/models`; models uploaded to that server via
 `POST /v1/audio/voices` appear in its Voice menu.
 
+## Troubleshooting
+
+**The generate screen sits on "Status: starting", and the server log says
+`Port 8100 is already in use ... Using port 37183 instead`.** Something
+else already holds the configured port (often a stale `sgl-omni` from an
+earlier attempt). The upstream launcher does not fail — it silently moves
+the server to a random port, where no client ever looks for it. The app
+detects this in the boot log, aborts the boot, and names both ports.
+Find and stop the process holding the port
+(`ss -tlnp 'sport = :8100'`), or move this server to a free port
+(Settings → SGLang-Omni port), then start again.
+
 ## Manual setup
 
 ```bash
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), \
diff --git a/app/ui/hub.py b/app/ui/hub.py
index 13f17e7..7a6bfcc 100644
--- a/app/ui/hub.py
+++ b/app/ui/hub.py
@@ -1133,6 +1133,32 @@ def _tui_confirm(stdscr) -> Callable:
     return confirm
 
 
+def _tui_chunk_clamp(stdscr, entry) -> Optional[int]:
+    """The chunk-cap popup for an sglomni run (None = no clamp).
+
+    A model whose catalog entry caps a sub-request below CHUNK_SIZE
+    (Higgs: the server pins each request's prompt plus generation at a
+    fixed window) asks before the run: clamp CHUNK_SIZE for this run,
+    keep it (audio may cut off mid-chunk), or go back to the form. The
+    answer rides the command's kwargs (``chunk_size``) so the converter
+    and its client see it; the config.CHUNK_SIZE setting itself is never
+    rewritten.
+    """
+    if not converter_mod.chunk_clamp_needed(entry):
+        return None
+    words = entry.chunk_words
+    answer = tui.menu(
+        stdscr, "Generation cap",
+        [(f"Set Chunk to {words} (this run)", "clamp"),
+         ("Try anyway", "anyway"),
+         ("Cancel", "cancel")],
+        help_lines=converter_mod.chunk_clamp_message(entry),
+        back_value="cancel")
+    if answer == "cancel":
+        raise _BackToForm()
+    return words if answer == "clamp" else None
+
+
 def _check_preflight_plan(stdscr, book_files: list, planned: dict) -> bool:
     """The shared nothing-to-convert flashes; True when there is a plan.
 
@@ -1180,6 +1206,12 @@ def _preflight(stdscr, cmd: tuple) -> bool:
             confirm=_tui_confirm(stdscr))
     if not _check_preflight_plan(stdscr, book_files, {"": planned}):
         return False
+    if backend == BACKEND_SGLOMNI:
+        entry = sglomni_backend.entry_by_key((kwargs.get("model_id")
+                                              or "").strip())
+        clamp = _tui_chunk_clamp(stdscr, entry)
+        if clamp is not None:
+            kwargs["chunk_size"] = clamp
     kwargs["book_files"] = book_files
     kwargs["planned"] = planned
     return True
diff --git a/app/ui/runview.py b/app/ui/runview.py
index 4fcaeb1..73ad315 100644
--- a/app/ui/runview.py
+++ b/app/ui/runview.py
@@ -193,6 +193,19 @@ class RunView(ScreenView):
             self.boot_hint = event.get("hint") or ""
             self._record_boot_failure()
             self._finish("error")
+        elif kind == "port_taken":
+            # The launcher moved the server to a random port (the
+            # configured one was taken) and the boot was killed as
+            # unreachable: same terminal treatment as a crash, with the
+            # port specifics as the message.
+            self.server = "error"
+            self.server_message = str(event.get("message")
+                                      or "the server moved itself to "
+                                         "another port")
+            self.log_tail = list(event.get("log_tail") or [])
+            self.boot_hint = event.get("hint") or ""
+            self._record_boot_failure()
+            self._finish("error")
         elif kind == "cancelled":
             self.cancelled = True
             if self.server in ("starting", "ready", "processing"):
diff --git a/audiobook.py b/audiobook.py
index 88e8dca..302bb17 100755
--- a/audiobook.py
+++ b/audiobook.py
@@ -259,7 +259,7 @@ def convert(backend: str, voice: str = None, clone: str = None,
             input_file: Path = None, output_file: Path = None,
             progress=None, cancel=None, confirm=None,
             book_files=None, planned=None, manage_server: bool = False,
-            model_ids=None, model_voices=None,
+            chunk_size: int = None, model_ids=None, model_voices=None,
             planned_by_model=None) -> int:
     """Run one conversion pass with explicit options (used by the CLI and hub).
 
@@ -382,6 +382,17 @@ def convert(backend: str, voice: str = None, clone: str = None,
         model_id = sg_entry.key
         voice_mode = voice_mode_for(backend, voice, clone, instructions,
                                     model=model_id)
+        if planned is None:
+            # CLI runs pre-flight here: a model whose engine caps one
+            # request below a full CHUNK_SIZE sub-chunk (Higgs) gets the
+            # clamp popup before anything converts; Cancel stops the run
+            # unstarted. Hub runs asked in the TUI pre-flight and carry
+            # the answer as chunk_size.
+            try:
+                chunk_size = _converter_mod.prompt_chunk_clamp(sg_entry)
+            except _converter_mod.ChunkClampCancelled:
+                print("[INFO] Conversion cancelled; nothing was converted")
+                return 0
     else:
         # qwen: instructions design the voice (VoiceDesign model), a
         # reference .wav clones one (Base), otherwise a built-in speaker.
@@ -435,6 +446,7 @@ def convert(backend: str, voice: str = None, clone: str = None,
             language=language, backend=backend, voice=voice, debug=debug,
             model_id=model_id, instructions=instructions,
             request_options=request_options, api_url=api_url,
+            chunk_size=chunk_size,
             progress=progress, cancel=cancel,
         )
         converter._book_files = book_files