"""
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")

# ── Karten-Belegung ─────────────────────────────────────────────────
# Grundregel: Modelle LADEN nur ueber die Diagnostic, und zwar pro Karte. Jede
# LLM-Karte haelt genau ein Modell, dauerhaft (ttl 0, beim Start vorgeladen).
# Clients — ARIA, Fremdprogramme ueber RVS, OpenHands ueber den OpenAI-Port —
# sehen nur die geladenen Modelle und waehlen daraus. Eine Anfrage fuer ein
# anderes Modell wird abgelehnt und laedt NIE etwas nach. So kann kein Client
# einem anderen das Modell unter den Fuessen wegtauschen.
#
# Welche Karten dem LLM gehoeren, steht in der .env der Box:
#   LLM_GPUS=0,1   (mehrere)   — oder wie bisher —   LLM_GPU=0   (eine)
# Die Belegung (Karte → Modell) liegt in SLOTS_PATH und wird nur von
# llm_provision_model / llm_unload_model geaendert.
SLOTS_PATH = os.getenv("LLM_SLOTS", "/models/aria_slots.json")


def _llm_cards() -> list:
    raw = os.getenv("LLM_GPUS", "").strip() or os.getenv("LLM_GPU", "0").strip() or "0"
    cards = [c.strip() for c in raw.split(",") if c.strip()]
    return [c for i, c in enumerate(cards) if c not in cards[:i]]   # Reihenfolge, ohne Doppelte


def _load_slots() -> dict:
    """{karte: modell}. Fehlt die Datei (erste Version mit Karten), wird das
    bisherige Standardmodell (LLM_MODEL) auf die erste Karte gelegt — so laeuft
    eine bestehende Box nach dem Update genau so weiter wie vorher."""
    try:
        with open(SLOTS_PATH) as f:
            data = json.load(f)
        if isinstance(data, dict):
            cards = set(_llm_cards())
            return {str(k): str(v) for k, v in data.items() if str(k) in cards and v}
    except FileNotFoundError:
        cards = _llm_cards()
        if cards and _model_cmd(LLM_MODEL):
            slots = {cards[0]: LLM_MODEL}
            _save_slots(slots)
            logger.info("Karten-Belegung neu angelegt: Karte %s → %s", cards[0], LLM_MODEL)
            return slots
    except Exception as e:
        logger.warning("Karten-Belegung nicht lesbar (%s) — keine Modelle", e)
    return {}


def _save_slots(slots: dict) -> None:
    tmp = SLOTS_PATH + ".tmp"
    with open(tmp, "w") as f:
        json.dump(slots, f, indent=2)
    os.replace(tmp, SLOTS_PATH)


def _card_of(model: str) -> str:
    for card, m in _load_slots().items():
        if m == model:
            return card
    return ""

# ── 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 _base_config() -> dict:
    try:
        with open(BASE_CONFIG_PATH) as f:
            return yaml.safe_load(f) or {}
    except Exception as e:
        logger.warning("Basis-Template %s nicht lesbar (%s)", BASE_CONFIG_PATH, e)
        return {}


def _model_cmd(key: str) -> str:
    """llama-server-Aufruf fuer ein Modell: aus der Registry (ueber die
    Diagnostic geladen) oder aus dem Basis-Template (qwen3-8b & Co.)."""
    for e in _load_registry():
        if (e.get("key") or "").strip() == key and (e.get("hfRepo") or "").strip():
            quant = (e.get("quant") or "Q4_K_M").strip()
            return (f"llama-server --port ${{PORT}} --host 127.0.0.1\n"
                    f"-hf {e['hfRepo'].strip()}:{quant}\n"
                    f"-ngl {int(e.get('ngl') or 99)} -c {int(e.get('ctx') or 8192)} --jinja")
    spec = (_base_config().get("models") or {}).get(key) or {}
    return str(spec.get("cmd") or "")


def _known_models() -> list:
    """Alles, was diese Box laden KOENNTE (Basis-Template + Registry)."""
    keys = list((_base_config().get("models") or {}).keys())
    keys += [e["key"] for e in _load_registry() if e.get("key") and e["key"] not in keys]
    return keys


def _generate_config() -> int:
    """Schreibt die llama-swap-Config: NUR die Modelle, die einer Karte
    zugeordnet sind. Was nicht drinsteht, kennt llama-swap nicht — auch der
    OpenAI-Port lehnt es dann ab (404 "no router for requested model"). Das ist
    der Kern der Regel "Clients laden nichts".

      env CUDA_VISIBLE_DEVICES=<karte>   pinnt das Modell auf seine Karte
      ttl 0                              nie wegen Leerlauf entladen
      groups.swap false                  alle gleichzeitig geladen, keins
                                         verdraengt ein anderes
      hooks.on_startup.preload           beim Start (und nach jedem Reload)
                                         sofort laden, nicht erst beim 1. Aufruf
    Getestet gegen llama-swap: genau dieses Verhalten.
    """
    base = _base_config()
    slots = _load_slots()
    key = _openai_key()
    models = {}
    for card in _llm_cards():
        m = slots.get(card)
        if not m:
            continue
        cmd = _model_cmd(m)
        if not cmd:
            logger.error("Karte %s: Modell %r unbekannt (weder Registry noch Basis) — uebersprungen", card, m)
            continue
        cmd = re.sub(r"\s*--api-key\s+\S+", "", cmd)
        if key:
            # llama-server prueft dann den Authorization-Header (OpenAI-Port).
            cmd = cmd.rstrip() + f" --api-key {shlex.quote(key)}"
        models[m] = {"cmd": cmd, "env": [f"CUDA_VISIBLE_DEVICES={card}"], "ttl": 0}

    out = {k: v for k, v in base.items() if k not in ("models", "groups", "hooks")}
    out["models"] = models
    if models:
        out["groups"] = {"aria-karten": {"swap": False, "exclusive": True,
                                         "persistent": True, "members": list(models)}}
        out["hooks"] = {"on_startup": {"preload": list(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: %s",
                    ", ".join(f"Karte {_card_of(m) or '?'} → {m}" for m in models) or "keine Karte belegt")
    except Exception as e:
        logger.error("Config schreiben fehlgeschlagen: %s", e)
    return len(models)


async def _reload_llama() -> None:
    """llama-swap laeuft mit --watch-config und liest die Datei selbst neu
    (Poll alle 2 s). Die frueher hier aufgerufenen Endpunkte /reload und
    /api/config/reload gibt es nicht (404, ausprobiert). Beim Neuladen startet
    llama-swap ALLE Modelle neu und laedt sie per Preload wieder — die anderen
    Karten sind also kurz nicht erreichbar."""
    await asyncio.sleep(3)   # Poll-Intervall + Puffer, bevor jemand fragt


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)
        # httpx haengt "For more information check: <Link>" an — fuer niemanden hilfreich.
        msg = re.sub(r"\s*For more information check:\s*\S+", "", str(e)).strip()
        return {"ok": False, "content": "", "error": msg[: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 _fetch_loaded() -> list:
    """Welche Modelle liegen GERADE in der Karte? llama-swap /running →
    [{"model", "state"}]. Ohne groups-Config ist das hoechstens eins — ein
    anderes anzufragen entlaedt es. None, wenn llama-swap nicht antwortet
    ("unbekannt" ist etwas anderes als "nichts geladen")."""
    try:
        async with httpx.AsyncClient(timeout=5) as client:
            r = await client.get(f"{LLAMA_URL}/running")
            r.raise_for_status()
            return [{"model": m.get("model", ""), "state": m.get("state", "")}
                    for m in (r.json().get("running") or []) if m.get("model")]
    except Exception as e:
        logger.debug("llama-swap /running nicht abfragbar: %s", e)
        return None


_gpu_name_cache: dict = {}


async def _gpu_names() -> dict:
    """{index: name} aller Karten, einmal per nvidia-smi geholt."""
    if _gpu_name_cache:
        return _gpu_name_cache
    try:
        proc = await asyncio.create_subprocess_exec(
            "nvidia-smi", "--query-gpu=index,name", "--format=csv,noheader",
            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL)
        out, _ = await asyncio.wait_for(proc.communicate(), 10)
        for line in out.decode().splitlines():
            if "," in line:
                i, n = line.split(",", 1)
                _gpu_name_cache[i.strip()] = n.strip()
    except Exception:
        pass
    return _gpu_name_cache


async def _slots_info() -> list:
    """Pro LLM-Karte: [{gpu, gpuName, model, state}]. state aus llama-swap
    /running: ready | starting | stopping | stopped; "unknown", wenn llama-swap
    nicht antwortet; "" bei leerer Karte."""
    slots = _load_slots()
    running = await _fetch_loaded()
    rmap = {r["model"]: r["state"] for r in running} if running is not None else None
    names = await _gpu_names()
    out = []
    for card in _llm_cards():
        m = slots.get(card, "")
        state = "" if not m else ("unknown" if rmap is None else rmap.get(m, "stopped"))
        out.append({"gpu": card, "gpuName": names.get(card, ""), "model": m, "state": state})
    return out


def _selectable(slots_info: list) -> list:
    """Was Clients auswaehlen duerfen: belegte Karten, deren Modell laeuft oder
    gerade (neu) startet — llama-swap haelt Anfragen bis "ready" fest. Auch
    "unknown": wenn llama-swap /running nicht kennt (aeltere Version), lieber
    die Belegung zeigen als die Box unbenutzbar machen."""
    return [s["model"] for s in slots_info
            if s["model"] and s["state"] in ("ready", "starting", "unknown")]


async def _ping(ws) -> None:
    """Lebenszeichen: busy + Karten-Belegung + auswaehlbare Modelle."""
    info = await _slots_info()
    await _send(ws, "worker_ping",
                {"instanceId": INSTANCE_ID, "busy": _inflight > 0,
                 "slots": info, "models": _selectable(info)})


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)."""
    info = await _slots_info()
    models = _selectable(info)
    await _send(ws, "worker_hello", {
        "instanceId": INSTANCE_ID, "service": WORKER_SERVICE,
        # Die LLM-Karten (LLM_GPUS/LLM_GPU). Die eigene NVIDIA_VISIBLE_DEVICES
        # des Adapters ist "all" bzw. wird vom Container-Toolkit ersetzt
        # (Anzeige "GPU void") — sagt nichts darueber, wo die Modelle laufen.
        "node": NODE_NAME, "gpus": ",".join(_llm_cards()),
        "model": models[0] if models else "",
        # NUR geladene Modelle — das ist die Auswahl fuer Clients (ARIA, Fremd-
        # programme). Laden geht ausschliesslich ueber die Diagnostic.
        "models": models,
        "slots": info,            # pro LLM-Karte: Modell + Zustand (Diagnostic)
        "knownModels": _known_models(),   # was in eine Karte geladen werden KOENNTE
        # 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: %s", ", ".join(f"Karte {x['gpu']}={x['model'] or '-'}({x['state'] or 'leer'})" for x in info))


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 _ping(ws)
            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
    slots = _load_slots()
    loaded = [slots[c] for c in _llm_cards() if slots.get(c)]
    targeted = payload.get("targetInstance") == INSTANCE_ID

    async def _refuse(code: str, msg: str) -> None:
        await _send(ws, "llm_response", {
            "requestId": req_id, "ok": False, "error": msg, "errorCode": code,
            "instanceId": INSTANCE_ID, "node": NODE_NAME, "loadedModels": loaded,
        })

    # Clients laden nichts: ein Modell, das auf keiner Karte liegt, gibt es
    # fuer sie nicht. Gezielt an uns adressiert → klare Absage mit der Liste,
    # was geladen IST. Als Broadcast → schweigen: eine andere Box hat es
    # vielleicht (sonst antworteten alle Boxen ohne das Modell mit Fehlern).
    if model and model not in loaded:
        if targeted:
            await _refuse("model_not_loaded",
                          f"Modell {model!r} ist auf {NODE_NAME} nicht geladen. Geladen: "
                          + (", ".join(loaded) or "nichts")
                          + ". Laden geht nur ueber die ARIA-Diagnostic.")
        return
    if not model:
        if not loaded:
            if targeted:
                await _refuse("no_model_loaded",
                              f"Auf {NODE_NAME} ist keine Karte mit einem Modell belegt.")
            return
        # Ohne Modellangabe: das der ersten LLM-Karte (Verhalten wie frueher,
        # als es nur eins gab). Clients sollten das Modell immer angeben.
        model = loaded[0]
    eff_model = 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 not res.get("ok"):
        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, 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,
        # Wer hat geantwortet: Box + Karte. Hilft Clients mit mehreren Boxen.
        "instanceId": INSTANCE_ID, "node": NODE_NAME, "gpu": _card_of(model),
        "usage": res.get("usage") or {},
        "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 (llama-swap laedt die Modelle neu)",
                "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 — greift in wenigen Sekunden. llama-swap startet "
                    "die Modelle dafuer kurz neu."
                    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()
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)


# Quantisierung aus einem Dateinamen: "gemma-4-12B-it-qat-UD-Q4_K_XL.gguf" → "UD-Q4_K_XL"
_QUANT_RE = re.compile(r"(?i)((?:UD-)?(?:IQ[1-4]_(?:XXS|XS|S|M|NL)|Q[2-8]_K(?:_(?:XL|L|M|S))?|Q[4-8]_[01]|BF16|F16|F32))")
# Wenn die gewuenschte fehlt: die erste vorhandene aus dieser Liste. 4-Bit zuerst
# (beste Groesse/Qualitaet fuer 12-GB-Karten), dann groessere.
QUANT_PREFERENCE = ["Q4_K_M", "UD-Q4_K_XL", "Q4_K_XL", "Q4_K_S", "IQ4_XS", "IQ4_NL", "Q4_0",
                    "Q5_K_M", "UD-Q5_K_XL", "Q6_K", "Q8_0"]


def _quant_of(path: str) -> str:
    base = os.path.basename(path)
    base = _SPLIT_RE.sub(".gguf", base)
    m = _QUANT_RE.findall(base[:-5] if base.lower().endswith(".gguf") else base)
    return m[-1].upper() if m else ""


def _is_main_gguf(path: str) -> bool:
    """Das eigentliche Modell — nicht Bild-Projektor (mmproj), nicht die
    Zusatzdatei fuers spekulative Dekodieren (mtp), keine imatrix."""
    low = path.lower()
    base = os.path.basename(low)
    return (low.endswith(".gguf") and "mmproj" not in base and not base.startswith("mtp")
            and "/mtp/" not in ("/" + low) and "imatrix" not in base)


async def _hf_repo_files(repo: str):
    """Alle Modell-GGUFs eines Repos: [{path, oid, size, quant}] — oder None,
    wenn Hugging Face nicht erreichbar ist (dann pruefen wir eben nicht)."""
    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)", repo, e)
        return None
    out = []
    for e in entries if isinstance(entries, list) else []:
        path = e.get("path") or ""
        if e.get("type") != "file" or not _is_main_gguf(path):
            continue
        lfs = e.get("lfs") or {}
        out.append({"path": path, "oid": lfs.get("oid") or e.get("oid") or "",
                    "size": int(lfs.get("size") or e.get("size") or 0), "quant": _quant_of(path)})
    return out


