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) and the
URL to probe for readiness. This module turns those specs into running
processes: ``start`` spawns the server, streams its output to
``app/logs/<name>-server.log``, records its pid, and polls the URL until it
accepts connections (model loads are slow, so the timeout is generous);
``stop`` terminates the process group the hub started.

Everything here runs in the plain console tail after the curses TUI returns
(matching the wizards' build/pip streaming), so progress and log tails appear
normally. Pid/log files live under ``app/logs/`` which is already gitignored.
"""

import os
import signal
import subprocess
import sys
import time
from pathlib import Path
from typing import List

from backends import common
from backends.common import APP_DIR

LOG_DIR = APP_DIR / "logs"

# 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


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 _tail_log(name: str, lines: int = 20) -> None:
    """Print the last LINES of the server's log (best-effort)."""
    path = _log_path(name)
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return
    tail = "\n".join(text.splitlines()[-lines:])
    if tail:
        print(f"--- last {lines} lines of {path} ---")
        print(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 _kill_pid(pid: int) -> bool:
    """Terminate PID (and its process group on POSIX). Returns True when dead."""
    if sys.platform == "win32":
        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)
        try:
            os.kill(pid, signal.SIGTERM)
        except OSError:
            pass
        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)):
        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 start(spec) -> bool:
    """Start the server described by SPEC (a ``backends.ServerSpec``).

    Spawns its argv with stdout/stderr to ``logs/<name>-server.log``, records
    the pid, and polls ``common.server_running(spec.url)`` until it accepts
    connections or ``SERVER_START_TIMEOUT`` elapses. Returns True when the
    server is up; on timeout or early exit, prints the log tail and returns
    False. A no-op (True) when the server is already running.
    """
    argv: List[str] = list(spec.argv)
    exe = Path(argv[0])
    if not exe.exists():
        print(f"[ERROR] server executable not found: {exe}")
        print("        run 'Set up a backend' for "
              f"{spec.name!r} first.")
        return False
    if common.server_running(spec.url):
        print(f"[INFO] {spec.name} server already running on {spec.url}")
        return True

    LOG_DIR.mkdir(parents=True, exist_ok=True)
    pid_file = _pid_path(spec.name)
    if pid_file.exists():
        try:
            pid_file.unlink()
        except OSError:
            pass

    print(f"[INFO] starting {spec.name} server: "
          + " ".join(str(a) for a in argv))
    log_handle = _log_path(spec.name).open("w", encoding="utf-8")
    popen_kwargs = {"stdout": log_handle, "stderr": subprocess.STDOUT}
    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:
        print(f"[ERROR] could not start server: {exc}")
        log_handle.close()
        return False

    pid_file.write_text(str(proc.pid), encoding="utf-8")
    print(f"[INFO] pid {proc.pid}; logs: {_log_path(spec.name)}")

    deadline = time.time() + SERVER_START_TIMEOUT
    while time.time() < deadline:
        if proc.poll() is not None:
            print(f"[ERROR] {spec.name} server exited with code "
                  f"{proc.returncode}")
            _tail_log(spec.name)
            try:
                pid_file.unlink()
            except OSError:
                pass
            return False
        if common.server_running(spec.url):
            print(f"[OK] {spec.name} server is up on {spec.url}")
            return True
        time.sleep(1)
    print(f"[ERROR] {spec.name} server did not start within "
          f"{SERVER_START_TIMEOUT}s")
    _tail_log(spec.name)
    # 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).
    """
    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").strip())
    except (OSError, ValueError):
        print(f"[WARNING] could not read pid file {pid_file}; removing it")
        try:
            pid_file.unlink()
        except OSError:
            pass
        return False
    if not _pid_alive(pid):
        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")
    else:
        print(f"[WARNING] could not stop pid {pid}; stop it manually")
    try:
        pid_file.unlink()
    except OSError:
        pass
    return killed


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"). 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 = pid_for(spec.name)
        if pid is not None and _pid_alive(pid):
            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").strip())
    except (OSError, ValueError):
        return None