feat(llm): Cluster — Programme fragen ein Modell, der Cluster waehlt die Karte

Programme sollen nicht Box und Karte aussuchen muessen. Die ARIA-Bridge kennt
alle Boxen samt Karten-Belegung und verteilt (bridge/llm_cluster.py):

- Auswahl: unter allen online Boxen mit dem Modell die mit den wenigsten
  laufenden Anfragen, bei Gleichstand reihum. Eine Karte / alles frei →
  alles dorthin ("verteilen, wenn noetig").
- Ausfall: geht eine Box waehrend der Anfrage offline (kein Ping mehr), wird
  sie EINMAL an eine andere Box mit demselben Modell gegeben. Modell-Fehler
  (z.B. Kontext zu lang) werden nicht wiederholt — waeren ueberall gleich.
- Fehler kommen als Fehler (ok=false + errorCode), nie als Antwort-Text: ein
  Programm wuerde ihn sonst als Aussage der KI in seinen Verlauf schreiben —
  so ist bei ARIA schon mal ein ganzer Chat kaputtgegangen.

RVS: llm_cluster_models_request → llm_cluster_models (jedes Modell EINMAL,
ctx = kleinstes Kontextfenster unter seinen Karten), llm_cluster_request →
llm_cluster_response. In der RVS-Allow-List freigegeben.

ARIA nutzt denselben Router (mit ihrer 30-s-Frist fuer den lokalen
Schnellweg) — vorher Round-Robin ohne Ausfall-Ersatz, und errorCode ging
verloren.

xtts/openai_rvs_wrapper: fuer Programme, die nur "OpenAI-kompatibel" koennen
(OpenHands & Co.). Kleiner Container NEBEN dem Programm, fuer das der Cluster
dann http://127.0.0.1:8099/v1 ist; braucht nur eine ausgehende Verbindung zum
RVS. /v1/models (jedes Modell einmal, context_length), /v1/chat/completions
(auch stream und tools), Fehler als HTTP mit OpenAI-Fehlerobjekt. Hilfsseite
auf / zeigt Status, Base URL und Modelle. Fuer Programme in Docker: Netz
aria-llm, Name openai-rvs-wrapper. .env.example mit RVS_HOST/PORT/TOKEN.
(Ein OpenAI-Endpunkt direkt auf der VM war kurz drin und ist wieder raus —
der Wrapper braucht keinen offenen Port auf der VM.)

docs/LLM-SCHNITTSTELLE.md neu gegliedert: A) Cluster ueber RVS (empfohlen),
B) Wrapper, C) direkt an eine Box (Sonderfall), plus "Umstellen" fuer
bestehende Clients (Software-Schmiede).

Tests: mit echtem RVS und zwei echten llm-adaptern als zwei Boxen —
Modell einmal gelistet, 6 parallele Anfragen 3/3 verteilt, Box friert
mitten in der Anfrage ein → nach 15 s von der anderen beantwortet
(attempts=2), danach direkt die gesunde, nicht geladenes Modell →
model_unavailable ohne content. Wrapper als Docker-Container gegen das
offizielle OpenAI-SDK: models.list, create mit/ohne stream, NotFoundError,
AuthenticationError. Das Doku-Beispiel lief unveraendert (nur Adresse).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

bridge/Dockerfile

@@ -21,5 +21,6 @@ RUN pip install --no-cache-dir -r requirements.txt
 
 COPY aria_bridge.py .
 COPY modes.py .
+COPY llm_cluster.py .
 
 CMD ["python", "aria_bridge.py"]

bridge/aria_bridge.py

@@ -41,6 +41,7 @@ import websockets
 from faster_whisper import WhisperModel
 from openwakeword.model import Model as WakeWordModel
 
+from llm_cluster import LlmCluster
 from modes import Mode, canonical_id, detect_mode_switch, mode_from_id, should_speak
 
 
@@ -806,6 +807,9 @@ class ARIABridge:
         # llm_request via RVS an den llm-adapter (AI-Box) und warten auf
         # llm_response.
         self._pending_llm: dict[str, asyncio.Future] = {}
