mission-control-v2/backend/routers/voice.py

"""
Voice-Endpoints für „Mit Hermes reden" (Browser-Voice + 3D-Avatar).

Dünner Layer: STT/TTS werden zum Voice-Sidecar (:8650) geproxyt; der Chat geht an den
Hermes-`api_server` (:8642, OpenAI-kompatibel) — denselben vollen Agenten mit Tools +
geteiltem Mem0 wie CLI/Telegram. Mit stabilem `X-Hermes-Session-Id` hält die Plattform den
Transcript server-seitig, daher schickt der Client je Turn nur die neue User-Nachricht.

LAN-only (kein Token in der 2.0-Phase), wie die übrigen MC2-Endpoints.
"""

import logging
import os
import time

import httpx
from fastapi import APIRouter, File, Form, HTTPException, UploadFile
from fastapi.responses import Response, StreamingResponse
from pydantic import BaseModel

from config import HERMES_API_KEY, HERMES_API_MODEL, HERMES_API_URL, LLAMA_SWAP_URL, VOICE_SERVICE_URL
from services.voice_metrics import Timer, get_metrics, record_stage  # Per-Stage-Latenz (C2)

# Injection-Schutz (Stufe 0): guard.py liegt im mcp/-Verzeichnis. Per Pfad laden (eigene MC2-Venv).
import sys as _sys
_GUARD_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "mcp")
if _GUARD_DIR not in _sys.path:
    _sys.path.insert(0, _GUARD_DIR)
try:
    from guard import wrap_untrusted
except Exception:  # den Voice-Pfad nie wegen des Filters lahmlegen
    def wrap_untrusted(text: str, label: str = "") -> str:
        return text

log = logging.getLogger(__name__)
router = APIRouter(prefix="/api")

# Bildschirm-Sicht: das DEDIZIERTE Vision-Modell (Qwen3-VL-8B) beschreibt das Bild; die Beschreibung
# geht als TEXT an Hermes -> Lucy behält ihr volles Hirn/Gedächtnis UND nutzt das bessere VL-Modell
# (statt der schwächeren Vision der fast-MoE). Per Env abschaltbar/umstellbar.
VISION_MODEL = os.environ.get("MC_VISION_MODEL", "vision")
# Knappe Beschreibung = schnellere VL-Generierung UND weniger Hermes-Kontext-Bloat (B2).
VISION_MAX_TOKENS = int(os.environ.get("MC_VISION_MAX_TOKENS", "280"))


async def _describe_images(image_urls: list[str], hint: str) -> str:
    """Lässt das Vision-Modell die Screenshots (1 je Monitor) knapp beschreiben (Deutsch).
    Mehrere Bilder gehen in EINER Nachricht ans VL-Modell. Leerer String bei Fehler."""
    multi = len(image_urls) > 1
    intro = (f"Hier sind {len(image_urls)} Screenshots (je ein Monitor). Beschreibe auf Deutsch in höchstens "
             "5 kurzen Sätzen das Wesentliche (pro Monitor: App/Fenster, wichtige Inhalte, sichtbarer Text/Code). "
             "Keine Einleitung, keine Wiederholung der Frage. "
             if multi else
             "Beschreibe auf Deutsch in höchstens 5 kurzen Sätzen das Wesentliche auf diesem Screenshot "
             "(App/Fenster, wichtige Inhalte, sichtbarer Text/Code). Keine Einleitung. ")
    content: list = [{"type": "text", "text": intro + "Frage des Nutzers dazu: " + hint}]
    for u in image_urls:
        content.append({"type": "image_url", "image_url": {"url": u}})
    try:
        # 45 s statt 120 s: Qwen3-VL braucht warm ~5 s; wenn es 45 s nicht schafft, ist etwas
        # kaputt und Lucy soll lieber ohne Bildschirm-Kontext antworten als ewig hängen.
        async with httpx.AsyncClient(timeout=httpx.Timeout(float(os.environ.get("MC_VISION_TIMEOUT", "45")), connect=5.0)) as client:
            r = await client.post(f"{LLAMA_SWAP_URL}/v1/chat/completions", json={
                "model": VISION_MODEL, "max_tokens": VISION_MAX_TOKENS, "stream": False,
                "messages": [{"role": "user", "content": content}],
            })
            r.raise_for_status()
            return (r.json().get("choices") or [{}])[0].get("message", {}).get("content", "").strip()
    except Exception as exc:
        log.warning("Vision-Beschreibung fehlgeschlagen: %s", exc)
        return ""

_TIMEOUT = httpx.Timeout(120.0, connect=5.0)   # Chatterbox-TTS auf CPU darf dauern


class TTSIn(BaseModel):
    text: str
    engine: str = "piper"
    voice: str = ""
    language: str = ""
    ref_path: str = ""


