tts-voice-creator-clone-and-design-2/routes/conversation.py

"""Conversation playground, LLM refinement, audio effects, export/import, speak, MCP."""
from __future__ import annotations

import asyncio
import base64
import io
import json
import time
import uuid
import wave
from pathlib import Path

import requests
from typing import Optional
from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile
from fastapi.responses import Response, StreamingResponse

from core.config import _load_settings, _save_settings, _clean_preview_backend
from core.constants import _VOICES_DIR_DEFAULT, _MAX_UPLOAD_BYTES
from core.registry import _registry_get, TEMP_DIR
from core.validation import _copy_limited
from core.audio import _to_wav_16k
from core.voice import (
    _AUDIO_EXTS, _UPLOAD_EXTS, _PICTURE_EXTS,
    _find_voice_audio, _load_meta, _active_voices_dir,
    _voice_audio_files, _is_internal_voice_file,
)
from core.tts_helpers import _preview_request_audio
from routes.stt import _transcribe_audio, _clean_stt_backend

router = APIRouter()

# ── LLM helpers ───────────────────────────────────────────────────────────────

def _rewrite_with_persona_sync(text: str, persona: str, llm_url: str, model: str = "") -> str:
    """Inline synchronous persona rewrite; raises RuntimeError on failure."""
    system = (
        f"Rephrase the user's text as if spoken by this character: {persona}\n"
        "Keep the same meaning but adapt vocabulary, tone, and style to the character. "
        "Return ONLY the rephrased text — no quotes, no explanation."
    )
    payload: dict = {
        "messages": [
            {"role": "system", "content": system},
            {"role": "user", "content": text},
        ],
        "temperature": 0.3,
        "max_tokens": 512,
    }
    if model:
        payload["model"] = model
    resp = requests.post(
        f"{llm_url.rstrip('/')}/chat/completions",
        json=payload,
        headers={"Authorization": "Bearer no-key"},
        timeout=60,
    )
    resp.raise_for_status()
    result = resp.json()["choices"][0]["message"]["content"].strip()
    if result.startswith('"') and result.endswith('"'):
        result = result[1:-1].strip()
    return result


def _resolve_speak_voice(settings: dict, client_id: str, explicit_voice: str) -> str:
    if explicit_voice:
        return explicit_voice
    bindings: dict = settings.get("client_voice_bindings") or {}
    if client_id and client_id in bindings:
        return bindings[client_id]
    return settings.get("captures_default_voice") or ""


# ── Routes ────────────────────────────────────────────────────────────────────

@router.post("/api/refine-text")
async def refine_text(request: Request):
    """Clean up raw STT transcription using a local OpenAI-compatible LLM."""
    data = await request.json()
    text: str = (data.get("text") or "").strip()
    llm_url: str = (data.get("llm_url") or "http://localhost:11434/v1").rstrip("/")
    model: str = (data.get("model") or "").strip()
    toggles: dict = data.get("toggles") or {}

    if not text:
        raise HTTPException(400, "No text to refine")

    rules = []
    if toggles.get("fillers", True):
        rules.append("Remove filler words (um, uh, like, you know, basically, literally, I mean, so, right, etc.)")
    if toggles.get("repetitions", True):
        rules.append("Remove repeated words and false starts (e.g. 'the the dog' → 'the dog', 'I was- I was going' → 'I was going')")
    if toggles.get("corrections", True):
        rules.append("Remove self-corrections and restarts, keeping only the final intended phrasing")
    if toggles.get("punctuation", True):
        rules.append("Fix punctuation, capitalisation, and sentence boundaries")
    if not rules:
        return {"text": text, "original": text}

    system = (
        "You are a transcription cleanup assistant. "
        "Apply ONLY the following rules to the user's text. "
        "Return ONLY the cleaned text — no explanations, no quotes, no markdown:\n"
        + "\n".join(f"- {r}" for r in rules)
    )
    payload: dict = {
        "messages": [
            {"role": "system", "content": system},
            {"role": "user", "content": text},
        ],
        "temperature": 0.1,
        "max_tokens": 2048,
    }
    if model:
        payload["model"] = model

    try:
        resp = requests.post(
            f"{llm_url}/chat/completions",
            json=payload,
            headers={"Authorization": "Bearer no-key"},
            timeout=60,
        )
        resp.raise_for_status()
        refined = resp.json()["choices"][0]["message"]["content"].strip()
        if refined.startswith('"') and refined.endswith('"'):
            refined = refined[1:-1].strip()
        return {"text": refined, "original": text}
    except Exception as e:
        raise HTTPException(502, f"LLM refinement failed: {e}")


