"""
ARIA Local-LLM-Adapter (AI-Box) — Plan B, Phase B0.

Bruecke zwischen RVS und dem lokalen llama.cpp-Server. Spiegelt das Muster der
whisper-bridge: verbindet sich per WebSocket mit dem RVS (Token-Room, TLS mit
ws-Fallback, Reconnect-Backoff), lauscht auf `llm_request` und ruft den lokalen
llama.cpp-`/v1/chat/completions`-Endpoint (OpenAI-kompatibel), antwortet mit
`llm_response` (korreliert per requestId).

Topologie: AI-Box steht zuhause, ARIA im RZ — die Kommunikation laeuft ueber
den RVS (wie TTS/STT), keine IPs zu pflegen. Nur URL + Token.

Env:
  RVS_HOST, RVS_PORT, RVS_TLS, RVS_TLS_FALLBACK, RVS_TOKEN  (wie f5tts/whisper)
  LLAMA_URL         Default http://llama:8081  (llama.cpp im selben Compose-Netz)
  LLM_MODEL         optionaler Modell-Name fuer llama (llama.cpp ignoriert ihn
                    meist, dient nur der Transparenz im Log)
  LLM_TIMEOUT_SEC   Default 60

Bewusst NICHT-streamend in B0 (volle llm_response). Token-Streaming (llm_partial)
kommt in B2 zusammen mit TTS-on-first-sentence.
"""

from __future__ import annotations

import asyncio
import json
import logging
import os
import re
import shlex
import time

import httpx
import websockets

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("llm-adapter")

RVS_HOST = os.getenv("RVS_HOST", "").strip()
RVS_PORT = os.getenv("RVS_PORT", "443").strip()
RVS_TLS = os.getenv("RVS_TLS", "true").lower() == "true"
RVS_TLS_FALLBACK = os.getenv("RVS_TLS_FALLBACK", "true").lower() == "true"
RVS_TOKEN = os.getenv("RVS_TOKEN", "").strip()
# TLS-Hostname (SNI + Cert), falls RVS_HOST eine IP ist (Box im selben Netz wie
# der RVS, direkt auf die interne IP). Leer = SNI = RVS_HOST.
RVS_SNI = os.getenv("RVS_SNI", "").strip()

# In-Process-DNS-Override: wenn RVS_SNI gesetzt ist, verbindet die URI ueber den
# HOSTNAMEN (Host-Header + SNI + Cert stimmen), waehrend getaddrinfo den Namen auf
# die echte IP in RVS_HOST aufloest. Versionsunabhaengig — host=/port= kollidiert
# in der Legacy-websockets-API mit dem aus der URI abgeleiteten Host.
if RVS_TLS and RVS_SNI:
    import socket as _socket
    _orig_getaddrinfo = _socket.getaddrinfo
    def _sni_getaddrinfo(host, *a, **k):
        return _orig_getaddrinfo(RVS_HOST if host == RVS_SNI else host, *a, **k)
    _socket.getaddrinfo = _sni_getaddrinfo


LLAMA_URL = os.getenv("LLAMA_URL", "http://llama:8081").rstrip("/")
LLM_MODEL = os.getenv("LLM_MODEL", "qwen3-8b")
LLM_TIMEOUT_SEC = float(os.getenv("LLM_TIMEOUT_SEC", "60"))

# ── Compute-Fleet: Worker-Identitaet & Registrierung ──────────────
# Meldet sich bei der aria-bridge (worker_hello) + periodischer worker_ping.
NODE_NAME = os.getenv("NODE_NAME", "node").strip() or "node"
GPU_IDS = os.getenv("NVIDIA_VISIBLE_DEVICES", "").strip()
WORKER_SERVICE = "llm"
INSTANCE_ID = f"{WORKER_SERVICE}@{NODE_NAME}"
WORKER_PING_INTERVAL_S = int(os.getenv("WORKER_PING_INTERVAL_S", "10"))
# Empfangs-Watchdog: kommt in RX_STALE_S kein Broadcast rein (ein echter Raum hat
# staendig Traffic, z.B. sat_hello alle 25s / Brain-Polling), gilt die Verbindung
# als halb-tot (Caddy pongt die WS-Pings selbst) -> Zwangs-Reconnect.
RX_STALE_S = int(os.getenv("RX_STALE_S", "60"))
_inflight = 0   # laufende llm_requests (busy-Report im ping)
# Qwen3 hat Thinking-Mode default AN — dann verbraet es Tokens in einem
# <think>-Block und liefert (bei kleinem max_tokens) leeren/abgeschnittenen
# content, ausserdem 3x langsamer. ARIAs schnelles Tier will KEIN Grübeln
# (grübeln = harter Turn = Claude). Wir schalten Thinking daher per
# chat_template_kwargs ab (Qwen3-Template versteht enable_thinking=false;
# andere Templates ignorieren das kwarg). Bei einem Modell, das darauf
# empfindlich reagiert: LLM_DISABLE_THINKING=false setzen.
LLM_DISABLE_THINKING = os.getenv("LLM_DISABLE_THINKING", "true").lower() == "true"

