fix(llm): fertige Antwort geht bei RVS-Abbruch nicht mehr verloren

28.10., 17:51:42: alle Clients verloren gleichzeitig die RVS-Verbindung
(RVS/Proxy kurz weg) — waehrend Gemma 31 s an einer Anfrage von Marina
rechnete. Nach 2 s war alles wieder verbunden, aber die fertige Antwort ging
ueber die tote Verbindung raus und wurde verworfen ("Send fehlgeschlagen
(llm_response)"); Marina wartete bis zu ihrer Frist (660 s) ins Leere.

llm-adapter: llm_response wird ueber die aktuelle Verbindung geschickt, ist
gerade keine offen, aufgehoben (5 min) und nach dem Wiederverbinden
nachgeschickt.

RVS: gezielte Antwort, deren Fragender nur neu verbunden ist (alte
Verbindung weg), geht bis 2 MB an den Raum statt verworfen zu werden — der
Fragende erkennt sie an der requestId. Dateien bleiben gezielt.

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

rvs/server.js

@@ -188,6 +188,9 @@ const TARGETED_REPLIES = new Set(TARGETED.values());
 // Nachrichten-ID) — darum ein Set, sonst bekaeme nur der letzte die Antwort.
 const pendingReplies = new Map();
 const PENDING_TTL_MS = 180_000;
+// Antworten bis zu dieser Groesse gehen an den Raum, wenn der Fragende nur
+// neu verbunden ist (siehe Weiterleitung). Dateien sind groesser.
+const REJOIN_BROADCAST_MAX = 2 * 1024 * 1024;
 
 function pendingKey(replyType, requestId) {
   return `${replyType}\u0000${requestId}`;
@@ -696,7 +699,16 @@ function registerClient(ws, token) {
     rememberRequester(ws, msg);
 
     // Gezielte Antwort? Dann NUR an den Fragenden — nicht an den ganzen Raum.
-    const targets = targetedRecipients(msg);
+    let targets = targetedRecipients(msg);
+    // Fragender inzwischen neu verbunden (RVS-/Proxy-Neustart waehrend die
+    // Antwort noch lief): seine alte Verbindung gibt es nicht mehr, die neue
+    // kennen wir nicht. Kleine Antworten (LLM-Antwort, Status) dann an den
+    // Raum — der Fragende erkennt sie an der requestId. Grosse (Dateien)
+    // bleiben verworfen, sonst gingen hunderte MB an alle.
+    if (targets && targets.length === 0 && raw.length <= REJOIN_BROADCAST_MAX) {
+      log(`${msg.type}: Fragender neu verbunden — ${(raw.length / 1024).toFixed(0)}KB an den Raum`);
+      targets = null;
+    }
     if (targets) {
       const data = raw.toString();
       let sent = 0;

xtts/llm-adapter/adapter.py

@@ -332,15 +332,49 @@ async def _reload_llama() -> None:
     await asyncio.sleep(3)   # Poll-Intervall + Puffer, bevor jemand fragt
 
 
+# Antworten, die nicht verloren gehen duerfen: Bricht die RVS-Verbindung ab,
+# waehrend eine Anfrage noch rechnet (z.B. RVS-/Proxy-Neustart), wuerde die
+# fertige Antwort sonst ueber die tote Verbindung geschickt und verworfen — der
+# Fragende (Schmiede, ARIA) wartet dann bis zu seiner Frist ins Leere (am
+# 28.10. 11 Minuten). Jetzt: ueber die NEUE Verbindung senden, notfalls bis zum
+# Wiederverbinden aufheben.
+_RELIABLE_TYPES = {"llm_response"}
+_OUTBOX_TTL_S = 300
+_outbox: list = []              # [(ts, mtype, payload)]
+_current_ws = None              # die aktuell offene RVS-Verbindung
+
+
 async def _send(ws, mtype: str, payload: dict) -> None:
-    try:
+    data = json.dumps({"type": mtype, "payload": payload, "timestamp": int(time.time() * 1000)})
-        await ws.send(json.dumps({
+    targets = [ws] if ws is _current_ws or _current_ws is None else [ws, _current_ws]
-            "type": mtype,
+    last_err = None
-            "payload": payload,
+    for t in targets:
-            "timestamp": int(time.time() * 1000),
+        try:
-        }))
+            await t.send(data)
-    except Exception as e:
+            return
-        logger.warning("Send fehlgeschlagen (%s): %s", mtype, e)
+        except Exception as e:
+            last_err = e
+    if mtype in _RELIABLE_TYPES:
+        _outbox.append((time.time(), mtype, payload))
+        logger.warning("Send fehlgeschlagen (%s): %s — aufgehoben, geht nach dem Wiederverbinden raus",
+                       mtype, last_err)
+    else:
+        logger.warning("Send fehlgeschlagen (%s): %s", mtype, last_err)
+
+
+async def _flush_outbox(ws) -> None:
+    """Nach dem (Wieder-)Verbinden: aufgehobene Antworten nachschicken."""
+    now = time.time()
+    keep = [(t, m, p) for (t, m, p) in _outbox if now - t <= _OUTBOX_TTL_S]
+    _outbox.clear()
+    for t, m, p in keep:
+        try:
+            await ws.send(json.dumps({"type": m, "payload": p, "timestamp": int(time.time() * 1000)}))
+            logger.info("Aufgehobene Antwort nachgeschickt (%s, %.0f s alt)", m, now - t)
+        except Exception as e:
+            _outbox.append((t, m, p))
+            logger.warning("Nachschicken fehlgeschlagen (%s): %s", m, e)
+            return
 
 
 def _sanitize_messages(messages: list) -> list:
@@ -1364,6 +1398,7 @@ async def _handle_models_request(ws, payload: dict) -> None:
 
 
 async def _run() -> None:
+    global _current_ws
     if not RVS_HOST:
         logger.error("RVS_HOST nicht gesetzt — Abbruch")
         return
@@ -1401,6 +1436,9 @@ async def _run() -> None:
                         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")
+                _current_ws = ws
+                if _outbox:
+                    await _flush_outbox(ws)
                 retry_s = 2
                 tls_fallback_tried = False
                 ping_task = asyncio.create_task(_worker_register(ws))