@router.post("/api/rewrite-with-persona")
async def rewrite_with_persona(request: Request):
    """Rewrite user text in a voice persona's character using a local LLM."""
    data = await request.json()
    text: str = (data.get("text") or "").strip()
    persona: str = (data.get("persona") or "").strip()
    llm_url: str = (data.get("llm_url") or "http://localhost:11434/v1").rstrip("/")
    model: str = (data.get("model") or "").strip()
    mode: str = (data.get("mode") or "rewrite").strip()

    if not persona:
        raise HTTPException(400, "No persona defined for this voice")
    if not text and mode != "compose":
        raise HTTPException(400, "No text provided")

    if mode == "compose":
        system = (
            f"You are a voice assistant with this character: {persona}\n"
            "Write a single natural utterance in this character's voice about the topic given. "
            "Return ONLY the utterance — no quotes, no explanation."
        )
        user_msg = text or "Introduce yourself briefly."
        temp = 0.9
    else:
        system = (
            f"Rephrase the user's text as if spoken by this character: {persona}\n"
            "Keep the same meaning but adapt vocabulary, tone, and style to the character. "
            "Return ONLY the rephrased text — no quotes, no explanation."
        )
        user_msg = text
        temp = 0.3

    payload: dict = {
        "messages": [
            {"role": "system", "content": system},
            {"role": "user", "content": user_msg},
        ],
        "temperature": temp,
        "max_tokens": 512,
    }
    if model:
        payload["model"] = model

    try:
        resp = requests.post(
            f"{llm_url}/chat/completions",
            json=payload,
            headers={"Authorization": "Bearer no-key"},
            timeout=60,
        )
        resp.raise_for_status()
        result = resp.json()["choices"][0]["message"]["content"].strip()
        if result.startswith('"') and result.endswith('"'):
            result = result[1:-1].strip()
        return {"text": result, "original": text, "persona": persona}
    except Exception as e:
        raise HTTPException(502, f"LLM persona rewrite failed: {e}")


# ── Audio effects ─────────────────────────────────────────────────────────────

def _apply_audio_effects(audio_bytes: bytes, effects: list) -> bytes:
    try:
        from pedalboard import Pedalboard, Reverb, Chorus, Delay, Compressor, Gain, HighpassFilter, LowpassFilter, PitchShift  # type: ignore
        import numpy as np  # type: ignore
    except ImportError:
        raise RuntimeError("pedalboard is not installed — run: pip install pedalboard numpy")

    with io.BytesIO(audio_bytes) as buf:
        with wave.open(buf, "rb") as wf:
            n_channels = wf.getnchannels()
            sample_rate = wf.getframerate()
            n_frames = wf.getnframes()
            raw = wf.readframes(n_frames)
            sampwidth = wf.getsampwidth()

    import numpy as np  # noqa: F811
    dtype = {1: np.int8, 2: np.int16, 4: np.int32}.get(sampwidth, np.int16)
    samples = np.frombuffer(raw, dtype=dtype).astype(np.float32) / float(np.iinfo(dtype).max)
    samples = samples.reshape(1, -1) if n_channels == 1 else samples.reshape(-1, n_channels).T

    board = []
    for fx in effects:
        t = fx.get("type", "")
        p = fx.get("params", {})
        if t == "reverb":
            board.append(Reverb(
                room_size=float(p.get("room_size", 0.35)),
                damping=float(p.get("damping", 0.5)),
                wet_level=float(p.get("wet", 0.25)),
                dry_level=float(p.get("dry", 0.8)),
            ))
        elif t == "chorus":
            board.append(Chorus(
                rate_hz=float(p.get("rate_hz", 1.0)),
                depth=float(p.get("depth", 0.25)),
                mix=float(p.get("mix", 0.5)),
            ))
        elif t == "delay":
            board.append(Delay(
                delay_seconds=float(p.get("delay_s", 0.25)),
                feedback=float(p.get("feedback", 0.3)),
                mix=float(p.get("mix", 0.4)),
            ))
        elif t == "compressor":
            board.append(Compressor(
                threshold_db=float(p.get("threshold_db", -20.0)),
                ratio=float(p.get("ratio", 4.0)),
                attack_ms=float(p.get("attack_ms", 10.0)),
                release_ms=float(p.get("release_ms", 100.0)),
            ))
        elif t == "gain":
            board.append(Gain(gain_db=float(p.get("gain_db", 0.0))))
        elif t == "highpass":
            board.append(HighpassFilter(cutoff_frequency_hz=float(p.get("cutoff_hz", 80.0))))
        elif t == "lowpass":
            board.append(LowpassFilter(cutoff_frequency_hz=float(p.get("cutoff_hz", 8000.0))))
        elif t == "pitch_shift":
            board.append(PitchShift(semitones=float(p.get("semitones", 0.0))))

    if board:
        samples = Pedalboard(board)(samples, sample_rate)

    out = np.clip(samples, -1.0, 1.0)
    pcm = ((out[0] if out.shape[0] == 1 else out.T.reshape(-1)) * 32767).astype(np.int16).tobytes()
    buf_out = io.BytesIO()
    with wave.open(buf_out, "wb") as wf:
        wf.setnchannels(n_channels)
        wf.setsampwidth(2)
        wf.setframerate(sample_rate)
        wf.writeframes(pcm)
    return buf_out.getvalue()