# ── Modell-Verwaltung (Stage D): Adapter besitzt llama-swaps Config ──
# llama-swap liest die GENERIERTE Config (beschreibbar, im /models-Bind). Wir
# erzeugen sie aus dem Basis-Template (kuratierte Defaults) + der persistenten
# Box-Registry (per Diagnostic hinzugefuegte Modelle). So werden neue Modelle
# ohne Image-Rebuild waehlbar.
import yaml  # pyyaml
BASE_CONFIG_PATH = os.getenv("LLAMA_BASE_CONFIG", "/llamaswap/config.yaml")
GEN_CONFIG_PATH = os.getenv("LLAMA_GEN_CONFIG", "/models/llama-swap.config.yaml")
REGISTRY_PATH = os.getenv("LLM_REGISTRY", "/models/aria_models.json")

# ── OpenAI-kompatibler Endpunkt (fuer OpenHands & Co.) ─────
# Der Port ist ein Docker-Port-Mapping und steht beim Container-Start fest —
# er kommt IMMER aus der .env. Der API-Key kann zusaetzlich zur Laufzeit aus der
# Diagnostic gesetzt werden, ABER: steht er in der .env, hat die .env Vorrang
# und das Feld ist in der Diagnostic schreibgeschuetzt.
OPENAI_PORT = os.getenv("LLM_OPENAI_PORT", "8081").strip() or "8081"
OPENAI_BIND = os.getenv("LLM_OPENAI_BIND", "127.0.0.1").strip() or "127.0.0.1"
_OPENAI_KEY_ENV = os.getenv("LLM_OPENAI_KEY", "").strip()
OPENAI_KEY_FILE = os.getenv("LLM_OPENAI_KEY_FILE", "/models/openai_key.json")


def _openai_key() -> str:
    """Gueltiger API-Key: .env gewinnt, sonst der in der Diagnostic gesetzte."""
    if _OPENAI_KEY_ENV:
        return _OPENAI_KEY_ENV
    try:
        with open(OPENAI_KEY_FILE) as f:
            return str((json.load(f) or {}).get("key") or "").strip()
    except Exception:
        return ""


def _openai_key_save(key: str) -> bool:
    """Speichert den in der Diagnostic gesetzten Key. Bei .env-Vorrang: nein."""
    if _OPENAI_KEY_ENV:
        return False
    try:
        os.makedirs(os.path.dirname(OPENAI_KEY_FILE), exist_ok=True)
        tmp = OPENAI_KEY_FILE + ".tmp"
        with open(tmp, "w") as f:
            json.dump({"key": key.strip()}, f)
        os.replace(tmp, OPENAI_KEY_FILE)
        os.chmod(OPENAI_KEY_FILE, 0o600)
        return True
    except Exception as exc:
        logger.error("OpenAI-Key speichern fehlgeschlagen: %s", exc)
        return False

# ── Auslastungs-Monitor (Stage E) ──────────────────────────
import node_stats
STATS_PATH = os.getenv("STATS_PATH", "/models/aria_stats.json")
_total_tokens = 0   # kumulativ, fuer den Token-Graph
_stats = node_stats.NodeStats(INSTANCE_ID, NODE_NAME, STATS_PATH,
                              token_getter=lambda: _total_tokens, logger=logger)


def _load_registry() -> list:
    try:
        with open(REGISTRY_PATH) as f:
            data = json.load(f)
        return data if isinstance(data, list) else []
    except Exception:
        return []


def _save_registry(reg: list) -> None:
    try:
        tmp = REGISTRY_PATH + ".tmp"
        with open(tmp, "w") as f:
            json.dump(reg, f, indent=2)
        os.replace(tmp, REGISTRY_PATH)
    except Exception as e:
        logger.warning("Registry speichern fehlgeschlagen: %s", e)