def _pick_quant(files: list, wanted: str) -> str:
    """Die gewuenschte Quantisierung, wenn es sie gibt — sonst die erste aus
    QUANT_PREFERENCE, die das Repo hat. "" wenn gar nichts Brauchbares da ist."""
    have = {f["quant"] for f in files if f["quant"]}
    if wanted and wanted.upper() in have:
        return wanted.upper()
    return next((q for q in QUANT_PREFERENCE if q in have), "")


def _files_for_quant(files: list, quant: str) -> list:
    """Die Datei(en), die llama.cpp fuer repo:quant zieht. Gesplittet → alle
    Teile; sonst die eine (bei mehreren die mit dem kuerzesten Namen)."""
    match = [f for f in files if f["quant"] == quant.upper()]
    split = [f for f in match if _SPLIT_RE.search(f["path"])]
    if split:
        return split
    return [min(match, key=lambda f: len(f["path"]))] if match else []


async def _hf_expected_files(repo: str, quant: str) -> list:
    """Welche GGUF-Dateien zieht 'repo:quant'? [{path, oid, size}]. Fail-soft: []."""
    files = await _hf_repo_files(repo)
    return _files_for_quant(files, quant) if files else []


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(card: str) -> dict:
    """Name + VRAM einer Karte nach dem Laden (fuer 'in Karte N geladen')."""
    if not card:
        return {}
    try:
        proc = await asyncio.create_subprocess_exec(
            "nvidia-smi", f"--id={card}",
            "--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": card}


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 _provision_result(ws, key: str, ok: bool, **extra) -> None:
    await _send(ws, "llm_provision_result",
                {"instanceId": INSTANCE_ID, "key": key, "ok": ok, **extra})


