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

"""STT-server — API FastAPI pour les clients STT.

Endpoints :
  GET  /healthz       → état du service
  POST /v1/ask {text} → {reply}   (requête AI, orchestrée vers LiteLLM)
"""

from __future__ import annotations

import logging
import time
from contextlib import asynccontextmanager

import httpx
from fastapi import BackgroundTasks, FastAPI, HTTPException
from pydantic import BaseModel

from stt_server import __version__
from stt_server import brain
from stt_server.brain import ask as brain_ask
from stt_server.config import settings
from stt_server import agent
from stt_server import tools
from stt_server.contexts import CONTEXTS, assemble, get_context
from stt_server.knowledge import Knowledge
from stt_server.longterm import LongTermMemory
from stt_server.memory import SessionStore
from stt_server.sources import fetch_blocks, ghostfolio_details, ghostfolio_phrase

log = logging.getLogger("stt_server")
# uvicorn ne configure que ses propres loggers : on attache notre handler en INFO
# pour que les lignes de timing (`ask … recall/gen/total`) sortent dans les logs du pod.
if not log.handlers:
    log.setLevel(logging.INFO)
    _h = logging.StreamHandler()
    _h.setFormatter(logging.Formatter("%(levelname)s:     %(name)s: %(message)s"))
    log.addHandler(_h)
    log.propagate = False

sessions = SessionStore()
longterm = LongTermMemory() if settings.memory_longterm else None
knowledge = Knowledge() if settings.docs_rag else None

# Contexte « agent » : action en attente de confirmation, par session.
pending_actions: dict[str, str] = {}


def _actions_available() -> bool:
    return settings.actions_enabled and bool(settings.hermes_exec_token)


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    # Fermeture propre des clients HTTP persistants (pooling).
    await brain.aclose()
    if longterm:
        await longterm.aclose()
    if knowledge:
        await knowledge.aclose()


app = FastAPI(title="STT-server", version=__version__, lifespan=lifespan)


class AskRequest(BaseModel):
    text: str
    model: str | None = None       # alias LiteLLM ; défaut serveur si absent
    session_id: str | None = None  # mémoire court-terme : fil de conversation
    context: str | None = None     # contexte présélectionné (funk/ghostfolio/…)
    secrets: dict | None = None    # jetons client (ex. {"ghostfolio_token": …})


class PortfolioRequest(BaseModel):
    token: str | None = None       # jeton Ghostfolio fourni par le client (sinon env)


class AskReply(BaseModel):
    reply: str
    model: str
    context_id: str                # contexte effectivement utilisé
    context: dict | None = None    # contexte assemblé (visualiseur HUD)


@app.get("/healthz")
async def healthz() -> dict:
    return {"status": "ok", "version": __version__}


@app.get("/v1/models")
async def v1_models() -> dict:
    return {"default": settings.model, "available": settings.allowed_models}


@app.get("/v1/contexts")
async def v1_contexts() -> dict:
    """Contextes présélectionnables (pour le sélecteur du HUD)."""
    return {
        "default": settings.default_context,
        "contexts": [
            {"id": c.id, "label": c.label, "icon": c.icon, "description": c.description}
            for c in CONTEXTS.values()
            # le contexte « agent » n'apparaît que si les actions sont activées + jeton
            if c.id != "agent" or _actions_available()
        ],
    }


@app.get("/v1/memory/health")
async def v1_memory_health() -> dict:
    """État de la mémoire long-terme (embeddings + Qdrant + collection), erreurs exposées."""
    if not longterm:
        return {"enabled": False}
    return await longterm.health()


@app.post("/v1/portfolio")
async def v1_portfolio(req: PortfolioRequest) -> dict:
    """Valeur du portefeuille Ghostfolio (intent vocal client → serveur).

    Le serveur fait l'appel Ghostfolio avec le jeton fourni par le client (sinon le
    jeton serveur `STT_GHOSTFOLIO_TOKEN`). Un seul code Ghostfolio, lecture seule.
    """
    async with httpx.AsyncClient() as c:
        summary = await ghostfolio_phrase(c, req.token)
    return {"summary": summary}


@app.post("/v1/portfolio/details")
async def v1_portfolio_details(req: PortfolioRequest) -> dict:
    """Aperçu structuré du portefeuille (page détail du portail HUD).

    Même source de jeton que /v1/portfolio (client prioritaire, sinon env serveur).
    """
    async with httpx.AsyncClient() as c:
        return await ghostfolio_details(c, req.token)


@app.post("/v1/reset")
async def v1_reset(req: AskRequest) -> dict:
    if req.session_id:
        sessions.reset(req.session_id)
    return {"status": "reset"}