class ChatIn(BaseModel):
    text: str                       # die neue User-Äußerung (STT-Ergebnis)
    session_id: str                 # stabiler Voice-Faden → server-seitiger Transcript
    session_key: str = ""           # optional: Langzeit-Memory-Scope
    system: str = ""                # optionaler ephemerer System-Prompt (z.B. „antworte knapp/gesprochen")
    model: str = ""
    images: list[str] = []          # optionale Bildschirm-Sicht: ein data:-URL je Monitor (Lucys „Augen")


@router.get("/voice/metrics")
def voice_metrics() -> dict:
    """Per-Stage-Latenz (STT/Vision/Chat-TTFB/TTS) — rollende Statistik, macht die Voice-Pipeline
    messbar (C2). Anzeige im Frontend-Overhaul (E)."""
    return get_metrics()


@router.get("/voice/health")
def voice_health() -> dict:
    """Erreichbarkeit des Voice-Sidecars + ob der Hermes-API-Key gesetzt ist."""
    out: dict = {"sidecar": False, "hermes_key": bool(HERMES_API_KEY)}
    try:
        r = httpx.get(f"{VOICE_SERVICE_URL}/health", timeout=httpx.Timeout(5.0))
        out["sidecar"] = r.status_code == 200
        out["detail"] = r.json() if r.status_code == 200 else None
    except Exception as exc:  # noqa: BLE001
        out["error"] = str(exc)
    return out


@router.get("/voice/voices")
def voice_voices() -> dict:
    try:
        r = httpx.get(f"{VOICE_SERVICE_URL}/voices", timeout=httpx.Timeout(10.0))
        r.raise_for_status()
        return r.json()
    except Exception as exc:  # noqa: BLE001
        raise HTTPException(502, f"Voice-Sidecar nicht erreichbar: {exc}")


@router.post("/voice/stt")
async def voice_stt(audio: UploadFile = File(...), language: str = Form(default="")) -> dict:
    """Mikro-Audio → Text (Proxy auf Sidecar /stt)."""
    data = await audio.read()
    if not data:
        raise HTTPException(400, "Leeres Audio.")
    files = {"audio": (audio.filename or "rec.webm", data, audio.content_type or "audio/webm")}
    try:
        async with httpx.AsyncClient(timeout=_TIMEOUT) as client:
            with Timer("stt"):
                r = await client.post(f"{VOICE_SERVICE_URL}/stt", files=files, data={"language": language})
            r.raise_for_status()
            return r.json()
    except httpx.HTTPError as exc:
        raise HTTPException(502, f"STT fehlgeschlagen: {exc}")


@router.post("/voice/turn")
async def voice_turn(audio: UploadFile = File(...)) -> dict:
    """Semantische Turn-Detection (Smart Turn v3): war die Äußerung fertig? Proxy → Sidecar."""
    data = await audio.read()
    if not data:
        raise HTTPException(400, "Leeres Audio.")
    files = {"audio": (audio.filename or "rec.wav", data, audio.content_type or "audio/wav")}
    try:
        async with httpx.AsyncClient(timeout=httpx.Timeout(10.0, connect=3.0)) as client:
            with Timer("turn"):
                r = await client.post(f"{VOICE_SERVICE_URL}/turn", files=files)
            r.raise_for_status()
            return r.json()
    except httpx.HTTPError as exc:
        # Turn-Check ist eine Optimierung — bei Ausfall lieber sofort antworten als hängen.
        log.warning("Turn-Check fehlgeschlagen: %s", exc)
        return {"complete": True, "probability": 1.0, "engine": "fallback"}


@router.post("/voice/reference")
async def voice_set_reference(audio: UploadFile = File(...)) -> dict:
    """Klon-Referenz (z.B. ElevenLabs-Erzeugnis) hochladen → Chatterbox nutzt sie. Proxy → Sidecar."""
    data = await audio.read()
    if not data:
        raise HTTPException(400, "Leeres Audio.")
    files = {"audio": (audio.filename or "ref.wav", data, audio.content_type or "audio/mpeg")}
    try:
        async with httpx.AsyncClient(timeout=_TIMEOUT) as client:
            r = await client.post(f"{VOICE_SERVICE_URL}/reference", files=files)
            r.raise_for_status()
            return r.json()
    except httpx.HTTPError as exc:
        raise HTTPException(502, f"Referenz-Upload fehlgeschlagen: {exc}")


@router.get("/voice/reference")
def voice_get_reference() -> dict:
    try:
        r = httpx.get(f"{VOICE_SERVICE_URL}/reference", timeout=httpx.Timeout(8.0))
        r.raise_for_status()
        return r.json()
    except Exception as exc:  # noqa: BLE001
        return {"active": False, "error": str(exc)}


