"""
Jarvis Brain Bridge
===================

A thin local HTTP service that exposes the existing Jarvis "brain"
(speech-to-text + reply engine + text-to-speech) to the Node/bun Discord bot.

The Discord layer (``bot/``) is responsible for everything Discord-specific:
joining voice channels, capturing user audio, playing audio back, slash
commands, and streaming the VNC screen. It does NOT contain any AI logic.
Instead it calls this bridge:

    POST /converse        (multipart wav)  -> { transcript, reply, audio_b64 }
    POST /text            (json {text})    -> { reply, audio_b64 }
    POST /stt             (multipart wav)  -> { text, language }
    POST /tts             (json {text})    -> { audio_b64 }
    GET  /health                            -> { ok, brain, stt, tts }

This keeps the mature ~39k-line Python brain intact while letting Node own the
Discord/voice/video integration (which is only feasible in the Node ecosystem).

Run:
    python -m bridge.server          # from repo root
    # or
    BRIDGE_HOST=127.0.0.1 BRIDGE_PORT=8765 python bridge/server.py
"""

from __future__ import annotations

import base64
import io
import json
import os
import sys
import threading
import wave
from pathlib import Path
from typing import Optional

# Ensure repo-root/src is importable (jarvis package lives in src/jarvis) and
# the repo root itself (so ``bridge.text_utils`` resolves whether this module is
# launched as ``python -m bridge.server`` or ``python bridge/server.py``).
_REPO_ROOT = Path(__file__).resolve().parent.parent
_SRC = _REPO_ROOT / "src"
if str(_SRC) not in sys.path:
    sys.path.insert(0, str(_SRC))
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

from flask import Flask, request, jsonify, Response, stream_with_context

try:  # package-relative when imported as ``bridge.server``
    from bridge.text_utils import split_sentences
    from bridge.stt_filter import filter_speech_segments, has_speech
    from bridge import settings_web
except ImportError:  # script-relative when run as ``bridge/server.py``
    from text_utils import split_sentences
    from stt_filter import filter_speech_segments, has_speech
    import settings_web

app = Flask(__name__)
# Settings web UI (/settings) — change models/language/TTS/instructions live.
try:
    settings_web.register(app)
except Exception as _e:  # pragma: no cover - never block the bridge on the UI
    print(f"[bridge] settings UI unavailable: {_e}", flush=True)

# ---------------------------------------------------------------------------
# Configuration (env-driven; see .env.example)
# ---------------------------------------------------------------------------
BRIDGE_HOST = os.environ.get("BRIDGE_HOST", "127.0.0.1")
BRIDGE_PORT = int(os.environ.get("BRIDGE_PORT", "8765"))
BRAIN_ENABLED = os.environ.get("JARVIS_BRAIN_ENABLED", "1") not in ("0", "false", "False")
TTS_ENABLED = os.environ.get("JARVIS_TTS_ENABLED", "1") not in ("0", "false", "False")

# Pre-STT speech gate (Silero VAD). Tunable for the Discord mic without a code
# change: raise VAD_THRESHOLD to reject more noise, lower it to catch quieter
# speech. VAD_MIN_SPEECH_MS is the shortest run of speech that counts (a brief
# loud blip shorter than this never reaches Whisper). Set VAD_ENABLED=0 to fall
# back to the old behaviour (always transcribe, rely on the post-filter only).
VAD_ENABLED = os.environ.get("VAD_ENABLED", "1") not in ("0", "false", "False")
VAD_THRESHOLD = float(os.environ.get("VAD_THRESHOLD", "0.4"))
VAD_MIN_SPEECH_MS = int(os.environ.get("VAD_MIN_SPEECH_MS", "200"))

# Lock STT to a single language (this deployment is Korean-only). Skipping
# Whisper's language auto-detect both fixes occasional mis-detection (e.g. a
# Korean phrase decoded as Chinese) and shaves a little latency. Empty = auto.
STT_LANGUAGE = os.environ.get("STT_LANGUAGE", "ko").strip() or None

