"""Pont vers hermes-exec (storage-01) — exécution d'actions via le vrai agent Hermes. Le STT-server appelle hermes-exec **après** confirmation vocale de l'utilisateur. Le jeton d'auth (env, depuis un secret k8s) protège l'endpoint. Tout est best-effort : une erreur renvoie un message parlable, jamais une exception. """ from __future__ import annotations import unicodedata import httpx from stt_server.config import settings # Mots de (dé)confirmation pour le handshake en 2 temps. _CONFIRM = {"confirme", "confirmer", "oui", "vas-y", "vas y", "valide", "ok", "go", "d'accord", "daccord", "c'est bon", "cest bon", "fais-le", "fais le"} _CANCEL = {"annule", "annuler", "non", "stop", "laisse tomber", "laisse", "oublie", "abandonne"} def _norm(s: str) -> str: s = unicodedata.normalize("NFKD", str(s)) s = "".join(c for c in s if not unicodedata.combining(c)) return " ".join(s.lower().split()).strip(" .!?") def is_confirmation(text: str) -> bool: return _norm(text) in _CONFIRM def is_cancellation(text: str) -> bool: return _norm(text) in _CANCEL async def run_action(prompt: str) -> str: """Exécute `prompt` via hermes-exec ; renvoie la sortie de l'agent (ou une erreur lisible).""" if not settings.hermes_exec_token: return "Les actions ne sont pas configurées côté serveur (jeton manquant)." url = settings.hermes_exec_url.rstrip("/") + "/exec" headers = {"Authorization": f"Bearer {settings.hermes_exec_token}"} try: async with httpx.AsyncClient(timeout=settings.hermes_exec_timeout) as c: r = await c.post(url, json={"prompt": prompt}, headers=headers) if r.status_code == 401: return "Refusé par l'exécuteur Hermes (jeton invalide)." r.raise_for_status() data = r.json() out = (data.get("output") or "").strip() return out or "Action effectuée (aucune sortie)." except httpx.HTTPError: return "Je n'arrive pas à joindre l'exécuteur Hermes pour le moment." except ValueError: return "Réponse inattendue de l'exécuteur Hermes."