@router.post("/api/audio/effects")
async def audio_effects(request: Request):
    """Apply an effects chain to a WAV file."""
    form = await request.form()
    audio_file = form.get("audio")
    effects_json = str(form.get("effects") or "[]")

    if audio_file is None:
        raise HTTPException(400, "No audio file provided")
    audio_bytes = await audio_file.read()
    try:
        effects = json.loads(effects_json)
    except Exception:
        raise HTTPException(400, "Invalid effects JSON")

    try:
        result = await asyncio.to_thread(_apply_audio_effects, audio_bytes, effects)
        return Response(content=result, media_type="audio/wav")
    except RuntimeError as e:
        raise HTTPException(501, str(e))
    except Exception as e:
        raise HTTPException(500, f"Effects processing failed: {e}")


# ── Voices export / import ────────────────────────────────────────────────────

_EXPORT_SKIP_KEYS = {"groq_api_key", "whisper_api_key", "tts_api_key", "voice_design_api_key", "elevenlabs_api_key"}
_IMPORT_ALLOWED_SUFFIXES = set(_AUDIO_EXTS + [".reference.txt", ".meta.json", ".jpg", ".jpeg", ".png", ".webp"])


@router.get("/api/voices/export")
async def voices_export():
    """Export all voices + non-sensitive settings as a ZIP archive."""
    import zipfile
    from datetime import datetime
    settings = _load_settings()
    scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))

    buf = io.BytesIO()
    with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
        safe_settings = {k: v for k, v in settings.items() if k not in _EXPORT_SKIP_KEYS}
        zf.writestr("settings.json", json.dumps(safe_settings, indent=2))
        if scan_dir.exists():
            for f in scan_dir.rglob("*"):
                if f.is_file():
                    try:
                        zf.write(f, str(f.relative_to(scan_dir.parent)))
                    except Exception:
                        pass
    buf.seek(0)
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    return Response(
        content=buf.read(),
        media_type="application/zip",
        headers={"Content-Disposition": f'attachment; filename="voices_export_{ts}.zip"'},
    )


@router.post("/api/voices/import")
async def voices_import(file: UploadFile = File(...)):
    """Import voices from a ZIP archive (skips settings.json and unsafe paths)."""
    import zipfile
    settings = _load_settings()
    scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
    scan_dir.mkdir(parents=True, exist_ok=True)

    content = await file.read()
    if len(content) > _MAX_UPLOAD_BYTES:
        raise HTTPException(413, "ZIP file too large")

    try:
        imported = 0
        with zipfile.ZipFile(io.BytesIO(content)) as zf:
            for info in zf.infolist():
                if info.is_dir():
                    continue
                parts = Path(info.filename).parts
                if any(p in ("..", "") for p in parts) or Path(info.filename).name == "settings.json":
                    continue
                if Path(info.filename).suffix.lower() not in _IMPORT_ALLOWED_SUFFIXES:
                    continue
                rel = parts[1:] if len(parts) > 1 else parts
                dest = scan_dir / Path(*rel)
                dest.parent.mkdir(parents=True, exist_ok=True)
                dest.write_bytes(zf.read(info))
                imported += 1
    except zipfile.BadZipFile:
        raise HTTPException(400, "Not a valid ZIP file")
    except Exception as e:
        raise HTTPException(500, f"Import failed: {e}")

    return {"ok": True, "imported": imported}