def _generate_config() -> int:
    """Schreibt die llama-swap-Config aus Basis-Template + Registry. Gibt die
    Anzahl Modelle zurueck. Idempotent, bei jeder Aenderung + beim Start."""
    base = {}
    try:
        with open(BASE_CONFIG_PATH) as f:
            base = yaml.safe_load(f) or {}
    except Exception as e:
        logger.warning("Basis-Template %s nicht lesbar (%s)", BASE_CONFIG_PATH, e)
    models = dict(base.get("models") or {})
    for e in _load_registry():
        key = (e.get("key") or "").strip()
        repo = (e.get("hfRepo") or "").strip()
        if not key or not repo:
            continue
        quant = (e.get("quant") or "Q4_K_M").strip()
        ctx = int(e.get("ctx") or 8192)
        ngl = int(e.get("ngl") or 99)
        models[key] = {
            "cmd": (f"llama-server --port ${{PORT}} --host 127.0.0.1\n"
                    f"-hf {repo}:{quant}\n-ngl {ngl} -c {ctx} --jinja"),
            "ttl": 3600,
        }
    # API-Key auf JEDES Modell anwenden (auch die aus dem Basis-Template) —
    # llama-server prueft dann den Authorization-Header. Ohne Key bleibt der
    # Endpunkt offen; deshalb ist er per Default nur an localhost gebunden.
    key = _openai_key()
    for k, spec in list(models.items()):
        cmd = str((spec or {}).get("cmd") or "")
        if not cmd:
            continue
        cmd = re.sub(r"\s*--api-key\s+\S+", "", cmd)     # alten Key entfernen
        if key:
            cmd = cmd.rstrip() + f" --api-key {shlex.quote(key)}"
        models[k] = dict(spec, cmd=cmd)

    out = dict(base)
    out["models"] = models
    try:
        os.makedirs(os.path.dirname(GEN_CONFIG_PATH), exist_ok=True)
        tmp = GEN_CONFIG_PATH + ".tmp"
        with open(tmp, "w") as f:
            yaml.safe_dump(out, f, sort_keys=False, default_flow_style=False)
        os.replace(tmp, GEN_CONFIG_PATH)
        logger.info("llama-swap-Config generiert: %d Modelle → %s", len(models), GEN_CONFIG_PATH)
    except Exception as e:
        logger.error("Config schreiben fehlgeschlagen: %s", e)
    return len(models)


async def _reload_llama() -> None:
    """Stoesst llama-swap-Reload an. Viele Builds watchen die Config-Datei ohnehin;
    zusaetzlich versuchen wir bekannte Reload-Endpunkte (Fehler ignoriert)."""
    for path in ("/api/config/reload", "/reload"):
        try:
            async with httpx.AsyncClient(timeout=10) as c:
                r = await c.post(f"{LLAMA_URL}{path}")
                if r.status_code < 400:
                    logger.info("llama-swap reload via %s", path)
                    return
        except Exception:
            pass
    logger.info("llama-swap reload: kein Endpoint — verlasse mich auf File-Watch")


async def _send(ws, mtype: str, payload: dict) -> None:
    try:
        await ws.send(json.dumps({
            "type": mtype,
            "payload": payload,
            "timestamp": int(time.time() * 1000),
        }))
    except Exception as e:
        logger.warning("Send fehlgeschlagen (%s): %s", mtype, e)


async def _call_llama(messages: list, *, max_tokens: int, temperature: float,
                      stop, tools=None, model=None, timeout: float | None = None) -> dict:
    """Ruft llama.cpp/llama-swap /v1/chat/completions (OpenAI-Format). Gibt
    {ok, content, tool_calls, error} zurueck — wirft nie.

    model: welches Modell llama-swap laden soll (B0.5). Kommt aus dem Request
    (Brain -> local_llm.json). Faellt auf LLM_MODEL (env) zurueck.
    tools: optionale OpenAI-Tool-Definitionen (B1b). Qwen3 (--jinja) kann
    natives Tool-Calling und liefert dann message.tool_calls."""
    body = {
        "model": model or LLM_MODEL,
        "messages": messages,
        "max_tokens": max_tokens,
        "temperature": temperature,
        "stream": False,
    }
    if stop:
        body["stop"] = stop
    if tools:
        body["tools"] = tools
        body["tool_choice"] = "auto"
    if LLM_DISABLE_THINKING:
        # llama.cpp (--jinja) reicht chat_template_kwargs an die Chat-Vorlage
        # weiter. Qwen3 unterdrueckt damit den <think>-Block.
        body["chat_template_kwargs"] = {"enable_thinking": False}
    try:
        async with httpx.AsyncClient(timeout=timeout or LLM_TIMEOUT_SEC) as client:
            r = await client.post(f"{LLAMA_URL}/v1/chat/completions", json=body)
            r.raise_for_status()
            data = r.json()
        msg = (data.get("choices") or [{}])[0].get("message", {}) or {}
        usage = data.get("usage") or {}
        try:
            global _total_tokens
            _total_tokens += int(usage.get("total_tokens") or 0)
        except Exception:
            pass
        return {
            "ok": True,
            "content": msg.get("content") or "",
            "tool_calls": msg.get("tool_calls") or None,
            "usage": usage,
        }
    except Exception as e:
        logger.warning("llama.cpp-Call fehlgeschlagen: %s", e)
        return {"ok": False, "content": "", "error": str(e)[:300]}