# Whisper decoding accuracy knobs. beam_size=1 is greedy decoding — fast but the
# least accurate; beam search (5 is the Whisper default) explores alternatives
# and noticeably improves recognition on short, accented, or noisy Discord-mic
# speech. condition_on_previous_text=False stops Whisper from feeding a previous
# clip's transcript back in as a prompt, which on isolated short utterances
# causes repetition loops and drift rather than helping. Both are env-tunable so
# accuracy/latency can be traded without a code change (lower STT_BEAM_SIZE for
# speed, raise it for accuracy).
STT_BEAM_SIZE = max(1, int(os.environ.get("STT_BEAM_SIZE", "5")))
STT_CONDITION_ON_PREV = os.environ.get("STT_CONDITION_ON_PREV", "0") in ("1", "true", "True", "yes", "on")

# TTS engine: "edge" (Microsoft Edge TTS, natural Korean neural voice) is the
# primary voice. "melo" (a warm MeloTTS worker) and "piper" remain selectable.
def _tts_engine_setting() -> str:
    """TTS engine: settings-UI value (runtime config JSON) wins, else env, else
    edge. Read at startup; the settings UI restarts the bridge on apply."""
    try:
        _cp = os.environ.get("JARVIS_CONFIG_PATH", "/app/config/jarvis.json")
        _v = json.loads(open(_cp, encoding="utf-8").read()).get("tts_engine")
        if _v:
            return str(_v).strip().lower()
    except Exception:
        pass
    return os.environ.get("TTS_ENGINE", "edge").strip().lower()


TTS_ENGINE = _tts_engine_setting()
# Edge TTS (online MS neural voice). Voice + rate are env-driven so they can be
# changed without code. Default: Korean "Hyunsu" multilingual voice at +45%
# (≈1.45×), the chosen settings. NOTE: edge synthesis sends the reply TEXT to
# Microsoft's servers and needs internet — an intentional privacy trade-off for
# the more natural voice.
EDGE_TTS_VOICE = os.environ.get("EDGE_TTS_VOICE", "ko-KR-HyunsuMultilingualNeural").strip()
EDGE_TTS_RATE = os.environ.get("EDGE_TTS_RATE", "+45%").strip()
MELO_WORKER_URL = os.environ.get("MELO_WORKER_URL", "http://127.0.0.1:8770")
MELO_TIMEOUT = float(os.environ.get("MELO_TIMEOUT", "30"))
# Do NOT silently fall back to the English Piper voice on a neural-voice failure:
# speaking Korean through an English voice produces mangled audio. Default is
# neural-only (return no audio on failure); set MELO_FALLBACK_PIPER=1 to opt in.
MELO_FALLBACK_PIPER = os.environ.get("MELO_FALLBACK_PIPER", "0") in ("1", "true", "True", "yes", "on")

# ---------------------------------------------------------------------------
# Lazy singletons. The first request pays the model-load cost; afterwards the
# brain stays warm. A lock guards initialization so concurrent Discord events
# don't double-load Whisper.
# ---------------------------------------------------------------------------
_init_lock = threading.Lock()
_cfg = None
_db = None
_dialogue_memory = None
_whisper = None
_piper_voice = None
_brain_error: Optional[str] = None