# ── /speak REST endpoint ──────────────────────────────────────────────────────

@router.post("/speak")
async def speak(request: Request):
    data = await request.json()
    text: str = str(data.get("text") or "").strip()
    if not text:
        raise HTTPException(400, "text is required")

    explicit_voice: str = str(data.get("voice") or data.get("profile_id") or data.get("profile") or "").strip()
    backend: str = _clean_preview_backend(str(data.get("backend") or "voice_clone"))
    apply_persona: bool = bool(data.get("apply_persona") or data.get("personality"))
    client_id: str = request.headers.get("X-Voice-Creator-Client-Id", "").strip()

    settings = _load_settings()
    voice = _resolve_speak_voice(settings, client_id, explicit_voice)
    if not voice:
        raise HTTPException(400, "No voice specified and no default voice configured")

    if apply_persona:
        scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
        wav = _find_voice_audio(voice, scan_dir)
        if wav:
            persona = _load_meta(wav).get("persona", "")
            if persona:
                llm_url = (settings.get("llm_url") or "http://localhost:11434/v1").rstrip("/")
                try:
                    text = await asyncio.to_thread(_rewrite_with_persona_sync, text, persona, llm_url)
                except Exception as e:
                    raise HTTPException(502, f"Persona rewrite failed: {e}")

    instruct = str(data.get("instruct") or "")
    try:
        audio, media_type = await asyncio.to_thread(
            _preview_request_audio, text, voice, settings, instruct, backend
        )
    except Exception as e:
        raise HTTPException(502, f"TTS error: {e}")

    return Response(content=audio, media_type=media_type)


@router.get("/speak/bindings")
async def get_speak_bindings():
    s = _load_settings()
    return {"bindings": s.get("client_voice_bindings") or {}}


@router.put("/speak/bindings/{client_id}")
async def put_speak_binding(client_id: str, request: Request):
    data = await request.json()
    voice = str(data.get("voice") or "").strip()
    if not voice:
        raise HTTPException(400, "voice is required")
    s = _load_settings()
    bindings: dict = dict(s.get("client_voice_bindings") or {})
    bindings[client_id] = voice
    s["client_voice_bindings"] = bindings
    _save_settings(s)
    return {"ok": True, "client_id": client_id, "voice": voice}


@router.delete("/speak/bindings/{client_id}")
async def delete_speak_binding(client_id: str):
    s = _load_settings()
    bindings: dict = dict(s.get("client_voice_bindings") or {})
    if client_id not in bindings:
        raise HTTPException(404, f"No binding for client '{client_id}'")
    del bindings[client_id]
    s["client_voice_bindings"] = bindings
    _save_settings(s)
    return {"ok": True, "client_id": client_id}


# ── MCP JSON-RPC 2.0 server ───────────────────────────────────────────────────

_MCP_SERVER_INFO = {"name": "tts-voice-creator", "version": "1.0.0"}

_MCP_TOOLS = [
    {
        "name": "speak",
        "description": "Generate speech audio from text using a cloned voice. Returns a data URI with the WAV audio.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "text":          {"type": "string",  "description": "Text to synthesize"},
                "voice":         {"type": "string",  "description": "Voice ID (optional — uses per-client binding or default if omitted)"},
                "apply_persona": {"type": "boolean", "description": "Rewrite text through the voice's persona before synthesis"},
            },
            "required": ["text"],
        },
    },
    {
        "name": "transcribe",
        "description": "Transcribe base64-encoded WAV audio to text.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "audio_base64": {"type": "string", "description": "Base64-encoded audio bytes (WAV preferred)"},
            },
            "required": ["audio_base64"],
        },
    },
    {
        "name": "list_captures",
        "description": "List the 20 most recently generated audio files.",
        "inputSchema": {"type": "object", "properties": {}},
    },
    {
        "name": "list_profiles",
        "description": "List all available voice profiles with their language, persona, and enabled state.",
        "inputSchema": {"type": "object", "properties": {}},
    },
]