# B0.5-2: Lade-Status ans Diagnostic (service_status, service="llm"). Wir kennen
# den Download-Fortschritt nicht (llama-swap gibt ihn nicht her), aber wir melden
# den Zustand bei Modellwechsel: loading -> ready/error. _last_model = aktuell
# geladenes; _ready_models = in dieser Session schon einmal bereit gewesene
# (fuer den "frisch geladen"-Hinweis 🎉 bei langem Erst-Load).
_last_model = None
_ready_models: set = set()


async def _emit_llm_status(ws, state: str, model: str, **extra) -> None:
    await _send(ws, "service_status",
                {"service": "llm", "state": state, "model": model, **extra})


async def _fetch_available_models() -> list:
    """Fragt llama-swap ab, welche Modelle diese Box fahren kann (GET /v1/models,
    OpenAI-kompatibel → {data:[{id},...]}). Das sind die config.yaml-Keys.
    Defensiv: bei Fehler Fallback auf [LLM_MODEL]."""
    try:
        async with httpx.AsyncClient(timeout=10) as client:
            r = await client.get(f"{LLAMA_URL}/v1/models")
            r.raise_for_status()
            data = r.json()
        ids = [m.get("id") for m in (data.get("data") or []) if m.get("id")]
        return ids or [LLM_MODEL]
    except Exception as e:
        logger.warning("llama-swap /v1/models nicht abfragbar (%s) — Fallback [%s]", e, LLM_MODEL)
        return [LLM_MODEL]


async def _announce(ws) -> None:
    """Sendet ein frisches worker_hello mit der aktuellen Modell-Liste (nach
    Provision/Remove aufrufen, damit Bridge+Diagnostic das neue Modell lernen)."""
    models = await _fetch_available_models()
    await _send(ws, "worker_hello", {
        "instanceId": INSTANCE_ID, "service": WORKER_SERVICE,
        "node": NODE_NAME, "gpus": GPU_IDS, "model": LLM_MODEL,
        "models": models,   # welche Modelle diese Box fahren kann (llama-swap-Keys)
        # OpenAI-Endpunkt fuer Fremd-Tools (OpenHands & Co.). Der Key selbst wird
        # NICHT gesendet (Broadcast!) — nur ob einer gesetzt ist und ob die .env
        # ihn vorgibt (dann ist das Feld in der Diagnostic schreibgeschuetzt).
        "openaiPort": OPENAI_PORT,
        "openaiBind": OPENAI_BIND,
        "openaiKeySet": bool(_openai_key()),
        "openaiEnvLocked": bool(_OPENAI_KEY_ENV),
    })
    logger.info("worker_hello: models=%s", models)


async def _worker_register(ws) -> None:
    """Meldet diesen Worker bei der aria-bridge an (worker_hello) und haelt die
    Flotten-Registry per periodischem worker_ping (mit busy-Status) frisch."""
    try:
        await _announce(ws)
        n = 0
        while True:
            await asyncio.sleep(WORKER_PING_INTERVAL_S)
            n += 1
            await _send(ws, "worker_ping",
                        {"instanceId": INSTANCE_ID, "busy": _inflight > 0})
            if n % 3 == 0:   # ~30s worker_hello wiederholen (wie der Satellit) →
                await _announce(ws)   # auch neu gestartetes Diagnostic/Bridge lernt uns
    except asyncio.CancelledError:
        raise
    except Exception:
        return   # Socket tot → still beenden; _run reconnectet + startet neu


async def _handle_llm_request(ws, payload: dict) -> None:
    global _last_model, _inflight
    _inflight += 1
    try:
        await _do_llm_request(ws, payload)
    finally:
        _inflight -= 1