def _ensure_brain():
    """Initialize cfg, db, dialogue memory, and Whisper once."""
    global _cfg, _db, _dialogue_memory, _whisper, _brain_error
    if _cfg is not None or _brain_error is not None:
        return
    with _init_lock:
        if _cfg is not None or _brain_error is not None:
            return
        try:
            from jarvis.config import load_settings
            from jarvis.memory.db import Database
            from jarvis.memory.conversation import DialogueMemory
            from faster_whisper import WhisperModel

            cfg = load_settings()
            db = Database(cfg.db_path, cfg.sqlite_vss_path)
            dialogue_memory = DialogueMemory(
                inactivity_timeout=getattr(cfg, "dialogue_memory_timeout", 300.0),
                max_interactions=20,
            )
            device = os.environ.get("WHISPER_DEVICE", "auto")
            compute = os.environ.get("WHISPER_COMPUTE_TYPE", "auto")
            try:
                whisper = WhisperModel(cfg.whisper_model, device=device, compute_type=compute)
                # Log the device actually resolved by CTranslate2 (device="auto"
                # picks cuda when available) so a silent CPU load is visible.
                resolved = str(getattr(getattr(whisper, "model", None), "device", device)).lower()
                print(f"[bridge] whisper loaded on {resolved} (compute={compute})", flush=True)
            except Exception as ge:
                # GPU not available / unsupported -> fall back to CPU so the
                # bridge still works without a GPU passed to the container.
                if device != "cpu":
                    print(f"[bridge] whisper device='{device}' failed ({ge}); falling back to CPU", flush=True)
                    whisper = WhisperModel(cfg.whisper_model, device="cpu", compute_type="int8")
                    print("[bridge] whisper loaded on cpu (compute=int8)", flush=True)
                else:
                    raise

            _cfg, _db, _dialogue_memory, _whisper = cfg, db, dialogue_memory, whisper
            print(f"[bridge] brain ready (chat={cfg.ollama_chat_model}, whisper={cfg.whisper_model})", flush=True)
        except Exception as e:  # pragma: no cover - depends on local models
            _brain_error = f"{type(e).__name__}: {e}"
            print(f"[bridge] brain init FAILED: {_brain_error}", flush=True)


def _ensure_piper():
    """Initialize the Piper TTS voice once (independent of the brain)."""
    global _piper_voice
    if _piper_voice is not None or not TTS_ENABLED:
        return
    with _init_lock:
        if _piper_voice is not None:
            return
        try:
            from piper import PiperVoice  # piper-tts package
            model_path = os.environ.get("TTS_PIPER_MODEL_PATH")
            if not model_path:
                # Fall back to jarvis' default piper model location.
                from jarvis.output.tts import _get_default_piper_model_path  # type: ignore
                model_path = _get_default_piper_model_path()
            if not model_path or not Path(model_path).exists():
                raise FileNotFoundError(
                    f"Piper voice model not found at '{model_path}'. "
                    f"Set TTS_PIPER_MODEL_PATH in .env or run scripts/setup_models.sh"
                )
            _piper_voice = PiperVoice.load(model_path)
            print(f"[bridge] piper TTS ready ({model_path})", flush=True)
        except Exception as e:  # pragma: no cover
            print(f"[bridge] piper init failed (TTS disabled): {e}", flush=True)


# ---------------------------------------------------------------------------
# Core operations
# ---------------------------------------------------------------------------
def _read_wav_pcm(raw: bytes) -> tuple[bytes, int]:
    """Decode an incoming WAV blob to mono 16-bit PCM @ its sample rate."""
    with wave.open(io.BytesIO(raw), "rb") as wf:
        sr = wf.getframerate()
        frames = wf.readframes(wf.getnframes())
    return frames, sr