async def _mcp_tool_speak(args: dict, client_id: str) -> dict:
    text = str(args.get("text") or "").strip()
    if not text:
        raise ValueError("text is required")
    explicit_voice = str(args.get("voice") or "").strip()
    apply_persona = bool(args.get("apply_persona", False))
    settings = _load_settings()
    voice = _resolve_speak_voice(settings, client_id, explicit_voice)
    if not voice:
        raise ValueError("No voice specified and no default voice configured")
    if apply_persona:
        scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
        wav = _find_voice_audio(voice, scan_dir)
        if wav:
            persona = _load_meta(wav).get("persona", "")
            if persona:
                llm_url = (settings.get("llm_url") or "http://localhost:11434/v1").rstrip("/")
                text = await asyncio.to_thread(_rewrite_with_persona_sync, text, persona, llm_url)
    audio, _media_type = await asyncio.to_thread(
        _preview_request_audio, text, voice, settings, "", "voice_clone"
    )
    audio_b64 = base64.b64encode(audio).decode()
    return {
        "content": [
            {"type": "text", "text": f"Generated {len(audio)} bytes of speech audio for voice '{voice}'."},
            {"type": "resource", "resource": {"uri": f"data:audio/wav;base64,{audio_b64}", "mimeType": "audio/wav"}},
        ]
    }


async def _mcp_tool_transcribe(args: dict) -> dict:
    raw = args.get("audio_base64") or ""
    try:
        audio_bytes = base64.b64decode(raw)
    except Exception:
        raise ValueError("audio_base64 is not valid base64")
    tmp = TEMP_DIR / f"{uuid.uuid4().hex}_mcp_transcribe.wav"
    tmp.write_bytes(audio_bytes)
    settings = _load_settings()
    try:
        text, used_backend = await asyncio.to_thread(_transcribe_audio, tmp, settings, "configured")
    finally:
        try:
            tmp.unlink(missing_ok=True)
        except Exception:
            pass
    return {"content": [{"type": "text", "text": text}], "backend": used_backend}


async def _mcp_tool_list_captures() -> dict:
    settings = _load_settings()
    out_dir = _active_voices_dir(settings)
    if not out_dir.exists():
        return {"content": [{"type": "text", "text": "[]"}]}
    files = sorted(
        (p for p in out_dir.rglob("*.wav") if not _is_internal_voice_file(p)),
        key=lambda p: p.stat().st_mtime,
        reverse=True,
    )[:20]
    result = [{"name": p.stem, "path": str(p), "mtime": p.stat().st_mtime} for p in files]
    return {"content": [{"type": "text", "text": json.dumps(result)}]}


async def _mcp_tool_list_profiles() -> dict:
    settings = _load_settings()
    scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
    profiles = []
    seen: set[str] = set()
    for audio in sorted(_voice_audio_files(scan_dir), key=lambda p: p.stem.lower()):
        if audio.stem in seen:
            continue
        seen.add(audio.stem)
        meta = _load_meta(audio)
        parts = audio.stem.split("_", 2)
        lang = parts[0].upper() if parts else ""
        profiles.append({
            "id":      audio.stem,
            "name":    audio.stem,
            "lang":    lang,
            "persona": meta.get("persona", ""),
            "enabled": meta.get("enabled", True),
        })
    return {"content": [{"type": "text", "text": json.dumps(profiles)}]}


def _mcp_error_response(code: int, message: str, rpc_id) -> Response:
    import logging
    body = {"jsonrpc": "2.0", "error": {"code": code, "message": message}, "id": rpc_id}
    return Response(content=json.dumps(body), media_type="application/json")


