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


def _sanitize_messages(messages: list) -> list:
    """Verlauf so aufbereiten, dass llama.cpp ihn nicht als Ganzes ablehnt.

    Steht im Verlauf ein frueherer Werkzeugaufruf, dessen arguments KEIN
    gueltiges JSON sind, antwortet llama.cpp auf die GANZE Anfrage mit 500
    ("Failed to parse tool call arguments as JSON") — und jede weitere Anfrage
    mit demselben Verlauf auch. Bei Programmen, die Code schreiben, passiert
    das leicht (Dateiinhalt mit Anfuehrungszeichen/Umbruechen). Statt den
    ganzen Auftrag scheitern zu lassen: solche Argumente als {"_raw": "…"}
    verpacken. Ein dict als arguments (statt String) wird zu JSON.
    Nur Kopien — die Nachrichten des Aufrufers bleiben unveraendert."""
    out = []
    for m in messages or []:
        tcs = m.get("tool_calls") if isinstance(m, dict) else None
        if not tcs:
            out.append(m)
            continue
        fixed = []
        for tc in tcs:
            fn = dict((tc or {}).get("function") or {})
            args = fn.get("arguments")
            if isinstance(args, (dict, list)):
                fn["arguments"] = json.dumps(args, ensure_ascii=False)
            elif isinstance(args, str):
                try:
                    json.loads(args or "{}")
                    if not args.strip():
                        fn["arguments"] = "{}"
                except Exception:
                    logger.warning("Werkzeugaufruf %r im Verlauf hat kein gueltiges JSON als "
                                   "arguments (%d Zeichen) — eingepackt", fn.get("name"), len(args))
                    fn["arguments"] = json.dumps({"_raw": args}, ensure_ascii=False)
            elif args is None:
                fn["arguments"] = "{}"
            fixed.append(dict(tc, function=fn))
        out.append(dict(m, tool_calls=fixed))
    return out


def _llama_error_info(exc: Exception) -> dict:
    """Maschinenlesbare Fehlerart. Wichtigster Fall: der Verlauf ist laenger
    als das Kontextfenster — dann soll ein Programm automatisch kuerzen und
    nochmal fragen koennen, statt einen Text auszuwerten."""
    resp = getattr(exc, "response", None)
    if resp is None:
        return {}
    try:
        e = (resp.json() or {}).get("error") or {}
    except Exception:
        return {}
    if isinstance(e, dict) and e.get("type") == "exceed_context_size_error":
        return {"errorCode": "context_exceeded",
                "promptTokens": int(e.get("n_prompt_tokens") or 0),
                "ctx": int(e.get("n_ctx") or 0)}
    return {}


def _llama_error_text(exc: Exception) -> str:
    """Bei HTTP-Fehlern den GRUND von llama.cpp mitnehmen. Bisher kam nur
    "Server error '500 …'" an — die eigentliche Erklaerung steht im Body."""
    resp = getattr(exc, "response", None)
    if resp is None:
        return re.sub(r"\s*For more information check:\s*\S+", "", str(exc)).strip()
    detail = ""
    try:
        j = resp.json()
        e = j.get("error") if isinstance(j, dict) else None
        detail = (e.get("message") if isinstance(e, dict) else e) or json.dumps(j)[:600]
    except Exception:
        detail = (resp.text or "")[:600]
    logger.warning("llama.cpp HTTP %s — Antwort: %s", resp.status_code, (resp.text or "")[:2000])
    return f"llama.cpp HTTP {resp.status_code}: {str(detail).strip()[:600]}"


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": _sanitize_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:
        msg = _llama_error_text(e)
        logger.warning("llama.cpp-Call fehlgeschlagen: %s", msg[:300])
        return {"ok": False, "content": "", "error": msg[:700], **_llama_error_info(e)}