def transcribe(wav_bytes: bytes) -> dict:
    _ensure_brain()
    if _whisper is None:
        return {"text": "", "language": None, "error": _brain_error or "stt unavailable"}
    import numpy as np

    pcm, sr = _read_wav_pcm(wav_bytes)
    audio = np.frombuffer(pcm, dtype=np.int16).astype(np.float32) / 32768.0
    # faster-whisper expects 16kHz mono float32; linearly resample if needed.
    if sr != 16000 and audio.size:
        n_out = int(round(audio.size * 16000 / sr))
        if n_out > 0:
            x_old = np.linspace(0.0, 1.0, num=audio.size, endpoint=False)
            x_new = np.linspace(0.0, 1.0, num=n_out, endpoint=False)
            audio = np.interp(x_new, x_old, audio).astype(np.float32)
    # Pre-STT speech gate: don't even invoke Whisper unless there is real speech
    # in the clip. Noise or a brief loud blip (no actual speech) is dropped here,
    # before transcription, so the model never gets a chance to hallucinate a
    # phrase from it. Fail-open inside has_speech() keeps a real utterance from
    # being swallowed if the VAD is unavailable.
    if VAD_ENABLED and not has_speech(
        audio,
        16000,
        threshold=VAD_THRESHOLD,
        min_speech_duration_ms=VAD_MIN_SPEECH_MS,
        log=lambda m: print(f"[bridge] {m}", flush=True),
    ):
        print("[bridge] no speech detected (VAD) — skipping STT", flush=True)
        return {"text": "", "language": None, "note": "음성 아님(VAD 차단)"}

    segments, info = _whisper.transcribe(
        audio,
        beam_size=STT_BEAM_SIZE,
        language=STT_LANGUAGE,
        condition_on_previous_text=STT_CONDITION_ON_PREV,
    )
    # Second line of defence: drop non-speech / hallucinated segments by
    # Whisper's own no_speech_prob. The no_speech_prob hard cutoff (plus the VAD
    # pre-gate above) is what rejects noise/hallucinations. The avg_logprob
    # CONFIDENCE floor is deliberately OFF by default (STT_MIN_CONFIDENCE=0):
    # short, accented, or quiet real speech over a Discord mic scores very low
    # avg_logprob (e.g. the wake word "자비스" at 0.0-0.3) and a confidence floor
    # silently eats it, making the bot need many tries to hear one utterance.
    # Raise STT_MIN_CONFIDENCE only if hallucinations slip past the no_speech gate.
    no_speech_threshold = float(os.environ.get("STT_NO_SPEECH_THRESHOLD", str(getattr(_cfg, "whisper_no_speech_threshold", 0.5))))
    min_confidence = float(os.environ.get("STT_MIN_CONFIDENCE", "0.0"))
    kept = filter_speech_segments(
        segments,
        no_speech_threshold=no_speech_threshold,
        min_confidence=min_confidence,
        log=lambda m: print(f"[bridge] {m}", flush=True),
    )
    text = "".join(seg.text for seg in kept).strip()
    note = "ok" if text else "인식 실패(빈 결과/필터)"
    return {"text": text, "language": getattr(info, "language", None), "note": note}


def think(text: str, language: Optional[str] = None, broadcasting: Optional[bool] = None) -> dict:
    """Run the Jarvis reply engine on a piece of text.

    ``broadcasting`` is the bot's live screen-share state for this turn; it is
    stashed in request-scoped state so the search routing can pick Chrome vs
    Gemini. If the reply engine calls the setBroadcast tool, the recorded
    directive is returned as ``broadcast_action`` for the bot to act on.
    """
    if not BRAIN_ENABLED:
        return {"reply": text, "error": "brain disabled (JARVIS_BRAIN_ENABLED=0)"}
    _ensure_brain()
    if _cfg is None:
        return {"reply": "", "error": _brain_error or "brain unavailable"}
    try:
        from jarvis.reply.engine import run_reply_engine
        from jarvis.reply import turn_state

        turn_state.reset()
        turn_state.set_broadcasting(broadcasting)

        # tts=None: we do our own Discord-side synthesis, the engine must not
        # try to speak to a local speaker that doesn't exist in this process.
        reply = run_reply_engine(
            _db, _cfg, None, text, _dialogue_memory, language=language
        )
        reply = (reply or "").strip()
        if reply:
            _dialogue_memory.add_interaction(text, reply)
        return {"reply": reply, "broadcast_action": turn_state.get_broadcast_action()}
    except Exception as e:  # pragma: no cover
        return {"reply": "", "error": f"{type(e).__name__}: {e}"}