@router.delete("/voice/reference")
def voice_clear_reference() -> dict:
    try:
        r = httpx.delete(f"{VOICE_SERVICE_URL}/reference", timeout=httpx.Timeout(8.0))
        r.raise_for_status()
        return r.json()
    except httpx.HTTPError as exc:
        raise HTTPException(502, f"Löschen fehlgeschlagen: {exc}")


@router.post("/voice/tts")
async def voice_tts(body: TTSIn) -> Response:
    """Text → Sprache (Proxy auf Sidecar /tts), liefert WAV-Bytes."""
    try:
        async with httpx.AsyncClient(timeout=_TIMEOUT) as client:
            with Timer("tts"):
                r = await client.post(f"{VOICE_SERVICE_URL}/tts", json=body.model_dump())
            r.raise_for_status()
            return Response(content=r.content, media_type=r.headers.get("content-type", "audio/wav"))
    except httpx.HTTPError as exc:
        raise HTTPException(502, f"TTS fehlgeschlagen: {exc}")


@router.post("/voice/chat")
async def voice_chat(body: ChatIn) -> StreamingResponse:
    """Neue User-Äußerung → Hermes-Agent (api_server, streamend). SSE wird 1:1 durchgereicht.

    Mit `X-Hermes-Session-Id` hält die Plattform den Verlauf — wir senden nur die neue Nachricht.
    Auth per Bearer (API_SERVER_KEY); ohne Key liefert :8642 ein 401."""
    if not HERMES_API_KEY:
        raise HTTPException(503, "HERMES_API_KEY/API_SERVER_KEY nicht gesetzt — Agent-Auth fehlt.")

    headers = {
        "Authorization": f"Bearer {HERMES_API_KEY}",
        "X-Hermes-Session-Id": body.session_id,
    }
    if body.session_key:
        headers["X-Hermes-Session-Key"] = body.session_key

    async def gen():
        t0 = time.perf_counter()
        first = True
        first_content = True
        # Bildschirm-Sicht INNERHALB des Streams (C2-Fix): so startet die SSE-Antwort sofort und
        # der Client bekommt ein Progress-Event (-> Lucy kann eine Warte-Ansage sprechen), statt
        # dass der Request bis zu 120 s "tot" hängt, während das Vision-Modell beschreibt.
        user_text = body.text
        imgs = [u for u in (body.images or []) if u]
        if imgs:
            yield b'event: hermes.vision.progress\ndata: {"note": "Bildschirm wird angeschaut"}\n\n'
            with Timer("vision"):
                desc = await _describe_images(imgs, body.text)
            if desc:
                safe_desc = wrap_untrusted(desc, "BILDSCHIRM")
                user_text = f"[Bildschirm-Sicht — das ist gerade auf dem/den Schirm(en) zu sehen:\n{safe_desc}\n]\n\n{body.text}"
        messages = []
        if body.system:
            messages.append({"role": "system", "content": body.system})
        messages.append({"role": "user", "content": user_text})
        payload = {"model": body.model or HERMES_API_MODEL, "messages": messages, "stream": True}
        # Lucys Hirn (Qwen3.6) ist ein Thinking-Modell -> für die gesprochene Assistentin Thinking AUS,
        # sonst generiert es tausende Reasoning-Token VOR der kurzen Antwort (gemessen: 11k Token, ~30s TTFB).
        # Gleiches Muster wie die fast-Spur im Gateway (gateway_proxy.py) und die Mem0-Extraktion.
        if os.environ.get("MC_VOICE_NO_THINK", "1") not in ("0", "false", "False"):
            payload["chat_template_kwargs"] = {"enable_thinking": False}
        try:
            async with httpx.AsyncClient(timeout=httpx.Timeout(None, connect=5.0)) as client:
                async with client.stream(
                    "POST", f"{HERMES_API_URL}/v1/chat/completions", json=payload, headers=headers,
                ) as r:
                    if r.status_code != 200:
                        detail = (await r.aread()).decode("utf-8", "replace")[:500]
                        yield f"data: {{\"error\": \"Hermes {r.status_code}: {detail}\"}}\n\n".encode()
                        return
                    async for chunk in r.aiter_raw():
                        if first:  # Time-To-First-Byte des Hermes-Streams (Verbindungs-Overhead)
                            record_stage("chat_ttfb", (time.perf_counter() - t0) * 1000.0)
                            first = False
                        # Erster CONTENT-Delta = echte Hirn-Latenz (Agent-Overhead + LLM-TTFT) —
                        # chat_ttfb misst nur den SSE-Start (~5 ms) und ist dafür blind.
                        if first_content and b'"content"' in chunk:
                            record_stage("chat_first_content", (time.perf_counter() - t0) * 1000.0)
                            first_content = False
                        yield chunk
        except httpx.HTTPError as exc:
            yield f"data: {{\"error\": \"Verbindung zu Hermes fehlgeschlagen: {exc}\"}}\n\n".encode()

    return StreamingResponse(gen(), media_type="text/event-stream")