async def _do_llm_request(ws, payload: dict) -> None:
    global _last_model
    req_id = payload.get("requestId", "")
    messages = payload.get("messages") or []
    if not isinstance(messages, list) or not messages:
        await _send(ws, "llm_response", {
            "requestId": req_id, "ok": False, "error": "leere/ungueltige messages",
        })
        return
    max_tokens = int(payload.get("max_tokens", 512) or 512)
    temperature = float(payload.get("temperature", 0.7) or 0.7)
    stop = payload.get("stop")
    tools = payload.get("tools") or None
    model = (payload.get("model") or "").strip() or None
    eff_model = model or LLM_MODEL

    # Modellwechsel (oder erster Request) → llama-swap laedt/swappt: Status melden.
    switching = eff_model != _last_model
    if switching:
        await _emit_llm_status(ws, "loading", eff_model)

    t0 = time.time()
    res = await _call_llama(messages, max_tokens=max_tokens,
                            temperature=temperature, stop=stop, tools=tools,
                            model=model)
    dt = time.time() - t0

    if switching:
        if res.get("ok"):
            fresh = (eff_model not in _ready_models) and dt > 25
            _ready_models.add(eff_model)
            _last_model = eff_model
            await _emit_llm_status(ws, "ready", eff_model,
                                   loadSeconds=round(dt, 1), freshlyDownloaded=fresh)
        else:
            # bei Fehler _last_model NICHT setzen → naechster Versuch meldet erneut loading
            await _emit_llm_status(ws, "error", eff_model,
                                   error=(res.get("error") or "")[:120])
    tc = res.get("tool_calls")
    logger.info("llm_request id=%s model=%s -> ok=%s %.2fs content_len=%d tool_calls=%d",
                (req_id[:8] if req_id else "?"), model or LLM_MODEL, res.get("ok"), dt,
                len(res.get("content") or ""), len(tc) if tc else 0)
    await _send(ws, "llm_response", {
        "requestId": req_id,
        "ok": res.get("ok", False),
        "content": res.get("content", ""),
        "tool_calls": tc,
        "error": res.get("error"),
        "model": model or LLM_MODEL,
        "elapsedMs": int(dt * 1000),
    })


async def _handle_openai_config(ws, payload: dict) -> None:
    """Setzt den API-Key des OpenAI-Endpunkts (aus der Diagnostic).

    Der PORT wird hier bewusst NICHT geaendert: er ist ein Docker-Port-Mapping
    und steht beim Container-Start fest — dafuer ist die .env zustaendig.
    Steht der Key in der .env, gewinnt sie und wir lehnen ab.
    """
    if (payload.get("instanceId") or INSTANCE_ID) != INSTANCE_ID:
        return
    req_id = payload.get("requestId", "")
    key = str(payload.get("apiKey") or "").strip()
    if _OPENAI_KEY_ENV:
        await _send(ws, "llm_openai_config_result", {
            "requestId": req_id, "instanceId": INSTANCE_ID, "ok": False,
            "error": "Der API-Key steht in der .env dieser Box — die hat Vorrang. "
                     "Zum Aendern dort anpassen und die Box neu starten.",
        })
        return
    if not _openai_key_save(key):
        await _send(ws, "llm_openai_config_result", {
            "requestId": req_id, "instanceId": INSTANCE_ID, "ok": False,
            "error": "Key konnte nicht gespeichert werden (Schreibrechte auf /models?)",
        })
        return
    # Config neu schreiben, damit llama-server den Key beim naechsten Laden nutzt.
    _generate_config()
    logger.info("OpenAI-API-Key %s (wirkt beim naechsten Modell-Laden)",
                "gesetzt" if key else "entfernt")
    await _send(ws, "llm_openai_config_result", {
        "requestId": req_id, "instanceId": INSTANCE_ID, "ok": True,
        "keySet": bool(key),
        "message": ("API-Key gesetzt. Er greift, sobald das Modell das naechste Mal "
                    "geladen wird — bereits laufende Modelle einmal entladen lassen "
                    "(TTL) oder ein anderes Modell anfragen."
                    if key else "API-Key entfernt — der Endpunkt ist jetzt offen."),
    })
    await _announce(ws)


# ── Download-Fortschritt beim Modell-Laden ─────────────────────────────
# llama.cpp zieht das GGUF per -hf selbst und meldet keinen Fortschritt. Wir
# messen von aussen: Hugging Face sagt vorab, welche Dateien kommen (Name,
# Groesse, sha256). llama.cpp legt sie im HF-Cache-Layout ab —
#   /models/models--org--repo/blobs/<sha256>.downloadInProgress  (waechst)
#   /models/models--org--repo/blobs/<sha256>                     (fertig)
# Genau diese Dateien werden gemessen. Das klappt auch bei gesplitteten
# Modellen und bei zwei Downloads gleichzeitig. Fallback fuer aeltere
# llama.cpp-Layouts: alle *.downloadInProgress unter /models.
MODELS_DIR = os.getenv("LLAMA_CACHE_DIR", "/models")
HF_TOKEN = os.getenv("HF_TOKEN", "").strip()
LLM_GPU = os.getenv("LLM_GPU", "").strip()          # Karte, auf die llama-swap gepinnt ist
WARMUP_TIMEOUT_S = float(os.getenv("LLM_WARMUP_TIMEOUT_S", "1900"))  # > healthCheckTimeout
_INPROGRESS = ".downloadInProgress"
_SPLIT_RE = re.compile(r"-(\d{5})-of-(\d{5})\.gguf$", re.IGNORECASE)