def _coerce_bool(value) -> Optional[bool]:
    """Parse a broadcasting flag from JSON (bool) or a query string."""
    if value is None:
        return None
    if isinstance(value, bool):
        return value
    return str(value).strip().lower() in ("1", "true", "yes", "on")


def _edge_synthesize(text: str) -> Optional[bytes]:
    """Synthesise via Microsoft Edge TTS (online neural voice) and return a
    16-bit PCM WAV, or None on any failure. Edge emits MP3; we transcode to
    PCM16 mono with the system ffmpeg, writing to a temp file (seekable) so the
    WAV header carries a correct length. Needs internet."""
    import asyncio
    import subprocess
    import tempfile

    try:
        import edge_tts  # type: ignore

        async def _gen() -> bytes:
            comm = edge_tts.Communicate(text, EDGE_TTS_VOICE, rate=EDGE_TTS_RATE)
            buf = bytearray()
            async for chunk in comm.stream():
                if chunk.get("type") == "audio":
                    buf.extend(chunk["data"])
            return bytes(buf)

        mp3 = asyncio.run(_gen())
        if not mp3:
            print("[bridge] edge TTS returned no audio", flush=True)
            return None
        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as t:
            out_path = t.name
        try:
            proc = subprocess.run(
                ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
                 "-i", "pipe:0", "-ac", "1", "-ar", "24000",
                 "-acodec", "pcm_s16le", out_path],
                input=mp3, capture_output=True,
            )
            if proc.returncode != 0:
                print(f"[bridge] edge ffmpeg transcode failed: {proc.stderr.decode('utf-8','ignore')[:200]}", flush=True)
                return None
            with open(out_path, "rb") as f:
                return f.read()
        finally:
            try:
                os.unlink(out_path)
            except OSError:
                pass
    except Exception as e:  # pragma: no cover - network / dep dependent
        print(f"[bridge] edge synth failed: {e}", flush=True)
    return None


def _melo_synthesize(text: str) -> Optional[bytes]:
    """Synthesise via the warm MeloTTS worker (separate /opt/melo venv, Korean
    speaker @ speed 1.5). Returns a 16-bit PCM WAV, or None on any failure so
    the caller can fall back to Piper."""
    import urllib.request

    try:
        req = urllib.request.Request(
            f"{MELO_WORKER_URL}/synth",
            data=json.dumps({"text": text}).encode("utf-8"),
            headers={"Content-Type": "application/json"},
        )
        with urllib.request.urlopen(req, timeout=MELO_TIMEOUT) as resp:
            if resp.status == 200:
                return resp.read()
            print(f"[bridge] melo worker HTTP {resp.status}", flush=True)
    except Exception as e:  # pragma: no cover - worker may be down
        print(f"[bridge] melo worker unreachable: {e}", flush=True)
    return None


def _piper_synthesize(text: str) -> Optional[bytes]:
    """Fallback: synthesise with Piper (English voice). Returns WAV bytes."""
    _ensure_piper()
    if _piper_voice is None:
        return None
    buf = io.BytesIO()
    with wave.open(buf, "wb") as wf:
        # piper-tts API: synthesize_wav(text, wav_file) writes a full WAV;
        # plain synthesize() returns AudioChunks and takes a SynthesisConfig
        # (NOT a wav file) as its 2nd arg.
        _piper_voice.synthesize_wav(text, wf)
    return buf.getvalue()


def _tts_ready() -> bool:
    """Whether the configured TTS voice can synthesise right now.

    The bot polls this before logging in so the very first spoken reply is not
    silently dropped while the voice is still warming up. For MeloTTS the worker
    only binds its HTTP port AFTER the model is loaded (``main()`` warms the
    model before ``serve_forever()``), so a successful /health ping is a precise
    "voice is warm" signal. Piper loads on first synth and was never gated, so
    it reports ready. TTS disabled means there is nothing to wait for.
    """
    if not TTS_ENABLED:
        return True
    if TTS_ENGINE == "melo":
        import urllib.request

        try:
            with urllib.request.urlopen(f"{MELO_WORKER_URL}/health", timeout=2) as resp:
                return resp.status == 200
        except Exception:
            return False
    return True