async def _handle_provision(ws, payload: dict) -> None:
    """Laedt ein Modell in eine LLM-Karte (nur aus der Diagnostic). Ersetzt,
    was dort lag. Ablauf: Registry + Belegung → Config (llama-swap laedt per
    --watch-config + Preload) → Warmup, der wartet bis es bereit ist; daneben
    misst _watch_download den Download."""
    key = (payload.get("key") or "").strip()
    repo = (payload.get("hfRepo") or "").strip()
    cards = _llm_cards()
    card = str(payload.get("gpu") if payload.get("gpu") is not None else "").strip()
    if not card and len(cards) == 1:
        card = cards[0]      # aeltere Diagnostic ohne Kartenwahl, Box mit einer Karte
    if not key:
        return await _provision_result(ws, key, False, error="key fehlt")
    if card not in cards:
        return await _provision_result(
            ws, key, False, gpu=card,
            error=(f"Karte {card or '?'} ist auf {NODE_NAME} nicht fuers LLM vorgesehen "
                   f"(LLM-Karten: {', '.join(cards)} — LLM_GPUS in der .env der Box)."))
    slots = _load_slots()
    elsewhere = next((c for c, m in slots.items() if m == key and c != card), None)
    if elsewhere is not None:
        return await _provision_result(
            ws, key, False, gpu=card,
            error=f"{key} liegt schon auf Karte {elsewhere} dieser Box — ein Modell pro Box nur einmal.")
    quant_note = ""
    repo_files = None
    if repo:
        wanted = (payload.get("quant") or "Q4_K_M").strip()
        quant = wanted
        # Gibt es die Datei ueberhaupt? Fehlt sie, beendet sich llama-server
        # sofort und llama-swap meldet nur "500" — der Grund geht verloren.
        # Lieber vorher pruefen und ggf. die naechstbeste Quantisierung nehmen
        # (unsloth z.B. hat statt Q4_K_M ein UD-Q4_K_XL).
        repo_files = await _hf_repo_files(repo)
        if repo_files is not None:
            if not repo_files:
                return await _provision_result(
                    ws, key, False, gpu=card,
                    error=f"In {repo} gibt es keine GGUF-Modelldatei.")
            quant = _pick_quant(repo_files, wanted)
            if not quant:
                have = sorted({f["quant"] for f in repo_files if f["quant"]}) or ["?"]
                return await _provision_result(
                    ws, key, False, gpu=card,
                    error=f"{repo}: keine brauchbare Quantisierung. Vorhanden: {', '.join(have)}")
            if quant != wanted.upper():
                quant_note = f"{wanted} gibt es in {repo} nicht — nehme {quant}"
                logger.info("Karte %s: %s", card, quant_note)
        entry = {
            "key": key, "hfRepo": repo, "quant": quant,
            "ctx": int(payload.get("ctx") or 8192),
            "ngl": int(payload.get("ngl") or 99),
        }
        _save_registry([e for e in _load_registry() if e.get("key") != key] + [entry])
    elif not _model_cmd(key):
        return await _provision_result(ws, key, False, gpu=card, error="hfRepo fehlt (unbekanntes Modell)")

    previous = slots.get(card, "")
    slots[card] = key
    _save_slots(slots)
    _generate_config()

    quant = next((e.get("quant") for e in _load_registry() if e.get("key") == key), "") or "Q4_K_M"
    files = _files_for_quant(repo_files, quant) if repo_files else []
    await _emit_progress(ws, key, "start", total=sum(f["size"] for f in files),
                         files=len(files), gpu=card, replaces=previous, quant=quant,
                         note=quant_note)
    await _reload_llama()
    await _announce(ws)
    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 meldete das frueher "Fehler", waehrend llama-swap 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),
                             replaces=previous, **(await _gpu_info(card)))
        await _ping(ws)
        await _provision_result(ws, key, True, gpu=card, elapsedMs=int(dt * 1000))
        logger.info("Karte %s: %s geladen (%.1fs)%s", card, key, dt,
                    f", ersetzt {previous}" if previous else "")
        return
    # Fehlgeschlagen (z.B. VRAM zu klein): Karte zuruecksetzen. Sonst stuende
    # ein kaputtes Modell in der Config, und jedes Neuladen von llama-swap
    # risse die anderen Karten mit.
    err = (res.get("error") or "Laden fehlgeschlagen")[:200]
    if "500" in err and dt < 120:
        # llama-swap reicht den Grund nicht weiter ("upstream command exited
        # prematurely"). Kam der Abbruch schnell, lief kein Download — dann ist
        # es fast immer eins von dreien.
        err = ("llama.cpp hat sich beim Start sofort beendet. Haeufigste Gruende: "
               "die Modell-Architektur ist neuer als die llama.cpp-Version der Box, "
               "zu wenig (V)RAM, oder die Datei passt nicht. (llama-swap gibt den "
               "genauen Grund nicht weiter.)")
    slots = _load_slots()
    if previous:
        slots[card] = previous
    else:
        slots.pop(card, None)
    _save_slots(slots)
    _generate_config()
    await _reload_llama()
    await _announce(ws)
    note = f" — Karte {card} wieder auf {previous}" if previous else f" — Karte {card} wieder leer"
    await _emit_llm_status(ws, "error", key, error=err)
    await _emit_progress(ws, key, "error", gpu=card, error=err + note)
    await _provision_result(ws, key, False, gpu=card, error=err + note, elapsedMs=int(dt * 1000))
    logger.warning("Karte %s: %s fehlgeschlagen (%s)%s", card, key, err, note)