def _agent_context_debug(detail: dict) -> dict:
    """Structure de visualisation pour le contexte agent (transparence du handshake)."""
    return {
        "id": "agent", "label": "Agent (actions)", "icon": "🤖",
        "system_prompt": "Flux agent : la demande est confirmée puis exécutée par "
        "le vrai agent Hermes (hermes -z --yolo).",
        "blocks": [{"source": "agent", "title": "Action", "text": str(detail)}],
        "docs": [], "memories": [],
    }


async def _handle_agent(req: AskRequest, text: str) -> AskReply:
    """Handshake de confirmation puis exécution via hermes-exec. Court-circuite le LLM."""
    sid = req.session_id or "anon"
    if not _actions_available():
        return AskReply(
            reply="Les actions sont désactivées côté serveur.",
            model=settings.model, context_id="agent",
            context=_agent_context_debug({"disabled": True}),
        )
    pending = pending_actions.get(sid)
    if pending and agent.is_confirmation(text):
        pending_actions.pop(sid, None)
        out = await agent.run_action(pending)
        log.info("agent EXEC sid=%s prompt=%r", sid, pending[:120])
        return AskReply(reply=out, model=settings.model, context_id="agent",
                        context=_agent_context_debug({"executed": pending, "result": out[:300]}))
    if pending and agent.is_cancellation(text):
        pending_actions.pop(sid, None)
        return AskReply(reply="C'est annulé.", model=settings.model, context_id="agent",
                        context=_agent_context_debug({"cancelled": pending}))
    # nouvelle demande (ou remplacement) → on stocke et on demande confirmation
    pending_actions[sid] = text
    reply = f"Tu veux que j'exécute : « {text} » ? Dis « confirme » ou « annule »."
    return AskReply(reply=reply, model=settings.model, context_id="agent",
                    context=_agent_context_debug({"pending": text}))


def _tool_blocks(trace: list[dict]) -> list[dict]:
    """Trace d'outils → blocs du visualiseur HUD (un bloc par appel)."""
    return [
        {
            "source": f"tool:{t['name']}",
            "title": f"🔧 {t['name']}({', '.join(f'{k}={v}' for k, v in t['args'].items())})",
            "text": t["result"],
        }
        for t in trace
    ]


def _agentic_ctx_debug(ctx, memories: list[str], trace: list[dict]) -> dict:
    return {
        "id": ctx.id, "label": ctx.label, "icon": ctx.icon,
        "system_prompt": ctx.system_prompt,
        "blocks": _tool_blocks(trace), "docs": [], "memories": memories,
    }


async def _handle_agentic(
    req: "AskRequest", text: str, ctx, model: str, background: BackgroundTasks
) -> "AskReply":
    """Contexte à outils : recall mémoire (perso) + boucle de function-calling (état live).

    La doc n'est plus injectée d'office (elle est devenue l'outil `search_docs`) ; on garde
    le recall long-terme pour la personnalisation. Le trace des outils alimente le visualiseur.
    Phase 3 : si une action admin a été proposée au tour précédent, ce tour la confirme/annule.
    """
    t0 = time.perf_counter()
    sid = req.session_id or "anon"
    # Confirmation d'une action admin proposée au tour précédent (handshake, hors LLM).
    pending = pending_actions.get(sid)
    if pending and _actions_available():
        if agent.is_confirmation(text):
            pending_actions.pop(sid, None)
            out = await agent.run_action(pending)
            log.info("asa EXEC sid=%s prompt=%r", sid, pending[:120])
            trace = [{"name": "admin_action", "args": {"description": pending}, "result": out}]
            return AskReply(reply=out, model=model, context_id=ctx.id,
                            context=_agentic_ctx_debug(ctx, [], trace))
        if agent.is_cancellation(text):
            pending_actions.pop(sid, None)
            return AskReply(reply="C'est annulé.", model=model, context_id=ctx.id,
                            context=_agentic_ctx_debug(ctx, [], []))
        # Ni confirmation ni annulation → l'utilisateur est passé à autre chose : on oublie.
        pending_actions.pop(sid, None)
    history = sessions.history(req.session_id) if req.session_id else None
    memories, qvec = await longterm.recall(text) if longterm else ([], None)
    system = ctx.system_prompt
    if memories:
        souvenirs = "\n".join(f"- {m}" for m in memories)
        system += (
            "\n\nÉléments de mémoire long-terme (peuvent aider, ignore si hors-sujet) :\n"
            + souvenirs
        )
    # admin_action n'est exposé que si les actions sont activées (opt-in + jeton).
    tool_names = tuple(
        t for t in ctx.tools if t != "admin_action" or _actions_available()
    )
    schemas = tools.schemas_for(tool_names)
    t_recall = time.perf_counter()
    async with httpx.AsyncClient() as client:
        deps = tools.ToolDeps(client=client, knowledge=knowledge, qvec=qvec)

        async def _dispatch(name: str, args: dict) -> str:
            # admin_action : on n'exécute PAS — on mémorise la proposition et on demande
            # confirmation (la boucle s'arrête via confirm_tools). Exécution = tour suivant.
            if name == "admin_action":
                desc = (args.get("description") or "").strip()
                if not desc:
                    return "Aucune action décrite."
                pending_actions[sid] = desc
                return (f"Tu veux que j'exécute : « {desc} » ? "
                        "Dis « confirme » pour lancer, ou « annule ».")
            return await tools.run(name, args, deps)

        try:
            reply, tool_trace = await brain.ask_with_tools(
                text, system, schemas=schemas, dispatch=_dispatch,
                model=model, history=history, confirm_tools=("admin_action",),
            )
        except httpx.HTTPError as e:
            raise HTTPException(status_code=502, detail=f"upstream LiteLLM : {e}") from e
    t_gen = time.perf_counter()
    if req.session_id:
        sessions.add(req.session_id, "user", text)
        sessions.add(req.session_id, "assistant", reply)
    if longterm:
        background.add_task(longterm.store, req.session_id or "anon", text, qvec)
    log.info(
        "ask ctx=%s model=%s recall=%.0fms loop=%.0fms total=%.0fms mem=%d tools=%d",
        ctx.id, model, (t_recall - t0) * 1000, (t_gen - t_recall) * 1000,
        (t_gen - t0) * 1000, len(memories), len(tool_trace),
    )
    # Visualiseur HUD : chaque appel d'outil = un bloc (réutilise l'UI existante).
    return AskReply(reply=reply, model=model, context_id=ctx.id,
                    context=_agentic_ctx_debug(ctx, memories, tool_trace))


