my_voice_assistant_v3_jamulix/app/admin_llm.py

"""Read-only Statusabfragen rund um das lokale LLM-Backend (fuer das Admin-Panel).

Alles nur lesend, ohne sudo: `docker ps`, `ollama ps`, `nvidia-smi`, `systemctl --user`.
Fehlende Tools oder Fehler fuehren zu sicheren Defaults (None/[]), nie zu Exceptions.
"""

from __future__ import annotations

import asyncio
import os
import re
import shutil
import subprocess
from pathlib import Path

from app.config import settings

ALLOWED_BACKENDS = {"ollama", "llamacpp"}
_SCRIPT = Path(__file__).resolve().parent.parent / "scripts" / "llm-server" / "switch-llm.sh"
# Erlaubtes Format fuer Ollama-Modellnamen (zusaetzlich zur Pruefung gegen 'ollama list').
_MODEL_RE = re.compile(r"^[A-Za-z0-9._:/-]{1,100}$")


async def _run(cmd: list[str], timeout: float = 6.0) -> str | None:
    """Fuehrt ein Kommando aus (shell=False) und liefert stdout, oder None bei Fehler."""
    if not shutil.which(cmd[0]):
        return None
    try:
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.DEVNULL,
            env={**os.environ, "XDG_RUNTIME_DIR": os.environ.get(
                "XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}")},
        )
        out, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout)
        if proc.returncode != 0:
            return None
        return out.decode("utf-8", "replace")
    except (asyncio.TimeoutError, OSError):
        return None


def _detect_backend(base_url: str) -> str:
    if ":11434" in base_url:
        return "ollama"
    if ":8001" in base_url:
        return "llamacpp"
    return "unknown"


async def _llamacpp_running(container: str = "va_llm") -> bool:
    out = await _run(["docker", "ps", "--filter", f"name=^{container}$", "--format", "{{.Names}}"])
    return bool(out and container in out)


async def _ollama_loaded() -> tuple[bool, list[dict]]:
    """(dienst_erreichbar, [geladene Modelle])."""
    out = await _run(["ollama", "ps"])
    if out is None:
        return False, []
    models: list[dict] = []
    lines = [ln for ln in out.splitlines() if ln.strip()]
    for ln in lines[1:]:  # Kopfzeile ueberspringen
        # Spalten sind durch 2+ Leerzeichen getrennt: NAME ID SIZE PROCESSOR CONTEXT UNTIL
        import re
        cols = re.split(r"\s{2,}", ln.strip())
        if cols:
            models.append({
                "name": cols[0],
                "size": cols[2] if len(cols) > 2 else "",
                "processor": cols[3] if len(cols) > 3 else "",
            })
    return True, models


async def _gpus() -> list[dict]:
    out = await _run([
        "nvidia-smi",
        "--query-gpu=index,memory.used,memory.total",
        "--format=csv,noheader,nounits",
    ])
    if out is None:
        return []
    gpus: list[dict] = []
    for ln in out.splitlines():
        parts = [p.strip() for p in ln.split(",")]
        if len(parts) == 3 and parts[0].isdigit():
            used, total = int(parts[1]), int(parts[2])
            gpus.append({
                "index": int(parts[0]),
                "used_mib": used,
                "total_mib": total,
                "percent": round(used / total * 100) if total else 0,
            })
    return gpus


async def _gateway_service_active() -> bool:
    out = await _run(["systemctl", "--user", "is-active", "voice-assistant.service"], timeout=4.0)
    return bool(out and out.strip() == "active")


async def llm_status() -> dict:
    """Aggregierter, read-only LLM-/System-Status fuer das Admin-Panel."""
    base_url = settings.local_llm_base_url
    backend = _detect_backend(base_url)
    llamacpp, (ollama_reachable, ollama_models), gpus, gw = await asyncio.gather(
        _llamacpp_running(),
        _ollama_loaded(),
        _gpus(),
        _gateway_service_active(),
    )
    return {
        "backend": backend,
        "model": settings.local_llm_model,
        "base_url": base_url,
        "llamacpp_running": llamacpp,
        "ollama_reachable": ollama_reachable,
        "ollama_loaded": ollama_models,
        "gpus": gpus,
        "gateway_service_active": gw,
    }


# ── Schreibende Steuerung (Backend-Wechsel / Gateway-Neustart) ───────────────

class LlmControlError(ValueError):
    """Validierungs-/Steuerfehler -> wird vom Endpoint als 400/422 gemeldet."""


async def available_ollama_models() -> list[str]:
    """Modellnamen aus `ollama list` (erste Spalte), oder []."""
    out = await _run(["ollama", "list"])
    if out is None:
        return []
    names: list[str] = []
    for ln in out.splitlines()[1:]:
        ln = ln.strip()
        if ln:
            names.append(ln.split()[0])
    return names


async def switch_backend(backend: str, model: str | None = None) -> dict:
    """Validiert streng und startet den Backend-Wechsel als losgelösten Prozess.

    Allowlist: backend ∈ {ollama, llamacpp}. Bei ollama muss `model` (falls gesetzt)
    exakt in `ollama list` vorkommen. Nie freie Strings an die Shell — Aufruf mit
    fester Argumentliste (shell=False); das Modell geht ausschließlich als Env-Var.
    Der Wechsel läuft detached weiter (er startet ggf. das Gateway neu).
    """
    if backend not in ALLOWED_BACKENDS:
        raise LlmControlError(f"Unbekanntes Backend: {backend!r}")
    if not _SCRIPT.exists():
        raise LlmControlError("switch-llm.sh nicht gefunden")

    env = {**os.environ}
    if backend == "ollama" and model:
        if not _MODEL_RE.match(model):
            raise LlmControlError("Ungültiger Modellname")
        available = await available_ollama_models()
        if available and model not in available:
            raise LlmControlError(f"Modell nicht in 'ollama list': {model!r}")
        env["OLLAMA_MODEL"] = model

    # Detached starten: der Wechsel kann das Gateway neu starten -> wir würden uns
    # sonst selbst killen, bevor die HTTP-Antwort raus ist.
    subprocess.Popen(
        ["bash", str(_SCRIPT), backend],
        env=env,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        start_new_session=True,
    )
    return {"status": "switching", "backend": backend, "model": model}


def restart_gateway_detached() -> dict:
    """Startet das Gateway als systemd-User-Dienst neu (losgelöst, Self-Restart-sicher)."""
    xdg = os.environ.get("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}")
    subprocess.Popen(
        ["bash", "-c",
         f"sleep 1; XDG_RUNTIME_DIR={xdg} systemctl --user restart voice-assistant.service"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        start_new_session=True,
    )
    return {"status": "restarting"}