Funk-lab/stt/server/stt_server/tools.py

"""Outils appelables par Asa (function calling) — Phase 1 : LECTURE SEULE.

Chaque outil = un schéma OpenAI (exposé au LLM) + un exécuteur async qui renvoie un
**texte court** réinjecté dans la boucle (`brain.ask_with_tools`). Tout est best-effort :
un exécuteur n'élève jamais d'exception fatale — il renvoie une note d'erreur lisible que
le modèle peut relayer à l'utilisateur.

Les outils réutilisent l'infra existante (Prometheus via `sources`, RAG via `knowledge`).
Aucune action d'écriture ici : le volet « agir sur le homelab » passe par `agent.py`
(hermes-exec, avec confirmation) et sera branché comme outil en Phase 3.
"""

from __future__ import annotations

import asyncio
from dataclasses import dataclass

import httpx

from stt_server import sources
from stt_server.config import settings

# job Prometheus du node_exporter par hôte hors-cluster (cf. sources.cluster_block).
_HOST_JOBS = {"gpu-01": "gpu-01-node", "storage-01": "storage-01"}


async def _q1(client: httpx.AsyncClient, expr: str) -> float | None:
    """Valeur scalaire d'une requête PromQL (1re série) ou None. Ne lève jamais."""
    try:
        rows = await sources._prom_query(client, expr)
    except httpx.HTTPError:
        return None
    return rows[0][1] if rows else None


@dataclass
class ToolDeps:
    """Dépendances injectées aux exécuteurs (client HTTP partagé, RAG, vecteur de requête)."""

    client: httpx.AsyncClient
    knowledge: object | None = None        # stt_server.knowledge.Knowledge | None
    qvec: list[float] | None = None        # embedding de la requête (réutilisé par search_docs)


# --- Exécuteurs -----------------------------------------------------------------

async def _search_docs(args: dict, deps: ToolDeps) -> str:
    query = (args.get("query") or "").strip()
    if not query:
        return "Aucune requête fournie."
    if deps.knowledge is None:
        return "La recherche documentaire est désactivée."
    # Réutilise le vecteur déjà calculé si la requête de l'outil == requête utilisateur.
    docs = await deps.knowledge.search(query, deps.qvec)
    if not docs:
        return "Aucun passage pertinent dans la documentation Funk."
    return "\n\n".join(docs[:4])


def _fmt_series(rows: list[tuple[dict, float]], label_key: str | None = None, limit: int = 10) -> str:
    out = []
    for metric, val in rows[:limit]:
        v = int(val) if float(val).is_integer() else round(val, 3)
        if label_key and metric.get(label_key):
            out.append(f"{metric[label_key]}={v}")
        elif metric:
            ident = metric.get("__name__") or next(iter(metric.values()), "")
            out.append(f"{ident}={v}")
        else:
            out.append(str(v))
    return ", ".join(out)