+        # LLM-Cluster: ein Eingang fuer alle lokalen Modelle aller Boxen
+        # (Auswahl nach Last, Ausfall-Ersatz). Siehe llm_cluster.py.
+        self._llm_cluster = LlmCluster(self)
         # User-Message-Counter fuer Auto-Compact. Bei zu langer Konversation
         # sprengt die argv-Liste beim Claude-Subprocess-Spawn (E2BIG). Bei
         # COMPACT_AFTER erreicht → Sessions reset + Container restart.
@@ -2453,6 +2457,9 @@ class ARIABridge:
 
         msg_type = message.get("type", "")
         payload = message.get("payload", {})
+        if msg_type.startswith("llm_cluster_") and isinstance(payload, dict):
+            if await self._llm_cluster.handle_rvs(msg_type, payload):
+                return
 
         if msg_type == "chat":
             # Nur User-Nachrichten weiterleiten — ARIA/Diagnostic-Antworten ignorieren (sonst Loop!)
@@ -4252,65 +4259,28 @@ class ARIABridge:
         if not isinstance(messages, list) or not messages:
             return {"ok": False, "error": "messages leer/ungueltig"}
 
-        request_id = str(uuid.uuid4())
-        loop = asyncio.get_event_loop()
-        future: asyncio.Future = loop.create_future()
-        self._pending_llm[request_id] = future
-        try:
-            req_payload = {
-                "requestId": request_id,
-                "messages": messages,
-                "max_tokens": max_tokens,
-                "temperature": temperature,
-            }
-            if stop:
-                req_payload["stop"] = stop
-            if tools:
-                req_payload["tools"] = tools
-            if model:
-                req_payload["model"] = model
-            # Redundanz/Multitasking: freie llm-Instanz gezielt adressieren, die
-            # das gewaehlte Modell fahren kann; None → Broadcast wie bisher.
-            # Mehrere Boxen mit demselben Modell → Round-Robin (pro Projekt verteilt).
-            llm_target = self._pick_worker("llm", model=model or None)
-            if llm_target:
-                req_payload["targetInstance"] = llm_target
-            elif model and self._llm_boxes_with_slots():
-                # Boxen mit Karten-Belegung laden nichts nach und schweigen auf
-                # Broadcasts fuer fremde Modelle. Ein Broadcast liefe also nur in
-                # den Timeout — lieber sofort ehrlich absagen.
-                self._pending_llm.pop(request_id, None)
-                loaded = sorted({m for w in self._llm_boxes_with_slots() for m in (w.get("models") or [])})
-                return {"ok": False, "errorCode": "model_not_loaded",
-                        "error": (f"Lokales Modell {model!r} ist auf keiner KI-Box geladen. "
-                                  f"Geladen: {', '.join(loaded) or 'nichts'}. "
-                                  "Laden: Diagnostic → Modell-Katalog.")}
-            logger.info("[rvs] llm_request → llm-adapter (id=%s, msgs=%d, max_tokens=%d, tools=%d, model=%s, target=%s)",
-                        request_id[:8], len(messages), max_tokens, len(tools) if tools else 0,
-                        model or "-", llm_target or "(broadcast)")
-            ok = await self._send_to_rvs({
-                "type": "llm_request",
-                "payload": req_payload,
-                "timestamp": int(time.time() * 1000),
-            })
-            if not ok:
-                return {"ok": False, "error": "llm_request konnte nicht gesendet werden"}
-            try:
-                result = await asyncio.wait_for(future, timeout=self._LLM_TIMEOUT_S)
-            except asyncio.TimeoutError:
-                return {"ok": False, "error": f"Timeout ({self._LLM_TIMEOUT_S:.0f}s) — AI-Box nicht erreichbar?"}
-            if not isinstance(result, dict) or not result.get("ok"):
-                err = (result or {}).get("error") if isinstance(result, dict) else "leeres Resultat"
-                return {"ok": False, "error": err or "llm-adapter Fehler"}
-            return {
-                "ok": True,
-                "content": result.get("content", ""),
-                "tool_calls": result.get("tool_calls"),
-                "model": result.get("model"),
-                "elapsedMs": result.get("elapsedMs"),
-            }
-        finally:
-            self._pending_llm.pop(request_id, None)
+        # Derselbe Router wie fuer Fremdprogramme: Box nach Last waehlen, bei
+        # Ausfall einmal anderswo versuchen, ehrliche Absage, wenn das Modell
+        # nirgends geladen ist (statt Broadcast + Timeout).
+        res = await self._llm_cluster.route({
+            "model": model or "", "messages": messages, "max_tokens": max_tokens,
+            "temperature": temperature, "stop": stop, "tools": tools,
+        }, timeout=self._LLM_TIMEOUT_S)
+        if not res.get("ok"):
+            err = res.get("error") or "llm-adapter Fehler"
+            if res.get("errorCode") == "model_unavailable":
+                err += " Laden: Diagnostic → Einstellungen → LLM-Karten."
+            return {"ok": False, "error": err, "errorCode": res.get("errorCode")}
+        logger.info("[cluster] ARIA ← %s (Karte %s, %s ms, Versuche %s)",
+                    res.get("node") or res.get("instanceId"), res.get("gpu", "?"),
+                    res.get("elapsedMs"), res.get("attempts"))
+        return {
+            "ok": True,
+            "content": res.get("content", ""),
+            "tool_calls": res.get("tool_calls"),
+            "model": res.get("model"),
+            "elapsedMs": res.get("elapsedMs"),
+        }
 
     async def _flux_generate(self, prompt: str, width: int, height: int,
                              steps: Optional[int], guidance: Optional[float],

bridge/llm_cluster.py

@@ -0,0 +1,218 @@
+"""LLM-Cluster: ein Eingang fuer alle lokalen Sprachmodelle aller KI-Boxen.
+
+Clients waehlen ein MODELL, nicht Box oder Karte. Wo es laeuft, entscheidet
+dieser Router:
+
+  * Auswahl     — unter allen online Boxen, die das Modell GELADEN haben
+                  (Karten-Belegung aus worker_hello/worker_ping), die mit den
+                  wenigsten laufenden Anfragen; bei Gleichstand reihum.
+                  Ist nur eine Karte da oder alles frei, landet alles dort —
+                  "verteilen, wenn noetig, sonst nicht".
+  * Ausfall     — verschwindet eine Box mitten in der Anfrage (kein Ping mehr)
+                  oder antwortet nicht, geht die Anfrage EINMAL an eine andere
+                  Box mit demselben Modell. Fehler des Modells selbst (z.B.
+                  Kontext zu lang) werden nicht wiederholt.
+  * Fehler      — kommen als FEHLER zurueck (ok=false / HTTP-Status), nie als
+                  normale Antwort. Ein Client wuerde einen Fehlertext sonst
+                  als Aussage der KI in seinen Verlauf schreiben — genau das
+                  hat bei ARIA schon mal einen ganzen Chat vergiftet.
+
+Eingang ist der RVS:
+
+  llm_cluster_models_request → llm_cluster_models
+  llm_cluster_request        → llm_cluster_response
+
+Programme brauchen nur eine ausgehende Verbindung zum RVS, egal in welchem
+Netz sie stehen. Werkzeuge, die nur "OpenAI-kompatibel" koennen (OpenHands
+& Co.), bekommen den OpenAI-RVS-Wrapper neben sich gestellt
+(xtts/openai_rvs_wrapper) — fuer sie ist der Cluster dann 127.0.0.1.
+
+Schnittstelle fuer Drittprogramme: docs/LLM-SCHNITTSTELLE.md
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import time
+import uuid
+from typing import Optional
+
+logger = logging.getLogger("aria-bridge.llm-cluster")
+
+# Diese errorCodes heissen "die BOX war das Problem" → anderswo nochmal versuchen.
+# Alles andere (Modell-Fehler, z.B. Kontext zu lang) waere ueberall gleich.
+RETRY_CODES = {"timeout", "box_offline", "send_failed", "model_not_loaded"}
+MAX_ATTEMPTS = 2
+
+
+class LlmCluster:
+    def __init__(self, bridge):
+        self.b = bridge
+        self.inflight: dict[str, int] = {}      # instanceId → laufende Anfragen (von uns)
+        self._rr: dict[str, int] = {}
+        self.timeout = float(os.getenv("CLUSTER_LLM_TIMEOUT_S", "600"))
+
+    # ── Flotte ──────────────────────────────────────────────────────────
+
+    def _online_llm(self) -> list:
+        now = time.time()
+        return [w for w in self.b._workers.values()
+                if w.get("service") == "llm"
+                and (now - w.get("last_seen", 0)) < self.b.WORKER_OFFLINE_S]
+
+    def _is_online(self, instance_id: str) -> bool:
+        return any(w.get("instanceId") == instance_id for w in self._online_llm())
+
+    def models(self) -> list:
+        """Jedes geladene Modell GENAU EINMAL, egal auf wie vielen Karten.
+        ctx = das kleinste Kontextfenster unter seinen Karten (darauf muss sich
+        ein Client verlassen koennen, egal wohin der Router verteilt)."""
+        out: dict[str, dict] = {}
+        for w in self._online_llm():
+            slots = w.get("slots")
+            if isinstance(slots, list):
+                for sl in slots:
+                    m = sl.get("model")
+                    if not m or m not in (w.get("models") or []):
+                        continue
+                    e = out.setdefault(m, {"model": m, "ctx": 0, "cards": []})
+                    e["cards"].append({"node": w.get("node", ""), "instanceId": w.get("instanceId", ""),
+                                       "gpu": sl.get("gpu", ""), "gpuName": sl.get("gpuName", ""),
+                                       "state": sl.get("state", "")})
+                    c = int(sl.get("ctx") or 0)
+                    if c and (not e["ctx"] or c < e["ctx"]):
+                        e["ctx"] = c
+            else:
+                # Box mit altem Adapter (ohne Karten): nur die Modell-Liste
+                for m in (w.get("models") or []):
+                    e = out.setdefault(m, {"model": m, "ctx": 0, "cards": []})
+                    e["cards"].append({"node": w.get("node", ""), "instanceId": w.get("instanceId", ""),
+                                       "gpu": "", "gpuName": "", "state": ""})
+        for e in out.values():
+            e["available"] = len(e["cards"])
+        return sorted(out.values(), key=lambda e: e["model"])
+
+    def candidates(self, model: str) -> list:
+        return [w for w in self._online_llm() if model in (w.get("models") or [])]
+
+    def pick(self, model: str, exclude=()) -> Optional[str]:
+        """Die Box mit den wenigsten laufenden Anfragen; Gleichstand reihum."""
+        cands = [w["instanceId"] for w in self.candidates(model) if w["instanceId"] not in exclude]
+        if not cands:
+            return None
+        low = min(self.inflight.get(c, 0) for c in cands)
+        tied = sorted(c for c in cands if self.inflight.get(c, 0) == low)
+        i = self._rr.get(model, 0) % len(tied)
+        self._rr[model] = i + 1
+        return tied[i]
+
+    # ── Anfrage an eine bestimmte Box ───────────────────────────────────
+
+    async def _send_to(self, target: str, req: dict, deadline: float) -> dict:
+        """llm_request an GENAU diese Box, Antwort abwarten. Bricht frueh ab,
+        wenn die Box unterwegs offline geht — dann kann route() woanders hin."""
+        rid = str(uuid.uuid4())
+        fut: asyncio.Future = asyncio.get_running_loop().create_future()
+        self.b._pending_llm[rid] = fut
+        self.inflight[target] = self.inflight.get(target, 0) + 1
+        try:
+            payload = {k: v for k, v in req.items() if v is not None}
+            payload.update(requestId=rid, targetInstance=target)
+            ok = await self.b._send_to_rvs({"type": "llm_request", "payload": payload,
+                                            "timestamp": int(time.time() * 1000)})
+            if not ok:
+                return {"ok": False, "errorCode": "send_failed", "error": "RVS: Anfrage nicht gesendet"}
+            while True:
+                left = deadline - time.time()
+                if left <= 0:
+                    return {"ok": False, "errorCode": "timeout",
+                            "error": f"Keine Antwort von {target} in der Frist"}
+                try:
+                    return await asyncio.wait_for(asyncio.shield(fut), timeout=min(5.0, left))
+                except asyncio.TimeoutError:
+                    if not self._is_online(target):
+                        return {"ok": False, "errorCode": "box_offline",
+                                "error": f"{target} ist waehrend der Anfrage offline gegangen"}
+        finally:
+            self.inflight[target] = max(0, self.inflight.get(target, 1) - 1)
+            self.b._pending_llm.pop(rid, None)
+
+    async def route(self, req: dict, timeout: Optional[float] = None) -> dict:
+        """Eine Anfrage an den Cluster. req: model, messages, max_tokens,
+        temperature, stop, tools. Antwort: wie llm_response der Box, plus
+        attempts; bei Fehler errorCode + error (+ available).
+        timeout: Gesamtfrist inkl. Ausfall-Ersatz. ARIA nutzt eine kurze (lokal
+        ist ihr Schnellweg — dauert es zu lange, nimmt sie Claude)."""
+        model = (req.get("model") or "").strip()
+        messages = req.get("messages")
+        if not isinstance(messages, list) or not messages:
+            return {"ok": False, "errorCode": "bad_request", "error": "messages fehlt oder ist leer"}
+        available = [m["model"] for m in self.models()]
+        if not model:
+            if len(available) == 1:
+                model = available[0]       # eindeutig — dann darf es fehlen
+            else:
+                return {"ok": False, "errorCode": "bad_request", "available": available,
+                        "error": "model fehlt. Geladen: " + (", ".join(available) or "nichts")}
+        body = {"model": model, "messages": messages,
+                "max_tokens": int(req.get("max_tokens") or 512),
+                "temperature": float(req["temperature"]) if req.get("temperature") is not None else 0.7,
+                "stop": req.get("stop") or None, "tools": req.get("tools") or None}
+        limit = timeout or self.timeout
+        deadline = time.time() + limit
+        tried: list = []
+        last: dict = {}
+        for _ in range(MAX_ATTEMPTS):
+            target = self.pick(model, exclude=tried)
+            if not target:
+                break
+            tried.append(target)
+            t0 = time.time()
+            last = await self._send_to(target, body, deadline)
+            if last.get("ok"):
+                last["attempts"] = len(tried)
+                return last
+            logger.warning("[cluster] %s auf %s: %s (%s)%s", model, target, last.get("errorCode") or "Fehler",
+                           (last.get("error") or "")[:120], " — versuche andere Box" if
+                           last.get("errorCode") in RETRY_CODES else "")
+            if last.get("errorCode") not in RETRY_CODES or time.time() >= deadline:
+                break
+        if not tried:
+            return {"ok": False, "errorCode": "model_unavailable", "available": available,
+                    "error": (f"Modell {model!r} ist gerade auf keiner KI-Box im Cluster geladen. "
+                              f"Geladen: {', '.join(available) or 'nichts'}.")}
+        last.setdefault("errorCode", "model_error")
+        last["attempts"] = len(tried)
+        if last["errorCode"] in RETRY_CODES and len(tried) >= 1:
+            others = [c for c in self.candidates(model) if c["instanceId"] not in tried]
+            if not others:
+                last["error"] = (last.get("error") or "") + (
+                    " — keine weitere Box mit diesem Modell erreichbar.")
+        return last
+
+    # ── RVS-Eingang ─────────────────────────────────────────────────────
+
+    async def handle_rvs(self, mtype: str, payload: dict) -> bool:
+        if mtype == "llm_cluster_models_request":
+            await self.b._send_to_rvs({"type": "llm_cluster_models", "payload": {
+                "requestId": payload.get("requestId", ""), "models": self.models()},
+                "timestamp": int(time.time() * 1000)})
+            return True
+        if mtype == "llm_cluster_request":
+            asyncio.create_task(self._rvs_request(payload))
+            return True
+        return False
+
+    async def _rvs_request(self, payload: dict) -> None:
+        rid = payload.get("requestId", "")
+        res = await self.route(payload)
+        out = {"requestId": rid, "ok": bool(res.get("ok"))}
+        for k in ("content", "tool_calls", "model", "node", "gpu", "instanceId", "usage",
+                  "elapsedMs", "attempts", "error", "errorCode", "available"):
+            if res.get(k) is not None:
+                out[k] = res[k]
+        await self.b._send_to_rvs({"type": "llm_cluster_response", "payload": out,
+                                   "timestamp": int(time.time() * 1000)})