async def _hf_expected_files(repo: str, quant: str) -> list:
    """Welche GGUF-Dateien zieht 'repo:quant'? [{path, oid, size}]. Fail-soft: []."""
    url = f"https://huggingface.co/api/models/{repo}/tree/main?recursive=1"
    headers = {"Authorization": f"Bearer {HF_TOKEN}"} if HF_TOKEN else {}
    try:
        async with httpx.AsyncClient(timeout=15, follow_redirects=True) as c:
            r = await c.get(url, headers=headers)
            r.raise_for_status()
            entries = r.json()
    except Exception as e:
        logger.warning("HF-Dateiliste fuer %s nicht abrufbar (%s) — Fortschritt ohne Gesamtgroesse", repo, e)
        return []
    q = (quant or "").lower()
    files = []
    for e in entries if isinstance(entries, list) else []:
        path = e.get("path") or ""
        low = path.lower()
        if e.get("type") != "file" or not low.endswith(".gguf") or "mmproj" in low:
            continue
        if q and q not in low:
            continue
        lfs = e.get("lfs") or {}
        files.append({"path": path, "oid": lfs.get("oid") or e.get("oid") or "",
                      "size": int(lfs.get("size") or e.get("size") or 0)})
    if not files:
        return []
    # Gesplittetes Modell → alle Teile; sonst die eine passende Datei (bei
    # mehreren Treffern die mit dem kuerzesten Namen, z.B. ohne "-imat").
    split = [f for f in files if _SPLIT_RE.search(f["path"])]
    if split:
        return split
    return [min(files, key=lambda f: len(f["path"]))]


def _scan_download(oids: set) -> tuple:
    """(bytes_da, laeuft_noch, gefunden) fuer die erwarteten Blobs."""
    done, active, found = 0, False, 0
    fallback, fallback_active = 0, False
    for root, _dirs, fns in os.walk(MODELS_DIR):
        for fn in fns:
            inprog = fn.endswith(_INPROGRESS)
            base = fn[:-len(_INPROGRESS)] if inprog else fn
            try:
                size = os.path.getsize(os.path.join(root, fn))
            except OSError:
                continue
            if base in oids:
                done += size
                found += 1
                active |= inprog
            elif inprog:
                fallback += size
                fallback_active = True
    if not found and fallback_active:
        return fallback, True, 0
    return done, active, found


async def _gpu_info() -> dict:
    """Name + VRAM der LLM-Karte nach dem Laden (fuer 'in Karte N geladen')."""
    if not LLM_GPU:
        return {}
    try:
        proc = await asyncio.create_subprocess_exec(
            "nvidia-smi", f"--id={LLM_GPU}",
            "--query-gpu=index,name,memory.used,memory.total", "--format=csv,noheader,nounits",
            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL)
        out, _ = await asyncio.wait_for(proc.communicate(), 10)
        idx, name, used, total = [x.strip() for x in out.decode().strip().split(",")[:4]]
        return {"gpu": idx, "gpuName": name, "vramUsedMB": int(used), "vramTotalMB": int(total)}
    except Exception as e:
        logger.info("GPU-Info nicht verfuegbar: %s", e)
        return {"gpu": LLM_GPU}


async def _emit_progress(ws, key: str, phase: str, **extra) -> None:
    await _send(ws, "llm_provision_progress",
                {"instanceId": INSTANCE_ID, "node": NODE_NAME, "key": key,
                 "phase": phase, "ts": int(time.time() * 1000), **extra})