async def _host_health(args: dict, deps: ToolDeps) -> str:
    host = (args.get("host") or "").strip()
    job = _HOST_JOBS.get(host)
    if not job:
        return f"Hôte inconnu : {host!r} (attendu : {', '.join(_HOST_JOBS)})."
    c = deps.client
    # Sondes hôte (concurrentes) : up, CPU%, RAM libre%, disque /%, temp CPU max, uptime h.
    up, cpu, mem, disk, temp, uptime = await asyncio.gather(
        _q1(c, f'up{{job="{job}"}}'),
        _q1(c, f'100-100*avg(rate(node_cpu_seconds_total{{job="{job}",mode="idle"}}[5m]))'),
        _q1(c, f'100*node_memory_MemAvailable_bytes{{job="{job}"}}'
               f'/node_memory_MemTotal_bytes{{job="{job}"}}'),
        _q1(c, f'100-100*node_filesystem_avail_bytes{{job="{job}",mountpoint="/"}}'
               f'/node_filesystem_size_bytes{{job="{job}",mountpoint="/"}}'),
        _q1(c, f'max(node_hwmon_temp_celsius{{job="{job}"}})'),
        _q1(c, f'(node_time_seconds{{job="{job}"}}-node_boot_time_seconds{{job="{job}"}})/3600'),
    )
    if up is None:
        return f"{host} : aucune métrique (cible Prometheus absente ou hôte éteint)."
    parts = [f"{host} : {'en ligne' if up == 1 else 'HORS LIGNE'}"]
    if cpu is not None:
        parts.append(f"CPU {round(cpu)}%")
    if mem is not None:
        parts.append(f"RAM libre {round(mem)}%")
    if disk is not None:
        parts.append(f"disque / {round(disk)}% utilisé")
    if temp is not None:
        parts.append(f"temp CPU {round(temp)}°C")
    if uptime is not None:
        parts.append(f"uptime {round(uptime)} h")
    if host == "gpu-01":
        # Métriques GPU (rocm_scraper → textfile collector node_exporter).
        gtemp, gutil, gvram, gpow, llama = await asyncio.gather(
            _q1(c, "rocm_gpu_temperature_celsius"),
            _q1(c, "rocm_gpu_utilization_percent"),
            _q1(c, "100*rocm_vram_used_bytes/rocm_vram_total_bytes"),
            _q1(c, "rocm_gpu_power_watts"),
            _q1(c, 'up{job="llama-server-gpu"}'),
        )
        gpu = []
        if gutil is not None:
            gpu.append(f"util {round(gutil)}%")
        if gtemp is not None:
            gpu.append(f"{round(gtemp)}°C")
        if gvram is not None:
            gpu.append(f"VRAM {round(gvram)}%")
        if gpow is not None:
            gpu.append(f"{round(gpow)} W")
        if gpu:
            parts.append("GPU (" + ", ".join(gpu) + ")")
        if llama is not None:
            parts.append("llama-server " + ("up" if llama == 1 else "DOWN"))
    return ", ".join(parts) + "."


async def _cluster_status(args: dict, deps: ToolDeps) -> str:
    # Réutilise le résumé déjà éprouvé (nœuds joignables, pods prêts / non prêts).
    return await sources.cluster_block(deps.client)


async def _prometheus_query(args: dict, deps: ToolDeps) -> str:
    expr = (args.get("expr") or "").strip()
    if not expr:
        return "Aucune expression PromQL fournie."
    try:
        rows = await sources._prom_query(deps.client, expr)
    except httpx.HTTPError:
        return "Données indisponibles : Prometheus injoignable."
    except Exception as e:  # noqa: BLE001 — PromQL invalide, etc.
        return f"Requête refusée par Prometheus : {e}"
    if not rows:
        return "La requête ne renvoie aucune série (résultat vide)."
    return _fmt_series(rows)


async def _web_search(args: dict, deps: ToolDeps) -> str:
    query = (args.get("query") or "").strip()
    if not query:
        return "Aucune requête de recherche fournie."
    url = settings.searxng_url.rstrip("/") + "/search"
    try:
        r = await deps.client.get(
            url,
            params={"q": query, "format": "json", "safesearch": 0, "language": "fr"},
            timeout=settings.web_search_timeout,
        )
        r.raise_for_status()
        data = r.json()
    except httpx.HTTPError:
        return "Recherche web indisponible : SearXNG injoignable."
    except ValueError:
        return "Recherche web indisponible : réponse SearXNG inattendue (API JSON activée ?)."
    results = data.get("results") or []
    if not results:
        answers = data.get("answers") or []
        if answers:
            return " ".join(str(a) for a in answers[:2])
        return "Aucun résultat web pour cette recherche."
    lines = []
    for res in results[: settings.web_search_results]:
        title = (res.get("title") or "").strip()
        content = (res.get("content") or "").strip()
        src = res.get("url") or ""
        lines.append(f"- {title} : {content} ({src})" if content else f"- {title} ({src})")
    return "\n".join(lines)


async def _admin_action_placeholder(args: dict, deps: ToolDeps) -> str:
    # Jamais atteint : app.py intercepte admin_action dans son dispatch (handshake de
    # confirmation). Présent uniquement pour que le schéma soit exposé au LLM via le registre.
    return "Action en attente de confirmation."