def synthesize(text: str) -> Optional[bytes]:
    """Synthesize text to a 16-bit PCM WAV. The primary voice is Edge TTS (a
    natural Korean neural voice); "melo" uses the warm MeloTTS worker. For a
    neural engine, Piper (English) is only used if explicitly enabled, since
    speaking Korean through an English voice mangles it. Returns None if off."""
    if not TTS_ENABLED or not text.strip():
        return None
    _neural = {"edge": _edge_synthesize, "melo": _melo_synthesize}.get(TTS_ENGINE)
    if _neural is not None:
        audio = _neural(text)
        if audio:
            return audio
        if not MELO_FALLBACK_PIPER:
            # Neural-only: better silent than mangled English for Korean text.
            print(f"[bridge] {TTS_ENGINE} synth failed; no audio (Piper fallback disabled)", flush=True)
            return None
        print(f"[bridge] {TTS_ENGINE} synth failed; falling back to Piper", flush=True)
    return _piper_synthesize(text)


# ---------------------------------------------------------------------------
# HTTP endpoints
# ---------------------------------------------------------------------------
@app.get("/health")
def health():
    return jsonify(
        {
            "ok": True,
            "brain_enabled": BRAIN_ENABLED,
            "brain_ready": _cfg is not None,
            "brain_error": _brain_error,
            "tts_enabled": TTS_ENABLED,
            "tts_engine": TTS_ENGINE,
            "tts_ready": _tts_ready(),
        }
    )


@app.post("/stt")
def http_stt():
    raw = request.get_data()
    if not raw:
        return jsonify({"error": "empty body; send a WAV blob"}), 400
    return jsonify(transcribe(raw))


@app.post("/text")
def http_text():
    data = request.get_json(silent=True) or {}
    text = (data.get("text") or "").strip()
    if not text:
        return jsonify({"error": "missing 'text'"}), 400
    result = think(text, data.get("language"), _coerce_bool(data.get("broadcasting")))
    audio = synthesize(result.get("reply", ""))
    if audio:
        result["audio_b64"] = base64.b64encode(audio).decode("ascii")
    return jsonify(result)


@app.post("/tts")
def http_tts():
    data = request.get_json(silent=True) or {}
    text = (data.get("text") or "").strip()
    if not text:
        return jsonify({"error": "missing 'text'"}), 400
    audio = synthesize(text)
    if not audio:
        return jsonify({"error": "tts unavailable"}), 503
    return jsonify({"audio_b64": base64.b64encode(audio).decode("ascii")})


@app.post("/converse")
def http_converse():
    """Full turn: speech in -> transcript -> reply -> speech out."""
    raw = request.get_data()
    if not raw:
        return jsonify({"error": "empty body; send a WAV blob"}), 400
    stt = transcribe(raw)
    transcript = stt.get("text", "")
    if not transcript:
        return jsonify({"transcript": "", "reply": "", "audio_b64": None})
    broadcasting = _coerce_bool(request.args.get("broadcasting"))
    result = think(transcript, stt.get("language"), broadcasting)
    audio = synthesize(result.get("reply", ""))
    return jsonify(
        {
            "transcript": transcript,
            "language": stt.get("language"),
            "reply": result.get("reply", ""),
            "error": result.get("error"),
            "broadcast_action": result.get("broadcast_action"),
            "audio_b64": base64.b64encode(audio).decode("ascii") if audio else None,
        }
    )