@router.post("/mcp")
async def mcp_jsonrpc(request: Request):
    import logging
    logger = logging.getLogger("uvicorn.error")
    try:
        body = await request.json()
    except Exception:
        return _mcp_error_response(-32700, "Parse error", None)

    rpc_id    = body.get("id")
    method    = body.get("method", "")
    params    = body.get("params") or {}
    client_id = request.headers.get("X-Voice-Creator-Client-Id", "").strip()

    try:
        if method == "initialize":
            result = {
                "protocolVersion": "2024-11-05",
                "serverInfo": _MCP_SERVER_INFO,
                "capabilities": {"tools": {}},
            }
        elif method == "notifications/initialized":
            return Response(status_code=204)
        elif method == "tools/list":
            result = {"tools": _MCP_TOOLS}
        elif method == "tools/call":
            tool_name = str(params.get("name") or "")
            tool_args  = params.get("arguments") or {}
            if tool_name == "speak":
                result = await _mcp_tool_speak(tool_args, client_id)
            elif tool_name == "transcribe":
                result = await _mcp_tool_transcribe(tool_args)
            elif tool_name == "list_captures":
                result = await _mcp_tool_list_captures()
            elif tool_name == "list_profiles":
                result = await _mcp_tool_list_profiles()
            else:
                return _mcp_error_response(-32601, f"Unknown tool: {tool_name}", rpc_id)
        else:
            return _mcp_error_response(-32601, f"Method not found: {method}", rpc_id)
    except ValueError as e:
        return _mcp_error_response(-32602, str(e), rpc_id)
    except Exception as e:
        logger.exception("MCP tool error in method %s", method)
        return _mcp_error_response(-32000, f"Server error: {e}", rpc_id)

    return Response(
        content=json.dumps({"jsonrpc": "2.0", "result": result, "id": rpc_id}),
        media_type="application/json",
    )