async def _handle_unload(ws, payload: dict) -> None:
    """Karte leeren (nur Diagnostic). Das GGUF bleibt im Cache — erneutes Laden
    geht dann ohne Download."""
    card = str(payload.get("gpu") if payload.get("gpu") is not None else "").strip()
    slots = _load_slots()
    key = slots.pop(card, "")
    if not key:
        return await _provision_result(ws, "", False, gpu=card, error=f"Karte {card} ist schon leer")
    _save_slots(slots)
    _generate_config()
    await _reload_llama()
    await _announce(ws)
    await _provision_result(ws, key, True, gpu=card, unloaded=True)
    logger.info("Karte %s geleert (war %s)", card, key)


async def _handle_remove(ws, payload: dict) -> None:
    """Modell aus der Registry entfernen (und von seiner Karte, falls belegt).
    Das GGUF bleibt im Cache."""
    key = (payload.get("key") or "").strip()
    if not key:
        return
    _save_registry([e for e in _load_registry() if e.get("key") != key])
    slots = _load_slots()
    for c in [c for c, m in slots.items() if m == key]:
        slots.pop(c)
    _save_slots(slots)
    _generate_config()
    await _reload_llama()
    await _announce(ws)
    await _provision_result(ws, key, True, removed=True)
    logger.info("removed model %s", key)


async def _handle_models_request(ws, payload: dict) -> None:
    """Erkundung fuer Clients: welche Modelle kann ich bei DIESER Box fragen?
    Jede Box antwortet einzeln (RVS ist ein Broadcast-Raum) — der Client
    sammelt die Antworten ein. Nur geladene Modelle, mit Karte."""
    info = await _slots_info()
    sel = set(_selectable(info))
    await _send(ws, "llm_models", {
        "requestId": payload.get("requestId", ""),
        "instanceId": INSTANCE_ID, "node": NODE_NAME,
        "models": [{"model": x["model"], "gpu": x["gpu"], "gpuName": x["gpuName"],
                    "state": x["state"]} for x in info if x["model"] in sel],
    })


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",
                                     "llm_unload_model", "llm_models_request"):
                        continue
                    if mtype == "llm_provision_model":
                        asyncio.create_task(_handle_provision(ws, payload))
                    elif mtype == "llm_unload_model":
                        asyncio.create_task(_handle_unload(ws, payload))
                    elif mtype == "llm_remove_model":
                        asyncio.create_task(_handle_remove(ws, payload))
                    elif mtype == "llm_models_request":
                        asyncio.create_task(_handle_models_request(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())