@app.post("/converse_stream")
def http_converse_stream():
    """Streaming full turn: speech in -> transcript -> reply -> speech out.

    Reduces perceived latency by synthesising the reply one sentence at a time
    and emitting each clip as soon as it is ready, so the Discord layer can play
    the first sentence while the rest are still being spoken. The response is
    newline-delimited JSON (NDJSON):

        {"type":"meta","transcript":..,"language":..,"reply":..,"error":..,"broadcast_action":..}
        {"type":"audio","seq":0,"audio_b64":..}
        {"type":"audio","seq":1,"audio_b64":..}
        {"type":"end"}

    STT and the reply engine still run to completion before the meta line; only
    TTS is pipelined. The non-streaming /converse endpoint is unchanged.
    """
    raw = request.get_data()
    if not raw:
        return jsonify({"error": "empty body; send a WAV blob"}), 400
    broadcasting = _coerce_bool(request.args.get("broadcasting"))

    def gen():
        import time

        def now_ms() -> int:
            # Wall-clock epoch ms so the Node side can line these up against its
            # own Date.now() capture timestamps (same host, same clock).
            return int(time.time() * 1000)

        # Length of the captured speech clip (16-bit mono PCM). This is the
        # "음성 인식(녹음)" portion — how long the user actually spoke (+ the
        # bot's trailing silence cutoff) — as opposed to "STT 처리", the Whisper
        # transcription time below. Splitting them shows whether a slow turn is
        # the listening/recording or the transcription.
        try:
            _frames, _sr = _read_wav_pcm(raw)
            audio_sec = (len(_frames) / 2) / _sr if _sr else 0.0
        except Exception:
            audio_sec = 0.0

        t0 = time.monotonic()
        stt = transcribe(raw)
        t_stt = time.monotonic()
        transcript = stt.get("text", "")
        if not transcript:
            print(
                f"[bridge] ⏱️ turn 녹음(음성)={audio_sec:.1f}s STT처리(whisper)={t_stt - t0:.1f}s "
                f"→ 인식 결과 없음 ({stt.get('note', '빈 결과')})",
                flush=True,
            )
            yield json.dumps({"type": "meta", "transcript": "", "language": stt.get("language"),
                              "reply": "", "error": stt.get("error"),
                              "note": stt.get("note", "빈 결과"),
                              "audio_sec": round(audio_sec, 1),
                              "stt_sec": round(t_stt - t0, 1), "broadcast_action": None}) + "\n"
            yield json.dumps({"type": "end"}) + "\n"
            return
        llm_start_ms = now_ms()
        result = think(transcript, stt.get("language"), broadcasting)
        t_think = time.monotonic()
        llm_end_ms = now_ms()
        reply = result.get("reply", "")
        yield json.dumps({
            "type": "meta",
            "transcript": transcript,
            "language": stt.get("language"),
            "reply": reply,
            "error": result.get("error"),
            "note": "ok" if reply.strip() else "답변 없음",
            "audio_sec": round(audio_sec, 1),
            "stt_sec": round(t_stt - t0, 1),
            "think_sec": round(t_think - t_stt, 1),
            # Wall-clock LLM window (epoch ms) for the transcript-channel timing
            # breakdown. STT shows up as the gap between the Node-side capture
            # end and llm_start_ms.
            "llm_start_ms": llm_start_ms,
            "llm_end_ms": llm_end_ms,
            "broadcast_action": result.get("broadcast_action"),
        }) + "\n"
        tts_total = 0.0
        tts_start_ms = None
        tts_end_ms = None
        for seq, sentence in enumerate(split_sentences(reply)):
            ts = time.monotonic()
            if tts_start_ms is None:
                tts_start_ms = now_ms()
            audio = synthesize(sentence)
            tts_total += time.monotonic() - ts
            tts_end_ms = now_ms()
            if audio:
                yield json.dumps({
                    "type": "audio",
                    "seq": seq,
                    "audio_b64": base64.b64encode(audio).decode("ascii"),
                }) + "\n"
        # The end event carries TTS timing because synthesis happens AFTER the
        # meta line (it is pipelined sentence-by-sentence).
        yield json.dumps({
            "type": "end",
            "tts_sec": round(tts_total, 1),
            "tts_start_ms": tts_start_ms,
            "tts_end_ms": tts_end_ms,
        }) + "\n"
        print(
            f"[bridge] ⏱️ turn 녹음(음성)={audio_sec:.1f}s STT처리(whisper)={t_stt - t0:.1f}s "
            f"think(LLM)={t_think - t_stt:.1f}s tts={tts_total:.1f}s "
            f"total(STT~TTS)={time.monotonic() - t0:.1f}s replylen={len(reply)} "
            f"transcript={transcript[:40]!r}",
            flush=True,
        )

    return Response(stream_with_context(gen()), mimetype="application/x-ndjson")