@app.post("/v1/ask", response_model=AskReply)
async def v1_ask(req: AskRequest, background: BackgroundTasks) -> AskReply:
    text = req.text.strip()
    if not text:
        raise HTTPException(status_code=400, detail="text vide")
    model = req.model or settings.model
    if model not in settings.allowed_models:
        raise HTTPException(
            status_code=400,
            detail=f"modèle '{model}' non autorisé ; dispo : {settings.allowed_models}",
        )
    ctx = get_context(req.context)
    # Contexte « agent » : flux dédié (confirmation 2 temps → hermes-exec), sans LLM.
    if ctx.id == "agent":
        return await _handle_agent(req, text)
    # Contexte à outils (« asa ») : boucle de function-calling (le modèle sonde l'état live).
    if ctx.tools:
        return await _handle_agentic(req, text, ctx, model, background)
    t0 = time.perf_counter()
    history = sessions.history(req.session_id) if req.session_id else None
    # recall : timeout serré, dégrade vite ; renvoie aussi le vecteur (réutilisé ci-dessous)
    memories, qvec = await longterm.recall(text) if longterm else ([], None)
    # RAG doc : seulement si le contexte le demande (réutilise qvec, même embedder nomic)
    docs = []
    if "docs" in ctx.sources and knowledge:
        docs = await knowledge.search(text, qvec)
    # Sources live du contexte (Ghostfolio / Prometheus / Alertmanager) — best-effort
    blocks: list = []
    live = tuple(s for s in ctx.sources if s != "docs")
    if live:
        async with httpx.AsyncClient() as c:
            blocks = await fetch_blocks(c, live, secrets=req.secrets)
    t_recall = time.perf_counter()
    system, ctx_debug = assemble(ctx, blocks=blocks, docs=docs, memories=memories)
    try:
        reply = await brain_ask(text, system, model, history)
    except httpx.HTTPError as e:
        raise HTTPException(status_code=502, detail=f"upstream LiteLLM : {e}") from e
    t_gen = time.perf_counter()
    if req.session_id:
        sessions.add(req.session_id, "user", text)
        sessions.add(req.session_id, "assistant", reply)
    # store : APRÈS la réponse (BackgroundTasks) → hors latence perçue, et on réutilise qvec
    if longterm:
        background.add_task(longterm.store, req.session_id or "anon", text, qvec)
    log.info(
        "ask ctx=%s model=%s recall+src=%.0fms gen=%.0fms total=%.0fms mem=%d docs=%d blocks=%d",
        ctx.id, model,
        (t_recall - t0) * 1000,
        (t_gen - t_recall) * 1000,
        (t_gen - t0) * 1000,
        len(memories), len(docs), len(blocks),
    )
    return AskReply(reply=reply, model=model, context_id=ctx.id, context=ctx_debug)


def run() -> None:
    """Entrypoint `stt-server` (dev local). En prod : uvicorn via le conteneur."""
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)  # noqa: S104 — service interne au cluster