@router.get("/mcp")
async def mcp_sse(request: Request):
    """SSE keep-alive stream (satisfies MCP spec GET /mcp requirement)."""
    from typing import AsyncGenerator

    async def _keepalive() -> AsyncGenerator[str, None]:
        yield ": mcp-sse-ready\n\n"
        while True:
            await asyncio.sleep(15)
            yield ": ping\n\n"
    return StreamingResponse(
        _keepalive(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


# ── Conversation playground ───────────────────────────────────────────────────

@router.get("/api/conversation/llm-models")
async def conversation_llm_models(url: str = ""):
    """List models from a local LLM endpoint (Ollama / vLLM / LM Studio)."""
    from core.validation import _validate_http_url
    settings = _load_settings()
    base = (url or settings.get("llm_url") or "http://localhost:11434/v1").rstrip("/")
    try:
        base = _validate_http_url(base, allow_private=True)
        r = requests.get(f"{base}/models", timeout=5, headers={"Authorization": "Bearer no-key"})
        if r.status_code == 200:
            payload = r.json()
            data = payload.get("data", []) if isinstance(payload, dict) else []
            models = [
                str(item["id"]) if isinstance(item, dict) and item.get("id") else str(item)
                for item in data if item
            ]
            return {"models": models, "url": base}
    except Exception:
        pass
    return {"models": [], "url": base}


@router.post("/api/conversation/turn")
async def conversation_turn(
    audio:         Optional[UploadFile] = None,
    text:          str = Form(""),
    stt_backend:   str = Form("configured"),
    llm_url:       str = Form(""),
    llm_model:     str = Form(""),
    tts_backend:   str = Form("voice_clone"),
    tts_voice:     str = Form(""),
    system_prompt: str = Form("You are a helpful voice assistant. Keep replies short and conversational."),
    history:       str = Form("[]"),
):
    """Stream a full conversation turn (STT → LLM → TTS) as Server-Sent Events.
    Pass either an audio file (runs STT first) or a plain text string (skips STT).
    """
    direct_text = text.strip()
    if not direct_text and (audio is None or not getattr(audio, "filename", None)):
        raise HTTPException(400, "Provide either an audio file or a text field")

    settings = _load_settings()
    eff_llm_url = (llm_url or settings.get("llm_url") or "http://localhost:11434/v1").rstrip("/")

    # Prepare audio temp files only when an audio upload was provided
    _tmp: Path | None = None
    _wav: Path | None = None
    if not direct_text and audio is not None:
        suffix = Path(audio.filename or "audio.webm").suffix.lower() or ".webm"
        if suffix not in _UPLOAD_EXTS:
            suffix = ".webm"
        _tmp = TEMP_DIR / f"{uuid.uuid4().hex}_conv{suffix}"
        try:
            with _tmp.open("wb") as f:
                _copy_limited(audio.file, f, _MAX_UPLOAD_BYTES)
            _wav = _to_wav_16k(_tmp)
        except Exception as e:
            _tmp.unlink(missing_ok=True)
            raise HTTPException(400, f"Audio upload failed: {e}")

    try:
        hist = json.loads(history) if history else []
        if not isinstance(hist, list):
            hist = []
    except Exception:
        hist = []

    stt_be = _clean_stt_backend(stt_backend)
    tts_be = _clean_preview_backend(tts_backend)

    async def generate():
        t0 = time.monotonic()
        stt_ms = llm_ttft_ms = llm_total_ms = tts_ms = None
        transcript = llm_text = ""

        def sse(obj: dict) -> str:
            return f"data: {json.dumps(obj)}\n\n"

        # 1. STT — skipped when caller sends direct text
        if direct_text:
            transcript = direct_text
            yield sse({"type": "transcript", "text": transcript, "stt_ms": None})
        else:
            try:
                t_stt = time.monotonic()
                transcript, _ = await asyncio.to_thread(_transcribe_audio, _wav, settings, stt_be)
                stt_ms = int((time.monotonic() - t_stt) * 1000)
                yield sse({"type": "transcript", "text": transcript, "stt_ms": stt_ms})
            except Exception as e:
                yield sse({"type": "error", "stage": "stt", "message": str(e)})
                return
            finally:
                for p in filter(None, {_tmp, _wav}):
                    with contextlib.suppress(Exception):
                        p.unlink(missing_ok=True)

        if not transcript.strip():
            yield sse({"type": "error", "stage": "stt", "message": "No speech detected."})
            return

        # 2. LLM stream
        messages = [{"role": "system", "content": system_prompt}]
        messages.extend(hist[-20:])
        messages.append({"role": "user", "content": transcript})
        llm_payload: dict = {"messages": messages, "stream": True, "temperature": 0.7, "max_tokens": 512}
        if llm_model:
            llm_payload["model"] = llm_model
        try:
            t_llm = time.monotonic()
            llm_resp = await asyncio.to_thread(lambda: requests.post(
                f"{eff_llm_url}/chat/completions", json=llm_payload,
                headers={"Authorization": "Bearer no-key"}, stream=True, timeout=120,
            ))
            llm_resp.raise_for_status()
            ttft_done = False
            for raw_line in llm_resp.iter_lines():
                if not raw_line:
                    continue
                line = raw_line.decode("utf-8") if isinstance(raw_line, bytes) else str(raw_line)
                if not line.startswith("data:"):
                    continue
                chunk = line[5:].strip()
                if chunk == "[DONE]":
                    break
                try:
                    obj = json.loads(chunk)
                    d_obj = ((obj.get("choices") or [{}])[0].get("delta") or {})
                    # Primary: visible content. Fallback: reasoning_content (Qwen3 /think tokens)
                    delta = d_obj.get("content") or d_obj.get("reasoning_content") or ""
                    if not delta and isinstance(obj.get("message"), dict):
                        delta = obj["message"].get("content") or ""
                    if delta:
                        if not ttft_done:
                            llm_ttft_ms = int((time.monotonic() - t_llm) * 1000)
                            ttft_done = True
                        llm_text += delta
                        yield sse({"type": "token", "delta": delta})
                except Exception:
                    continue
            llm_total_ms = int((time.monotonic() - t_llm) * 1000)
            yield sse({"type": "llm_done", "text": llm_text,
                       "llm_ttft_ms": llm_ttft_ms, "llm_total_ms": llm_total_ms})
        except Exception as e:
            yield sse({"type": "error", "stage": "llm", "message": str(e)})
            return

        if not llm_text.strip():
            yield sse({"type": "error", "stage": "llm",
                       "message": "LLM returned empty response. "
                                  "If using a Qwen3 model, try adding /no-think to your system prompt "
                                  "or pick a non-thinking model in the Language Model dropdown."})
            return

        # 3. TTS
        try:
            t_tts = time.monotonic()
            audio_bytes, mime = await asyncio.to_thread(
                _preview_request_audio, llm_text, tts_voice, settings, "", tts_be
            )
            tts_ms = int((time.monotonic() - t_tts) * 1000)
            total_ms = int((time.monotonic() - t0) * 1000)
            yield sse({"type": "audio", "b64": base64.b64encode(audio_bytes).decode(), "mime": mime})
            yield sse({"type": "stats", "stt_ms": stt_ms, "llm_ttft_ms": llm_ttft_ms,
                       "llm_total_ms": llm_total_ms, "tts_ms": tts_ms, "total_ms": total_ms})
        except Exception as e:
            yield sse({"type": "error", "stage": "tts", "message": str(e)})
            return

        yield sse({"type": "done"})

    return StreamingResponse(generate(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})