# 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", "vramMB"}} aller Karten, einmal per nvidia-smi geholt.
    Der VRAM steht fest — die Diagnostic warnt damit vor Modellen, die nicht
    auf die Karte passen."""
    if _gpu_name_cache:
        return _gpu_name_cache
    try:
        proc = await asyncio.create_subprocess_exec(
            "nvidia-smi", "--query-gpu=index,name,memory.total", "--format=csv,noheader,nounits",
            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL)
        out, _ = await asyncio.wait_for(proc.communicate(), 10)
        for line in out.decode().splitlines():
            parts = [x.strip() for x in line.split(",")]
            if len(parts) >= 3:
                _gpu_name_cache[parts[0]] = {"name": parts[1], "vramMB": int(float(parts[2]))}
    except Exception:
        pass
    return _gpu_name_cache


def _model_ctx(key: str) -> int:
    """Kontextfenster, mit dem llama-server das Modell startet (-c). Clients
    MUESSEN das wissen: rechnen sie mit mehr, scheitern lange Auftraege oder
    werden falsch gekuerzt."""
    m = re.search(r"(?:^|\s)(?:-c|--ctx-size)\s+(\d+)", _model_cmd(key) or "")
    return int(m.group(1)) if m else 0


# ── Kontextfenster: was das Modell kann + was es an Grafikspeicher kostet ──
#
# Aus dem Kopf der GGUF-Datei (nur Metadaten, die Gewichte werden nicht
# gelesen): trainiertes Maximum (context_length) und die Groesse des KV-Caches
# pro Token. Modelle mit Sliding-Window-Attention (Gemma) halten in den
# "lokalen" Schichten nur ein festes Fenster — nur die globalen Schichten
# wachsen mit dem Kontext. Annahme: KV-Cache in f16 (llama.cpp-Standard).

_GGUF_T = {0: "B", 1: "b", 2: "H", 3: "h", 4: "I", 5: "i", 6: "f", 7: "?", 10: "Q", 11: "q", 12: "d"}
_gguf_cache: dict = {}          # path → (mtime, meta)
CTX_STEPS = [4096, 8192, 16384, 32768, 65536, 131072, 262144, 524288, 1048576]
# Fester Aufschlag (CUDA-Kontext + Rechenpuffer von llama.cpp) — gemessen auf
# einer RTX 3060 mit Gemma 4 12B bei 32k: ~1,4 GB ueber Modell + KV-Cache.
VRAM_OVERHEAD_MB = 1400
UBATCH = 512


def _gguf_meta(path: str) -> dict:
    """Metadaten-Block einer GGUF-Datei (gecacht). Lange String-Listen (der
    Tokenizer) werden uebersprungen, nicht gespeichert."""
    try:
        mt = os.path.getmtime(path)
    except OSError:
        return {}
    hit = _gguf_cache.get(path)
    if hit and hit[0] == mt:
        return hit[1]
    import struct
    meta: dict = {}

    def rd(f, fmt):
        return struct.unpack("<" + fmt, f.read(struct.calcsize("<" + fmt)))[0]

    def rstr(f):
        return f.read(rd(f, "Q")).decode("utf-8", "replace")

    def rval(f, t):
        if t == 8:
            return rstr(f)
        if t == 9:
            et, n = rd(f, "I"), rd(f, "Q")
            if et == 8:
                for _ in range(n):
                    f.seek(rd(f, "Q"), 1)
                return None
            if et == 9:
                raise ValueError("verschachtelte Arrays")
            fmt = _GGUF_T[et]
            if n > 4096:
                f.seek(struct.calcsize("<" + fmt) * n, 1)
                return None
            return [rd(f, fmt) for _ in range(n)]
        return rd(f, _GGUF_T[t])

    try:
        with open(path, "rb") as f:
            if f.read(4) != b"GGUF":
                return {}
            rd(f, "I"); rd(f, "Q")
            for _ in range(rd(f, "Q")):
                k = rstr(f)
                v = rval(f, rd(f, "I"))
                if v is not None:
                    meta[k] = v
    except Exception as e:
        logger.debug("GGUF %s nicht lesbar: %s", path, e)
        meta = {}
    _gguf_cache[path] = (mt, meta)
    return meta


def _model_files(key: str) -> tuple:
    """(Gewichte.gguf, mmproj.gguf|None) eines Modells im HF-Cache unter /models
    — aus "-hf org/repo:QUANT" bzw. "-m pfad" im llama-server-Aufruf."""
    import glob
    cmd = _model_cmd(key) or ""
    m = re.search(r"(?:^|\s)-m\s+(\S+)", cmd)
    if m and os.path.isfile(m.group(1)):
        return m.group(1), None
    m = re.search(r"(?:^|\s)-hf\s+([^\s:]+)(?::(\S+))?", cmd)
    if not m:
        return None, None
    repo, quant = m.group(1), (m.group(2) or "").lower()
    base = os.path.join(MODELS_DIR, "models--" + repo.replace("/", "--"), "snapshots")
    files = [p for p in glob.glob(os.path.join(base, "*", "*.gguf")) if os.path.isfile(p)]
    weights = [p for p in files if "mmproj" not in os.path.basename(p).lower()]
    if quant:
        weights = [p for p in weights if quant in os.path.basename(p).lower()] or weights
    mmproj = [p for p in files if "mmproj" in os.path.basename(p).lower()]
    return (weights[0] if weights else None), (mmproj[0] if mmproj else None)


def _ctx_profile(key: str) -> dict:
    """{ctxMax, options: [{ctx, estMB}]} fuer die Kontext-Auswahl der Diagnostic.
    Leer, wenn die Modelldatei (noch) nicht da oder nicht lesbar ist."""
    weights, mmproj = _model_files(key)
    if not weights:
        return {}
    meta = _gguf_meta(weights)
    arch = meta.get("general.architecture")
    if not arch:
        return {}
    g = lambda k, d=None: meta.get(f"{arch}.{k}", d)
    n_layer = int(g("block_count", 0) or 0)
    if not n_layer:
        return {}

    def per_layer(v, d=0):
        return list(v) if isinstance(v, list) else [v if v is not None else d] * n_layer

    kv_heads = per_layer(g("attention.head_count_kv", g("attention.head_count", 0)))
    k_len = int(g("attention.key_length", 0) or 0)
    v_len = int(g("attention.value_length", 0) or 0)
    if not (k_len and v_len):
        emb, heads = int(g("embedding_length", 0) or 0), int(g("attention.head_count", 0) or 0)
        k_len = v_len = emb // heads if heads else 0
    k_swa = int(g("attention.key_length_swa", k_len) or k_len)
    v_swa = int(g("attention.value_length_swa", v_len) or v_len)
    window = int(g("attention.sliding_window", 0) or 0)
    pattern = g("attention.sliding_window_pattern")
    swa = per_layer(pattern, False) if isinstance(pattern, list) else [False] * n_layer
    if not window:
        swa = [False] * n_layer

    per_tok = sum(int(kv_heads[i]) * (k_len + v_len) * 2 for i in range(n_layer) if not swa[i])
    swa_tok = sum(int(kv_heads[i]) * (k_swa + v_swa) * 2 for i in range(n_layer) if swa[i])
    files_mb = (os.path.getsize(weights) + (os.path.getsize(mmproj) if mmproj else 0)) / 1048576
    ctx_max = int(g("context_length", 0) or 0)
    steps = [c for c in CTX_STEPS if not ctx_max or c <= ctx_max]
    if ctx_max and ctx_max not in steps:
        steps.append(ctx_max)          # krummes Maximum (z.B. 40960) mit anbieten
    opts = []
    for c in steps:
        kv_mb = (per_tok * c + swa_tok * min(c, window + UBATCH)) / 1048576
        opts.append({"ctx": c, "estMB": int(files_mb + kv_mb + VRAM_OVERHEAD_MB)})
    return {"ctxMax": ctx_max, "options": opts}


async def _vram_used() -> dict:
    """{index: belegte MB} — frisch bei jedem Aufruf (aendert sich laufend)."""
    try:
        proc = await asyncio.create_subprocess_exec(
            "nvidia-smi", "--query-gpu=index,memory.used", "--format=csv,noheader,nounits",
            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL)
        out, _ = await asyncio.wait_for(proc.communicate(), 10)
        return {a.strip(): int(float(b)) for a, b in
                (l.split(",", 1) for l in out.decode().splitlines() if "," in l)}
    except Exception:
        return {}


# Ein geladenes Modell belegt GB auf seiner Karte. Bleibt die Karte trotz
# "ready" praktisch leer, rechnet llama.cpp auf der CPU — typisch, wenn der
# Treiber zu alt fuer das CUDA im Image ist ("forward compatibility …").
# llama.cpp meldet das nicht nach aussen; wir sehen es nur am Speicher.
ON_CPU_BELOW_MB = 400


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
    gpus = await _gpu_names()
    used = await _vram_used()
    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"))
        g = gpus.get(card) or {}
        u = used.get(card)
        on_cpu = bool(m and state == "ready" and u is not None and u < ON_CPU_BELOW_MB)
        out.append({"gpu": card, "gpuName": g.get("name", ""), "vramMB": g.get("vramMB", 0),
                    "vramUsedMB": u if u is not None else -1, "onCpu": on_cpu,
                    "model": m, "state": state, "ctx": _model_ctx(m) if m else 0,
                    **(_ctx_profile(m) if m else {})})
    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'}{', CPU!' if x.get('onCpu') else ''})" 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,
        # Fehlerart fuer Programme (z.B. context_exceeded + promptTokens + ctx)
        **{k: res[k] for k in ("errorCode", "promptTokens", "ctx") if res.get(k)},
        # 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
    # Fuer ein Zuruecksetzen: der bisherige Registry-Eintrag dieses Modells.
    # Noetig, wenn nur das Kontextfenster geaendert wird (gleiches Modell, gleiche
    # Karte) — sonst bliebe nach einem Fehlschlag der zu grosse Wert stehen.
    old_entry = next((dict(e) for e in _load_registry() if e.get("key") == key), 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)
    reg = [e for e in _load_registry() if e.get("key") != key]
    if old_entry:
        reg.append(old_entry)
    _save_registry(reg)
    _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"], "ctx": x["ctx"],
                    "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())