# --- Registre : nom → (schéma OpenAI, exécuteur) --------------------------------

_TOOLS: dict[str, tuple[dict, object]] = {
    "search_docs": (
        {
            "type": "function",
            "function": {
                "name": "search_docs",
                "description": "Recherche dans la documentation du homelab Funk "
                "(machines, services, configuration, procédures). À utiliser pour toute "
                "question sur la conception ou la config du homelab.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "Question ou mots-clés"}
                    },
                    "required": ["query"],
                },
            },
        },
        _search_docs,
    ),
    "host_health": (
        {
            "type": "function",
            "function": {
                "name": "host_health",
                "description": "État de santé LIVE d'une machine hors-cluster (en ligne ?, CPU, "
                "RAM libre, disque, température, uptime ; pour gpu-01 aussi GPU température/"
                "utilisation/VRAM/puissance et l'état de llama-server). À utiliser pour "
                "« est-ce que gpu-01 / storage-01 tourne bien ? » ou une métrique machine précise.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "host": {
                            "type": "string",
                            "enum": list(_HOST_JOBS),
                            "description": "Nom de la machine",
                        }
                    },
                    "required": ["host"],
                },
            },
        },
        _host_health,
    ),
    "cluster_status": (
        {
            "type": "function",
            "function": {
                "name": "cluster_status",
                "description": "État LIVE du cluster Kubernetes Funk : hôtes joignables, "
                "nombre de pods prêts et liste des pods non prêts. À utiliser pour "
                "« est-ce que le cluster va bien ? ».",
                "parameters": {"type": "object", "properties": {}},
            },
        },
        _cluster_status,
    ),
    "prometheus_query": (
        {
            "type": "function",
            "function": {
                "name": "prometheus_query",
                "description": "Exécute une requête PromQL arbitraire (lecture seule) contre "
                "Prometheus et renvoie les séries. À utiliser pour une métrique précise non "
                "couverte par les autres outils.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "expr": {"type": "string", "description": "Expression PromQL"}
                    },
                    "required": ["expr"],
                },
            },
        },
        _prometheus_query,
    ),
    "web_search": (
        {
            "type": "function",
            "function": {
                "name": "web_search",
                "description": "Recherche sur INTERNET (méta-moteur SearXNG) : actualité, faits "
                "généraux, documentation externe, sujets hors du homelab Funk. À utiliser quand "
                "l'information n'est ni dans le homelab ni dans ta connaissance, ou pour vérifier "
                "un fait récent.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "Termes de recherche"}
                    },
                    "required": ["query"],
                },
            },
        },
        _web_search,
    ),
    "admin_action": (
        {
            "type": "function",
            "function": {
                "name": "admin_action",
                "description": "PROPOSE une action d'administration qui MODIFIE le homelab "
                "(redémarrer un service/déploiement, relancer un pod, appliquer un changement…). "
                "N'exécute RIEN directement : l'utilisateur devra confirmer vocalement. À utiliser "
                "uniquement pour une demande explicite d'AGIR, jamais pour une simple lecture.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "description": {
                            "type": "string",
                            "description": "Action à effectuer, en une phrase claire et impérative "
                            "(ex. « redémarrer le déploiement open-webui »).",
                        }
                    },
                    "required": ["description"],
                },
            },
        },
        _admin_action_placeholder,
    ),
}


def schemas_for(names: tuple[str, ...]) -> list[dict]:
    """Schémas OpenAI des outils demandés (ignore les inconnus)."""
    return [_TOOLS[n][0] for n in names if n in _TOOLS]


async def run(name: str, args: dict, deps: ToolDeps) -> str:
    """Exécute un outil ; ne lève jamais — renvoie une note d'erreur lisible."""
    entry = _TOOLS.get(name)
    if entry is None:
        return f"Outil inconnu : {name}."
    try:
        return await entry[1](args or {}, deps)
    except Exception as e:  # noqa: BLE001 — robustesse : la boucle ne doit jamais casser
        return f"Erreur de l'outil {name} : {e}"