path: root/app/backends/servers.py

"""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/<name>-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 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"),
)

# 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
#   "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 == "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/<name>-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("---")


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) -> 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.
    """
    if not common.server_running(spec.url):
        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/<name>-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.

    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
    if _server_ready(spec):
        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
    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
    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
        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)