async def _watch_download(ws, key: str, files: list, stop: asyncio.Event) -> None:
    """Meldet jede Sekunde den Download-Stand, bis der Warmup fertig ist.
    Ist alles da, wechselt die Phase auf 'load' (llama.cpp laedt in die Karte)."""
    oids = {f["oid"] for f in files if f.get("oid")}
    total = sum(f["size"] for f in files)
    phase = ""
    downloaded = False        # gab es einen echten Download? sonst: lag schon im Cache
    last_b, last_t, speed = None, time.time(), 0.0
    while not stop.is_set():
        b, active, found = await asyncio.to_thread(_scan_download, oids)
        now = time.time()
        if last_b is not None and now > last_t and b >= last_b:
            inst = (b - last_b) / (now - last_t)
            speed = inst if speed == 0 else 0.7 * speed + 0.3 * inst
        last_b, last_t = b, now
        if active:
            downloaded = True
            phase = "download"
            pct = round(min(100.0, 100.0 * b / total), 1) if total else None
            eta = int(max(0, total - b) / speed) if total and speed > 1 else None
            await _emit_progress(ws, key, "download", bytes=b, total=total, percent=pct,
                                 speedBps=int(speed), etaSeconds=eta)
        else:
            # Nichts laeuft mehr: entweder fertig geladen (alle Blobs da) oder
            # es lag schon im Cache. Im Notfallmodus (keine HF-Liste) zaehlt
            # nur: es lief ein Download, und jetzt nicht mehr.
            all_there = bool(oids) and found >= len(oids)
            if phase != "load" and (all_there or downloaded):
                phase = "load"
                await _emit_progress(ws, key, "load", bytes=total or b, total=total,
                                     percent=100, cached=not downloaded)
        try:
            await asyncio.wait_for(stop.wait(), 1.0)
        except asyncio.TimeoutError:
            pass


async def _handle_provision(ws, payload: dict) -> None:
    """Fuegt ein Modell hinzu: Registry+Config schreiben, reload, dann Warmup
    (zieht das GGUF via -hf beim ersten Load). Meldet die neue Modell-Liste."""
    key = (payload.get("key") or "").strip()
    repo = (payload.get("hfRepo") or "").strip()
    if not key or not repo:
        await _send(ws, "llm_provision_result",
                    {"instanceId": INSTANCE_ID, "key": key, "ok": False, "error": "key/hfRepo fehlt"})
        return
    entry = {
        "key": key, "hfRepo": repo,
        "quant": (payload.get("quant") or "Q4_K_M").strip(),
        "ctx": int(payload.get("ctx") or 8192),
        "ngl": int(payload.get("ngl") or 99),
    }
    reg = [e for e in _load_registry() if e.get("key") != key]
    reg.append(entry)
    _save_registry(reg)
    _generate_config()
    await _reload_llama()
    await _announce(ws)   # Bridge/Diagnostic lernen das neue Modell
    # Warmup: Mini-Request → llama-swap zieht das GGUF (falls noch nicht da)
    # und laedt es in die Karte. Daneben misst _watch_download den Download.
    files = await _hf_expected_files(repo, entry["quant"])
    await _emit_progress(ws, key, "start", total=sum(f["size"] for f in files),
                         files=len(files), gpu=LLM_GPU)
    await _emit_llm_status(ws, "loading", key)
    stop = asyncio.Event()
    watcher = asyncio.create_task(_watch_download(ws, key, files, stop))
    t0 = time.time()
    # Langer Timeout: ein GGUF mit mehreren GB braucht Minuten. Mit dem normalen
    # LLM_TIMEOUT_SEC (60 s) meldete das hier "Fehler", waehrend llama-swap im
    # Hintergrund munter weiterlud.
    res = await _call_llama([{"role": "user", "content": "hi"}],
                            max_tokens=1, temperature=0.0, stop=None, model=key,
                            timeout=WARMUP_TIMEOUT_S)
    dt = time.time() - t0
    stop.set()
    try:
        await watcher
    except Exception:
        pass
    if res.get("ok"):
        _ready_models.add(key)
        await _emit_llm_status(ws, "ready", key, loadSeconds=round(dt, 1),
                               freshlyDownloaded=dt > 25)
        await _emit_progress(ws, key, "ready", percent=100, loadSeconds=round(dt, 1),
                             **(await _gpu_info()))
    else:
        await _emit_llm_status(ws, "error", key, error=(res.get("error") or "")[:160])
        await _emit_progress(ws, key, "error", error=(res.get("error") or "")[:200])
    await _send(ws, "llm_provision_result",
                {"instanceId": INSTANCE_ID, "key": key, "ok": res.get("ok", False),
                 "error": res.get("error"), "elapsedMs": int(dt * 1000)})
    logger.info("provision %s (%s) → ok=%s %.1fs", key, repo, res.get("ok"), dt)


async def _handle_remove(ws, payload: dict) -> None:
    """Entfernt ein Modell aus Registry+Config (GGUF bleibt im Cache)."""
    key = (payload.get("key") or "").strip()
    if not key:
        return
    reg = [e for e in _load_registry() if e.get("key") != key]
    _save_registry(reg)
    _generate_config()
    await _reload_llama()
    await _announce(ws)
    await _send(ws, "llm_provision_result",
                {"instanceId": INSTANCE_ID, "key": key, "ok": True, "removed": True})
    logger.info("removed model %s", key)


