"""Start and stop TTS backend servers from the TUI hub. Each backend's ``detect()`` returns a list of ``ServerSpec`` — the exact argv (absolute binaries in the managed venv, no shell activation needed), the working directory to spawn it in, and the URL to probe for readiness. This module turns those specs into running processes: ``start`` spawns the server (streaming its output to ``app/logs/-server.log``, in the spec's cwd when it has one — audio.cpp resolves model_specs/ relative to its process working directory), records its pid, and polls the URL until it accepts connections (model loads are slow, so the timeout is generous). With a spec ``identity`` the poll also verifies the server answers HTTP as that backend, so readiness means "serving", not just "listening". Progress reporting goes through an optional ``progress`` callback (see ``_console_progress`` for the event shapes); the default callback prints the same lines as before, so the plain-console flow is unchanged. ``stop`` terminates the process group the hub started. Everything here can run in the plain console tail after the curses TUI returns (matching the wizards' build/pip streaming) or behind the run view's boot screen, which renders the same events. Pid/log files live under ``app/logs/`` which is already gitignored. """ import os import re import signal import subprocess import sys import time from datetime import datetime from pathlib import Path from typing import Callable, List, Optional from backends import common, probe from backends.common import LOG_DIR # How long to wait for a server to accept connections on its URL. First-time # model loads (especially qwen-tts / faster-qwen3-tts pulling weights into # VRAM) can take minutes, so this is deliberately generous. SERVER_START_TIMEOUT = 600 # Grace period after SIGTERM before escalating to SIGKILL (POSIX). STOP_GRACE_SECONDS = 10 # How often the start poll re-checks readiness (seconds). POLL_INTERVAL = 1 # Known crash signatures in a failed boot's log tail, each with a # plain-language hint appended to the failure report (the raw tail alone # is often a wall of framework traceback). _BOOT_HINTS = ( # sglang fused-MoE fp8e4nv kernel on pre-sm_89 GPUs (e.g. an FP8 # checkpoint or a default FP8 pipeline on Ampere). ("fp8e4nv not supported", "the server crashed compiling an FP8 MoE kernel: FP8 needs compute " "capability 8.9+ (RTX 4090/5090, Hopper) and cannot run on this GPU"), # A model companion package missing from the backend venv (e.g. the # sglang-omni Qwen3-TTS models need qwen-tts; a shared-cache weight # install or a failed pip run leaves it absent). ("ModuleNotFoundError", "the backend venv is missing a Python module this model needs — " "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: # "starting" {name, argv, cwd, log_path, pid} spawned, waiting for boot # "elapsed" {name, seconds} heartbeat while waiting # "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 ProgressCallback = Optional[Callable[[dict], None]] def _console_progress(event: dict) -> None: """Print PROGRESS events as the plain-console output (the old behavior).""" kind = event.get("kind") if kind == "starting": print(f"[INFO] starting {event['name']} server: {event['argv']}") if event.get("cwd"): print(f"[INFO] working directory: {event['cwd']}") print(f"[INFO] pid {event['pid']}; logs: {event['log_path']}") elif kind == "elapsed": print(f"[INFO] still waiting for the server ({int(event['seconds'])}s)...") elif kind == "ready": print(f"[OK] {event['name']} server is up on {event['url']}") elif kind == "running": print(f"[INFO] {event['name']} server already running on {event['url']}") elif kind == "cancelled": print(f"[INFO] {event['name']} server start cancelled") elif kind == "exited": print(f"[ERROR] {event['name']} server exited with code " f"{event['returncode']}") _print_tail(event.get("log_tail")) if event.get("hint"): print(f"[WARNING] hint: {event['hint']}") elif kind == "timeout": print(f"[ERROR] {event['name']} server did not start within " f"{int(event['seconds'])}s") _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']}") def server_log_path(name: str) -> Path: """The full path of the managed NAME server's log file. ``app/logs/-server.log`` (the same file ``start`` streams stdout and stderr into). Clients use it to surface a server's own runtime detail — e.g. the exact allocation size a failed ggml graph build attempted — in their error messages. The audio.cpp server's spec name is ``"audiocpp"``. """ return _log_path(name) def _log_path(name: str) -> Path: return LOG_DIR / f"{name}-server.log" def _pid_path(name: str) -> Path: return LOG_DIR / f"{name}-server.pid" def _read_log_tail(name: str, lines: int = 20) -> List[str]: """Return the last LINES of the server's log (best-effort).""" try: text = _log_path(name).read_text(encoding="utf-8", errors="replace") except OSError: return [] return text.splitlines()[-lines:] def _boot_hint(log_tail: List[str]) -> Optional[str]: """A plain-language hint for a known crash signature in LOG_TAIL.""" text = "\n".join(log_tail) for signature, hint in _BOOT_HINTS: if signature in text: return hint return None def _print_tail(tail: List[str]) -> None: """Print a log-tail event payload (used by the console callback).""" if tail: print(f"--- last {len(tail)} lines of the server log ---") print("\n".join(tail)) 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\d+) is already in use.*?" r"Using port (?P\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": try: import ctypes kernel32 = ctypes.windll.kernel32 # type: ignore[attr-defined] PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 handle = kernel32.OpenProcess( PROCESS_QUERY_LIMITED_INFORMATION, False, pid) if not handle: return False kernel32.CloseHandle(handle) return True except OSError: return False try: os.kill(pid, 0) except ProcessLookupError: return False except PermissionError: return True return True def _process_start_token(pid: int) -> Optional[str]: """A token that changes when the OS recycles PID (best effort). On Linux the token is /proc's process start time (field 22): after a crash and pid reuse, a bare signal-0 probe would happily bless the new unrelated process as "our server", so the recorded token no longer matching is what breaks that identification. Platforms without an equivalent (macOS, Windows) return None — bare-pid probing stays, as before. """ if not sys.platform.startswith("linux"): return None try: stat = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8") # Fields 1/2 ("comm") may contain spaces and parentheses; starttime # is field 22, i.e. the 20th entry after the closing paren. fields = stat.rsplit(")", 1)[1].split() return fields[19] except (OSError, IndexError): return None def _pid_owned(pid: int, token: Optional[str]) -> bool: """True when PID is alive AND still the process the token recorded. An empty/missing token (legacy pid files, non-Linux platforms) falls back to the bare liveness probe; a recorded token must match the process's current start time, so a recycled pid is reported dead. """ if not _pid_alive(pid): return False if not token: return True current = _process_start_token(pid) return current is None or current == token def _reap_exited(pid: int) -> bool: """Reap PID when it is our exited child; True when confirmed dead. An exited child stays visible to signal-0 probes until its parent waits for it, and the hub never reaps between ``start`` and ``stop`` — so a server that honored SIGTERM would still count as alive and every stop would burn the whole grace period before escalating to SIGKILL. Returns False when the process may still be running or was not our child (the caller then falls back to its own liveness probes). """ if not hasattr(os, "waitpid") or not hasattr(os, "WNOHANG"): return False try: waited, _status = os.waitpid(pid, os.WNOHANG) except ChildProcessError: # Not our child (or someone reaped it already): not ours to judge. return False except OSError: return False return waited == pid def _kill_pid(pid: int) -> bool: """Terminate PID (and its process tree). Returns True when dead.""" if sys.platform == "win32": # TerminateProcess hits a single pid; the server may have spawned # children that would survive as orphans. taskkill /T kills the # whole tree (the same mechanism run_console_subprocess uses). try: subprocess.run(["taskkill", "/T", "/F", "/PID", str(pid)], capture_output=True, timeout=STOP_GRACE_SECONDS) except (OSError, subprocess.SubprocessError): try: os.kill(pid, signal.SIGTERM) except (ProcessLookupError, PermissionError, OSError): return not _pid_alive(pid) for _ in range(int(STOP_GRACE_SECONDS * 10)): if not _pid_alive(pid): return True time.sleep(0.1) return not _pid_alive(pid) # POSIX: kill the whole process group (started with start_new_session=True). try: pgid = os.getpgid(pid) except ProcessLookupError: return True try: os.killpg(pgid, signal.SIGTERM) except ProcessLookupError: return True except PermissionError: return False for _ in range(int(STOP_GRACE_SECONDS * 10)): # Reap first so an exited (zombie) child ends the wait immediately # instead of keeping the killpg(0) probe "alive" until SIGKILL. if _reap_exited(pid): return True try: os.killpg(pgid, 0) except ProcessLookupError: return True except PermissionError: return False time.sleep(0.1) try: os.killpg(pgid, signal.SIGKILL) except (ProcessLookupError, PermissionError): pass return True 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. LISTENING, when given, is a caller's fresh ``common.server_running`` result — reused so one start pass probes the port only once. """ if listening is None: listening = common.server_running(spec.url) if not listening: return False identity = getattr(spec, "identity", None) if identity is None: return True if probe.identify_server(spec.url) != identity: return False if identity == probe.IDENTITY_FASTER: return probe.faster_model_loaded(spec.url) return True def start(spec, progress: ProgressCallback = None, cancel=None) -> bool: """Start the server described by SPEC (a ``backends.ServerSpec``). Spawns its argv with stdout/stderr to ``logs/-server.log``, in the spec's CWD when it has one (audio.cpp discovers model_specs/ from its process working directory), records the pid, and polls readiness — ``_server_ready``, so an IDENTITY spec must actually answer HTTP — until it is up or the spec's start timeout elapses (``ServerSpec.start_timeout`` overrides SERVER_START_TIMEOUT; the sglang-omni pipeline needs the longer budget). Returns True when the server is up; on timeout or early 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 is terminated and False is reported (event kind "cancelled"). """ report = progress if progress is not None else _console_progress argv: List[str] = list(spec.argv) exe = Path(argv[0]) if not exe.exists(): report({"kind": "error", "message": f"server executable not found: {exe}. Run 'Set " "up a backend' first."}) return False # 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 start_timeout = getattr(spec, "start_timeout", None) \ or SERVER_START_TIMEOUT LOG_DIR.mkdir(parents=True, exist_ok=True) # Refuse to double-start: a live pid file means a previous start is # still booting (or its process is wedged). Spawning a second server # on the same port would orphan the first with no pid record left. if alive(spec.name): report({"kind": "error", "message": f"a {spec.name} server (pid " 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: pid_file.unlink() except OSError: pass # Reserve the slot atomically (O_EXCL): two concurrent starters can # both pass the liveness check above, but only one wins the create — # the loser refuses instead of spawning a duplicate server on the port. try: pid_handle = pid_file.open("x") except FileExistsError: report({"kind": "error", "message": f"a {spec.name} server is already starting " "(its pid file appeared while this start was " "running); stop it first"}) return False except OSError: pid_handle = None cwd = getattr(spec, "cwd", None) # Append so an earlier boot's output survives (crash-loop debugging); # the child inherits the handle and the parent's copy is closed right # after the spawn, so nothing leaks here. log_handle = _log_path(spec.name).open("a", encoding="utf-8") popen_kwargs = {"stdout": log_handle, "stderr": subprocess.STDOUT} if cwd is not None: popen_kwargs["cwd"] = str(cwd) if sys.platform == "win32": popen_kwargs["creationflags"] = \ subprocess.CREATE_NEW_PROCESS_GROUP # type: ignore[attr-defined] else: popen_kwargs["start_new_session"] = True try: proc = subprocess.Popen(argv, **popen_kwargs) except OSError as exc: report({"kind": "error", "message": f"could not start server: {exc}"}) log_handle.close() if pid_handle is not None: pid_handle.close() try: pid_file.unlink() except OSError: pass return False log_handle.write(f"\n=== boot {datetime.now():%Y-%m-%d %H:%M:%S} " f"(pid {proc.pid}) ===\n") log_handle.flush() log_handle.close() # "pid token": the process's start time where the platform provides # one, so a recycled pid is never mistaken for our server (see # _pid_owned). Empty token = bare-pid probing. token = _process_start_token(proc.pid) or "" if pid_handle is not None: try: pid_handle.write(f"{proc.pid} {token}\n".strip() + "\n") pid_handle.close() except OSError: pass report({"kind": "starting", "name": spec.name, "argv": " ".join(str(a) for a in argv), "cwd": str(cwd) if cwd is not None else None, "log_path": str(_log_path(spec.name)), "pid": proc.pid}) 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 # server we started is not left loading in the background). _kill_pid(proc.pid) try: pid_file.unlink() except OSError: pass report({"kind": "cancelled", "name": spec.name}) return False if proc.poll() is not None: tail = _read_log_tail(spec.name) report({"kind": "exited", "name": spec.name, "returncode": proc.returncode, "log_tail": tail, "hint": _boot_hint(tail)}) try: pid_file.unlink() 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 if time.time() >= next_heartbeat: report({"kind": "elapsed", "seconds": time.time() - started}) next_heartbeat += 15 time.sleep(POLL_INTERVAL) tail = _read_log_tail(spec.name) report({"kind": "timeout", "name": spec.name, "seconds": start_timeout, "log_tail": tail, "hint": _boot_hint(tail)}) # Leave the pid file in place so stop() can kill it (it may still load). return False def stop(name: str) -> bool: """Stop a server previously started by ``start`` (identified by pid file). Returns True when the process was terminated (or already gone). Returns False when there is no pid file — the server was not started by this tool, so the user must stop it manually (e.g. close its terminal). The pid file is removed only once the kill succeeded: while the process is still running, the record is what keeps a later ``start`` from spawning a duplicate onto the same port. """ pid_file = _pid_path(name) if not pid_file.exists(): print(f"[INFO] no pid file for '{name}' " "(not started by this tool — stop it manually)") return False try: pid = int(pid_file.read_text(encoding="utf-8").split()[0]) except (OSError, ValueError, IndexError): print(f"[WARNING] could not read pid file {pid_file}; removing it") try: pid_file.unlink() except OSError: pass return False if not _pid_owned_recorded(pid, pid_file): print(f"[INFO] {name} server (pid {pid}) already stopped") try: pid_file.unlink() except OSError: pass return True print(f"[INFO] stopping {name} server (pid {pid})...") killed = _kill_pid(pid) if killed: print(f"[OK] {name} server stopped") try: pid_file.unlink() except OSError: pass else: print(f"[WARNING] could not stop pid {pid}; stop it manually — the " "pid file is kept so the server is not started twice") return killed def _pid_owned_recorded(pid: int, pid_file: Path) -> bool: """PID alive and still the process recorded in PID_FILE (token aware).""" try: fields = pid_file.read_text(encoding="utf-8").split() except OSError: return _pid_alive(pid) token = fields[1] if len(fields) > 1 else None return _pid_owned(pid, token) def manages(specs) -> bool: """True when any SPEC in the list was started (and is kept alive) by us. A server counts as ours when ``start`` recorded a pid file for it and that pid is still alive — the same ownership rule ``stop`` applies before refusing ("not started by this tool"). Where the platform provides a start-time token, a recycled pid no longer counts as ours. Used by the backends' ``detect()`` so the hub's status table can tag an up server as "[remote]" when it was launched outside this tool. """ for spec in specs: pid_file = _pid_path(spec.name) pid = pid_for(spec.name) if pid is not None and _pid_owned_recorded(pid, pid_file): return True return False def pid_for(name: str): """Return the recorded pid for NAME, or None when no pid file exists.""" pid_file = _pid_path(name) if not pid_file.exists(): return None try: return int(pid_file.read_text(encoding="utf-8").split()[0]) except (OSError, ValueError, IndexError): return None def alive(name: str) -> bool: """True when the server named NAME was started by us and is still alive.""" pid_file = _pid_path(name) pid = pid_for(name) return pid is not None and _pid_owned_recorded(pid, pid_file)