def _warm_ollama(base_url: str, model: str) -> None:
    """Load ``model`` into Ollama (GPU if available) with a long keep_alive so it
    is resident before the first real turn. Best-effort.

    Warms at the SAME num_ctx the reply engine uses (OLLAMA_NUM_CTX, default
    8192). Ollama keeps a distinct loaded instance per (model, num_ctx), so
    warming at the default context would load the wrong instance and the first
    real chat call (8192) would still cold-reload (~3.4s)."""
    if not base_url or not model:
        return
    import urllib.request

    num_ctx = int(os.environ.get("OLLAMA_NUM_CTX", "8192"))
    try:
        req = urllib.request.Request(
            f"{base_url.rstrip('/')}/api/chat",
            data=json.dumps(
                {"model": model,
                 "messages": [{"role": "user", "content": "."}],
                 "stream": False, "keep_alive": "30m",
                 "options": {"num_ctx": num_ctx, "num_predict": 1}}
            ).encode("utf-8"),
            headers={"Content-Type": "application/json"},
        )
        with urllib.request.urlopen(req, timeout=120) as resp:
            ok = resp.status == 200
        print(f"[bridge] {'✅' if ok else '⚠️'} ollama warm (model={model}, num_ctx={num_ctx})", flush=True)
    except Exception as e:  # pragma: no cover - depends on local ollama
        print(f"[bridge] ollama warmup skipped (model={model}): {e}", flush=True)


def _warmup() -> None:
    """Pre-load Whisper + the chat model + TTS so the FIRST real utterance does
    not pay the cold-start cost (observed ~10s on the first STT). Best-effort and
    runs in a background thread so the HTTP server (and /health) is up
    immediately."""
    try:
        _ensure_brain()
        # JIT the Whisper transcribe path on a short silent buffer. We call the
        # model directly (not transcribe()) because the VAD gate short-circuits
        # silence before Whisper would run, leaving the model un-warmed.
        if _whisper is not None:
            try:
                import numpy as np

                dummy = np.zeros(8000, dtype=np.float32)  # 0.5s @ 16kHz
                segs, _info = _whisper.transcribe(dummy, beam_size=1, language=STT_LANGUAGE)
                for _ in segs:
                    pass
                print("[bridge] ✅ whisper warm", flush=True)
            except Exception as e:  # pragma: no cover
                print(f"[bridge] whisper warmup skipped: {e}", flush=True)
        if _cfg is not None:
            _warm_ollama(getattr(_cfg, "ollama_base_url", ""), getattr(_cfg, "ollama_chat_model", ""))
        # Nudge the TTS worker to warm (MeloTTS loads its model before binding
        # its port, so a ready ping confirms it; Piper loads on first synth).
        if _tts_ready():
            print("[bridge] ✅ tts warm", flush=True)
    except Exception as e:  # pragma: no cover
        print(f"[bridge] warmup error: {e}", flush=True)


def main():
    print(f"[bridge] listening on http://{BRIDGE_HOST}:{BRIDGE_PORT}", flush=True)
    # Warm the models in the background so the first spoken turn is fast while
    # the server is already accepting requests.
    threading.Thread(target=_warmup, name="bridge-warmup", daemon=True).start()
    # threaded=True so STT (slow) on one request doesn't block /health, etc.
    app.run(host=BRIDGE_HOST, port=BRIDGE_PORT, threaded=True)


if __name__ == "__main__":
    main()