async def _run() -> None:
    if not RVS_HOST:
        logger.error("RVS_HOST nicht gesetzt — Abbruch")
        return
    if not RVS_TOKEN:
        logger.error("RVS_TOKEN nicht gesetzt — Abbruch")
        return

    # llama-swap-Config aus Basis-Template + Registry erzeugen, BEVOR llama-swap
    # sie braucht (llama-swap restart: unless-stopped faengt die Erst-Boot-
    # Reihenfolge ab, falls es kurz vor uns startet).
    _generate_config()

    # Auslastungs-Sampler (GPU + Tokens) laeuft unabhaengig vom RVS.
    asyncio.create_task(_stats.run_sampler())

    use_tls = RVS_TLS
    retry_s = 2
    tls_fallback_tried = False

    while True:
        scheme = "wss" if use_tls else "ws"
        uri_host = RVS_SNI if (use_tls and RVS_SNI) else RVS_HOST
        url = f"{scheme}://{uri_host}:{RVS_PORT}?token={RVS_TOKEN}"
        masked = url.replace(RVS_TOKEN, "***") if RVS_TOKEN else url
        # max_size MUSS zum RVS/zur Bridge passen (1500 MB). Der RVS ist ein
        # BROADCAST-Relay: eine grosse Nachricht (Datei-Transfer, Voice-Upload,
        # APK-Update) geht an JEDEN Client im Raum. Wer ein kleineres Limit hat,
        # schliesst die Verbindung mit Code 1009 ("message too big") und landet in
        # einer Reconnect-Schleife — real gesehen mit einem 209-MB-Frame, der
        # whisper und f5tts im Minutentakt neu starten liess.
        _MAX_WS_BYTES = int(os.getenv("RVS_MAX_SIZE_MB", "1500")) * 1024 * 1024
        connect_kwargs = {"ping_interval": 20, "ping_timeout": 10, "max_size": _MAX_WS_BYTES}
        try:
            logger.info("Verbinde zu RVS: %s%s (llama=%s)", masked,
                        f" (TCP {RVS_HOST})" if (use_tls and RVS_SNI) else "", LLAMA_URL)
            async with websockets.connect(url, **connect_kwargs) as ws:
                logger.info("RVS verbunden — llm-adapter online")
                retry_s = 2
                tls_fallback_tried = False
                ping_task = asyncio.create_task(_worker_register(ws))
                while True:
                    try:
                        raw = await asyncio.wait_for(ws.recv(), timeout=RX_STALE_S)
                    except asyncio.TimeoutError:
                        logger.warning("Kein RVS-Traffic seit %ds — Verbindung halb-tot, reconnect", RX_STALE_S)
                        raise ConnectionError("rvs-stale")
                    try:
                        msg = json.loads(raw)
                    except Exception:
                        continue
                    mtype = msg.get("type")
                    payload = msg.get("payload", {}) or {}
                    # Redundanz-Routing: gezielt an eine andere Instanz adressiert
                    # → ignorieren. Ohne targetInstance → wie bisher (jeder nimmt).
                    tgt = payload.get("targetInstance")
                    if tgt and tgt != INSTANCE_ID:
                        continue
                    # Auslastungs-Monitor (node_stats_*) abfangen.
                    if await _stats.handle(ws, mtype, payload, _send):
                        continue
                    if mtype == "llm_openai_config":
                        asyncio.create_task(_handle_openai_config(ws, payload))
                        continue
                    if mtype not in ("llm_request", "llm_provision_model", "llm_remove_model"):
                        continue
                    if mtype == "llm_provision_model":
                        asyncio.create_task(_handle_provision(ws, payload))
                    elif mtype == "llm_remove_model":
                        asyncio.create_task(_handle_remove(ws, payload))
                    else:
                        # Jede Anfrage nebenlaeufig — llama.cpp serialisiert intern,
                        # aber wir blockieren so nicht den Empfang weiterer Messages.
                        asyncio.create_task(_handle_llm_request(ws, payload))
        except Exception as e:
            logger.warning("RVS-Verbindung verloren/fehlgeschlagen: %s", e)
            try:
                ping_task.cancel()
            except NameError:
                pass
            if use_tls and RVS_TLS_FALLBACK and not tls_fallback_tried:
                tls_fallback_tried = True
                use_tls = False
                logger.info("TLS fehlgeschlagen — Fallback auf ws://")
                continue
            await asyncio.sleep(min(retry_s, 30))
            retry_s = min(retry_s * 2, 30)
            use_tls = RVS_TLS


if __name__ == "__main__":
    asyncio.run(_run())