"""
ARIA Voice Bridge — Hauptmodul.

Verbindet die Android App (via RVS) mit ARIA-Core. Spracheingabe laeuft
ueber die whisper-bridge (Gamebox, faster-whisper auf CUDA), Sprachausgabe
ueber die f5tts-bridge (Voice Cloning, satzweises PCM-Streaming).

Nachrichtenfluss:
  App  → RVS → Bridge → aria-core
  aria-core → Bridge → f5tts-bridge → PCM → RVS → App
"""

from __future__ import annotations

import asyncio
import base64
import json
import logging
import mimetypes
import os
import re
import signal
import socket
import ssl
import threading
import time
import sys
import tempfile
import uuid
from collections import OrderedDict
from pathlib import Path
from typing import Optional

import subprocess
import urllib.request
import numpy as np
import sounddevice as sd
import websockets
from faster_whisper import WhisperModel
from openwakeword.model import Model as WakeWordModel

from modes import Mode, canonical_id, detect_mode_switch, mode_from_id, should_speak

# ── Logging ──────────────────────────────────────────────────

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    stream=sys.stdout,
)
logger = logging.getLogger("aria-bridge")


# ── TCP-Keepalive Helper ────────────────────────────────────
#
# Aktiviert TCP-Level Keepalive auf einer websockets-Verbindung mit
# aggressiven Intervallen: 30s idle bis erster Probe, 10s zwischen
# Probes, 3 verfehlte → Verbindung tot. Das deckt den Fall ab dass
# NAT-Tabellen-Verfall die TCP-Verbindung still kills ohne RST — Linux-
# Default braeucht sonst 2 Stunden idle bis der Kernel selber probt.
def _enable_tcp_keepalive(ws) -> None:
    try:
        sock = ws.transport.get_extra_info("socket")
        if sock is None:
            return
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
        # Linux-spezifisch — TCP_KEEPIDLE/INTVL/CNT existieren auf macOS
        # mit anderem Namen; im Container ist Linux garantiert.
        for opt, val in (
            ("TCP_KEEPIDLE", 30),
            ("TCP_KEEPINTVL", 10),
            ("TCP_KEEPCNT", 3),
        ):
            const = getattr(socket, opt, None)
            if const is not None:
                sock.setsockopt(socket.IPPROTO_TCP, const, val)
        logger.info("[rvs] TCP-Keepalive aktiviert (idle=30s, intvl=10s, cnt=3)")
    except Exception as exc:
        logger.warning("[rvs] TCP-Keepalive konnte nicht aktiviert werden: %s", exc)


# ── Konfiguration ───────────────────────────────────────────

VOICES_DIR = Path("/voices")
CORE_WS_URL = os.getenv("ARIA_CORE_WS", "ws://127.0.0.1:18789")
CORE_AUTH_TOKEN = os.getenv("ARIA_AUTH_TOKEN", "")  # OpenClaw Gateway Token
RVS_HOST = os.getenv("RVS_HOST", "")          # z.B. rvs.hackersoft.de
RVS_PORT = os.getenv("RVS_PORT", "443")      # Port des RVS
RVS_TLS = os.getenv("RVS_TLS", "true")       # true = wss://, false = ws://
RVS_TLS_FALLBACK = os.getenv("RVS_TLS_FALLBACK", "true")  # Bei TLS-Fehler ws:// versuchen
RVS_TOKEN = os.getenv("RVS_TOKEN", "")       # Pairing-Token (gleich wie in der App)
DIAGNOSTIC_URL = os.getenv("DIAGNOSTIC_URL", "http://127.0.0.1:3001")  # Diagnostic API
WHISPER_MODEL = os.getenv("WHISPER_MODEL", "medium")
WHISPER_LANGUAGE = os.getenv("WHISPER_LANGUAGE", "de")

# Audio-Parameter
SAMPLE_RATE = 16000
CHANNELS = 1
BLOCK_SIZE = 1280  # 80ms bei 16kHz — gut fuer Wake-Word-Erkennung
RECORD_SECONDS = 8  # Max. Aufnahmedauer nach Wake-Word

def load_config() -> dict[str, str]:
    """Laedt Konfiguration ausschliesslich aus /shared/config/runtime.json
    (zentral gepflegt ueber Diagnostic UI). Tokens + RVS-Settings kommen
    via ENV (siehe docker-compose).
    """
    config: dict[str, str] = {}
    runtime_path = Path("/shared/config/runtime.json")
    if runtime_path.exists():
        try:
            runtime = json.loads(runtime_path.read_text())
            config = {k: str(v) for k, v in runtime.items() if v not in (None, "")}
            if config:
                logger.info("Runtime-Config geladen: %s", sorted(config.keys()))
        except Exception as e:
            logger.warning("runtime.json konnte nicht gelesen werden: %s", e)
    else:
        logger.info("Keine runtime.json — Diagnostic schreibt sie beim ersten Konfigurieren")
    return config


# ── Voice Engine ─────────────────────────────────────────────


import re as _re_tts

_NUM_WORDS_DE = {
    0: "null", 1: "eins", 2: "zwei", 3: "drei", 4: "vier", 5: "fuenf",
    6: "sechs", 7: "sieben", 8: "acht", 9: "neun", 10: "zehn",
    11: "elf", 12: "zwoelf", 13: "dreizehn", 14: "vierzehn", 15: "fuenfzehn",
    16: "sechzehn", 17: "siebzehn", 18: "achtzehn", 19: "neunzehn", 20: "zwanzig",
}
_TENS_DE = {30: "dreissig", 40: "vierzig", 50: "fuenfzig"}


def _num_to_words_de(n: int) -> str:
    """Zahlen 0-59 als deutsches Wort — fuer Uhrzeiten und kleine Bereiche."""
    if n in _NUM_WORDS_DE:
        return _NUM_WORDS_DE[n]
    if 21 <= n <= 29:
        return f"{_NUM_WORDS_DE[n - 20]}undzwanzig"
    if 30 <= n <= 59:
        tens = (n // 10) * 10
        ones = n % 10
        tens_word = _TENS_DE.get(tens, str(tens))
        if ones == 0:
            return tens_word
        return f"{_NUM_WORDS_DE.get(ones, str(ones))}und{tens_word}"
    return str(n)


def _time_range_to_words(m):
    """'8:00-9:00 Uhr' → 'acht bis neun Uhr', '8-9 Uhr' → 'acht bis neun Uhr'."""
    h1 = int(m.group(1))
    h2 = int(m.group(3))
    return f"{_num_to_words_de(h1)} bis {_num_to_words_de(h2)} Uhr"


def _small_range_to_words(m):
    """'5-6' → 'fuenf bis sechs' (nur wenn beide Zahlen ≤ 24)."""
    a, b = int(m.group(1)), int(m.group(2))
    if a > 24 or b > 24 or a >= b:
        return m.group(0)
    return f"{_num_to_words_de(a)} bis {_num_to_words_de(b)}"


def _decimal_to_words(m):
    """'0.1' / '0,1' → 'null komma eins', '1,25' → 'eins komma zwei fuenf'."""
    int_part = int(m.group(1))
    dec_part = m.group(2)
    int_word = _num_to_words_de(int_part) if 0 <= int_part <= 59 else str(int_part)
    dec_words = " ".join(_num_to_words_de(int(d)) for d in dec_part)
    return f"{int_word} komma {dec_words}"


_UNIT_WORDS = [
    (r'\bTB\b', 'Terabyte'),
    (r'\bGB\b', 'Gigabyte'),
    (r'\bMB\b', 'Megabyte'),
    (r'\bKB\b', 'Kilobyte'),
    (r'\bkB\b', 'Kilobyte'),
    (r'\bms\b', 'Millisekunden'),
    (r'\bkm/h\b', 'Kilometer pro Stunde'),
    (r'\bkm\b', 'Kilometer'),
    (r'\bm/s\b', 'Meter pro Sekunde'),
    (r'\bkg\b', 'Kilogramm'),
    (r'\b°C\b', 'Grad Celsius'),
    (r'°C', ' Grad Celsius'),
    (r'\bMbps\b', 'Megabit pro Sekunde'),
    (r'\bGbps\b', 'Gigabit pro Sekunde'),
    (r'\bMhz\b|\bMHz\b', 'Megahertz'),
    (r'\bGhz\b|\bGHz\b', 'Gigahertz'),
    (r'%', ' Prozent'),
    (r'\bCPU\b', 'C P U'),
    (r'\bGPU\b', 'G P U'),
    (r'\bRAM\b', 'R A M'),
    (r'\bSSD\b', 'S S D'),
    (r'\bHDD\b', 'H D D'),
    (r'\bURL\b', 'U R L'),
    (r'\bAPI\b', 'A P I'),
    (r'\bRVS\b', 'R V S'),
    (r'\bSSH\b', 'S S H'),
    (r'\bVM\b', 'V M'),
    (r'\bUI\b', 'U I'),
    (r'\bTTS\b', 'T T S'),
    (r'\bSTT\b', 'S T T'),
    (r'\bTLS\b', 'T L S'),
]


def strip_voice_tag_for_display(text: str) -> str:
    """Entfernt `<voice>...</voice>`-Bloecke aus dem Chat-Display-Text.

    ARIA kann einen <voice>-Block ANHAENGEN um eine TTS-freundliche Variante
    ihrer Antwort zu liefern (Zahlen ausgeschrieben, Einheiten als Wort,
    Markdown entfernt). Der Block wird dann von clean_text_for_tts als
    TTS-Quelle benutzt — fuer die Chat-Bubble in der App soll er aber NICHT
    sichtbar sein, sonst sieht Stefan literal '<voice>...' in seinem Chat.

    Beispiel-Input (Stefan-typisch fuer Wetterbericht):
        '**Wetter:** 23,5°C, Wind 15 kt NW\\n<voice>Wetter: dreiundzwanzig
         komma fuenf Grad, Wind fuenfzehn Knoten Nordwest.</voice>'
    Output:
        '**Wetter:** 23,5°C, Wind 15 kt NW'

    Mehrere Voice-Bloecke werden alle entfernt (ARIA koennte theoretisch
    mehrere setzen, machen wir robust). Trailing-Whitespace nach dem Block
    auch wegtrimmen.
    """
    if not text or "<voice>" not in text.lower():
        return text
    return _re_tts.sub(r'\s*<voice>[\s\S]*?</voice>\s*', '\n', text, flags=_re_tts.IGNORECASE).strip()


def clean_text_for_tts(text: str) -> str:
    """Bereitet Chat-Text fuer Sprachausgabe auf.

    - `<voice>...</voice>` Tag: wenn vorhanden, NUR dieser Inhalt wird gelesen
    - Code-Bloecke (```...``` und `...`) werden komplett entfernt
    - Markdown (Fett, Kursiv, Links, Headings, Listen, Zitate) wird abgeraeumt
    - Einheiten und gaengige Abkuerzungen werden ausgeschrieben (22GB → 22 Gigabyte)
    - URLs werden durch "ein Link" ersetzt
    - Mehrfach-Leerzeichen/Umbrueche normalisiert
    """
    if not text:
        return ""

    # <voice>...</voice> wenn vorhanden → nur das nehmen
    voice_match = _re_tts.search(r'<voice>([\s\S]*?)</voice>', text, _re_tts.IGNORECASE)
    if voice_match:
        text = voice_match.group(1)

    t = text

    # Code-Bloecke komplett raus (Zeilenumbruch statt Platzhalter — sonst bricht Satzlogik)
    t = _re_tts.sub(r'```[\s\S]*?```', '. ', t)
    t = _re_tts.sub(r'`[^`]+`', '', t)

    # Markdown
    t = _re_tts.sub(r'\*\*([^*]+)\*\*', r'\1', t)
    t = _re_tts.sub(r'\*([^*]+)\*', r'\1', t)
    t = _re_tts.sub(r'__([^_]+)__', r'\1', t)
    t = _re_tts.sub(r'\[([^\]]+)\]\((https?://[^)]+)\)', r'\1, ein Link', t)
    t = _re_tts.sub(r'\[([^\]]+)\]\([^)]+\)', r'\1', t)
    t = _re_tts.sub(r'https?://\S+', 'ein Link', t)
    t = _re_tts.sub(r'^#{1,6}\s*', '', t, flags=_re_tts.MULTILINE)
    t = _re_tts.sub(r'^>\s*', '', t, flags=_re_tts.MULTILINE)
    t = _re_tts.sub(r'^[\-\*]\s+', '', t, flags=_re_tts.MULTILINE)

    # Zeitbereiche: "8:00-9:00 Uhr" / "8-9 Uhr" → "acht bis neun Uhr"
    t = _re_tts.sub(r'\b(\d{1,2})(:\d{2})?\s*[-–]\s*(\d{1,2})(:\d{2})?\s*Uhr\b', _time_range_to_words, t)
    # Uhrzeiten mit Minuten: "8:30 Uhr" → "acht Uhr dreissig", "8:00 Uhr" → "acht Uhr"
    def _single_time(m):
        h = int(m.group(1))
        mn = int(m.group(2)) if m.group(2) else 0
        words = _num_to_words_de(h) + " Uhr"
        if mn > 0:
            words += " " + _num_to_words_de(mn)
        return words
    t = _re_tts.sub(r'\b(\d{1,2}):(\d{2})\s*Uhr\b', _single_time, t)
    # Volle Uhrzeiten ohne ":" — "15 Uhr" → "fuenfzehn Uhr"
    t = _re_tts.sub(r'\b(\d{1,2})\s+Uhr\b', lambda m: f"{_num_to_words_de(int(m.group(1)))} Uhr", t)
    # Kleine Zahlen-Bereiche ohne "Uhr": "5-6" → "fuenf bis sechs"
    t = _re_tts.sub(r'\b(\d{1,2})\s*[-–]\s*(\d{1,2})\b', _small_range_to_words, t)

    # Dezimalzahlen: "0.1" / "0,5" / "1,25" → "null komma eins" / "null komma fuenf" / ...
    # Muss vor "Zahl+Einheit" laufen, sonst frisst die Unit-Regel den Nachkommaanteil.
    # Lookahead verhindert Match auf IP-artigen Strings wie 192.168.1.1.
    t = _re_tts.sub(r'\b(\d+)[.,](\d+)(?![.,\d])', _decimal_to_words, t)

    # Zahlen + Einheit: "22GB" → "22 Gigabyte" (Leerzeichen einfuegen)
    t = _re_tts.sub(r'(\d+)([A-Za-z]{1,4})\b', r'\1 \2', t)

    # Einheiten/Abkuerzungen ausschreiben
    for pat, repl in _UNIT_WORDS:
        t = _re_tts.sub(pat, repl, t)

    # Generisches Buchstabieren: alle verbleibenden 2-5-Zeichen-Grossbuchstaben-Woerter
    # (XTTS, USB, DNS, JSON, HTML, ...) → "X T T S". Laeuft NACH der expliziten Liste,
    # damit TTS/GPU/... schon aufgeloest sind. "WLAN"-artige, die als Wort gesprochen
    # werden, koennen bei Bedarf explizit in _UNIT_WORDS uebersteuert werden.
    t = _re_tts.sub(r'\b([A-Z]{2,5})\b', lambda m: " ".join(m.group(1)), t)

    # Anfuehrungszeichen
    t = _re_tts.sub(r'["""„`]', '', t)

    # Absaetze/Zeilenumbrueche normalisieren
    t = _re_tts.sub(r'\n{2,}', '. ', t)
    t = _re_tts.sub(r'\n', ', ', t)
    t = _re_tts.sub(r'\s{2,}', ' ', t)
    t = _re_tts.sub(r'\s*\.\s*\.\s*', '. ', t)

    return t.strip()


# ── STT Engine ───────────────────────────────────────────────


class STTEngine:
    """Whisper Speech-to-Text — laeuft komplett lokal."""

    def __init__(self, model_size: str = "small", language: str = "de") -> None:
        self.model_size = model_size
        self.language = language
        self.model: Optional[WhisperModel] = None

    def initialize(self) -> None:
        """Laedt das Whisper-Modell."""
        logger.info(
            "Lade Whisper-Modell '%s' (Sprache: %s)...",
            self.model_size,
            self.language,
        )
        self.model = WhisperModel(self.model_size, device="cpu", compute_type="int8")
        logger.info("Whisper-Modell geladen")

    def reload(self, model_size: str) -> bool:
        """Laedt ein anderes Whisper-Modell (bei Config-Aenderung)."""
        if model_size == self.model_size and self.model is not None:
            return False
        allowed = {"tiny", "base", "small", "medium", "large-v3"}
        if model_size not in allowed:
            logger.warning("Ungueltiges Whisper-Modell: %s (erlaubt: %s)", model_size, allowed)
            return False
        logger.info("Lade Whisper-Modell neu: %s -> %s", self.model_size, model_size)
        self.model_size = model_size
        self.model = None
        try:
            self.model = WhisperModel(model_size, device="cpu", compute_type="int8")
            logger.info("Whisper-Modell '%s' geladen", model_size)
            return True
        except Exception:
            logger.exception("Whisper-Modell '%s' konnte nicht geladen werden", model_size)
            return False

    def transcribe(self, audio_data: np.ndarray) -> str:
        """Transkribiert Audio-Daten zu Text.

        Args:
            audio_data: NumPy-Array mit Audio (float32, 16kHz, mono).

        Returns:
            Erkannter Text oder leerer String.
        """
        if self.model is None:
            # Lazy-Load: normalerweise laeuft STT remote auf der Gamebox.
            # Erst wenn das Fallback hier zuschlaegt, laden wir lokal.
            logger.info("Lokales Whisper-Fallback — Modell wird nachgeladen...")
            try:
                self.initialize()
            except Exception:
                logger.exception("Lokales Whisper konnte nicht geladen werden")
                return ""
            if self.model is None:
                return ""

        try:
            # Audio als float32 normalisieren
            if audio_data.dtype != np.float32:
                audio_data = audio_data.astype(np.float32) / 32768.0

            segments, info = self.model.transcribe(
                audio_data,
                language=self.language,
                beam_size=5,
                vad_filter=True,
            )

            text = " ".join(segment.text.strip() for segment in segments)
            logger.info("STT: '%s' (Sprache: %s, Dauer: %.1fs)", text, info.language, info.duration)
            return text

        except Exception:
            logger.exception("STT-Fehler")
            return ""


# ── Wake-Word Erkennung ──────────────────────────────────────


class WakeWordDetector:
    """Erkennt das Wake-Word im Audio-Stream.

    Nutzt ein Custom-Modell aus /voices/wake_aria.onnx falls vorhanden,
    sonst das eingebaute 'hey_jarvis' als Fallback.
    """

    CUSTOM_MODEL_PATH = "/voices/wake_aria.onnx"
    FALLBACK_MODEL = "hey_jarvis"
    THRESHOLD = 0.5

    def __init__(self) -> None:
        self.model: Optional[WakeWordModel] = None
        self.wake_word_key: str = ""

    def initialize(self) -> None:
        """Laedt das Wake-Word-Modell.

        Hinweis: WakeWordModel() wird OHNE Argumente aufgerufen.
        Aeltere openwakeword-Versionen leiten unbekannte kwargs
        an AudioFeatures weiter, was zum Crash fuehrt.
        """
        logger.info("Lade Wake-Word-Modell...")

        # Alle eingebauten Modelle laden (ohne kwargs — Kompatibilitaet!)
        self.model = WakeWordModel()

        # Verfuegbare Modelle ermitteln
        available = list(self.model.models.keys()) if hasattr(self.model, 'models') else []
        logger.info("Verfuegbare Wake-Words: %s", ", ".join(available) if available else "(keine)")

        # Bestes Modell auswaehlen
        if self.FALLBACK_MODEL in available:
            self.wake_word_key = self.FALLBACK_MODEL
        elif available:
            self.wake_word_key = available[0]
        else:
            self.wake_word_key = self.FALLBACK_MODEL

        logger.info("Wake-Word aktiv: '%s'", self.wake_word_key)
        logger.info(
            "Tipp: Custom 'aria' Wake-Word trainieren → "
            "https://github.com/dscripka/openWakeWord#training-new-models"
        )

    def detect(self, audio_chunk: np.ndarray) -> bool:
        """Prueft ob das Wake-Word im Audio-Chunk enthalten ist.

        Args:
            audio_chunk: Audio-Daten (int16, 16kHz).

        Returns:
            True wenn Wake-Word erkannt wurde.
        """
        if self.model is None:
            return False

        prediction = self.model.predict(audio_chunk)

        # openwakeword gibt Scores pro Modell zurueck
        for key, score in prediction.items():
            if score > self.THRESHOLD:
                logger.info("Wake-Word '%s' erkannt! (Score: %.2f)", key, score)
                return True

        return False


# ── Audio-Aufnahme ───────────────────────────────────────────


def record_audio(duration: float = RECORD_SECONDS) -> np.ndarray:
    """Nimmt Audio vom Mikrofon auf.

    Args:
        duration: Aufnahmedauer in Sekunden.

    Returns:
        NumPy-Array mit Audio-Daten (int16, 16kHz, mono).
    """
    logger.info("Aufnahme laeuft... (%d Sekunden)", duration)
    audio = sd.rec(
        int(duration * SAMPLE_RATE),
        samplerate=SAMPLE_RATE,
        channels=CHANNELS,
        dtype="int16",
    )
    sd.wait()
    logger.info("Aufnahme beendet")
    return audio.flatten()


# ── Bridge Hauptklasse ───────────────────────────────────────


class ARIABridge:
    """ARIA Voice Bridge — verbindet App (via RVS) und Sprache mit ARIA-Core.

    Drei parallele Aufgaben:
      1. connect_to_core()  — WebSocket zu aria-core (lokal)
      2. connect_to_rvs()   — WebSocket zum RVS (oeffentlich, fuer die App)
      3. audio_loop()       — Wake-Word + STT (lokales Mikrofon)
    """

    def __init__(self) -> None:
        self.config = load_config()
        self.ws_url = self.config.get("ARIA_CORE_WS", CORE_WS_URL)
        self.core_auth_token = self.config.get("ARIA_AUTH_TOKEN", CORE_AUTH_TOKEN)
        self._req_id_counter = 0
        self._session_key = "main"  # Fallback, wird per Diagnostic API aktualisiert
        self._diagnostic_url = self.config.get("DIAGNOSTIC_URL", DIAGNOSTIC_URL)
        # RVS-Verbindungsinfo aus Config oder Env
        rvs_host = self.config.get("RVS_HOST", RVS_HOST)
        rvs_port = self.config.get("RVS_PORT", RVS_PORT)
        rvs_tls = self.config.get("RVS_TLS", RVS_TLS).lower() == "true"
        self.rvs_tls_fallback = self.config.get("RVS_TLS_FALLBACK", RVS_TLS_FALLBACK).lower() == "true"
        self.rvs_token = self.config.get("RVS_TOKEN", RVS_TOKEN)
        # URLs zusammenbauen (primaer + fallback)
        if rvs_host:
            proto = "wss" if rvs_tls else "ws"
            self.rvs_url = f"{proto}://{rvs_host}:{rvs_port}"
            # Fallback-URL (ohne TLS) nur wenn TLS aktiv und Fallback erlaubt
            if rvs_tls and self.rvs_tls_fallback:
                self.rvs_url_fallback = f"ws://{rvs_host}:{rvs_port}"
            else:
                self.rvs_url_fallback = ""
        else:
            self.rvs_url = ""
            self.rvs_url_fallback = ""
        # Mode aus Shared Config laden (persistiert ueber Container-Restarts)
        self.current_mode = self._load_persisted_mode()
        self.running = False

        # Idempotenz: zuletzt gesehene clientMsgIds (App-seitig generiert).
        # Beim Reconnect/Retry sendet die App dieselbe ID nochmal — wir
        # antworten erneut mit ACK aber leiten NICHT doppelt an Brain weiter.
        # OrderedDict als FIFO mit Capping (Insertion-Order).
        self._seen_client_msg_ids: "OrderedDict[str, float]" = OrderedDict()
        self._SEEN_CLIENT_MSG_LIMIT = 200

        # Komponenten (TTS: F5-TTS remote auf der Gamebox, lokales TTS wurde entfernt)
        self.tts_enabled = True
        self.xtts_voice = ""
        self._f5tts_config: dict = {}
        self._flux_config: dict = {}
        # Persistente TTS-Speed (App-Setting), wird aus voice_config.json
        # gelesen + bei config-Broadcasts (siehe handle config in chat)
        # geupdated. Fallback wenn der Per-Request-Override fehlt.
        self._persistent_xtts_speed: Optional[float] = None
        vc: dict = {}
        # Gespeicherte Voice-Config laden
        try:
            vc_path = "/shared/config/voice_config.json"
            if os.path.exists(vc_path):
                with open(vc_path) as f:
                    vc = json.load(f)
                self.tts_enabled = vc.get("ttsEnabled", True)
                self.xtts_voice = vc.get("xttsVoice", "")
                # Persistente TTS-Speed: vorher war's nur per-Chat-Override
                # (App schickte speed mit jeder Nachricht). Bei Diagnostic-Chat
                # OHNE App-Vor-Chat blieb _next_speed_override=None → 1.0.
                # Jetzt persistent — Bridge greift bei TTS immer auf den
                # zuletzt von der App gesetzten Wert zurueck.
                try:
                    persisted_speed = float(vc.get("xttsSpeed", 1.0))
                    if 0.1 <= persisted_speed <= 5.0:
                        self._persistent_xtts_speed: Optional[float] = persisted_speed
                    else:
                        self._persistent_xtts_speed = None
                except (TypeError, ValueError):
                    self._persistent_xtts_speed = None
                # F5-TTS-Felder aufsammeln (werden spaeter via RVS rebroadcastet,
                # damit die f5tts-bridge auf der Gamebox die Settings auch nach
                # Restart wiederbekommt — sonst stuende sie auf Hard-Defaults)
                for k in ("f5ttsModel", "f5ttsCkptFile", "f5ttsVocabFile",
                          "f5ttsCfgStrength", "f5ttsNfeStep"):
                    if k in vc:
                        self._f5tts_config[k] = vc[k]
                # FLUX-Felder (Default-Modell + Keywords) gleicher Mechanismus
                for k in ("fluxDefaultModel", "fluxKeywordRaw", "fluxKeywordSwitch", "huggingfaceToken"):
                    if k in vc:
                        self._flux_config[k] = vc[k]
                # Debug-Log-Toggles fuer Whisper / F5TTS Bridges (Diagnostic-Toggle).
                # Default: aus — sonst muellen wir uns volle Disk wenn alles laeuft.
                self._debug_log_config: dict = {}
                for k in ("whisperDebugLog", "f5ttsDebugLog"):
                    if k in vc:
                        self._debug_log_config[k] = bool(vc[k])
                logger.info("Voice-Config geladen: tts=%s voice=%s f5tts=%s flux=%s",
                            self.tts_enabled, self.xtts_voice or "default",
                            self._f5tts_config or "defaults",
                            self._flux_config or "defaults")
        except Exception as e:
            logger.warning("Voice-Config laden fehlgeschlagen: %s", e)
        # Whisper-Modell: Config hat Vorrang, dann env/Default (medium)
        whisper_model = vc.get("whisperModel") or self.config.get("WHISPER_MODEL", WHISPER_MODEL)
        self.stt_engine = STTEngine(
            model_size=whisper_model,
            language=self.config.get("WHISPER_LANGUAGE", WHISPER_LANGUAGE),
        )
        self.wake_word = WakeWordDetector()

        # WebSocket-Verbindungen
        self.ws_core: Optional[websockets.WebSocketClientProtocol] = None
        self.ws_rvs: Optional[websockets.WebSocketClientProtocol] = None

        # Letzter gesendeter agent_activity-State (zum Entduplizieren)
        self._last_activity_state: Optional[tuple] = None
        # Zeitstempel des letzten chat:final — waehrend 3s danach werden
        # trailing Agent-Events unterdrueckt (Core raeumt manchmal nach).
        self._last_chat_final_at: float = 0.0
        # requestId → messageId Map fuer XTTS-Audio-Cache (App-seitige Zuordnung)
        self._xtts_request_to_message: dict[str, str] = {}
        # Voice-Router (Multi-Threading, 06/2026): sticky Projekt-Kontext fuer
        # STT-Voice-Nachrichten. Wechselt via „fuer <name>:"-Prefix, faellt nach
        # STICKY_TIMEOUT_SEC ohne neue Voice-Message zurueck auf Hauptchat.
        # Meta-Kommandos („zurueck zum hauptchat") werden client-seitig
        # interceptiert und aendern hier den Sticky OHNE Brain-Roundtrip.
        self._voice_sticky_project_id: str = ""
        self._voice_sticky_expires_at: float = 0.0
        # Focused-project pro Stream: die App schickt bei stt_stream_start
        # die projectId ihres aktuellen Focus mit. Wenn das Voice-Ergebnis
        # weder Meta-Kommando noch Prefix ist und der Sticky abgelaufen,
        # nutzen wir das als Default (Voice folgt dem sichtbaren Kontext).
        self._stt_stream_projects: dict[str, str] = {}
        # Voice-Override aus letzter Chat-Nachricht einer App.
        # Wird fuer die direkt folgende ARIA-Antwort genutzt und dann zurueckgesetzt.
        # So kann jedes Geraet seine bevorzugte Stimme bekommen (pro Request).
        self._next_voice_override: Optional[str] = None
        # Gleiche Logik fuer die Wiedergabegeschwindigkeit (F5-TTS speed-Param,
        # App-Setting aria_tts_speed, 1.0 = normal).
        self._next_speed_override: Optional[float] = None
        # STT-Requests die aktuell auf Antwort von der whisper-bridge (Gamebox) warten.
        # requestId → Future mit dem Text (oder None bei Fehler).
        self._pending_stt: dict[str, asyncio.Future] = {}
        # whisper-bridge service_status: True wenn ready, False/None wenn loading/unbekannt.
        # Beeinflusst das Timeout fuer stt_request — bei "loading" warten wir laenger,
        # weil das Modell beim ersten Request noch ~1-2 Min runtergeladen werden kann.
        self._remote_stt_ready: bool = False
        # FLUX-Render-Requests die aktuell auf Antwort der flux-bridge (Gamebox) warten.
        # requestId → Future mit dem flux_response-Payload (oder None bei Fehler).
        self._pending_flux: dict[str, asyncio.Future] = {}
        # flux-bridge service_status: True wenn ready. Render-Timeouts werden
        # bei 'loading' deutlich grosszuegiger gesetzt (Modell-Download ~24 GB).
        self._remote_flux_ready: bool = False
        # Lokales LLM (Plan B): requestId → Future mit dem llm_response-Payload.
        # Analog zu _pending_flux — Brain ruft /internal/local-llm, wir relayen
        # llm_request via RVS an den llm-adapter (Gamebox) und warten auf
        # llm_response.
        self._pending_llm: dict[str, asyncio.Future] = {}
        # 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.
        # Counter ueberlebt Bridge-Restart nicht (frischer Zaehler beim Start ok).
        # _user_message_count + _compact_after entfallen — Auto-Compact war
        # aria-core-spezifisch (E2BIG-Schutz). Der neue Brain-Loop kennt
        # diese Begrenzung nicht.
        # Pending Files: wenn die App ein Bild + Text gleichzeitig schickt, kommen
        # zwei separate RVS-Events ('file' und 'chat') — wir buffern die Files
        # kurz und mergen sie mit dem nachfolgenden Chat-Text zu einer einzigen
        # Anfrage an aria-core. Sonst antwortet ARIA zweimal (einmal "warte auf
        # Anweisung" beim file, einmal auf den Chat-Text).
        # Liste von Tuples: (file_path, name, file_type, size_kb, width, height)
        self._pending_files: list[tuple[str, str, str, int, int, int]] = []
        self._pending_files_flush_task: Optional[asyncio.Task] = None
        # Projekt-Kontext der gerade gepufferten Anhaenge (aus dem file-Upload).
        # Wird beim Flush an send_to_core gegeben, damit Anhaenge im richtigen
        # Projekt landen statt im Hauptchat.
        self._pending_files_project_id: str = ""
        self._PENDING_FILES_WINDOW_SEC: float = 0.8

    def initialize(self) -> None:
        """Initialisiert alle Komponenten.

        Audio-Komponenten (TTS, STT, Wake-Word) sind optional —
        wenn kein Audio-Geraet vorhanden ist (z.B. VM ohne Soundkarte),
        laeuft die Bridge trotzdem als reiner RVS-Relay.
        """
        logger.info("=" * 50)
        logger.info("ARIA Voice Bridge startet...")
        logger.info("=" * 50)

        # STT wird standardmaessig von der whisper-bridge (Gamebox) erledigt.
        # Lokales Whisper ist nur Fallback und wird lazy geladen wenn remote nicht
        # antwortet. Das spart RAM auf der VM und Startup-Zeit.

        # Audio-Hardware pruefen (fuer lokales Mikro/Lautsprecher)
        self.audio_available = False
        try:
            devices = sd.query_devices()
            sd.query_devices(kind='output')
            self.audio_available = True
            logger.info("Audio-Geraet gefunden — Wake-Word und lokale TTS aktiv")
            self.wake_word.initialize()
        except (sd.PortAudioError, Exception):
            logger.warning("Kein Audio-Geraet — Wake-Word und lokale Wiedergabe deaktiviert")
            logger.info("TTS rendert fuer App (via RVS), STT verarbeitet App-Audio")

        logger.info("Alle Komponenten initialisiert")
        logger.info("aria-core:  %s", self.ws_url)
        if self.rvs_url and self.rvs_token:
            logger.info("RVS:        %s (Token: %s...)", self.rvs_url, self.rvs_token[:8])
        else:
            logger.warning("RVS nicht konfiguriert — App-Verbindung deaktiviert")
            logger.warning("  Setze RVS_HOST, RVS_PORT, RVS_TOKEN in der .env auf der VM")
        logger.info("Modus:      %s %s", self.current_mode.config.emoji, self.current_mode.config.name)

    # ── aria-core Verbindung (OpenClaw Gateway Protokoll) ───

    def _next_req_id(self) -> str:
        """Erzeugt eine eindeutige Request-ID fuer das OpenClaw-Protokoll."""
        self._req_id_counter += 1
        return f"bridge-{self._req_id_counter}"

    async def _openclaw_handshake(self, ws: websockets.WebSocketClientProtocol) -> bool:
        """Fuehrt den OpenClaw Gateway Handshake durch.

        1. Wartet auf connect.challenge Event vom Gateway
        2. Sendet connect Request mit Auth-Token
        3. Wartet auf hello-ok Response

        Returns:
            True wenn Handshake erfolgreich, sonst False.
        """
        try:
            # Schritt 1: Auf Challenge warten (max 10s)
            raw = await asyncio.wait_for(ws.recv(), timeout=10.0)
            challenge = json.loads(raw)

            if challenge.get("type") != "event" or challenge.get("event") != "connect.challenge":
                logger.error("[core] Unerwartete erste Nachricht: %s", raw[:200])
                return False

            nonce = challenge.get("payload", {}).get("nonce", "")
            logger.info("[core] Challenge empfangen (nonce: %s...)", nonce[:8] if nonce else "?")

            # Schritt 2: Connect Request senden
            connect_req = {
                "type": "req",
                "id": self._next_req_id(),
                "method": "connect",
                "params": {
                    "minProtocol": 3,
                    "maxProtocol": 3,
                    "client": {
                        "id": "gateway-client",
                        "version": "0.0.3",
                        "platform": "linux",
                        "mode": "backend",
                    },
                    "role": "operator",
                    "scopes": ["operator.read", "operator.write"],
                    "caps": ["voice"],
                    "commands": [],
                    "permissions": {},
                    "auth": {"token": self.core_auth_token} if self.core_auth_token else {},
                    "locale": "de-DE",
                    "userAgent": "aria-bridge/0.0.3",
                },
            }
            await ws.send(json.dumps(connect_req))
            logger.info("[core] Connect-Request gesendet")

            # Schritt 3: Auf hello-ok warten (max 10s)
            raw = await asyncio.wait_for(ws.recv(), timeout=10.0)
            response = json.loads(raw)

            if response.get("type") == "res" and response.get("ok"):
                logger.info("[core] Handshake erfolgreich — hello-ok empfangen")
                return True
            else:
                error = response.get("error", response)
                logger.error("[core] Handshake fehlgeschlagen: %s", json.dumps(error)[:200])
                return False

        except asyncio.TimeoutError:
            logger.error("[core] Handshake-Timeout (10s)")
            return False
        except Exception:
            logger.exception("[core] Handshake-Fehler")
            return False

    async def connect_to_core(self) -> None:
        """Persistente WebSocket-Verbindung zu aria-core (OpenClaw Gateway)."""
        retry_delay = 2
        max_conn_failures = 6
        conn_fail_count = 0

        while self.running:
            try:
                logger.info("[core] Verbinde: %s", self.ws_url)
                # max_size=50MB damit grosse Bilder/Voice-Uploads durchgehen.
                # Python-websockets Default ist nur 1 MiB → 5MB JPEG sprengt
                # das Limit, Connection wird silent gedroppt.
                async with websockets.connect(self.ws_url, max_size=50 * 1024 * 1024) as ws:
                    # OpenClaw Handshake durchfuehren
                    if not await self._openclaw_handshake(ws):
                        logger.error("[core] Handshake fehlgeschlagen — Reconnect")
                        await asyncio.sleep(retry_delay)
                        retry_delay = min(retry_delay * 2, 30)
                        continue

                    self.ws_core = ws
                    retry_delay = 2
                    conn_fail_count = 0
                    logger.info("[core] Verbunden und authentifiziert")

                    async for message in ws:
                        await self._handle_core_message(message)

            except websockets.ConnectionClosed:
                logger.warning("[core] Verbindung verloren")
                conn_fail_count += 1
            except ConnectionRefusedError:
                logger.warning("[core] Nicht erreichbar (%s)", self.ws_url)
                conn_fail_count += 1
            except Exception:
                logger.exception("[core] WebSocket-Fehler")
                conn_fail_count += 1
            finally:
                self.ws_core = None

            # Nach N aufeinanderfolgenden Fehlern: Exit damit Docker neustartet
            # (bekommt neuen Network-Namespace wenn aria-core restarted wurde)
            if conn_fail_count >= max_conn_failures:
                logger.error(
                    "[core] %dx nicht erreichbar — Exit fuer Docker-Restart",
                    conn_fail_count,
                )
                sys.exit(1)

            if self.running:
                logger.info("[core] Reconnect in %ds...", retry_delay)
                await asyncio.sleep(retry_delay)
                retry_delay = min(retry_delay * 2, 30)

    async def _handle_core_message(self, raw_message: str) -> None:
        """Verarbeitet Nachrichten von aria-core (OpenClaw Gateway Protokoll).

        Unterstuetzte Frame-Typen:
          - event: chat:delta (Streaming-Tokens), chat:final (fertige Antwort)
          - res: Antworten auf Requests (chat.send Acknowledgment)
        """
        try:
            message = json.loads(raw_message)
        except json.JSONDecodeError:
            logger.error("[core] Ungueltige JSON: %s", raw_message[:100])
            return

        frame_type = message.get("type", "")
        logger.info("[core] <<< Frame: type=%s event=%s method=%s | %s",
                     frame_type, message.get("event", "-"), message.get("method", "-"),
                     raw_message[:200])

        # ── Response auf unsere Requests (z.B. chat.send Ack) ──
        if frame_type == "res":
            req_id = message.get("id", "")
            if message.get("ok"):
                logger.debug("[core] Request %s bestätigt", req_id)
            else:
                error = message.get("error", "Unbekannt")
                logger.error("[core] Request %s fehlgeschlagen: %s", req_id, error)
            return

        # ── Events vom Gateway ──
        if frame_type != "event":
            logger.debug("[core] Unbekannter Frame-Typ: %s", frame_type)
            return

        event_name = message.get("event", "")
        payload = message.get("payload", {})

        # ── agent Events: Streaming-Deltas vom LLM ──
        if event_name == "agent":
            data = payload.get("data", {})
            delta = data.get("delta", "")
            stream = payload.get("stream", "")
            if delta and stream == "assistant":
                logger.debug("[core] Delta: '%s'", delta[:40])
            # Activity-Signal zur App (entdupliziert)
            tool_name = data.get("name") or data.get("tool") or payload.get("tool") or ""
            if stream == "tool_use" or data.get("type") == "tool_use":
                activity = "tool"
            elif stream == "assistant":
                activity = "assistant"
            else:
                activity = "thinking"
            await self._emit_activity(activity, tool_name)
            return

        # ── chat Events: Snapshots mit state=delta|final|error ──
        if event_name == "chat":
            state = payload.get("state", "")

            if state == "final":
                text = self._extract_chat_text(payload)
                self._last_chat_final_at = asyncio.get_event_loop().time()
                await self._emit_activity("idle", "")
                if not text:
                    logger.warning("[core] chat final ohne Text: %s", json.dumps(payload)[:200])
                    # App+Diagnostic informieren statt stumm — sonst wartet die
                    # UI ewig auf eine Antwort die nicht kommt. Passiert z.B.
                    # wenn Claude-Vision das Bild ablehnt (leere Antwort)
                    # oder die Antwort nur aus Tool-Calls ohne Final-Text bestand.
                    await self._send_to_rvs({
                        "type": "chat",
                        "payload": {
                            "text": "[Hinweis] Antwort ohne Text — moeglicherweise Bild zu gross fuer Vision-API oder reine Tool-Ausfuehrung.",
                            "sender": "aria",
                        },
                        "timestamp": int(asyncio.get_event_loop().time() * 1000),
                    })
                    return
                logger.info("[core] Antwort: '%s'", text[:80])
                await self._process_core_response(text, payload)
                return

            if state == "error":
                # OpenClaw nutzt errorMessage statt error bei state=error.
                error = (payload.get("error")
                         or payload.get("errorMessage")
                         or payload.get("message")
                         or "Unbekannt")
                logger.error("[core] Chat-Fehler: %s", error)
                self._last_chat_final_at = asyncio.get_event_loop().time()
                await self._emit_activity("idle", "")
                await self._send_to_rvs({
                    "type": "chat",
                    "payload": {
                        "text": f"[Fehler] {error}",
                        "sender": "aria",
                    },
                    "timestamp": int(asyncio.get_event_loop().time() * 1000),
                })
                return

            # state=delta — periodischer Snapshot, ignorieren
            return

        # ── Legacy event names (chat:delta, chat:final, chat:error) ──
        if event_name == "chat:delta":
            delta = payload.get("delta", payload.get("text", ""))
            if delta:
                logger.debug("[core] Delta (legacy): '%s'", delta[:40])
            return

        if event_name == "chat:final":
            text = payload.get("text", payload.get("message", ""))
            if not text:
                text = self._extract_chat_text(payload)
            if not text:
                logger.warning("[core] chat:final ohne Text: %s", json.dumps(payload)[:200])
                return
            logger.info("[core] Antwort (legacy): '%s'", text[:80])
            await self._process_core_response(text, payload)
            return

        if event_name == "chat:error":
            # OpenClaw legt den echten Text manchmal in errorMessage ab
            # (state=error). Vorher wurde nur error/message gechecked → "Unbekannt".
            error = (payload.get("error")
                     or payload.get("errorMessage")
                     or payload.get("message")
                     or "Unbekannt")
            logger.error("[core] Chat-Fehler (legacy): %s", error)
            await self._send_to_rvs({
                "type": "chat",
                "payload": {
                    "text": f"[Fehler] {error}",
                    "sender": "aria",
                },
                "timestamp": int(asyncio.get_event_loop().time() * 1000),
            })
            return

        # tick, health, etc. — ignorieren
        if event_name in ("tick", "health"):
            return

        logger.debug("[core] Event: %s", event_name)

    @staticmethod
    def _extract_chat_text(payload: dict) -> str:
        """Extrahiert Text aus OpenClaw chat-Event message.content Array."""
        try:
            content = payload.get("message", {}).get("content", [])
            if isinstance(content, list):
                return "".join(
                    c.get("text", "") for c in content if c.get("type") == "text"
                )
            if isinstance(content, str):
                return content
        except (AttributeError, TypeError):
            pass
        return payload.get("text", "")

    # File-Marker-Pattern: `[FILE: /pfad/zur/datei.ext]` (Pfad kann Spaces
    # enthalten, Endung beliebig). Mehrfach im Text moeglich.
    _FILE_MARKER_RE = re.compile(r"\[FILE:\s*(/shared/uploads/[^\]]+?)\s*\]", re.IGNORECASE)

    def _extract_file_markers(self, text: str) -> tuple[str, list[dict], list[str]]:
        """Sucht [FILE: /shared/uploads/...]-Marker.
        Returns (cleaned_text, valid_files, missing_paths)."""
        files: list[dict] = []
        missing: list[str] = []
        for m in self._FILE_MARKER_RE.finditer(text):
            path = m.group(1).strip()
            if not path.startswith("/shared/uploads/"):
                logger.warning("[core] FILE-Marker mit unerlaubtem Pfad ignoriert: %s", path)
                continue
            if not os.path.isfile(path):
                logger.warning("[core] FILE-Marker zeigt auf nicht existente Datei: %s", path)
                missing.append(path)
                continue
            name = os.path.basename(path)
            mime, _ = mimetypes.guess_type(path)
            size = os.path.getsize(path)
            files.append({
                "serverPath": path,
                "name": name,
                "mimeType": mime or "application/octet-stream",
                "size": size,
            })
        cleaned = self._FILE_MARKER_RE.sub("", text).strip()
        # Zwei aufeinanderfolgende Leerzeilen → eine
        cleaned = re.sub(r"\n{3,}", "\n\n", cleaned)
        return cleaned, files, missing

    def _tag_file_to_project(self, file_path: str, project_id: str) -> None:
        """Schreibt file_path → project_id in /shared/config/file_projects.json.
        Best-effort, fail-silent. project_id leer = Eintrag entfernen (Hauptchat)."""
        try:
            manifest_path = "/shared/config/file_projects.json"
            os.makedirs("/shared/config", exist_ok=True)
            try:
                with open(manifest_path) as f:
                    manifest = json.load(f)
                if not isinstance(manifest, dict):
                    manifest = {}
            except FileNotFoundError:
                manifest = {}
            except Exception:
                manifest = {}
            if project_id:
                manifest[file_path] = project_id
            else:
                manifest.pop(file_path, None)
            tmp = manifest_path + ".tmp"
            with open(tmp, "w") as f:
                json.dump(manifest, f, indent=2, ensure_ascii=False)
            os.replace(tmp, manifest_path)
            logger.info("[file-project] %s → %s", file_path, project_id or "(main)")
        except Exception as exc:
            logger.warning("[file-project] tag failed (%s): %s", file_path, exc)

    def _tag_file_to_active_project(self, file_path: str) -> None:
        """Convenience: Brain nach aktivem Projekt fragen + taggen.
        Wird vom App-Upload-Handler genutzt (dort wissen wir die Projekt-ID
        noch nicht aus dem Payload — Stefan kann ja zwischen App-Upload und
        Chat-Send das Projekt gewechselt haben). ARIA-eigene Dateien gehen
        ueber _tag_file_to_project mit turn_project_id direkt."""
        try:
            brain_url = os.environ.get("BRAIN_URL", "http://aria-brain:8080")
            with urllib.request.urlopen(f"{brain_url}/projects/status", timeout=5) as r:
                data = json.loads(r.read())
            active_id = (data.get("active_id") or "").strip()
            if not active_id:
                return
            self._tag_file_to_project(file_path, active_id)
        except Exception as exc:
            logger.warning("[file-project] active-query failed (%s): %s", file_path, exc)

    async def _broadcast_aria_file(self, file_info: dict) -> None:
        """ARIA hat eine Datei fuer den User erstellt — App+Diagnostic informieren."""
        logger.info("[rvs] ARIA-Datei rausgeben: %s (%s, %dKB)",
                    file_info["name"], file_info["mimeType"], file_info["size"] // 1024)
        try:
            await self._send_to_rvs({
                "type": "file_from_aria",
                "payload": file_info,
                "timestamp": int(asyncio.get_event_loop().time() * 1000),
            })
        except Exception as e:
            logger.warning("[rvs] file_from_aria broadcast fehlgeschlagen: %s", e)

    def _persist_state(self, key: str, data: dict) -> None:
        """Atomic-Write in /shared/state/<key>.json — fuer Brain-Watcher.
        Wird genutzt fuer location + activity-Tracking."""
        try:
            import time as _time
            data = dict(data)
            data["ts_unix"] = int(_time.time())
            Path("/shared/state").mkdir(parents=True, exist_ok=True)
            target = Path(f"/shared/state/{key}.json")
            tmp = target.with_suffix(".tmp")
            tmp.write_text(json.dumps(data), encoding="utf-8")
            tmp.replace(target)
        except Exception as e:
            logger.warning("[state] %s schreiben fehlgeschlagen: %s", key, e)

    def _persist_location(self, location: Optional[dict]) -> None:
        """Speichert die letzte bekannte GPS-Position fuer Watcher.
        Erwartet {lat, lon} oder {lat, lng}. Nicht-Dicts und fehlende
        Koordinaten werden ignoriert.

        Plus: triggert sofort einen on-demand Trigger-Check im Brain
        (POST /triggers/check-now). Ohne das wartet der Watcher-Loop
        bis zu TICK_SEC Sekunden — bei Auto-Vorbeifahrt durch einen
        300m-Radius (18-43s drin) kann das den Trigger verpassen."""
        if not isinstance(location, dict):
            return
        try:
            lat = location.get("lat")
            lon = location.get("lon") or location.get("lng")
            if lat is None or lon is None:
                return
            self._persist_state("location", {
                "lat": float(lat),
                "lon": float(lon),
            })
        except Exception:
            return
        # Fire-and-forget: Brain-on-demand-Tick. Wenn Brain nicht antwortet
        # oder langsam ist, blockt das nicht den GPS-Pfad.
        try:
            asyncio.create_task(self._trigger_brain_check_now())
        except Exception:
            pass

    async def _trigger_brain_check_now(self) -> None:
        """Brain-Endpoint POST /triggers/check-now anstossen."""
        brain_url = os.environ.get("BRAIN_URL", "http://aria-brain:8080")
        def _post():
            try:
                req = urllib.request.Request(
                    f"{brain_url}/triggers/check-now",
                    data=b"", method="POST",
                    headers={"Content-Type": "application/json"},
                )
                with urllib.request.urlopen(req, timeout=8) as r:
                    return r.status
            except Exception:
                return None
        await asyncio.get_event_loop().run_in_executor(None, _post)

    def _persist_user_activity(self) -> None:
        """Markiert dass der User gerade etwas gemacht hat (Chat/Voice).
        Watcher: last_user_message_ago_sec basiert darauf."""
        self._persist_state("activity", {"last_user_ts": int(time.time())})

    def _append_chat_backup(self, entry: dict) -> int:
        """Schreibt eine Zeile in /shared/config/chat_backup.jsonl.
        Wird von Diagnostic + App als History-Quelle gelesen.
        entry braucht mindestens {role, text}; ts wird ergaenzt.
        Returns den ts (auch fuer Bubble-Loeschen-Tracking).

        WICHTIG: ts ist UNIX-ms (time.time()*1000), NICHT loop-time.
        Loop-time ist Container-monotonic — bei jedem Restart wieder 0.
        Das brach die App-History-Sortierung weil App-side Date.now()
        (echtes UNIX-ms) mit Bridge-Container-Uptime gemischt wurde."""
        ts = int(time.time() * 1000)
        try:
            line = {"ts": ts}
            line.update(entry)
            Path("/shared/config").mkdir(parents=True, exist_ok=True)
            with open("/shared/config/chat_backup.jsonl", "a", encoding="utf-8") as f:
                f.write(json.dumps(line, ensure_ascii=False) + "\n")
        except Exception as e:
            logger.warning("[backup] chat_backup-Write fehlgeschlagen: %s", e)
        return ts

    def _read_chat_backup_since(self, since_ms: int, limit: int = 100) -> list[dict]:
        """Liest chat_backup.jsonl, gibt Eintraege > since_ms zurueck, max limit neueste.
        File-deleted-Marker werden honoriert: vor einem file_deleted-Marker liegende
        Eintraege mit gleichem Pfad werden als deleted markiert."""
        path = Path("/shared/config/chat_backup.jsonl")
        if not path.exists():
            return []
        try:
            lines = path.read_text(encoding="utf-8").splitlines()
        except Exception as e:
            logger.warning("[backup] Lesen fehlgeschlagen: %s", e)
            return []
        out: list[dict] = []
        for raw in lines:
            raw = raw.strip()
            if not raw:
                continue
            try:
                obj = json.loads(raw)
            except Exception:
                continue
            ts = obj.get("ts") or 0
            if ts <= since_ms:
                continue
            # file_deleted-Marker: nicht als Chat ausliefern, aber an die App schicken
            # damit sie ihre Bubbles updaten kann (separater Pfad existiert ja schon)
            if obj.get("type") == "file_deleted":
                continue
            role = obj.get("role")
            if role not in ("user", "assistant"):
                continue
            out.append(obj)
        # Auf "limit" neueste cappen
        if len(out) > limit:
            out = out[-limit:]
        return out

    async def _process_core_response(self, text: str, payload: dict) -> None:
        """Verarbeitet eine fertige Antwort von aria-core.

        - Leitet Antwort an die App weiter (via RVS)
        - Sprachausgabe ueber TTS (wenn Modus erlaubt)
        """
        # NO_REPLY Token: ARIA signalisiert explizit "nicht antworten"
        # → komplett verwerfen (keine Chat-Nachricht, kein TTS)
        # Toleranz fuer Variationen: "NO_REPLY", "no_reply", mit Punkt/Anfuehrungszeichen
        stripped = text.strip().strip('."\'`*').upper()
        if stripped == "NO_REPLY" or stripped.startswith("NO_REPLY"):
            logger.info("[core] NO_REPLY empfangen — Antwort still verworfen")
            return

        # Antwort in chat_backup.jsonl loggen (cleaned text, ohne File-Marker)
        # — passiert weiter unten nach extract_file_markers

        # File-Marker `[FILE: /shared/uploads/aria_xyz.pdf]` extrahieren —
        # ARIA legt damit Dateien fuer den User bereit (Bilder, PDFs, etc.).
        # Der Marker wird aus dem Antworttext entfernt (TTS soll ihn nicht
        # vorlesen) und parallel als file_from_aria-Event geschickt.
        text, aria_files, missing_files = self._extract_file_markers(text)
        # ARIA-Dateien dem aktiven Projekt zuordnen (falls eines aktiv war).
        # turn_project_id kommt vom Brain mit dem /chat-Response und reflektiert
        # den Stand NACH dem Turn — passt fuer Dateien die ARIA waehrend des
        # Turns geschrieben hat (sie sind „im selben Projekt entstanden").
        turn_pid = (payload.get("projectId") or "").strip() if isinstance(payload, dict) else ""
        for f in aria_files:
            server_path = f.get("serverPath")
            if turn_pid and server_path:
                self._tag_file_to_project(server_path, turn_pid)
            # projectId mitschicken, damit App+Diagnostic die Datei-Bubble dem
            # richtigen Kontext zuordnen (sonst faellt sie im Diagnostic-Focus-
            # Filter durch = wird nur im Hauptchat angezeigt).
            f["projectId"] = turn_pid
            await self._broadcast_aria_file(f)
        # Bei fehlenden Files: User informieren (sonst sieht er nur stille
        # Verluste — ARIA hat den Marker hingeschrieben aber das File nicht
        # tatsaechlich angelegt).
        if missing_files:
            missing_list = "\n".join(f"  • {os.path.basename(p)}" for p in missing_files)
            text = (text + "\n\n[Hinweis] Folgende Dateien hat ARIA zwar erwaehnt "
                    f"aber nicht erstellt:\n{missing_list}\n"
                    "Bitte ARIA bitten, sie wirklich zu schreiben.").strip()

        # Antwort in chat_backup.jsonl loggen (gecleanter Text, ohne File-Marker
        # UND ohne <voice>-Tag — der ist eine TTS-Annotation, gehoert nicht in
        # die Chat-Historie weil ARIA ihre eigene Vorgaenger-Antwort sonst mit
        # Voice-Tag-Noise als Kontext sieht).
        # File-Marker werden separat als file_from_aria-Events ausgeliefert.
        display_text = strip_voice_tag_for_display(text)
        assistant_backup_ts = self._append_chat_backup({
            "role": "assistant",
            "text": display_text,
            "files": [{"serverPath": f["serverPath"], "name": f["name"],
                       "mimeType": f["mimeType"], "size": f["size"]} for f in aria_files],
            "project_id": turn_pid,
        })

        metadata = payload.get("metadata", {})
        is_critical = metadata.get("critical", False)
        requested_voice = metadata.get("voice")

        # Modus-Wechsel pruefen (Sprachbefehl im Text)
        new_mode = detect_mode_switch(text)
        if new_mode is not None:
            self.current_mode = new_mode
            self._persist_mode()
            logger.info(
                "[core] Modus → %s %s",
                self.current_mode.config.emoji,
                self.current_mode.config.name,
            )
            await self._broadcast_current_mode()

        # Eindeutige Message-ID fuer Audio-Cache-Zuordnung
        message_id = str(uuid.uuid4())

        # TTS-aufbereitete Variante fuer Debug (Diagnostic zeigt optional)
        tts_text_preview = clean_text_for_tts(text)

        # Antwort an die App weiterleiten (als Chat-Nachricht).
        # display_text == text aber ohne <voice>-Tag — der lebt nur transient
        # in `text` damit clean_text_for_tts weiter unten daraus die TTS-
        # Variante zieht. Im Chat-Bubble soll der Tag nicht erscheinen.
        await self._send_to_rvs({
            "type": "chat",
            "payload": {
                "text": display_text,
                "sender": "aria",
                "messageId": message_id,
                # backupTs = der ts in chat_backup.jsonl. Wird von Clients als
                # Bubble-ID fuer das Mülltonne-Loeschen verwendet (delete_message_request).
                "backupTs": assistant_backup_ts,
                # Debug: aufbereiteter Text fuer TTS (App ignoriert, Diagnostic zeigt optional)
                "ttsText": tts_text_preview if tts_text_preview != text else "",
                # Projekt-Zuordnung — App + Diagnostic sortieren die Bubble in
                # den passenden Projekt-Block. Leer = Hauptchat.
                "projectId": (payload.get("projectId") or "") if isinstance(payload, dict) else "",
            },
            "timestamp": int(asyncio.get_event_loop().time() * 1000),
        })

        # TTS ueber XTTS (XTTS-Bridge auf Gaming-PC)
        if not (getattr(self, 'tts_enabled', True) and should_speak(self.current_mode, is_critical)):
            logger.info("[core] TTS unterdrueckt (Modus: %s)", self.current_mode.config.name)
            return

        # Voice bestimmen: App-Override (gesetzt durch letzten chat-Event) > globale
        # Default-Voice. Der Override wird NICHT pro Antwort verbraucht — sonst nutzt
        # eine Multi-Turn-Antwort von ARIA (Tool-Use + finale Antwort) ab dem zweiten
        # TTS-Call wieder die alte Default-Stimme. Der Override bleibt gueltig bis
        # zum naechsten chat-Event, wo er entweder ueberschrieben oder geloescht wird.
        xtts_voice = self._next_voice_override or getattr(self, 'xtts_voice', '')
        # Speed-Reihenfolge: Per-Request-Override (App schickte gerade) >
        # persistierter App-Setting (voice_config.json xttsSpeed) > 1.0 default.
        # Damit greift die App-Speed auch bei Diagnostic-Chats / Trigger-
        # Replies / Bridge-Restart, ohne dass die App vorher noch mal getippt
        # haben muss.
        xtts_speed = (
            self._next_speed_override
            or getattr(self, "_persistent_xtts_speed", None)
            or 1.0
        )

        # Ein vorhandener <voice>-Tag ist die EXPLIZITE TTS-Vorgabe von ARIA —
        # auch wenn er leer ist. Leeres <voice></voice> = bewusst stumm: die
        # Bubble erscheint im Chat, aber es wird NICHTS gesprochen. Use-Case:
        # Steuerbefehl-Quittungen (Spotify next/pause, Licht an) — dort darf die
        # Sprachausgabe NICHT feuern, weil das TTS-Playback auf dem Handy den
        # Audio-Fokus klaut, Spotify duckt/pausiert und (bei kurzem Text) nicht
        # sauber weiterspielt. Nur OHNE Voice-Tag faellt es auf den vollen Text
        # zurueck (normale Antwort ohne TTS-Annotation wird komplett gelesen).
        has_voice_tag = "<voice>" in (text or "").lower()
        tts_text = tts_text_preview if has_voice_tag else (tts_text_preview or text)
        if not tts_text:
            logger.info("[core] TTS-Text leer (bewusst stumm via <voice></voice> oder nach Cleanup) — uebersprungen")
            return
        try:
            xtts_request_id = str(uuid.uuid4())
            self._xtts_request_to_message[xtts_request_id] = message_id
            if len(self._xtts_request_to_message) > 100:
                oldest = next(iter(self._xtts_request_to_message))
                self._xtts_request_to_message.pop(oldest, None)
            await self._send_to_rvs({
                "type": "xtts_request",
                "payload": {
                    "text": tts_text,
                    "voice": xtts_voice,
                    "speed": xtts_speed,
                    "language": "de",
                    "requestId": xtts_request_id,
                    "messageId": message_id,
                },
                "timestamp": int(asyncio.get_event_loop().time() * 1000),
            })
            logger.info("[core] XTTS-Request gesendet (voice=%s, speed=%.2fx): '%s'",
                        xtts_voice or "default", xtts_speed, tts_text[:60])
        except Exception as e:
            logger.error("[core] XTTS-Request fehlgeschlagen: %s — kein Audio", e)

        # ARIA ist fertig — App's "ARIA denkt..." Indicator zurueck auf idle.
        # _last_chat_final_at bewusst NICHT setzen: die 3s-Cooldown war fuer
        # trailing OpenClaw-Activity-Events; bei Voice-Chat wuerde sie die
        # naechste thinking-Welle unterdruecken.
        await self._emit_activity("idle", "", project_id=turn_pid)

    # ── Mode Persistence (global, nicht pro Geraet) ──────
    _MODE_FILE = "/shared/config/mode.json"

    def _load_persisted_mode(self) -> Mode:
        """Laedt den zuletzt aktiven Modus aus Shared Config oder NORMAL."""
        try:
            if os.path.exists(self._MODE_FILE):
                data = json.loads(Path(self._MODE_FILE).read_text())
                mode_name = data.get("mode", "NORMAL")
                for m in Mode:
                    if m.name == mode_name:
                        logger.info("[mode] Persistierter Modus geladen: %s", m.config.name)
                        return m
        except Exception as e:
            logger.warning("[mode] Laden fehlgeschlagen: %s", e)
        return Mode.NORMAL

    def _persist_mode(self) -> None:
        """Speichert den aktuellen Modus in Shared Config."""
        try:
            os.makedirs("/shared/config", exist_ok=True)
            Path(self._MODE_FILE).write_text(json.dumps({"mode": self.current_mode.name}))
        except Exception as e:
            logger.warning("[mode] Speichern fehlgeschlagen: %s", e)

    async def _broadcast_current_mode(self) -> None:
        """Broadcastet den aktuellen Modus an alle RVS-Clients (App + Diagnostic)."""
        try:
            await self._send_to_rvs({
                "type": "mode",
                "payload": {
                    "mode": canonical_id(self.current_mode),
                    "name": self.current_mode.config.name,
                    "emoji": self.current_mode.config.emoji,
                    "sender": "bridge",  # Filter in mode-Handler gegen Loops
                },
                "timestamp": int(asyncio.get_event_loop().time() * 1000),
            })
        except Exception as e:
            logger.debug("[mode] Broadcast fehlgeschlagen: %s", e)

    async def _broadcast_persisted_config(self) -> None:
        """Broadcastet die aktuelle voice_config.json einmalig nach RVS-Connect.

        Damit bekommen frisch verbundene Bridges (insbesondere die f5tts-bridge
        auf der Gamebox nach Container-Restart) die zuletzt in Diagnostic
        gewaehlten Settings — ohne dass der User in Diagnostic was klicken muss.
        """
        try:
            payload = {
                "ttsEnabled": getattr(self, "tts_enabled", True),
                "xttsVoice": getattr(self, "xtts_voice", ""),
                "whisperModel": self.stt_engine.model_size,
            }
            if getattr(self, "_persistent_xtts_speed", None) is not None:
                payload["xttsSpeed"] = self._persistent_xtts_speed
            payload.update(getattr(self, "_f5tts_config", {}) or {})
            payload.update(getattr(self, "_flux_config", {}) or {})
            payload.update(getattr(self, "_debug_log_config", {}) or {})
            await self._send_to_rvs({
                "type": "config",
                "payload": payload,
                "timestamp": int(asyncio.get_event_loop().time() * 1000),
            })
            logger.info("[rvs] Persistierte Config broadcastet: %s", payload)
        except Exception as e:
            logger.debug("[rvs] Config-Broadcast fehlgeschlagen: %s", e)

    async def _persist_speed_change(self, speed: float) -> None:
        """Schreibt nur den xttsSpeed-Eintrag in voice_config.json — der
        Rest bleibt unangetastet. Wird gerufen wenn App per chat-Event
        einen neuen Speed mitschickt (kein config-Broadcast)."""
        try:
            path = "/shared/config/voice_config.json"
            data: dict = {}
            if os.path.exists(path):
                with open(path) as f:
                    data = json.load(f) or {}
            data["xttsSpeed"] = speed
            os.makedirs("/shared/config", exist_ok=True)
            with open(path, "w") as f:
                json.dump(data, f, indent=2)
            logger.info("[speed] Persistiert: %.2fx", speed)
        except Exception as exc:
            logger.warning("[speed] Persistierung fehlgeschlagen: %s", exc)

    def _fetch_active_session(self) -> None:
        """Holt die aktive Session vom Diagnostic-Endpoint."""
        try:
            req = urllib.request.Request(f"{self._diagnostic_url}/api/session", method="GET")
            with urllib.request.urlopen(req, timeout=2) as resp:
                data = json.loads(resp.read().decode())
                new_key = data.get("sessionKey", "")
                if new_key and new_key != self._session_key:
                    logger.info("[session] Aktive Session gewechselt: %s -> %s", self._session_key, new_key)
                    self._session_key = new_key
        except Exception as e:
            logger.debug("[session] Diagnostic nicht erreichbar (%s) — nutze '%s'", e, self._session_key)

    def _build_core_text(self, text: str, interrupted: bool = False,
                          location: Optional[dict] = None) -> str:
        """Baut den Text fuer aria-core mit allen relevanten Hints (Barge-In,
        GPS-Position). Hints sind in eckigen Klammern, der eigentliche User-
        Text folgt unverandert."""
        parts: list[str] = []
        if interrupted:
            parts.append(
                "[Hinweis: Stefan hat dich gerade unterbrochen waehrend du noch "
                "gesprochen oder gearbeitet hast. Folgendes ist eine Korrektur, "
                "Ergaenzung oder ein Themenwechsel zu deiner letzten Antwort.]"
            )
        if location and isinstance(location, dict):
            lat = location.get("lat")
            lon = location.get("lon") or location.get("lng")
            if lat is not None and lon is not None:
                parts.append(
                    f"[Stefans aktuelle GPS-Position: {float(lat):.6f}, {float(lon):.6f}. "
                    f"Nutze die nur wenn die Frage sich auf seinen Standort bezieht. "
                    f"Erwaehne sie nicht von dir aus, ausser er fragt explizit danach.]"
                )
        if parts:
            return " ".join(parts) + " " + text
        return text

    def _build_pending_files_message(self, user_text: str) -> str:
        """Baut eine Anweisung an aria-core aus den gepufferten Files + optionalem
        User-Text. user_text leer → 'warte auf Anweisung'-Variante."""
        parts: list[str] = []
        for fp, name, ftype, kb, w, h in self._pending_files:
            dim = f" {w}x{h}px" if (w and h) else ""
            kind = "Bild" if ftype.startswith("image/") else "Datei"
            parts.append(f"- {kind}: {name}{dim} ({ftype}, {kb}KB) liegt unter {fp}")
        files_summary = "\n".join(parts)
        n = len(self._pending_files)
        anhang = "Anhang" if n == 1 else "Anhaenge"
        if user_text:
            return (f"Stefan hat dir {n} {anhang} geschickt:\n{files_summary}\n\n"
                    f"Er sagt dazu: \"{user_text}\"")
        return (f"Stefan hat dir {n} {anhang} geschickt:\n{files_summary}\n\n"
                f"Warte auf seine Anweisung was du damit tun sollst.")

    async def _flush_pending_files_after(self, delay: float) -> None:
        """Wenn nach `delay`s kein chat-Text gekommen ist: Files alleine an
        aria-core senden ('warte auf Anweisung'-Variante)."""
        try:
            await asyncio.sleep(delay)
        except asyncio.CancelledError:
            return
        if not self._pending_files:
            return
        text = self._build_pending_files_message("")
        self._pending_files = []
        self._pending_files_flush_task = None
        pid = self._pending_files_project_id
        self._pending_files_project_id = ""
        await self.send_to_core(text, source="app-file", project_id=pid)

    async def _flush_pending_files_with_text(self, user_text: str,
                                             project_id: str = "") -> bool:
        """Wenn ein chat-Text reinkommt waehrend Files gepuffert sind:
        Files + Text zu einer einzigen aria-core-Nachricht mergen.
        Returns True wenn gemerged wurde (Caller soll dann nicht nochmal senden).

        project_id: Projekt-Kontext aus dem chat-Payload (der sichtbare Focus
        beim Absenden). Faellt auf den beim File-Upload gemerkten Kontext
        zurueck, damit Anhaenge im richtigen Projekt landen statt im Hauptchat."""
        if not self._pending_files:
            return False
        if self._pending_files_flush_task and not self._pending_files_flush_task.done():
            self._pending_files_flush_task.cancel()
        self._pending_files_flush_task = None
        text = self._build_pending_files_message(user_text)
        self._pending_files = []
        pid = (project_id or "").strip() or self._pending_files_project_id
        self._pending_files_project_id = ""
        # create_task statt await — sonst blockt der RVS-recv-Loop bis Brain
        # fertig ist (siehe chat-handler oben).
        asyncio.create_task(self.send_to_core(text, source="app-file+chat", project_id=pid))
        return True

    async def send_to_core(self, text: str, source: str = "bridge",
                           client_msg_id: Optional[str] = None,
                           project_id: str = "") -> None:
        """Sendet Text an aria-brain (HTTP /chat) und broadcastet die Antwort.

        Nicht-Streaming: wir warten bis Brain fertig ist, dann pushen wir
        die komplette Reply via RVS an alle Clients (App + Diagnostic).
        TTS wird vom Bridge-Code separat angestossen (gleiche Logik wie
        vorher mit aria-core).
        """
        brain_url = os.environ.get("BRAIN_URL", "http://aria-brain:8080")
        url = f"{brain_url}/chat"
        # project_id kommt jetzt IM /chat-Body an das Brain (Multi-Threading:
        # per-Request-Routing statt globaler active_project-State).
        payload = json.dumps({
            "message": text, "source": source,
            "project_id": project_id or "",
        }).encode("utf-8")
        logger.info("[brain] chat ← %s '%s' project=%s", source, text[:80], project_id or "(main)")

        # User-Nachricht in chat_backup.jsonl loggen — wird beim App-Reconnect
        # / Diagnostic-Reload als History-Quelle gelesen. clientMsgId speichern
        # damit die App beim chat_history_response ihre lokale Bubble
        # dedupen kann (sonst verschwindet sie nach Offline→Online-Race).
        entry: dict = {"role": "user", "text": text, "source": source}
        if client_msg_id:
            entry["clientMsgId"] = client_msg_id
        if project_id:
            entry["project_id"] = project_id
        self._append_chat_backup(entry)

        # agent_activity → thinking. _emit_activity statt direktem _send_to_rvs
        # damit der State-Cache fuer die spaetere idle-Dedup richtig steht.
        await self._emit_activity("thinking", "", project_id=project_id)

        def _do_call():
            try:
                req = urllib.request.Request(
                    url, data=payload, method="POST",
                    headers={"Content-Type": "application/json"},
                )
                # 20 Min Timeout — lange Multi-Tool-Workflows (Karten,
                # PDFs, viele curl-Calls) brauchen das. 5 Min waren chronisch
                # zu knapp und haben ARIA mitten in der Arbeit gekappt.
                with urllib.request.urlopen(req, timeout=1200) as resp:
                    return resp.status, resp.read().decode("utf-8", errors="ignore")
            except Exception as exc:
                return None, str(exc)

        status, body = await asyncio.get_event_loop().run_in_executor(None, _do_call)
        if status != 200:
            logger.error("[brain] /chat fehlgeschlagen: status=%s body=%s", status, body[:200])
            await self._emit_activity("idle", "", project_id=project_id)
            await self._send_to_rvs({
                "type": "chat",
                "payload": {
                    "text": f"[Brain-Fehler] {body[:200] or 'unbekannt'}",
                    "sender": "aria",
                },
                "timestamp": int(asyncio.get_event_loop().time() * 1000),
            })
            return

        try:
            data = json.loads(body)
        except Exception:
            logger.error("[brain] /chat lieferte ungueltiges JSON: %s", body[:200])
            await self._emit_activity("idle", "", project_id=project_id)
            return

        reply = (data.get("reply") or "").strip()
        if not reply:
            logger.warning("[brain] /chat: leerer Reply")
            await self._emit_activity("idle", "", project_id=project_id)
            return

        # Projekt-Kontext des Turns — wird an _process_core_response weiter-
        # gegeben damit der chat-Broadcast die Bubble dem richtigen Projekt-
        # Block in App + Diagnostic zuordnen kann.
        turn_project_id = (data.get("project_id") or "").strip()

        # Side-Channel-Events VOR der Chat-Bubble broadcasten (z.B. skill_created)
        # damit sie in der UI vor der Reply auftauchen
        for event in data.get("events", []) or []:
            etype = event.get("type")
            if etype == "skill_created":
                await self._send_to_rvs({
                    "type": "skill_created",
                    "payload": event.get("skill", {}),
                    "timestamp": int(asyncio.get_event_loop().time() * 1000),
                })
                logger.info("[brain] ARIA hat einen Skill erstellt: %s",
                            event.get("skill", {}).get("name"))
            elif etype == "trigger_created":
                await self._send_to_rvs({
                    "type": "trigger_created",
                    "payload": event.get("trigger", {}),
                    "timestamp": int(asyncio.get_event_loop().time() * 1000),
                })
                logger.info("[brain] ARIA hat einen Trigger angelegt: %s",
                            event.get("trigger", {}).get("name"))
            elif etype == "location_tracking":
                # ARIA bittet die App das GPS-Tracking ein-/auszuschalten
                await self._send_to_rvs({
                    "type": "location_tracking",
                    "payload": {
                        "on": bool(event.get("on")),
                        "reason": event.get("reason") or "",
                    },
                    "timestamp": int(asyncio.get_event_loop().time() * 1000),
                })
                logger.info("[brain] location_tracking Request: on=%s (%s)",
                            event.get("on"), event.get("reason", ""))
            elif etype == "memory_saved":
                # ARIA hat selber etwas in die Vector-DB gespeichert.
                # Eigene Bubble in App + Diagnostic (gelb wie skill/trigger).
                await self._send_to_rvs({
                    "type": "memory_saved",
                    "payload": event.get("memory", {}),
                    "timestamp": int(asyncio.get_event_loop().time() * 1000),
                })
                logger.info("[brain] ARIA hat eine Memory angelegt: %s (type=%s)",
                            event.get("memory", {}).get("title"),
                            event.get("memory", {}).get("type"))
            elif etype == "project_changed":
                # ARIA hat ein Projekt erstellt / betreten / verlassen / beendet.
                # App + Diagnostic refreshen ihren Projekt-Banner anhand des Events.
                await self._send_to_rvs({
                    "type": "project_changed",
                    "payload": {
                        "action": event.get("action") or "",
                        **(event.get("project") or {}),
                    },
                    "timestamp": int(asyncio.get_event_loop().time() * 1000),
                })
                proj = event.get("project") or {}
                logger.info("[brain] Projekt %s: %s (id=%s)",
                            event.get("action") or "?", proj.get("name"), proj.get("id"))

        # _process_core_response uebernimmt alles weitere:
        # File-Marker extrahieren + broadcasten, NO_REPLY-Check, Chat-
        # Broadcast an RVS, TTS, agent_activity idle. Wir geben das
        # raw payload mit dem reply rein damit Mode/voice-Metadata
        # passend behandelt wird (hier minimal, weil Brain noch keine
        # metadata mitschickt).
        try:
            await self._process_core_response(reply, {"projectId": turn_project_id})
        except Exception:
            logger.exception("[brain] _process_core_response Fehler")
            await self._emit_activity("idle", "", project_id=project_id)
            # Originaler Fallback-Send (toter Code, _emit_activity uebernimmt jetzt)
            await self._send_to_rvs({
                "type": "agent_activity",
                "payload": {"activity": "idle"},
                "timestamp": int(asyncio.get_event_loop().time() * 1000),
            })

        if data.get("distilling"):
            logger.info("[brain] Destillat laeuft im Hintergrund")

    # ── RVS Verbindung (App-Relay) ──────────────────────────

    async def connect_to_rvs(self) -> None:
        """Persistente WebSocket-Verbindung zum RVS mit Auto-Reconnect.

        Bei TLS-Fehler wird automatisch auf ws:// gefallbackt
        (wenn RVS_TLS_FALLBACK=true).
        """
        if not self.rvs_url or not self.rvs_token:
            logger.info("[rvs] Nicht konfiguriert — ueberspringe")
            return

        retry_delay = 2
        current_url = self.rvs_url
        using_fallback = False

        while self.running:
            try:
                url = f"{current_url}?token={self.rvs_token}"
                logger.info("[rvs] Verbinde: %s", current_url)
                # max_size=1500MB synchron zum RVS-Server (siehe rvs/server.js).
                # File-Re-Download fuer Anhaenge braucht Platz fuer base64-
                # inflate (~1.33×) — 1 GB binaer ≈ 1.34 GB base64, plus Margin.
                # Groessere Files lehnt der file_request-Handler proaktiv ab
                # bevor's zur 1009-Disconnection kommt.
                async with websockets.connect(url, max_size=1500 * 1024 * 1024) as ws:
                    self.ws_rvs = ws
                    retry_delay = 2
                    logger.info("[rvs] Verbunden — warte auf App-Nachrichten")

                    # TCP-Keepalive auf dem unterliegenden Socket aktivieren —
                    # damit NAT-Tabellen-Verfall oder "halb-tote" Verbindungen
                    # (kein RST, kein FIN) innerhalb von ~1 Minute erkannt
                    # werden statt nach Linux-Default (2h idle). Ohne das
                    # hat die Bridge schon mal 5+h auf einer toten Connection
                    # gehangen ohne dass irgendeine Exception kam.
                    _enable_tcp_keepalive(ws)

                    # Heartbeat-Watchdog: jeden erfolgreichen Ping markieren wir
                    # in _last_heartbeat_ok. Ein separater Watchdog killt die
                    # WS-Verbindung wenn diese Marke > 60s stale ist — schuetzt
                    # gegen den Fall dass ws.ping() selbst nie zurueckkommt.
                    self._last_heartbeat_ok = time.monotonic()

                    # Aktuellen Modus broadcasten damit gerade verbundene Apps/Diagnostic
                    # ihren UI-State sofort syncen koennen
                    await self._broadcast_current_mode()

                    # Persistierte Voice-Config broadcasten — die f5tts-bridge auf
                    # der Gamebox bekommt damit nach Restart die zuletzt in
                    # Diagnostic gewaehlten Settings wieder (sonst stuende sie auf
                    # ihren Hard-Defaults).
                    asyncio.create_task(self._broadcast_persisted_config())

                    # Heartbeat senden (RVS erwartet Ping alle 30s)
                    heartbeat_task = asyncio.create_task(self._rvs_heartbeat())
                    watchdog_task = asyncio.create_task(self._rvs_heartbeat_watchdog())

                    try:
                        async for raw_message in ws:
                            await self._handle_rvs_message(raw_message)
                    finally:
                        heartbeat_task.cancel()
                        watchdog_task.cancel()

            except websockets.ConnectionClosed:
                logger.warning("[rvs] Verbindung verloren")
                # Bei Reconnect wieder primary (wss://) versuchen — die
                # Bedingungen die zum Fallback gefuehrt haben sind transient
                # (z.B. Caddy noch nicht fertig mit ACME).
                if using_fallback:
                    logger.info("[rvs] Reset auf primary URL fuer Reconnect-Versuch")
                    current_url = self.rvs_url
                    using_fallback = False
            except ConnectionRefusedError:
                logger.warning("[rvs] Nicht erreichbar")
                if using_fallback:
                    current_url = self.rvs_url
                    using_fallback = False
            except (ssl.SSLError, OSError) as e:
                # TLS-Fehler — Fallback auf ws:// nur einmal pro Connect-Versuch,
                # bei naechstem Reconnect wieder primary probieren.
                if not using_fallback and self.rvs_url_fallback:
                    logger.warning("[rvs] TLS-Fehler: %s", e)
                    logger.warning("[rvs] TLS gewollt aber nicht verfuegbar — Fallback auf ws://")
                    current_url = self.rvs_url_fallback
                    using_fallback = True
                    retry_delay = 1  # Sofort versuchen
                else:
                    logger.error("[rvs] SSL-Fehler (kein Fallback): %s", e)
                    # Auch hier: nach gescheitertem Fallback wieder primary probieren
                    current_url = self.rvs_url
                    using_fallback = False
            except Exception as e:
                logger.exception("[rvs] WebSocket-Fehler")
                # InvalidMessage (HTTP 400 von TLS-Endpoint bei ws-Connect)
                # → wir kleben auf dem falschen Fallback, zurueck zu primary.
                if using_fallback:
                    logger.warning("[rvs] Fallback liefert auch nichts — schalte zurueck auf primary")
                    current_url = self.rvs_url
                    using_fallback = False
            finally:
                self.ws_rvs = None

            if self.running:
                logger.info("[rvs] Reconnect in %ds...", retry_delay)
                await asyncio.sleep(retry_delay)
                retry_delay = min(retry_delay * 2, 30)

    async def _rvs_heartbeat(self) -> None:
        """Sendet Heartbeats + WebSocket Pings an den RVS damit die Verbindung offen bleibt.

        Markiert nach jedem erfolgreichen Ping `_last_heartbeat_ok` —
        `_rvs_heartbeat_watchdog` schaut darauf und killt die Verbindung
        wenn die Marke stale ist (Fallback fuer den Fall dass ping() selbst
        in einer halb-toten TCP-Verbindung ewig blockt)."""
        while True:
            await asyncio.sleep(15)
            if self.ws_rvs:
                try:
                    # WebSocket Protocol-Level Ping (haelt TCP-Verbindung am Leben)
                    pong = await self.ws_rvs.ping()
                    await asyncio.wait_for(pong, timeout=10)
                    # Erfolgreicher Pong → Watchdog-Marke updaten
                    self._last_heartbeat_ok = time.monotonic()
                except Exception:
                    logger.warning("[rvs] Ping fehlgeschlagen — Verbindung tot, erzwinge Reconnect")
                    try:
                        await self.ws_rvs.close()
                    except Exception:
                        pass
                    self.ws_rvs = None
                    break
                try:
                    await self.ws_rvs.send(json.dumps({
                        "type": "heartbeat",
                        "timestamp": int(asyncio.get_event_loop().time() * 1000),
                    }))
                except Exception:
                    break

    # Heartbeat-Watchdog: wenn der letzte erfolgreiche Ping > HEARTBEAT_STALE_SEC
    # her ist (z.B. weil ws.ping() im Limbo haengt), erzwingen wir ein hartes
    # Schliessen der Verbindung. Das wirft den `async for raw_message in ws`-
    # Loop aus, der Reconnect-Loop in connect_to_rvs greift dann.
    HEARTBEAT_STALE_SEC = 60.0
    HEARTBEAT_WATCHDOG_INTERVAL_SEC = 20.0

    async def _rvs_heartbeat_watchdog(self) -> None:
        """Independent watchdog der den Heartbeat-Status ueberwacht und
        bei staleness die WS-Verbindung haert killt. Wird parallel zu
        `_rvs_heartbeat` gestartet, ist aber unabhaengig davon — auch wenn
        die heartbeat-Coroutine in einem await ewig haengen wuerde, laeuft
        diese hier weiter (eigene Coroutine, eigener await-Slot)."""
        while True:
            try:
                await asyncio.sleep(self.HEARTBEAT_WATCHDOG_INTERVAL_SEC)
            except asyncio.CancelledError:
                return
            if not self.ws_rvs:
                return
            stale = time.monotonic() - getattr(self, "_last_heartbeat_ok", time.monotonic())
            if stale > self.HEARTBEAT_STALE_SEC:
                logger.error(
                    "[rvs] Heartbeat stale (%.0fs > %.0fs) — erzwinge harten Reconnect",
                    stale, self.HEARTBEAT_STALE_SEC,
                )
                ws = self.ws_rvs
                self.ws_rvs = None
                try:
                    # close mit Reason — falls's hängt killen wir via Underlying-Transport
                    await asyncio.wait_for(ws.close(code=1011, reason="heartbeat-stale"), timeout=3.0)
                except Exception:
                    # Letzte Option: Transport direkt schliessen, das wirft den recv-Loop
                    try:
                        ws.transport.close()  # type: ignore[attr-defined]
                    except Exception:
                        pass
                return

    async def _send_chat_ack(self, client_msg_id: Optional[str]) -> None:
        """Bestaetigt der App den Empfang einer chat/audio-Nachricht.
        App nutzt das fuer Delivery-Status (✓ = sent). Ohne ACK wuerde die
        App nach Timeout retryen — gegen Verlust bei Netz-Hicksern.
        """
        if not client_msg_id:
            return
        await self._send_to_rvs({
            "type": "chat_ack",
            "payload": {"clientMsgId": client_msg_id},
            "timestamp": int(asyncio.get_event_loop().time() * 1000),
        })

    def _is_duplicate_client_msg(self, client_msg_id: Optional[str]) -> bool:
        """Prueft ob wir diese clientMsgId schon verarbeitet haben.
        Wenn ja → True (Caller soll ACK senden aber NICHT an Brain forwarden).
        Wenn nein → in den Seen-Cache aufnehmen + False zurueck.
        """
        if not client_msg_id:
            return False
        if client_msg_id in self._seen_client_msg_ids:
            logger.info("[rvs] Idempotenz: cmid=%s bereits verarbeitet, ignoriere",
                        client_msg_id)
            return True
        self._seen_client_msg_ids[client_msg_id] = time.time()
        # Capping: aelteste Eintraege rauswerfen
        while len(self._seen_client_msg_ids) > self._SEEN_CLIENT_MSG_LIMIT:
            self._seen_client_msg_ids.popitem(last=False)
        return False

    async def _handle_rvs_message(self, raw_message: str) -> None:
        """Verarbeitet Nachrichten von der App (via RVS).

        Unterstuetzte Typen:
          - chat:     Text-Nachricht → an aria-core weiterleiten
          - audio:    Audio-Daten → STT → an aria-core
          - mode:     Moduswechsel
          - location: GPS-Daten (loggen, spaeter fuer Skills)
          - file:     Datei-Upload (an aria-core weiterleiten)
        """
        try:
            message = json.loads(raw_message)
        except json.JSONDecodeError:
            logger.error("[rvs] Ungueltige JSON: %s", raw_message[:100])
            return

        msg_type = message.get("type", "")
        payload = message.get("payload", {})

        if msg_type == "chat":
            # Nur User-Nachrichten weiterleiten — ARIA/Diagnostic-Antworten ignorieren (sonst Loop!)
            sender = payload.get("sender", "")
            if sender in ("aria", "stt"):
                return
            # Delivery-ACK: immer zurueckschicken (auch bei Idempotenz-Hit),
            # damit die App den Status auf 'sent' setzen kann. Idempotenz-
            # Check VERHINDERT aber die Doppel-Weiterleitung an Brain.
            client_msg_id = payload.get("clientMsgId") or None
            await self._send_chat_ack(client_msg_id)
            if self._is_duplicate_client_msg(client_msg_id):
                return
            text = payload.get("text", "")
            # Voice-Override fuer Folgenachrichten setzen — gilt bis zum naechsten
            # chat-Event. Leerer String "" = explizit Default-Voice (override loeschen).
            # Field nicht gesendet = vorherigen Override unveraendert lassen (z.B. wenn
            # cancel_request oder anderer Service die App umgeht).
            if "voice" in payload:
                voice_override = payload.get("voice", "") or ""
                self._next_voice_override = voice_override or None
                logger.info("[rvs] Voice fuer Antworten: %s",
                            self._next_voice_override or "(Default)")
            # Speed-Override (TTS-Wiedergabegeschwindigkeit, pro Geraet) —
            # plus persistente Spiegelung damit der Wert nach Bridge-Restart
            # erhalten bleibt und Diagnostic-Chats / Trigger-Replies den
            # zuletzt von der App gesetzten Speed bekommen.
            if "speed" in payload:
                try:
                    speed = float(payload.get("speed", 0) or 0)
                    if 0.1 <= speed <= 5.0:
                        self._next_speed_override = speed
                        # Persistieren wenn der Wert sich gegenueber dem
                        # gespeicherten geaendert hat — vermeidet voice_config.json
                        # auf jeder Nachricht zu schreiben.
                        if speed != getattr(self, "_persistent_xtts_speed", None):
                            self._persistent_xtts_speed = speed
                            asyncio.create_task(self._persist_speed_change(speed))
                    else:
                        self._next_speed_override = None
                except (TypeError, ValueError):
                    self._next_speed_override = None
            if text:
                interrupted = bool(payload.get("interrupted", False))
                location = payload.get("location") or None
                # State persist fuer Brain-Watcher (current_lat, ..., last_user_ts)
                self._persist_location(location)
                self._persist_user_activity()
                # Wenn Files gerade gepuffert sind (Bild + Text gleichzeitig
                # gesendet), mergen wir sie zu einer einzigen Anfrage statt
                # zwei separater send_to_core-Calls.
                merged = await self._flush_pending_files_with_text(
                    text, project_id=str(payload.get("projectId") or ""))
                if merged:
                    logger.info("[rvs] App-Chat (mit Anhaengen) project=%s: '%s'",
                                str(payload.get("projectId") or "") or "(main)", text[:80])
                else:
                    core_text = self._build_core_text(text, interrupted, location)
                    logger.info("[rvs] App-Chat%s%s: '%s'",
                                " [BARGE-IN]" if interrupted else "",
                                " [GPS]" if location else "",
                                text[:80])
                    # KEIN await: send_to_core kann 20 Min dauern. Wenn wir
                    # hier awaiten, blockt der `async for raw_message in ws`-
                    # Loop solange → RVS-Server droppt uns nach ~4 Min idle.
                    # Als Task: Brain laeuft im Hintergrund, RVS-recv bleibt
                    # bedienbar, Pings werden beantwortet, Verbindung lebt.
                    asyncio.create_task(self.send_to_core(
                        core_text,
                        source="app" + (" [barge-in]" if interrupted else ""),
                        client_msg_id=client_msg_id,
                        project_id=str(payload.get("projectId") or ""),
                    ))
            return

        if msg_type == "cancel_request":
            hard = bool(payload.get("hard"))
            if hard:
                logger.warning("[rvs] NOT-AUS — hard cancel: Diagnostic /api/cancel + Proxy /cancel-all")
                await self._cancel_via_diagnostic()
                await self._cancel_proxy_subprocesses()
                await self._emit_activity("idle", "")
            else:
                # Barge-In: nur den fokussierten Kontext abbrechen (projectId von
                # der App), damit parallele Arbeit in anderen Kontexten weiterlaeuft.
                cancel_pid = str(payload.get("projectId") or "")
                logger.info("[rvs] Cancel-Request (kontext-scoped) project=%s", cancel_pid or "(main)")
                await self._cancel_proxy_for_project(cancel_pid)
                await self._emit_activity("idle", "", project_id=cancel_pid)
            return

        elif msg_type == "audio_pcm":
            # Audio-PCM geht direkt von XTTS-Bridge an die App.
            # Die aria-bridge darf es NICHT rebroadcasten — sonst bekommt die App
            # jeden Chunk doppelt (einmal direkt von XTTS-Bridge via RVS-Broadcast,
            # einmal indirekt via uns).
            # Wir ignorieren diese Message hier einfach — messageId wird von
            # XTTS-Bridge selbst im Payload mitgeliefert.
            return

        elif msg_type == "xtts_response":
            # Legacy-Pfad (alte XTTS-Bridge mit WAV-Response). Weiterleiten als
            # type "audio" — App nutzt den bestehenden WAV-Queue-Spieler.
            audio_b64 = payload.get("base64", "")
            error = payload.get("error", "")
            req_id_full = payload.get("requestId", "")
            req_id_base = req_id_full.rsplit("_", 1)[0] if "_" in req_id_full else req_id_full
            linked_message_id = self._xtts_request_to_message.get(req_id_base, "")
            if error:
                logger.warning("[rvs] XTTS Fehler: %s", error)
                return
            if audio_b64:
                logger.info("[rvs] XTTS-Audio legacy empfangen: %dKB", len(audio_b64) // 1365)
                await self._send_to_rvs({
                    "type": "audio",
                    "payload": {
                        "base64": audio_b64,
                        "mimeType": payload.get("mimeType", "audio/wav"),
                        "voice": payload.get("voice", "xtts"),
                        "messageId": linked_message_id,
                    },
                    "timestamp": int(asyncio.get_event_loop().time() * 1000),
                })
            return

        elif msg_type == "tts_request":
            # App fordert TTS-Audio fuer einen Text an (Play-Button) → immer XTTS.
            text = payload.get("text", "")
            message_id = payload.get("messageId", "")
            if not text:
                return
            tts_text = clean_text_for_tts(text) or text
            # Voice + Speed aus App-Payload gewinnen, sonst global/default
            xtts_voice = payload.get("voice", "") or getattr(self, 'xtts_voice', '')
            try:
                xtts_speed = float(payload.get("speed", 0) or 0)
                if not (0.1 <= xtts_speed <= 5.0):
                    xtts_speed = 1.0
            except (TypeError, ValueError):
                xtts_speed = 1.0
            try:
                xtts_request_id = str(uuid.uuid4())
                if message_id:
                    self._xtts_request_to_message[xtts_request_id] = message_id
                await self._send_to_rvs({
                    "type": "xtts_request",
                    "payload": {
                        "text": tts_text,
                        "voice": xtts_voice,
                        "speed": xtts_speed,
                        "language": "de",
                        "requestId": xtts_request_id,
                        "messageId": message_id,
                    },
                    "timestamp": int(asyncio.get_event_loop().time() * 1000),
                })
                logger.info("[rvs] TTS on-demand via XTTS: '%s'", tts_text[:60])
            except Exception as e:
                logger.warning("[rvs] TTS on-demand fehlgeschlagen: %s", e)
            return

        elif msg_type == "config":
            # Konfiguration von App/Diagnostic empfangen + persistent speichern.
            # Felder die nicht direkt zur aria-bridge gehoeren (f5tts*) werden
            # nur persistiert; die f5tts-bridge auf der Gamebox empfaengt den
            # gleichen RVS-Broadcast und reagiert selber.
            changed = False
            if "ttsEnabled" in payload:
                self.tts_enabled = bool(payload["ttsEnabled"])
                logger.info("[rvs] TTS %s", "aktiviert" if self.tts_enabled else "deaktiviert")
                changed = True
            if "xttsVoice" in payload:
                self.xtts_voice = payload["xttsVoice"]
                logger.info("[rvs] XTTS-Stimme: %s", self.xtts_voice or "default")
                changed = True
            if "xttsSpeed" in payload:
                try:
                    new_speed = float(payload["xttsSpeed"])
                    if 0.1 <= new_speed <= 5.0:
                        self._persistent_xtts_speed = new_speed
                        logger.info("[rvs] XTTS-Speed (persistent): %.2fx", new_speed)
                        changed = True
                except (TypeError, ValueError):
                    pass
            if "whisperModel" in payload:
                new_model = payload["whisperModel"]
                allowed = {"tiny", "base", "small", "medium", "large-v3"}
                if new_model in allowed and new_model != self.stt_engine.model_size:
                    logger.info("[rvs] Whisper-Modell → %s (nur Config; Modell laedt Gamebox)",
                                new_model)
                    self.stt_engine.model_size = new_model
                    self.stt_engine.model = None
                    changed = True
            # F5-TTS-Felder: einfach persistieren, f5tts-bridge applied selber.
            for k in ("f5ttsModel", "f5ttsCkptFile", "f5ttsVocabFile",
                      "f5ttsCfgStrength", "f5ttsNfeStep"):
                if k in payload:
                    if not hasattr(self, "_f5tts_config"):
                        self._f5tts_config = {}
                    self._f5tts_config[k] = payload[k]
                    changed = True
            # FLUX-Felder: gleiche Logik wie F5-TTS. flux-bridge applied
            # fluxDefaultModel selbst (Pipeline-Swap). Keywords nutzt Brain
            # via /shared/config/voice_config.json.
            for k in ("fluxDefaultModel", "fluxKeywordRaw", "fluxKeywordSwitch", "huggingfaceToken"):
                if k in payload:
                    if not hasattr(self, "_flux_config"):
                        self._flux_config = {}
                    self._flux_config[k] = payload[k]
                    changed = True
            # Debug-Log-Toggles fuer Whisper- und F5TTS-Bridge — werden via
            # naechstem config-Broadcast an die jeweiligen Bridges weitergegeben.
            # Persistent damit Toggle einen Container-Restart ueberlebt.
            for k in ("whisperDebugLog", "f5ttsDebugLog"):
                if k in payload:
                    if not hasattr(self, "_debug_log_config"):
                        self._debug_log_config = {}
                    self._debug_log_config[k] = bool(payload[k])
                    changed = True
            # Persistent speichern in Shared Volume
            if changed:
                try:
                    os.makedirs("/shared/config", exist_ok=True)
                    config_data = {
                        "ttsEnabled": getattr(self, "tts_enabled", True),
                        "xttsVoice": getattr(self, "xtts_voice", ""),
                        "whisperModel": self.stt_engine.model_size,
                    }
                    if getattr(self, "_persistent_xtts_speed", None) is not None:
                        config_data["xttsSpeed"] = self._persistent_xtts_speed
                    config_data.update(getattr(self, "_f5tts_config", {}))
                    config_data.update(getattr(self, "_flux_config", {}))
                    config_data.update(getattr(self, "_debug_log_config", {}))
                    with open("/shared/config/voice_config.json", "w") as f:
                        json.dump(config_data, f, indent=2)
                    logger.info("[rvs] Voice-Config gespeichert: %s", config_data)
                except Exception as e:
                    logger.warning("[rvs] Config speichern fehlgeschlagen: %s", e)
            return

        elif msg_type == "mode":
            # Moduswechsel von der App — ID ('normal', 'dnd', ...) ODER Aktivierungsphrase
            mode_name = payload.get("mode", "")
            # Sender kann der Broadcast der Bridge selbst sein — den ignorieren damit
            # andere Apps nicht in eine Loop geraten
            if payload.get("sender") == "bridge":
                return
            new_mode = mode_from_id(mode_name) or detect_mode_switch(mode_name)
            if new_mode is not None and new_mode != self.current_mode:
                self.current_mode = new_mode
                self._persist_mode()
                logger.info(
                    "[rvs] Modus → %s %s (von App)",
                    self.current_mode.config.emoji,
                    self.current_mode.config.name,
                )
                # Broadcast an ALLE Clients (App + Diagnostic) damit UI ueberall sync ist
                await self._broadcast_current_mode()
            elif new_mode is None:
                logger.warning("[rvs] Unbekannter Modus: '%s'", mode_name)

        elif msg_type == "location":
            # GPS-Daten von der App
            lat = payload.get("lat")
            lng = payload.get("lng")
            speed = payload.get("speed")
            logger.info("[rvs] GPS: lat=%.4f lng=%.4f speed=%s", lat or 0, lng or 0, speed)
            # An aria-core weiterleiten (fuer kontextbasierte Skills)
            if self.ws_core:
                await self.ws_core.send(raw_message)

        elif msg_type == "file":
            # Datei von der App: speichern + zu Pending-Queue hinzufuegen.
            # Wird mit dem nachfolgenden chat-Event (innerhalb PENDING_FILES_WINDOW)
            # zu einer einzigen aria-core-Anfrage gemerged. Sonst antwortet ARIA
            # zweimal: einmal "warte auf Anweisung" beim file, einmal auf den Chat.
            file_name = payload.get("name", "unbekannt")
            file_type = payload.get("type", "")
            file_b64 = payload.get("base64", "")
            width = payload.get("width", 0)
            height = payload.get("height", 0)
            # Projekt-Kontext des Uploads (sichtbarer App-Focus). Merken, damit
            # der spaetere Flush (Files+Text oder Files-only) die Anfrage im
            # richtigen Projekt an das Brain schickt statt im Hauptchat.
            file_project_id = str(payload.get("projectId") or "")
            self._pending_files_project_id = file_project_id
            logger.info("[rvs] Datei empfangen: %s (%s, %dKB) project=%s",
                        file_name, file_type, len(file_b64) // 1365 if file_b64 else 0,
                        file_project_id or "(main)")

            SHARED_DIR = "/shared/uploads"
            os.makedirs(SHARED_DIR, exist_ok=True)

            if not file_b64:
                text = f"Stefan hat eine Datei gesendet ({file_name}, {file_type}) aber die Daten sind leer angekommen."
                # create_task statt await — RVS-recv darf nicht blocken
                asyncio.create_task(self.send_to_core(text, source="app-file",
                                                      project_id=file_project_id))
                return

            if file_type.startswith("image/"):
                ext = ".jpg" if "jpeg" in file_type or "jpg" in file_type else ".png"
                safe_name = f"img_{int(asyncio.get_event_loop().time())}_{file_name.replace('/', '_')}"
                file_path = os.path.join(SHARED_DIR, safe_name if safe_name.endswith(ext) else safe_name + ext)
            else:
                safe_name = f"file_{int(asyncio.get_event_loop().time())}_{file_name.replace('/', '_')}"
                file_path = os.path.join(SHARED_DIR, safe_name)
            with open(file_path, "wb") as f:
                f.write(base64.b64decode(file_b64))
            size_kb = len(file_b64) // 1365
            logger.info("[rvs] Datei gespeichert: %s (%dKB)", file_path, size_kb)
            # Datei dem Projekt des Uploads zuordnen (Multi-Threading: explizit
            # aus dem file-Payload, kein globaler active_project-State mehr).
            # Manifest in /shared/config/file_projects.json — File-Manager
            # in App + Diagnostic filtert danach. Leer = Hauptchat.
            if file_project_id:
                self._tag_file_to_project(file_path, file_project_id)

            # Pixel-Bilder fuer Claude-Vision shrinken wenn > 2 MB. SVG/PDF/ZIP
            # bleiben unangetastet (Vision laeuft eh nur auf Raster-Formaten).
            CLAUDE_VISION_FORMATS = ("image/jpeg", "image/jpg", "image/png", "image/webp", "image/gif")
            if file_type.lower() in CLAUDE_VISION_FORMATS:
                file_size_bytes = os.path.getsize(file_path)
                if file_size_bytes > 2 * 1024 * 1024:
                    try:
                        from PIL import Image
                        with Image.open(file_path) as img:
                            orig_w, orig_h = img.size
                            # Anthropic-Empfehlung: max 1568px lange Seite. RGB-Konvertierung
                            # falls RGBA/Palette (JPEG braucht RGB).
                            img.thumbnail((1568, 1568), Image.Resampling.LANCZOS)
                            if img.mode in ("RGBA", "P"):
                                img = img.convert("RGB")
                            img.save(file_path, "JPEG", quality=85, optimize=True)
                        new_size_bytes = os.path.getsize(file_path)
                        logger.info("[rvs] Bild verkleinert: %dx%d → %dx%d, %.1fMB → %.1fMB",
                                    orig_w, orig_h, img.size[0], img.size[1],
                                    file_size_bytes / 1024 / 1024,
                                    new_size_bytes / 1024 / 1024)
                    except Exception as e:
                        logger.warning("[rvs] Bild-Resize fehlgeschlagen (%s) — Original wird genutzt: %s",
                                       file_name, e)

            # In Pending-Queue + Flush-Timer (anti-spam Buffering)
            self._pending_files.append((file_path, file_name, file_type, size_kb, int(width or 0), int(height or 0)))
            if self._pending_files_flush_task and not self._pending_files_flush_task.done():
                self._pending_files_flush_task.cancel()
            self._pending_files_flush_task = asyncio.create_task(
                self._flush_pending_files_after(self._PENDING_FILES_WINDOW_SEC)
            )

            try:
                await self._send_to_rvs({
                    "type": "file_saved",
                    "payload": {"name": file_name, "serverPath": file_path, "mimeType": file_type},
                    "timestamp": int(asyncio.get_event_loop().time() * 1000),
                })
            except Exception as e:
                logger.warning("[rvs] file_saved konnte nicht an App gesendet werden: %s", e)

        elif msg_type == "chat_history_request":
            # App holt verpasste Nachrichten beim Reconnect.
            # payload: {since: <ts_ms>}, default 0 = alles
            since = int(payload.get("since") or 0)
            limit = int(payload.get("limit") or 100)
            logger.info("[rvs] chat_history_request since=%d limit=%d", since, limit)
            messages = self._read_chat_backup_since(since, limit=limit)
            await self._send_to_rvs({
                "type": "chat_history_response",
                "payload": {"messages": messages, "since": since},
                "timestamp": int(asyncio.get_event_loop().time() * 1000),
            })
            return

        elif msg_type == "delete_message_request":
            # App oder Diagnostic loescht eine einzelne Bubble.
            # payload: {ts: <chat_backup-ts>}. Bridge entfernt aus
            # chat_backup.jsonl + Brain conversation.jsonl, broadcastet
            # danach chat_message_deleted an alle Clients.
            ts = payload.get("ts")
            if not isinstance(ts, (int, float)):
                logger.warning("[rvs] delete_message_request ohne valide ts: %r", payload)
                return
            logger.info("[rvs] delete_message_request ts=%s", ts)
            result = await self._delete_chat_message(int(ts))
            if not result.get("ok"):
                logger.warning("[rvs] delete_message fehlgeschlagen: %s", result.get("error"))
            return

        elif msg_type == "app_log":
            # App schickt Crash/Error/Info-Log via RVS — wir schreiben das
            # in /shared/logs/app.log (JSONL) damit Diagnostic + Claude
            # mitlesen koennen, auch ohne ADB-Zugriff aufs Handy.
            try:
                log_dir = Path("/shared/logs")
                log_dir.mkdir(parents=True, exist_ok=True)
                line = {
                    "ts": payload.get("ts") or int(time.time() * 1000),
                    "platform": payload.get("platform", "?"),
                    "level": payload.get("level", "info"),
                    "scope": payload.get("scope", ""),
                    "message": payload.get("message", ""),
                    "stack": payload.get("stack", ""),
                }
                with (log_dir / "app.log").open("a", encoding="utf-8") as f:
                    f.write(json.dumps(line, ensure_ascii=False) + "\n")
                logger.info("[app-log] %s %s: %s",
                            line["level"], line["scope"], line["message"][:120])
            except Exception as exc:
                logger.warning("[app-log] schreiben fehlgeschlagen: %s", exc)
            return

        elif msg_type == "brain_request":
            # Generischer RVS-Proxy fuer die Brain-HTTP-API.
            # payload: {requestId, method, path, body?, bodyBase64?, contentType?}
            #   - method: GET | POST | PATCH | DELETE
            #   - path: z.B. "/memory/list" oder "/memory/get/<id>"
            #   - body: JSON-Objekt (wird als JSON encoded)
            #   - bodyBase64: rohe Bytes als Base64 (fuer Upload mit contentType)
            #   - contentType: default application/json
            # Antwort als brain_response {requestId, status, json?, base64?}.
            req_id = payload.get("requestId") or ""
            method = (payload.get("method") or "GET").upper()
            path = payload.get("path") or ""
            if not req_id or not path or not path.startswith("/"):
                logger.warning("[rvs] brain_request ungueltig: %r", payload)
                return
            brain_url = os.environ.get("BRAIN_URL", "http://aria-brain:8080")
            url = brain_url.rstrip("/") + path
            headers: dict[str, str] = {}
            data: Optional[bytes] = None
            ct = payload.get("contentType") or "application/json"
            if payload.get("bodyBase64"):
                try:
                    data = base64.b64decode(payload["bodyBase64"])
                except Exception:
                    data = None
                if data is not None:
                    headers["Content-Type"] = ct
            elif payload.get("body") is not None:
                data = json.dumps(payload["body"]).encode("utf-8")
                headers["Content-Type"] = "application/json"
            logger.info("[rvs] brain_request %s %s (%d Byte)", method, path, len(data or b""))

            def _do_call():
                try:
                    req = urllib.request.Request(url, data=data, method=method, headers=headers)
                    with urllib.request.urlopen(req, timeout=120) as r:
                        return r.status, r.read(), r.headers.get("Content-Type", "")
                except urllib.error.HTTPError as e:
                    try:
                        body = e.read()
                    except Exception:
                        body = b""
                    return e.code, body, e.headers.get("Content-Type", "") if e.headers else ""
                except Exception as exc:
                    return None, str(exc).encode("utf-8"), "text/plain"

            status, body_bytes, response_ct = await asyncio.get_event_loop().run_in_executor(None, _do_call)
            out: dict = {"requestId": req_id, "status": status or 0}
            if response_ct and "json" in response_ct:
                try:
                    out["json"] = json.loads(body_bytes.decode("utf-8", errors="ignore"))
                except Exception:
                    out["text"] = body_bytes.decode("utf-8", errors="ignore")[:2000]
            elif response_ct and "text" in response_ct:
                out["text"] = body_bytes.decode("utf-8", errors="ignore")[:4000]
            else:
                # Binaer (z.B. attachment-download) → base64 zurueck
                out["base64"] = base64.b64encode(body_bytes).decode("ascii")
                out["contentType"] = response_ct or "application/octet-stream"
            await self._send_to_rvs({
                "type": "brain_response",
                "payload": out,
                "timestamp": int(asyncio.get_event_loop().time() * 1000),
            })
            return

        elif msg_type == "file_list_request":
            # App fragt die Liste aller /shared/uploads/-Dateien an.
            logger.info("[rvs] file_list_request von App")
            try:
                req = urllib.request.Request(
                    "http://localhost:3001/api/files-list",
                    method="GET",
                )
                def _do_list():
                    try:
                        with urllib.request.urlopen(req, timeout=10) as resp:
                            return json.loads(resp.read().decode("utf-8", errors="ignore"))
                    except Exception as e:
                        return {"ok": False, "error": str(e)}
                d = await asyncio.get_event_loop().run_in_executor(None, _do_list)
                await self._send_to_rvs({
                    "type": "file_list_response",
                    "payload": d,
                    "timestamp": int(asyncio.get_event_loop().time() * 1000),
                })
            except Exception as e:
                logger.warning("[rvs] file_list_request: %s", e)
            return

        elif msg_type == "file_delete_batch_request":
            # App will mehrere Dateien auf einmal loeschen.
            paths = payload.get("paths") or []
            req_id = payload.get("requestId", "")
            logger.warning("[rvs] file_delete_batch_request: %d Pfade", len(paths))
            try:
                body_bytes = json.dumps({"paths": paths}).encode("utf-8")
                req = urllib.request.Request(
                    "http://localhost:3001/api/files-delete-batch",
                    data=body_bytes, method="POST",
                    headers={"Content-Type": "application/json"},
                )
                def _do_delete():
                    try:
                        with urllib.request.urlopen(req, timeout=30) as resp:
                            return resp.status, resp.read().decode("utf-8", errors="ignore")
                    except Exception as e:
                        return None, str(e)
                status, body = await asyncio.get_event_loop().run_in_executor(None, _do_delete)
                logger.info("[rvs] file_delete_batch result: status=%s", status)
                # Server broadcastet file_deleted pro Pfad — App kriegt das via persistente RVS.
                # Wir bestaetigen zusaetzlich mit Counts.
                try: d = json.loads(body or "{}")
                except: d = {}
                await self._send_to_rvs({
                    "type": "file_delete_batch_response",
                    "payload": {
                        "requestId": req_id,
                        "deleted": len(d.get("deleted", [])),
                        "errors": d.get("errors", []),
                    },
                    "timestamp": int(asyncio.get_event_loop().time() * 1000),
                })
            except Exception as e:
                logger.warning("[rvs] file_delete_batch_request: %s", e)
            return

        elif msg_type == "file_zip_request":
            # App will mehrere Dateien als ZIP. Bridge holt ZIP von Diagnostic
            # via HTTP, kodiert base64 und schickt zurueck. Cap auf 30 MB
            # ZIP-Groesse damit RVS nicht erstickt.
            paths = payload.get("paths") or []
            req_id = payload.get("requestId", "")
            logger.warning("[rvs] file_zip_request: %d Pfade (req=%s)", len(paths), req_id)

            def _do_zip():
                try:
                    body_bytes = json.dumps({"paths": paths}).encode("utf-8")
                    req = urllib.request.Request(
                        "http://localhost:3001/api/files-download-zip",
                        data=body_bytes, method="POST",
                        headers={"Content-Type": "application/json"},
                    )
                    with urllib.request.urlopen(req, timeout=120) as resp:
                        if resp.status != 200:
                            return None, f"HTTP {resp.status}"
                        data = resp.read()
                        if len(data) > 30 * 1024 * 1024:
                            return None, f"ZIP zu gross ({len(data) // (1024*1024)} MB > 30 MB)"
                        return data, None
                except Exception as e:
                    return None, str(e)

            data, err = await asyncio.get_event_loop().run_in_executor(None, _do_zip)
            if err or data is None:
                await self._send_to_rvs({
                    "type": "file_zip_response",
                    "payload": {"requestId": req_id, "ok": False, "error": err or "leer"},
                    "timestamp": int(asyncio.get_event_loop().time() * 1000),
                })
                return
            import base64 as _b64
            await self._send_to_rvs({
                "type": "file_zip_response",
                "payload": {
                    "requestId": req_id, "ok": True,
                    "size": len(data),
                    "data": _b64.b64encode(data).decode("ascii"),
                },
                "timestamp": int(asyncio.get_event_loop().time() * 1000),
            })
            return

        elif msg_type == "file_delete_request":
            # App will eine Datei loeschen — leite an Diagnostic.
            p = payload.get("path", "")
            logger.warning("[rvs] file_delete_request von App: %s", p)
            try:
                body_bytes = json.dumps({"path": p}).encode("utf-8")
                req = urllib.request.Request(
                    "http://localhost:3001/api/files-delete",
                    data=body_bytes,
                    method="POST",
                    headers={"Content-Type": "application/json"},
                )
                def _do_delete():
                    try:
                        with urllib.request.urlopen(req, timeout=10) as resp:
                            return resp.status, resp.read().decode("utf-8", errors="ignore")
                    except Exception as e:
                        return None, str(e)
                status, body = await asyncio.get_event_loop().run_in_executor(None, _do_delete)
                logger.info("[rvs] file_delete_request %s: status=%s", p, status)
                # Diagnostic broadcastet file_deleted via sendToRVS_raw — kommt
                # ueber den persistenten WS-Path zur App. Wir bestaetigen
                # zusaetzlich, damit der Caller sicher ist dass es durch ist.
            except Exception as e:
                logger.warning("[rvs] file_delete_request: %s", e)
            return

        elif msg_type == "file_version_list_request":
            # Versions-Historie einer Datei (App-Side Dateimanager).
            # Pfad ist relativ-zu-/shared/uploads, kommt vom App-File-Manager
            # der eh nur diesen flachen Bereich anzeigt. Diagnostic hat die
            # git-Logik — wir proxien.
            req_path = payload.get("path", "")
            logger.info("[rvs] file_version_list_request: %s", req_path)
            try:
                qs = urllib.parse.urlencode({"path": req_path})
                req = urllib.request.Request(
                    f"http://localhost:3001/api/files-versions?{qs}",
                    method="GET",
                )
                def _do_list():
                    try:
                        with urllib.request.urlopen(req, timeout=10) as resp:
                            return json.loads(resp.read().decode("utf-8", errors="ignore"))
                    except Exception as e:
                        return {"ok": False, "error": str(e)}
                d = await asyncio.get_event_loop().run_in_executor(None, _do_list)
                await self._send_to_rvs({
                    "type": "file_version_list_response",
                    "payload": d,
                    "timestamp": int(asyncio.get_event_loop().time() * 1000),
                })
            except Exception as e:
                logger.warning("[rvs] file_version_list_request: %s", e)
            return

        elif msg_type == "file_version_download_request":
            # Inhalt einer alten Version holen, base64 zurueck. Diagnostic
            # liefert Binary, wir wrappen als base64 in der Response damit
            # die App's RVS-WS damit umgehen kann.
            req_path = payload.get("path", "")
            req_hash = payload.get("hash", "")
            req_id = payload.get("requestId", "")
            logger.info("[rvs] file_version_download_request: %s @ %s",
                        req_path, req_hash[:7])
            try:
                qs = urllib.parse.urlencode({"path": req_path, "hash": req_hash})
                req = urllib.request.Request(
                    f"http://localhost:3001/api/files-version-content?{qs}",
                    method="GET",
                )
                def _do_dl():
                    try:
                        with urllib.request.urlopen(req, timeout=30) as resp:
                            return resp.status, resp.read()
                    except urllib.error.HTTPError as e:
                        return e.code, e.read()
                    except Exception as e:
                        return None, str(e).encode("utf-8")
                status, body = await asyncio.get_event_loop().run_in_executor(None, _do_dl)
                if status == 200 and isinstance(body, (bytes, bytearray)):
                    await self._send_to_rvs({
                        "type": "file_version_download_response",
                        "payload": {
                            "ok": True,
                            "requestId": req_id,
                            "path": req_path,
                            "hash": req_hash,
                            "base64": base64.b64encode(body).decode("ascii"),
                            "size": len(body),
                            "name": (req_path.rsplit("/", 1)[-1] or "file"),
                        },
                        "timestamp": int(asyncio.get_event_loop().time() * 1000),
                    })
                else:
                    err = body.decode("utf-8", "ignore") if isinstance(body, (bytes, bytearray)) else str(body)
                    await self._send_to_rvs({
                        "type": "file_version_download_response",
                        "payload": {
                            "ok": False,
                            "requestId": req_id,
                            "path": req_path,
                            "hash": req_hash,
                            "error": f"HTTP {status}: {err[:200]}",
                        },
                        "timestamp": int(asyncio.get_event_loop().time() * 1000),
                    })
            except Exception as e:
                logger.warning("[rvs] file_version_download_request: %s", e)
            return

        elif msg_type == "file_version_restore_request":
            # Eine Version als neue aktive setzen — non-destructive
            # (diagnostic schreibt den alten Inhalt + macht einen neuen Commit).
            req_path = payload.get("path", "")
            req_hash = payload.get("hash", "")
            logger.warning("[rvs] file_version_restore_request: %s <- %s",
                           req_path, req_hash[:7])
            try:
                body_bytes = json.dumps({"path": req_path, "hash": req_hash}).encode("utf-8")
                req = urllib.request.Request(
                    "http://localhost:3001/api/files-version-restore",
                    data=body_bytes,
                    method="POST",
                    headers={"Content-Type": "application/json"},
                )
                def _do_restore():
                    try:
                        with urllib.request.urlopen(req, timeout=15) as resp:
                            return resp.status, resp.read().decode("utf-8", errors="ignore")
                    except urllib.error.HTTPError as e:
                        return e.code, e.read().decode("utf-8", errors="ignore")
                    except Exception as e:
                        return None, str(e)
                status, body = await asyncio.get_event_loop().run_in_executor(None, _do_restore)
                try:
                    parsed = json.loads(body) if body else {"ok": False, "error": "leer"}
                except Exception:
                    parsed = {"ok": False, "error": body[:200]}
                if status != 200 and "ok" not in parsed:
                    parsed = {"ok": False, "error": f"HTTP {status}"}
                await self._send_to_rvs({
                    "type": "file_version_restore_response",
                    "payload": parsed,
                    "timestamp": int(asyncio.get_event_loop().time() * 1000),
                })
            except Exception as e:
                logger.warning("[rvs] file_version_restore_request: %s", e)
            return

        elif msg_type == "location_update":
            # Live-GPS-Update von der App (nicht an Chat gekoppelt). Wird in
            # /shared/state/location.json geschrieben, damit Watcher-Trigger
            # near()-Conditions auswerten koennen.
            lat = payload.get("lat")
            lon = payload.get("lon") or payload.get("lng")
            if lat is not None and lon is not None:
                self._persist_location({"lat": lat, "lon": lon})
                logger.debug("[gps] location_update: %.5f, %.5f", float(lat), float(lon))
            return

        elif msg_type == "container_restart":
            # App-Button "Container neu" — leitet generisch an Diagnostic
            # weiter. Whitelist ist im Diagnostic-Server.
            name = payload.get("name", "")
            logger.warning("[rvs] container_restart Request von App: %s", name)
            try:
                body_bytes = json.dumps({"name": name}).encode("utf-8")
                req = urllib.request.Request(
                    "http://localhost:3001/api/container-restart",
                    data=body_bytes,
                    method="POST",
                    headers={"Content-Type": "application/json"},
                )
                def _do_restart():
                    try:
                        with urllib.request.urlopen(req, timeout=45) as resp:
                            return resp.status, resp.read().decode("utf-8", errors="ignore")
                    except Exception as e:
                        return None, str(e)
                status, body = await asyncio.get_event_loop().run_in_executor(None, _do_restart)
                logger.info("[rvs] container_restart %s Result: status=%s", name, status)
                ok = status == 200
                await self._send_to_rvs({
                    "type": "chat",
                    "payload": {
                        "text": f"[Container] {name} neu gestartet." if ok
                                else f"[Container] Restart {name} fehlgeschlagen: {body[:200]}",
                        "sender": "aria",
                    },
                    "timestamp": int(asyncio.get_event_loop().time() * 1000),
                })
            except Exception as e:
                logger.warning("[rvs] container_restart Weiterleitung fehlgeschlagen: %s", e)
            return

        elif msg_type == "file_request":
            # App fordert eine Datei an (Re-Download nach Cache-Leerung)
            server_path = payload.get("serverPath", "")
            req_id = payload.get("requestId", "")
            if not server_path or not server_path.startswith("/shared/"):
                logger.warning("[rvs] Ungueltiger file_request: %s", server_path)
                return
            if not os.path.isfile(server_path):
                logger.warning("[rvs] Datei nicht gefunden: %s", server_path)
                await self._send_to_rvs({
                    "type": "file_response",
                    "payload": {"requestId": req_id, "error": "Datei nicht gefunden"},
                    "timestamp": int(asyncio.get_event_loop().time() * 1000),
                })
                return
            # Groessen-Check VOR base64-Encode + Send. Sonst zerreisst's bei
            # grossen Files (>~70 MB binaer) die WebSocket-Verbindung mit
            # Code 1009 (message too big) — RVS-Server droppt, Bridge crasht
            # im cleanup (websockets-Lib-Bug). Limit deckt typische Videos
            # und Bilder ab; alles drueber soll der User per SSH abholen.
            FILE_MAX_BYTES = 1024 * 1024 * 1024  # 1 GB binaer
            try:
                file_size = os.path.getsize(server_path)
            except OSError as exc:
                logger.warning("[rvs] getsize fehlgeschlagen: %s", exc)
                file_size = 0
            if file_size > FILE_MAX_BYTES:
                logger.warning("[rvs] Re-Download abgelehnt: %s zu gross (%dMB > %dMB)",
                               server_path, file_size // (1024 * 1024),
                               FILE_MAX_BYTES // (1024 * 1024))
                await self._send_to_rvs({
                    "type": "file_response",
                    "payload": {
                        "requestId": req_id,
                        "serverPath": server_path,
                        "name": os.path.basename(server_path),
                        "error": f"Datei zu gross fuer Transfer ({file_size // (1024 * 1024)} MB, Limit {FILE_MAX_BYTES // (1024 * 1024)} MB)",
                        "sizeBytes": file_size,
                    },
                    "timestamp": int(asyncio.get_event_loop().time() * 1000),
                })
                return
            with open(server_path, "rb") as f:
                file_b64 = base64.b64encode(f.read()).decode("ascii")
            mime, _ = mimetypes.guess_type(server_path)
            logger.info("[rvs] Re-Download: %s (%dKB)", server_path, len(file_b64) // 1365)
            await self._send_to_rvs({
                "type": "file_response",
                "payload": {
                    "requestId": req_id,
                    "serverPath": server_path,
                    "base64": file_b64,
                    "name": os.path.basename(server_path),
                    "mimeType": mime or "application/octet-stream",
                },
                "timestamp": int(asyncio.get_event_loop().time() * 1000),
            })

        elif msg_type == "audio":
            # Audio von der App → decodieren → STT → an aria-core
            # Delivery-ACK + Idempotenz wie bei chat — App nutzt die ACKs
            # auch fuer Sprach-Bubbles (Status auf der Bubble: ✓ sent).
            client_msg_id = payload.get("clientMsgId") or None
            await self._send_chat_ack(client_msg_id)
            if self._is_duplicate_client_msg(client_msg_id):
                return
            audio_b64 = payload.get("base64", "")
            mime_type = payload.get("mimeType", "audio/mp4")
            duration_ms = payload.get("durationMs", 0)
            if not audio_b64:
                logger.warning("[rvs] Audio ohne Daten empfangen")
                return
            # Voice-Override fuer Folgenachrichten — gleiche Semantik wie beim chat-Event.
            if "voice" in payload:
                voice_override = payload.get("voice", "") or ""
                self._next_voice_override = voice_override or None
                logger.info("[rvs] Voice fuer Antworten (via Audio): %s",
                            self._next_voice_override or "(Default)")
            if "speed" in payload:
                try:
                    speed = float(payload.get("speed", 0) or 0)
                    self._next_speed_override = speed if 0.1 <= speed <= 5.0 else None
                except (TypeError, ValueError):
                    self._next_speed_override = None
            interrupted = bool(payload.get("interrupted", False))
            audio_request_id = payload.get("audioRequestId", "") or ""
            location = payload.get("location") or None
            # State persist fuer Brain-Watcher (current_lat etc.)
            self._persist_location(location)
            self._persist_user_activity()
            logger.info("[rvs] Audio empfangen: %s, %dms, %dKB%s%s%s",
                        mime_type, duration_ms, len(audio_b64) // 1365,
                        " [BARGE-IN]" if interrupted else "",
                        " [GPS]" if location else "",
                        f" reqId={audio_request_id[:16]}" if audio_request_id else "")
            asyncio.create_task(self._process_app_audio(
                audio_b64, mime_type, interrupted, audio_request_id, location,
                client_msg_id=client_msg_id))

        elif msg_type == "stt_response":
            # Antwort der whisper-bridge auf unseren stt_request
            request_id = payload.get("requestId", "")
            future = self._pending_stt.get(request_id)
            if future is None or future.done():
                return
            error = payload.get("error", "")
            if error:
                logger.warning("[rvs] stt_response Fehler: %s", error)
                future.set_result(None)
            else:
                text = payload.get("text", "")
                stt_ms = payload.get("sttMs", 0)
                model = payload.get("model", "?")
                logger.info("[rvs] Remote-STT OK (%s, %dms): '%s'", model, stt_ms, (text or "")[:80])
                future.set_result(text)
            return

        elif msg_type == "stt_stream_start":
            # App startet eine neue Streaming-STT-Session. Wir merken uns
            # ihre Focus-projectId damit der Voice-Router beim spaeteren
            # stt_endpoint einen sinnvollen Default hat (Voice folgt dem
            # visuellen Focus).
            req_id = payload.get("requestId", "") or ""
            focused_pid = str(payload.get("projectId") or "")
            if req_id:
                self._stt_stream_projects[req_id] = focused_pid
                logger.info("[rvs] stt_stream_start id=%s focus=%s",
                            req_id[:12], focused_pid or "(main)")
            return

        elif msg_type == "stt_stream_end":
            # Session vorbei — ABER nicht sofort aufraeumen: nach einem manuellen
            # Stop folgt noch der finale stt_endpoint (Whisper-Final-Transcribe),
            # der die Focus-projectId aus dieser Registry braucht. Wuerden wir hier
            # sofort poppen, bekaeme der Endpoint "" → die Nachricht (und ARIA's
            # Antwort) landet im Hauptchat statt im fokussierten Projekt. Der
            # stt_endpoint-Handler popt selbst; hier nur ein verzoegerter Cleanup
            # als Leak-Schutz, falls gar kein Endpoint mehr kommt.
            req_id = payload.get("requestId", "") or ""
            if req_id:
                try:
                    loop = asyncio.get_event_loop()
                    loop.call_later(
                        20.0,
                        lambda rid=req_id: self._stt_stream_projects.pop(rid, None),
                    )
                except Exception:
                    # Kein laufender Loop (sollte im ws-Handler nie passieren) —
                    # dann lieber gar nicht aufraeumen als crashen.
                    pass
            return

        elif msg_type == "stt_endpoint":
            # Phase 2 Brain-Shortcut: die whisper-bridge hat im Streaming-Modus
            # einen Endpoint erkannt und schickt den finalen Text direkt.
            # Wir uebernehmen die Rolle die sonst _process_app_audio NACH dem
            # STT-Schritt hat: STT-Text fuer UI broadcasten + send_to_core.
            # Kein Audio-Roundtrip mehr — App-Latenz sinkt deutlich.
            text = (payload.get("text") or "").strip()
            if not text:
                logger.info("[rvs] stt_endpoint mit leerem Text — ignoriert (reason=%s)",
                            payload.get("reason", ""))
                return
            audio_request_id = payload.get("audioRequestId", "") or ""
            voice = payload.get("voice", "") or ""
            speed_raw = payload.get("speed")
            interrupted = bool(payload.get("interrupted", False))
            location = payload.get("location") or None

            # Voice-Override fuer Folgenachrichten — gleiche Semantik wie beim
            # 'audio'-Event. Nur setzen wenn vom App-Stream mitgegeben.
            if voice:
                self._next_voice_override = voice or None
                logger.info("[rvs] Voice fuer Antworten (via stt_endpoint): %s",
                            self._next_voice_override or "(Default)")
            if speed_raw is not None:
                try:
                    sp = float(speed_raw)
                    self._next_speed_override = sp if 0.1 <= sp <= 5.0 else None
                except (TypeError, ValueError):
                    self._next_speed_override = None

            # State-Persist wie bei _process_app_audio
            self._persist_location(location)
            self._persist_user_activity()

            logger.info("[rvs] stt_endpoint: '%s' (%dms, reason=%s)%s%s reqId=%s",
                        text[:80],
                        payload.get("sttMs", 0),
                        payload.get("reason", ""),
                        " [BARGE-IN]" if interrupted else "",
                        " [GPS]" if location else "",
                        audio_request_id[:16] if audio_request_id else "?")

            # Idempotenz ueber audioRequestId — falls App den Stream irgendwie
            # nochmal triggern sollte (Reconnect-Race etc.).
            client_msg_id = audio_request_id or None
            if self._is_duplicate_client_msg(client_msg_id):
                return

            # App-Focus aus stt_stream_start-Registry auflösen (falls die
            # App-Version die projectId noch nicht mitschickt: leer = Hauptchat).
            stream_req_id = payload.get("requestId", "") or ""
            focused_pid = self._stt_stream_projects.pop(stream_req_id, "")

            asyncio.create_task(self._process_endpoint_text(
                text, interrupted, audio_request_id, location,
                client_msg_id=client_msg_id,
                focused_project_id=focused_pid))
            return

        elif msg_type == "oauth_callback":
            # RVS hat einen OAuth-Provider-Callback empfangen (z.B. Spotify
            # nach User-Authorize) und broadcastet ihn. Wir forwarden an Brain,
            # das den state-Match macht + code gegen access_token tauscht.
            asyncio.create_task(self._forward_oauth_callback(payload))
            return

        elif msg_type == "flux_response":
            # Antwort der flux-bridge auf unseren flux_request. Erste Nachricht
            # mit state='rendering' ist nur Progress-Ping — die echte Antwort
            # kommt mit state='done' (oder error).
            request_id = payload.get("requestId", "")
            future = self._pending_flux.get(request_id)
            if future is None or future.done():
                return
            error = payload.get("error", "")
            if error:
                logger.warning("[rvs] flux_response Fehler: %s", error)
                future.set_result({"error": error})
                return
            state = payload.get("state", "")
            if state == "rendering":
                # Nur Progress-Info, future bleibt offen
                logger.info("[rvs] flux: rendering %dx%d steps=%d ...",
                            payload.get("width", 0), payload.get("height", 0),
                            payload.get("steps", 0))
                return
            # state == "done" oder fehlt → final
            logger.info("[rvs] flux fertig: %dx%d, %.1fs, %d KB",
                        payload.get("width", 0), payload.get("height", 0),
                        payload.get("renderSeconds", 0),
                        (payload.get("sizeBytes", 0)) // 1024)
            future.set_result(payload)
            return

        elif msg_type == "llm_response":
            # Antwort des llm-adapter (Gamebox) auf unseren llm_request.
            request_id = payload.get("requestId", "")
            future = self._pending_llm.get(request_id)
            if future is None or future.done():
                return
            future.set_result(payload)
            return

        elif msg_type == "service_status":
            # Gamebox-Bridges (whisper / f5tts / flux) melden ihren Lade-Status.
            # Wir nutzen das fuer den dynamischen STT-Timeout: solange whisper
            # im 'loading' steckt, geben wir der Bridge mehr Zeit (Modell-Download
            # kann 1-2 Min dauern), statt nach 45s lokal zu fallbacken.
            svc = payload.get("service", "")
            state = payload.get("state", "")
            if svc == "whisper":
                was_ready = self._remote_stt_ready
                self._remote_stt_ready = (state == "ready")
                if self._remote_stt_ready != was_ready:
                    logger.info("[rvs] whisper-bridge -> %s", state)
            elif svc == "flux":
                was_ready = self._remote_flux_ready
                self._remote_flux_ready = (state == "ready")
                if self._remote_flux_ready != was_ready:
                    logger.info("[rvs] flux-bridge -> %s", state)
            return

        elif msg_type == "config_request":
            # Eine andere Bridge (whisper/f5tts) bittet um die aktuelle Voice-
            # Config — passiert wenn sie sich connected, weil sie sonst die
            # Diagnostic-Settings nicht kennt. Wir broadcasten die persistierte
            # Config (auch beim normalen Connect von aria-bridge selber, aber
            # da war eventuell die andere Bridge noch nicht connected).
            requester = payload.get("service", "?")
            logger.info("[rvs] config_request von %s — broadcaste Voice-Config", requester)
            asyncio.create_task(self._broadcast_persisted_config())
            return

        else:
            logger.debug("[rvs] Unbekannter Typ: %s", msg_type)

    # STT-Orchestrierung: zuerst Remote (Gamebox), Fallback lokal.
    # Zwei Timeouts:
    #   ready=True   → 45s reicht selbst fuer lange Audios
    #   ready=False  → 300s, weil das Modell evtl. noch heruntergeladen wird
    #                  (large-v3 ~3GB, kann auf der Gamebox 1-2 Min dauern).
    _STT_REMOTE_TIMEOUT_READY_S = 45.0
    _STT_REMOTE_TIMEOUT_LOADING_S = 300.0

    async def _process_app_audio(self, audio_b64: str, mime_type: str,
                                  interrupted: bool = False,
                                  audio_request_id: str = "",
                                  location: Optional[dict] = None,
                                  client_msg_id: Optional[str] = None) -> None:
        """App-Audio → STT → aria-core. Primaer via whisper-bridge (RVS), Fallback lokal.

        interrupted=True wenn der User waehrend ARIA noch sprach/dachte aufgenommen hat
        (Barge-In). Wird als Hinweis-Praefix an aria-core mitgegeben damit ARIA die
        Korrektur/Unterbrechung in den Kontext einordnen kann statt als reine
        Folgefrage zu behandeln.

        audio_request_id: Korrelations-ID die die App im audio-Event mitschickt — wird
        unveraendert ans STT-Result zurueckgegeben damit die App die EXAKT richtige
        'wird verarbeitet'-Bubble ersetzen kann (auch bei mehreren parallelen Aufnahmen).

        location: Optional GPS-Position {lat, lon} — wird als Hinweis-Praefix mitgegeben
        damit ARIA bei standortbezogenen Fragen sie nutzen kann."""
        # Erst Remote versuchen
        text = await self._stt_remote(audio_b64, mime_type)
        if text is None:
            # Remote hat nicht geantwortet → lokales Whisper
            logger.warning("[rvs] Remote-STT nicht verfuegbar — Fallback auf lokales Whisper")
            text = await self._stt_local(audio_b64, mime_type)
        if text is None:
            return

        if text.strip():
            logger.info("[rvs] STT Ergebnis: '%s'", text[:80])

            # Reihenfolge wichtig: STT-Text ZUERST broadcasten damit die App
            # die Voice-Bubble sofort mit dem erkannten Text aktualisieren
            # kann — send_to_core blockt danach synchron auf Brain (kann
            # dauern), wuerde sonst die Anzeige verzoegern.
            try:
                stt_payload = {
                    "text": text,
                    "sender": "stt",
                }
                if audio_request_id:
                    stt_payload["audioRequestId"] = audio_request_id
                # GPS aus dem Original-Audio-Payload mitgeben — Diagnostic
                # zeigt sie sonst nicht an (App sendet location nur einmal,
                # die im audio-Payload). Reine Anzeige-Information.
                if location:
                    stt_payload["location"] = location
                ok = await self._send_to_rvs({
                    "type": "chat",
                    "payload": stt_payload,
                    "timestamp": int(asyncio.get_event_loop().time() * 1000),
                })
                if ok:
                    logger.info("[rvs] STT-Text an RVS broadcastet (sender=stt)")
                else:
                    logger.warning("[rvs] STT-Text NICHT broadcastet — _send_to_rvs lieferte False")
            except Exception as e:
                logger.warning("[rvs] STT-Text konnte nicht an RVS gesendet werden: %s", e)

            # Dann an Brain — der blockt synchron bis ARIA fertig ist.
            core_text = self._build_core_text(text, interrupted, location)
            await self.send_to_core(core_text,
                                    source="app-voice" + (" [barge-in]" if interrupted else ""),
                                    client_msg_id=client_msg_id)
        else:
            logger.info("[rvs] Keine Sprache erkannt — ignoriert")

    # Voice-Router-Konstanten
    _VOICE_STICKY_TIMEOUT_SEC = 30.0
    _VOICE_META_BACK_TO_MAIN = re.compile(
        r"^\s*(?:aria[,.]?\s+)?"
        r"(?:"
        # „zurück zum hauptchat / hauptmenü / haupt / menü / main"
        r"zur(?:ü|ue)ck\s+(?:zum|zur|ins?|in\s+den)\s+"
        r"(?:hauptchat|hauptmen(?:ü|ue)|haupt|men(?:ü|ue)|main)"
        r"|"
        # „zurück hauptchat / zurück haupt"
        r"zur(?:ü|ue)ck\s+(?:hauptchat|hauptmen(?:ü|ue)|haupt|main)"
        r"|"
        # „hauptchat bitte", „aria hauptchat" (auch mit Menü/Main)
        r"(?:hauptchat|hauptmen(?:ü|ue)|main)\s+bitte"
        r"|"
        r"aria[,.]?\s+(?:hauptchat|hauptmen(?:ü|ue)|haupt|main)"
        r")\s*[.!?]?\s*$",
        re.IGNORECASE,
    )
    _VOICE_META_PROJECT_PREFIX = re.compile(
        r"^\s*(?:aria[,.]?\s+)?(?:f(?:ü|ue)r|ins?)\s+([\w\-äöüßÄÖÜ]{2,40})[:\-,]\s*(.+?)\s*$",
        re.IGNORECASE | re.DOTALL,
    )

    def _apply_voice_router(self, text: str,
                             default_project_id: str = "") -> tuple[bool, str, str, str]:
        """Voice-Router: entscheidet ob ein STT-Text ans Brain geht und wenn ja
        an welchen Projekt-Kontext.

        Returns (should_forward, cleaned_text, project_id, meta_action):
          - should_forward=False: reines Meta-Kommando, kein Brain-Call.
            meta_action beschreibt was passiert ist (broadcastet an UI).
          - should_forward=True: cleaned_text ans Brain, project_id ist Focus.
            Bei Prefix wird der Prefix aus dem Text entfernt.

        Prioritaets-Reihenfolge:
          1. Meta „zurueck zum hauptchat" → Sticky reset, kein Forward.
          2. „fuer <name>:"-Prefix → Sticky auf gematchtes Projekt.
          3. Sticky aktiv (<=30s alt) → dessen Projekt.
          4. default_project_id (App-Focus) — Voice folgt dem sichtbaren
             Kontext. Wenn App in Projekt X guckt, geht die STT-Nachricht
             ohne weitere Marker dort rein.
          5. Fallback: Hauptchat.
        """
        import time as _time
        now = _time.time()
        stripped = text.strip()

        # 1) Meta: zurueck zum Hauptchat
        if self._VOICE_META_BACK_TO_MAIN.match(stripped):
            self._voice_sticky_project_id = ""
            self._voice_sticky_expires_at = 0.0
            return (False, "", "", "back_to_main")

        # 2) Prefix: "fuer <name>: <text>"
        m = self._VOICE_META_PROJECT_PREFIX.match(stripped)
        if m:
            name = m.group(1)
            remainder = m.group(2).strip()
            # Fuzzy-Match auf Projekt via Brain-API
            try:
                brain_url = os.environ.get("BRAIN_URL", "http://aria-brain:8080")
                with urllib.request.urlopen(f"{brain_url}/projects/list", timeout=3) as r:
                    projects = json.loads(r.read()).get("projects", [])
                from difflib import SequenceMatcher
                best, best_score = None, 0.0
                q = name.lower()
                for p in projects:
                    pname = p.get("name", "").lower()
                    if q == pname or q == p.get("id", ""):
                        best, best_score = p, 1.0
                        break
                    s = SequenceMatcher(None, q, pname).ratio()
                    if s > best_score:
                        best, best_score = p, s
                if best and best_score >= 0.6:
                    pid = best["id"]
                    self._voice_sticky_project_id = pid
                    self._voice_sticky_expires_at = now + self._VOICE_STICKY_TIMEOUT_SEC
                    logger.info("[voice-router] Prefix → Projekt '%s' (id=%s, score=%.2f)",
                                best.get("name"), pid, best_score)
                    return (True, remainder or stripped, pid, "project_prefix")
            except Exception as exc:
                logger.warning("[voice-router] Prefix-Match fehlgeschlagen: %s", exc)
            # Kein Match → als normale Nachricht weiter (Sticky wenn aktiv)
        # 3) Kein Meta / Prefix → Sticky oder Default
        if self._voice_sticky_project_id and now < self._voice_sticky_expires_at:
            # Sticky refreshen
            self._voice_sticky_expires_at = now + self._VOICE_STICKY_TIMEOUT_SEC
            return (True, stripped, self._voice_sticky_project_id, "sticky")
        # Sticky abgelaufen — zurücksetzen
        self._voice_sticky_project_id = ""
        # 4) App-Focus als Default: Voice folgt dem sichtbaren Kontext
        if default_project_id:
            return (True, stripped, default_project_id, "app_focus")
        # 5) Fallback Hauptchat
        return (True, stripped, "", "default")

    async def _process_endpoint_text(self, text: str,
                                      interrupted: bool = False,
                                      audio_request_id: str = "",
                                      location: Optional[dict] = None,
                                      client_msg_id: Optional[str] = None,
                                      focused_project_id: str = "") -> None:
        """Phase-2 Brain-Shortcut: Streaming-Whisper hat den finalen Text
        schon ermittelt — wir uebernehmen den Pfad ab broadcast-STT + brain.

        Spiegel-Methode zu _process_app_audio NACH dem STT-Schritt. Bewusst
        eigene Methode statt Code-Pfade in _process_app_audio aufdroeseln,
        damit der Legacy-Pfad (App schickt 'audio') unangetastet bleibt.

        Voice-Router: interceptiert Meta-Kommandos (zurueck zum Hauptchat)
        + Prefix-Adressierung („fuer Frankreich: ...") + 30s-Sticky. Meta
        selbst geht NICHT ans Brain, sondern broadcastet als project_changed-
        Event → App+Diagnostic wechseln den Focus.
        """
        should_forward, cleaned, project_id, meta_action = self._apply_voice_router(
            text, default_project_id=focused_project_id,
        )

        if meta_action in ("back_to_main", "project_prefix"):
            # UI-Focus-Update broadcasten
            payload = {"action": "entered" if meta_action == "project_prefix" else "exited"}
            if meta_action == "project_prefix" and project_id:
                # Namen aus dem Cache holen — best effort
                try:
                    brain_url = os.environ.get("BRAIN_URL", "http://aria-brain:8080")
                    with urllib.request.urlopen(f"{brain_url}/projects/list", timeout=2) as r:
                        for p in json.loads(r.read()).get("projects", []):
                            if p.get("id") == project_id:
                                payload["id"] = project_id
                                payload["name"] = p.get("name", "")
                                break
                except Exception:
                    payload["id"] = project_id
            await self._send_to_rvs({
                "type": "project_changed",
                "payload": payload,
                "timestamp": int(asyncio.get_event_loop().time() * 1000),
            })

        if not should_forward:
            logger.info("[voice-router] Meta-Kommando '%s' intercepted, kein Brain-Call",
                        meta_action)
            return

        try:
            stt_payload = {
                "text": cleaned,
                "sender": "stt",
            }
            if audio_request_id:
                stt_payload["audioRequestId"] = audio_request_id
            if location:
                stt_payload["location"] = location
            if project_id:
                stt_payload["projectId"] = project_id
            ok = await self._send_to_rvs({
                "type": "chat",
                "payload": stt_payload,
                "timestamp": int(asyncio.get_event_loop().time() * 1000),
            })
            if ok:
                logger.info("[rvs] STT-Text (endpoint) broadcastet")
            else:
                logger.warning("[rvs] STT-Text (endpoint) NICHT broadcastet")
        except Exception as e:
            logger.warning("[rvs] STT-Text (endpoint) konnte nicht broadcastet werden: %s", e)

        core_text = self._build_core_text(cleaned, interrupted, location)
        await self.send_to_core(core_text,
                                source="app-voice-stream" + (" [barge-in]" if interrupted else ""),
                                client_msg_id=client_msg_id,
                                project_id=project_id)

    async def _stt_remote(self, audio_b64: str, mime_type: str) -> Optional[str]:
        """Schickt Audio an die whisper-bridge und wartet auf stt_response.

        Rueckgabe:
            str  — erkannter Text (kann leer sein)
            None — Remote-STT nicht erreichbar oder Fehler/Timeout (→ Fallback)
        """
        if self.ws_rvs is None:
            return None

        request_id = str(uuid.uuid4())
        loop = asyncio.get_event_loop()
        future: asyncio.Future = loop.create_future()
        self._pending_stt[request_id] = future

        try:
            model = getattr(self.stt_engine, "model_size", "small")
            logger.info("[rvs] stt_request → whisper-bridge (id=%s, model=%s, %dKB)",
                        request_id[:8], model, len(audio_b64) // 1365)
            ok = await self._send_to_rvs({
                "type": "stt_request",
                "payload": {
                    "requestId": request_id,
                    "audio": audio_b64,
                    "mimeType": mime_type,
                    "model": model,
                    "language": getattr(self.stt_engine, "language", "de"),
                },
                "timestamp": int(loop.time() * 1000),
            })
            if not ok:
                logger.warning("[rvs] stt_request konnte nicht gesendet werden — skip Remote")
                return None
            timeout_s = (self._STT_REMOTE_TIMEOUT_READY_S
                         if self._remote_stt_ready
                         else self._STT_REMOTE_TIMEOUT_LOADING_S)
            logger.info("[rvs] STT-Timeout %ds (whisper-bridge %s)",
                        int(timeout_s), "ready" if self._remote_stt_ready else "loading")
            return await asyncio.wait_for(future, timeout=timeout_s)
        except asyncio.TimeoutError:
            logger.warning("[rvs] Remote-STT Timeout (%.0fs)", self._STT_REMOTE_TIMEOUT_S)
            return None
        except Exception as e:
            logger.warning("[rvs] Remote-STT Fehler: %s", e)
            return None
        finally:
            self._pending_stt.pop(request_id, None)

    async def _stt_local(self, audio_b64: str, mime_type: str) -> Optional[str]:
        """Lokales Whisper-Fallback: FFmpeg → float32 → stt_engine.transcribe."""
        loop = asyncio.get_event_loop()
        tmp_in = None
        tmp_out = None
        try:
            ext = ".mp4" if "mp4" in mime_type else ".wav" if "wav" in mime_type else ".ogg"
            tmp_in = tempfile.NamedTemporaryFile(suffix=ext, delete=False)
            tmp_in.write(base64.b64decode(audio_b64))
            tmp_in.close()

            tmp_out = tempfile.NamedTemporaryFile(suffix=".raw", delete=False)
            tmp_out.close()

            cmd = [
                "ffmpeg", "-y", "-i", tmp_in.name,
                "-ar", "16000", "-ac", "1", "-f", "f32le",
                tmp_out.name,
            ]
            result = await loop.run_in_executor(
                None,
                lambda: subprocess.run(cmd, capture_output=True, timeout=30),
            )
            if result.returncode != 0:
                logger.error("[rvs] FFmpeg Fehler: %s", result.stderr.decode()[:200])
                return None

            audio_data = np.fromfile(tmp_out.name, dtype=np.float32)
            if len(audio_data) == 0:
                logger.warning("[rvs] Leere Audio-Daten nach Konvertierung")
                return None

            duration_s = len(audio_data) / 16000.0
            logger.info("[rvs] Lokal-STT: %.1fs Audio, model=%s", duration_s, self.stt_engine.model_size)
            return await loop.run_in_executor(None, self.stt_engine.transcribe, audio_data)
        except Exception:
            logger.exception("[rvs] Lokales STT fehlgeschlagen")
            return None
        finally:
            for f in (tmp_in, tmp_out):
                if f:
                    try:
                        os.unlink(f.name)
                    except OSError:
                        pass

    # ── Flux-Roundtrip: Brain → Bridge → RVS → flux-bridge → zurueck ──
    # FLUX-Render auf der 3060 dauert je nach Aufloesung/Steps 20-90 s.
    # Beim 1. Render frisch nach Container-Start muss zudem das ~24 GB
    # Modell von HF geladen werden — daher der grosse Loading-Timeout.
    _FLUX_TIMEOUT_READY_S = 240.0      # 4 min nach erstem Render
    _FLUX_TIMEOUT_LOADING_S = 900.0    # 15 min beim allerersten Mal (Modell-Download)

    # ── Local-LLM-Roundtrip: Brain → Bridge → RVS → llm-adapter → zurueck ──
    # Qwen3 auf der Gamebox antwortet auf kurze Turns in <1 s. Grosszuegiger
    # Timeout deckt Kaltstart / laengere Antworten / Netz-Jitter (Gamebox@home)
    # ab. Bei Timeout faellt der Router im Brain per Escalation auf Claude.
    _LLM_TIMEOUT_S = 30.0

    async def _local_llm(self, messages: list, max_tokens: int = 512,
                         temperature: float = 0.7, stop=None) -> dict:
        """Schickt einen llm_request an den llm-adapter (Gamebox), wartet auf
        llm_response. Rueckgabe: {ok, content, model, elapsedMs} oder {ok:False, error}."""
        if self.ws_rvs is None:
            return {"ok": False, "error": "RVS-Verbindung nicht aktiv"}
        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
            logger.info("[rvs] llm_request → llm-adapter (id=%s, msgs=%d, max_tokens=%d)",
                        request_id[:8], len(messages), max_tokens)
            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) — Gamebox 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", ""),
                "model": result.get("model"),
                "elapsedMs": result.get("elapsedMs"),
            }
        finally:
            self._pending_llm.pop(request_id, None)

    async def _flux_generate(self, prompt: str, width: int, height: int,
                             steps: Optional[int], guidance: Optional[float],
                             seed: Optional[int], model: Optional[str] = None) -> dict:
        """Schickt einen flux_request an die flux-bridge, wartet auf das fertige
        PNG, speichert es nach /shared/uploads/aria_generated_<ts>.png.

        Rueckgabe:
            {ok: True,  path, sizeBytes, width, height, steps, guidance, seed, model, renderSeconds}
            {ok: False, error}
        """
        if self.ws_rvs is None:
            return {"ok": False, "error": "RVS-Verbindung nicht aktiv"}

        request_id = str(uuid.uuid4())
        loop = asyncio.get_event_loop()
        future: asyncio.Future = loop.create_future()
        self._pending_flux[request_id] = future

        try:
            req_payload: dict = {"requestId": request_id, "prompt": prompt,
                                 "width": width, "height": height}
            if steps is not None:
                req_payload["steps"] = steps
            if guidance is not None:
                req_payload["guidance_scale"] = guidance
            if seed is not None:
                req_payload["seed"] = seed
            if model:
                # 'dev' | 'schnell' — flux-bridge mappt das auf HF-IDs.
                # Ohne Angabe nimmt die flux-bridge ihren konfigurierten Default.
                req_payload["model"] = model

            logger.info("[rvs] flux_request → flux-bridge (id=%s, %dx%d, steps=%s, model=%s, prompt=%r)",
                        request_id[:8], width, height, steps, model or "default", prompt[:60])
            ok = await self._send_to_rvs({
                "type": "flux_request",
                "payload": req_payload,
                "timestamp": int(time.time() * 1000),
            })
            if not ok:
                return {"ok": False, "error": "flux_request konnte nicht gesendet werden"}

            timeout_s = (self._FLUX_TIMEOUT_READY_S
                         if self._remote_flux_ready
                         else self._FLUX_TIMEOUT_LOADING_S)
            result = await asyncio.wait_for(future, timeout=timeout_s)

            if not isinstance(result, dict) or result.get("error"):
                err = (result or {}).get("error") if isinstance(result, dict) else "leeres Resultat"
                return {"ok": False, "error": err or "flux-bridge Fehler"}

            b64 = result.get("base64") or ""
            if not b64:
                return {"ok": False, "error": "flux_response ohne Bilddaten"}

            try:
                png_bytes = base64.b64decode(b64)
            except Exception as e:
                return {"ok": False, "error": f"PNG-Decode fehlgeschlagen: {e}"}

            SHARED_DIR = "/shared/uploads"
            os.makedirs(SHARED_DIR, exist_ok=True)
            ts_ms = int(time.time() * 1000)
            file_name = f"aria_generated_{ts_ms}.png"
            path = os.path.join(SHARED_DIR, file_name)
            try:
                with open(path, "wb") as f:
                    f.write(png_bytes)
            except Exception as e:
                return {"ok": False, "error": f"Speichern fehlgeschlagen: {e}"}

            logger.info("[rvs] flux PNG gespeichert: %s (%d KB)", path, len(png_bytes) // 1024)
            return {
                "ok": True,
                "path": path,
                "sizeBytes": len(png_bytes),
                "width": result.get("width", width),
                "height": result.get("height", height),
                "steps": result.get("steps"),
                "guidance": result.get("guidance"),
                "seed": result.get("seed"),
                "model": result.get("model", ""),
                "renderSeconds": result.get("renderSeconds", 0),
            }
        except asyncio.TimeoutError:
            return {"ok": False, "error": f"Render-Timeout ({int(timeout_s)}s) — flux-bridge offline?"}
        except Exception as e:
            logger.exception("[rvs] _flux_generate Fehler")
            return {"ok": False, "error": str(e)[:200]}
        finally:
            self._pending_flux.pop(request_id, None)

    async def _send_to_rvs(self, message: dict) -> bool:
        """Sendet eine Nachricht an die App (via RVS) mit Verbindungs-Check.

        Rueckgabe: True wenn erfolgreich gesendet, False wenn Verbindung tot.
        """
        if self.ws_rvs is None:
            return False

        # Ping-Check: Verbindung wirklich aktiv?
        try:
            pong = await self.ws_rvs.ping()
            await asyncio.wait_for(pong, timeout=5)
        except Exception:
            logger.warning("[rvs] Ping fehlgeschlagen — Verbindung tot, erzwinge Reconnect")
            try:
                await self.ws_rvs.close()
            except Exception:
                pass
            self.ws_rvs = None
            # Reconnect wird vom connect_to_rvs Loop uebernommen
            return False

        try:
            await self.ws_rvs.send(json.dumps(message))
            return True
        except Exception:
            logger.warning("[rvs] Sendefehler — RVS nicht erreichbar")
            return False

    # ── Log-Streaming an die App ─────────────────────────────

    async def _cancel_via_diagnostic(self) -> None:
        """Ruft das Diagnostic /api/cancel an — dort laeuft die volle Abbruch-Logik
        (openclaw doctor --fix mit Docker-Socket)."""
        def _do_request():
            try:
                req = urllib.request.Request(
                    f"{self._diagnostic_url}/api/cancel",
                    method="POST",
                    data=b"",
                )
                with urllib.request.urlopen(req, timeout=5) as resp:
                    return resp.status
            except Exception as e:
                return f"error: {e}"

        status = await asyncio.get_event_loop().run_in_executor(None, _do_request)
        logger.info("[cancel] Diagnostic /api/cancel: %s", status)

    async def _forward_oauth_callback(self, payload: dict) -> None:
        """Forwarded den OAuth-Callback (kommt via RVS vom RVS-HTTP-Handler)
        per HTTP an Brain. Brain hat den pending-state + macht den token-
        exchange. Fire-and-forget — bei Failure loggen wir nur."""
        service = (payload.get("service") or "").strip()
        if not service:
            logger.warning("[oauth] callback ohne service, ignoriert")
            return
        brain_url = os.environ.get("BRAIN_URL", "http://aria-brain:8080")
        url = f"{brain_url}/internal/oauth-callback"

        def _do_request():
            try:
                data = json.dumps(payload).encode("utf-8")
                req = urllib.request.Request(
                    url, data=data, method="POST",
                    headers={"Content-Type": "application/json"},
                )
                with urllib.request.urlopen(req, timeout=10) as resp:
                    return resp.status, resp.read().decode("utf-8", "ignore")[:200]
            except Exception as e:
                return f"error: {e}", ""

        status, body = await asyncio.get_event_loop().run_in_executor(None, _do_request)
        logger.info("[oauth] Forward %s → brain: %s %s", service, status, body)

    async def _cancel_proxy_subprocesses(self) -> None:
        """Not-Aus: ruft den proxy-internen /cancel-all Side-Channel auf
        (siehe proxy-patches/routes.js). Killt alle aktiven Claude-Code-
        Subprocesses sofort. Bridge ist auf aria-net, Proxy auch — also
        per Container-Name + Side-Channel-Port (Default 3457) erreichbar."""
        url = os.environ.get("PROXY_INTERNAL_URL", "http://aria-proxy:3457") + "/cancel-all"

        def _do_request():
            try:
                req = urllib.request.Request(url, method="POST", data=b"")
                with urllib.request.urlopen(req, timeout=3) as resp:
                    return resp.status, resp.read().decode("utf-8", "ignore")[:200]
            except Exception as e:
                return f"error: {e}", ""

        status, body = await asyncio.get_event_loop().run_in_executor(None, _do_request)
        logger.warning("[NOT-AUS] proxy /cancel-all: %s %s", status, body)

    async def _cancel_proxy_for_project(self, project_id: str) -> None:
        """Kontext-scoped Barge-In: killt NUR die Subprozesse EINES Kontexts
        (leer = Hauptchat) ueber den proxy-internen /cancel-Endpoint. So bricht
        eine Nachricht in Kontext A nicht die parallele Arbeit in Kontext B ab."""
        url = os.environ.get("PROXY_INTERNAL_URL", "http://aria-proxy:3457") + "/cancel"
        data = json.dumps({"projectId": project_id or ""}).encode("utf-8")

        def _do_request():
            try:
                req = urllib.request.Request(
                    url, method="POST", data=data,
                    headers={"Content-Type": "application/json"},
                )
                with urllib.request.urlopen(req, timeout=3) as resp:
                    return resp.status, resp.read().decode("utf-8", "ignore")[:200]
            except Exception as e:
                return f"error: {e}", ""

        status, body = await asyncio.get_event_loop().run_in_executor(None, _do_request)
        logger.info("[cancel] proxy /cancel project=%s: %s %s",
                    project_id or "(main)", status, body)

    async def _emit_activity(self, activity: str, tool: str = "", force: bool = False,
                             project_id: str = "") -> None:
        """Sendet agent_activity an die App — nur wenn sich der State geaendert hat.

        Trailing Agent-Events nach chat:final werden 3s lang unterdrueckt
        (nur 'idle' kommt immer durch).

        force=True: kein State-Dedup — wird vom Proxy-Tool-Hook genutzt
        damit auch wiederholte gleiche Tool-Aufrufe (z.B. 3x Bash
        hintereinander) im Gedanken-Stream als eigene Eintraege sichtbar
        bleiben.

        project_id: welcher Kontext arbeitet (leer = Hauptchat). App/Diagnostic
        zeigen den Indikator damit pro Kontext statt global (Multi-Threading)."""
        if activity != "idle" and self._last_chat_final_at > 0:
            since_final = asyncio.get_event_loop().time() - self._last_chat_final_at
            if since_final < 3.0:
                return
        # Dedup schliesst project_id ein — sonst wuerde ein Kontext-Wechsel bei
        # gleichem (activity, tool) verschluckt.
        state = (activity, tool, project_id)
        if not force and state == self._last_activity_state:
            return
        self._last_activity_state = state
        await self._send_to_rvs({
            "type": "agent_activity",
            "payload": {"activity": activity, "tool": tool, "projectId": project_id or ""},
            "timestamp": int(asyncio.get_event_loop().time() * 1000),
        })

    async def send_log_to_app(self, source: str, message: str, level: str = "info") -> None:
        """Sendet einen Log-Eintrag an die App (erscheint im Log-Viewer)."""
        await self._send_to_rvs({
            "type": "log",
            "payload": {
                "source": source,
                "message": message,
                "level": level,
            },
            "timestamp": int(asyncio.get_event_loop().time() * 1000),
        })

    async def send_event_to_app(self, title: str, description: str) -> None:
        """Sendet ein Event an die App (erscheint im Event-Feed)."""
        await self._send_to_rvs({
            "type": "event",
            "payload": {
                "title": title,
                "description": description,
            },
            "timestamp": int(asyncio.get_event_loop().time() * 1000),
        })

    # ── Audio-Schleife (lokales Mikrofon) ────────────────────

    async def audio_loop(self) -> None:
        """Wake-Word erkennen, aufnehmen, transkribieren, an aria-core senden."""
        logger.info("Audio-Schleife gestartet — warte auf Wake-Word '%s'...", self.wake_word.wake_word_key)

        loop = asyncio.get_event_loop()

        while self.running:
            try:
                audio_chunk = sd.rec(
                    BLOCK_SIZE,
                    samplerate=SAMPLE_RATE,
                    channels=CHANNELS,
                    dtype="int16",
                )
                sd.wait()

                detected = await loop.run_in_executor(
                    None, self.wake_word.detect, audio_chunk.flatten()
                )

                if detected:
                    logger.info("Wake-Word erkannt — starte Aufnahme")
                    await self.send_event_to_app(
                        "Wake-Word erkannt",
                        "ARIA hoert zu...",
                    )

                    audio_data = await loop.run_in_executor(None, record_audio)

                    text = await loop.run_in_executor(
                        None, self.stt_engine.transcribe, audio_data
                    )

                    if text.strip():
                        new_mode = detect_mode_switch(text)
                        if new_mode is not None:
                            self.current_mode = new_mode

                        await self.send_to_core(text, source="bridge")
                    else:
                        logger.info("Keine Sprache erkannt — ignoriert")

            except sd.PortAudioError:
                if not hasattr(self, '_audio_warned'):
                    logger.warning("Audio-Geraet nicht verfuegbar — lokales Mikrofon deaktiviert (kein Spam mehr)")
                    self._audio_warned = True
                await asyncio.sleep(60)  # 60s statt 5s — spart Log-Spam
            except Exception:
                logger.exception("Fehler in der Audio-Schleife")
                await asyncio.sleep(1)

    # ── Internal HTTP (Brain → Bridge: Trigger-Feuer-Push) ───

    async def _serve_internal_http(self) -> None:
        """Kleiner asyncio HTTP-Listener auf Port 8090.

        Empfaengt Push-Events vom Brain wenn ein Trigger feuert. Nicht
        nach aussen exposed — nur erreichbar im docker-internen aria-net.
        Endpoint:
          POST /internal/trigger-fired
          { "reply": "...", "trigger_name": "...", "type": "timer",
            "events": [{"type":"trigger_created",...}, ...] }
        """
        host, port = "0.0.0.0", 8090

        async def _send_response(writer, status: int, payload: dict) -> None:
            body = json.dumps(payload).encode("utf-8")
            status_text = "OK" if status == 200 else "Error"
            writer.write(
                f"HTTP/1.1 {status} {status_text}\r\n"
                f"Content-Type: application/json\r\n"
                f"Content-Length: {len(body)}\r\n"
                f"Connection: close\r\n\r\n".encode("utf-8")
            )
            writer.write(body)
            await writer.drain()

        async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
            try:
                request_line = await asyncio.wait_for(reader.readline(), timeout=10)
                if not request_line:
                    return
                try:
                    method, path, _ver = request_line.decode("utf-8", "ignore").strip().split(" ", 2)
                except ValueError:
                    await _send_response(writer, 400, {"error": "bad request line"})
                    return
                headers: dict[str, str] = {}
                while True:
                    line = await asyncio.wait_for(reader.readline(), timeout=5)
                    if not line or line in (b"\r\n", b"\n"):
                        break
                    name, _, value = line.decode("utf-8", "ignore").partition(":")
                    headers[name.strip().lower()] = value.strip()
                content_length = int(headers.get("content-length", "0") or "0")
                body = await reader.readexactly(content_length) if content_length else b""

                if method == "POST" and path == "/internal/trigger-fired":
                    try:
                        data = json.loads(body.decode("utf-8", "ignore"))
                    except Exception as exc:
                        await _send_response(writer, 400, {"error": f"bad json: {exc}"})
                        return
                    reply = (data.get("reply") or "").strip()
                    trigger_name = data.get("trigger_name", "")
                    ttype = data.get("type", "trigger")
                    events = data.get("events") or []
                    logger.info("[bridge ← brain] Trigger '%s' (%s) gefeuert, reply=%d chars, events=%d",
                                trigger_name, ttype, len(reply), len(events))
                    # Async-spawn — HTTP-Antwort nicht durch RVS-Broadcast blockieren
                    asyncio.create_task(
                        self._handle_trigger_fired(reply, trigger_name, ttype, events)
                    )
                    await _send_response(writer, 200, {"ok": True})
                elif method == "POST" and path == "/internal/agent-activity":
                    # Vom Proxy gefeuert bei jedem Claude-Code-tool_use-Event
                    # (Bash, Read, Edit, Grep, ...). Wir spiegeln das als
                    # RVS agent_activity an App+Diagnostic damit der Gedanken-
                    # Stream live mitlaufen kann.
                    try:
                        data = json.loads(body.decode("utf-8", "ignore"))
                    except Exception as exc:
                        await _send_response(writer, 400, {"error": f"bad json: {exc}"})
                        return
                    tool = (data.get("tool") or "").strip()
                    if not tool:
                        await _send_response(writer, 400, {"error": "tool erforderlich"})
                        return
                    tool_pid = str(data.get("projectId") or "")
                    # Force-emit (kein Dedup): User soll JEDEN Tool-Call sehen
                    # selbst wenn derselbe Name zweimal in Folge kommt.
                    asyncio.create_task(self._emit_activity("tool", tool, force=True,
                                                            project_id=tool_pid))
                    await _send_response(writer, 200, {"ok": True})
                elif method == "POST" and path == "/internal/agent-stream":
                    # Vom Proxy gefeuert: voller Live-Stream der Claude-Code-
                    # Session (assistant_text, tool_use mit Input, tool_result
                    # mit truncated Output, start/end Markers). Wir leiten 1:1
                    # als RVS agent_stream an Diagnostic (ARIA-Live-View) und
                    # App weiter — read-only Mirror der gerade laufenden
                    # ARIA-Aktivitaet.
                    try:
                        data = json.loads(body.decode("utf-8", "ignore"))
                    except Exception as exc:
                        await _send_response(writer, 400, {"error": f"bad json: {exc}"})
                        return
                    asyncio.create_task(self._send_to_rvs({
                        "type": "agent_stream",
                        "payload": data,
                        "timestamp": int(time.time() * 1000),
                    }))
                    await _send_response(writer, 200, {"ok": True})
                elif method == "POST" and path == "/internal/flux-generate":
                    # Vom Brain (flux_generate-Tool) gefeuert. Wir routen den
                    # Render-Request via RVS an die flux-bridge (Gamebox),
                    # warten synchron auf die PNG-Antwort, speichern sie nach
                    # /shared/uploads/ und melden Pfad + Render-Stats zurueck.
                    # Brain referenziert das Bild dann mit [FILE:]-Marker in
                    # seiner Antwort, die Bridge broadcastet daraufhin
                    # automatisch ein file_from_aria-Event an App+Diagnostic.
                    try:
                        data = json.loads(body.decode("utf-8", "ignore"))
                    except Exception as exc:
                        await _send_response(writer, 400, {"error": f"bad json: {exc}"})
                        return
                    prompt = (data.get("prompt") or "").strip()
                    if not prompt:
                        await _send_response(writer, 400, {"error": "prompt erforderlich"})
                        return
                    try:
                        width = int(data.get("width") or 1024)
                        height = int(data.get("height") or 1024)
                    except (TypeError, ValueError):
                        width, height = 1024, 1024
                    steps_raw = data.get("steps")
                    guidance_raw = data.get("guidance_scale")
                    seed_raw = data.get("seed")
                    steps = int(steps_raw) if isinstance(steps_raw, (int, float)) else None
                    guidance = float(guidance_raw) if isinstance(guidance_raw, (int, float)) else None
                    seed = int(seed_raw) if isinstance(seed_raw, (int, float)) else None
                    model_raw = data.get("model")
                    model = model_raw.strip() if isinstance(model_raw, str) and model_raw.strip() in ("dev", "schnell") else None

                    result = await self._flux_generate(
                        prompt=prompt, width=width, height=height,
                        steps=steps, guidance=guidance, seed=seed, model=model,
                    )
                    status = 200 if result.get("ok") else 502
                    await _send_response(writer, status, result)
                elif method == "POST" and path == "/internal/local-llm":
                    # Vom Brain (Router / Testchat) gefeuert. Wir relayen den
                    # Chat-Request via RVS an den llm-adapter (Gamebox Qwen3),
                    # warten synchron auf llm_response und geben content zurueck.
                    try:
                        data = json.loads(body.decode("utf-8", "ignore"))
                    except Exception as exc:
                        await _send_response(writer, 400, {"error": f"bad json: {exc}"})
                        return
                    messages = data.get("messages")
                    if not isinstance(messages, list) or not messages:
                        await _send_response(writer, 400, {"error": "messages (nicht-leere Liste) erforderlich"})
                        return
                    try:
                        max_tokens = int(data.get("max_tokens") or 512)
                    except (TypeError, ValueError):
                        max_tokens = 512
                    try:
                        temperature = float(data.get("temperature"))
                    except (TypeError, ValueError):
                        temperature = 0.7
                    result = await self._local_llm(
                        messages=messages, max_tokens=max_tokens,
                        temperature=temperature, stop=data.get("stop"),
                    )
                    status = 200 if result.get("ok") else 502
                    await _send_response(writer, status, result)
                elif method == "POST" and path == "/internal/delete-chat-message":
                    try:
                        data = json.loads(body.decode("utf-8", "ignore"))
                    except Exception as exc:
                        await _send_response(writer, 400, {"error": f"bad json: {exc}"})
                        return
                    ts = data.get("ts")
                    if not isinstance(ts, (int, float)):
                        await _send_response(writer, 400, {"error": "ts (number) erforderlich"})
                        return
                    result = await self._delete_chat_message(int(ts))
                    if result.get("ok"):
                        await _send_response(writer, 200, result)
                    else:
                        await _send_response(writer, 404, result)
                elif method == "GET" and path == "/health":
                    await _send_response(writer, 200, {"ok": True, "service": "bridge-internal"})
                else:
                    await _send_response(writer, 404, {"error": "not found"})
            except asyncio.TimeoutError:
                logger.warning("[bridge http] Timeout beim Request-Lesen")
            except Exception as exc:
                logger.exception("[bridge http] Fehler: %s", exc)
                try:
                    await _send_response(writer, 500, {"error": str(exc)[:200]})
                except Exception:
                    pass
            finally:
                try:
                    writer.close()
                    await writer.wait_closed()
                except Exception:
                    pass

        try:
            server = await asyncio.start_server(handle, host, port)
            logger.info("[bridge] Internal HTTP-Listener auf %s:%d (Brain-Push)", host, port)
            async with server:
                await server.serve_forever()
        except Exception:
            logger.exception("[bridge] Internal HTTP-Listener konnte nicht starten")

    async def _delete_chat_message(self, ts: int) -> dict:
        """Entfernt eine Bubble: aus chat_backup.jsonl + Brain conversation,
        broadcastet chat_message_deleted via RVS.
        Returns {ok, role, content_preview} oder {ok:False, error}.
        """
        path = Path("/shared/config/chat_backup.jsonl")
        if not path.exists():
            return {"ok": False, "error": "chat_backup.jsonl existiert nicht"}

        try:
            lines = path.read_text(encoding="utf-8").splitlines()
        except Exception as exc:
            return {"ok": False, "error": f"Lesen fehlgeschlagen: {exc}"}

        kept: list[str] = []
        removed_entry: Optional[dict] = None
        for raw in lines:
            raw = raw.strip()
            if not raw:
                continue
            try:
                obj = json.loads(raw)
            except Exception:
                kept.append(raw)
                continue
            if obj.get("ts") == ts and removed_entry is None:
                removed_entry = obj
                continue
            kept.append(raw)

        if removed_entry is None:
            return {"ok": False, "error": f"Kein Eintrag mit ts={ts} gefunden"}

        # chat_backup.jsonl neu schreiben (atomar via tmp)
        try:
            tmp = path.with_suffix(".jsonl.tmp")
            tmp.write_text("\n".join(kept) + ("\n" if kept else ""), encoding="utf-8")
            tmp.replace(path)
        except Exception as exc:
            return {"ok": False, "error": f"Schreiben fehlgeschlagen: {exc}"}

        role = removed_entry.get("role", "")
        content = removed_entry.get("text", "")
        logger.info("[chat-del] chat_backup ts=%s role=%s content[:40]=%r entfernt",
                    ts, role, content[:40])

        # Brain conversation.jsonl auch entrümpeln (best-effort).
        # ts in chat_backup ist asyncio-loop-time-ms, im Brain ist's eine ISO-UTC-Time.
        # Die kann man nicht direkt mappen — wir uebergeben nur role+content
        # und hoffen dass das eindeutig matched. Bei mehrfach gleichem content
        # entfernt remove_by_match den juengsten passenden Turn.
        if role in ("user", "assistant") and content:
            try:
                brain_url = os.environ.get("BRAIN_URL", "http://aria-brain:8080")
                payload = json.dumps({"role": role, "content": content}).encode("utf-8")
                def _post():
                    req = urllib.request.Request(
                        f"{brain_url}/conversation/delete-turn",
                        data=payload, method="POST",
                        headers={"Content-Type": "application/json"},
                    )
                    try:
                        with urllib.request.urlopen(req, timeout=10) as r:
                            return r.status
                    except urllib.error.HTTPError as e:
                        return e.code
                    except Exception:
                        return None
                status = await asyncio.get_event_loop().run_in_executor(None, _post)
                logger.info("[chat-del] Brain conversation/delete-turn → %s", status)
            except Exception as exc:
                logger.warning("[chat-del] Brain-Call fehlgeschlagen: %s", exc)

        # RVS-Broadcast damit alle Clients die Bubble entfernen
        try:
            await self._send_to_rvs({
                "type": "chat_message_deleted",
                "payload": {"ts": ts, "role": role},
                "timestamp": int(asyncio.get_event_loop().time() * 1000),
            })
        except Exception as exc:
            logger.warning("[chat-del] RVS-Broadcast fehlgeschlagen: %s", exc)

        return {"ok": True, "role": role, "content_preview": content[:80]}

    async def _handle_trigger_fired(self, reply: str, trigger_name: str,
                                    ttype: str, events: list) -> None:
        """Spiegelt eine Brain-Trigger-Antwort wie eine normale ARIA-Antwort.

        Side-Channel-Events zuerst (trigger_created, location_tracking, ...),
        dann _process_core_response (Chat-Bubble, TTS, chat_backup).
        """
        # Side-Channel-Events erst (gleich wie in send_to_core)
        for event in events or []:
            etype = event.get("type")
            try:
                if etype == "skill_created":
                    await self._send_to_rvs({
                        "type": "skill_created",
                        "payload": event.get("skill", {}),
                        "timestamp": int(asyncio.get_event_loop().time() * 1000),
                    })
                elif etype == "trigger_created":
                    await self._send_to_rvs({
                        "type": "trigger_created",
                        "payload": event.get("trigger", {}),
                        "timestamp": int(asyncio.get_event_loop().time() * 1000),
                    })
                elif etype == "location_tracking":
                    await self._send_to_rvs({
                        "type": "location_tracking",
                        "payload": {
                            "on": bool(event.get("on")),
                            "reason": event.get("reason") or "",
                        },
                        "timestamp": int(asyncio.get_event_loop().time() * 1000),
                    })
                elif etype == "memory_saved":
                    mem = event.get("memory", {})
                    if event.get("action"):
                        mem = {**mem, "action": event.get("action")}
                    await self._send_to_rvs({
                        "type": "memory_saved",
                        "payload": mem,
                        "timestamp": int(asyncio.get_event_loop().time() * 1000),
                    })
            except Exception:
                logger.exception("[trigger-fire] Side-Channel-Event %s fehlgeschlagen", etype)

        if not reply:
            logger.info("[trigger-fire] Trigger '%s' hat leeren Reply — nichts zu broadcasten",
                        trigger_name)
            return

        # Reply wie eine normale ARIA-Antwort behandeln
        try:
            await self._process_core_response(
                reply,
                {"metadata": {"trigger_name": trigger_name, "trigger_type": ttype}},
            )
        except Exception:
            logger.exception("[trigger-fire] _process_core_response fehlgeschlagen")

    # ── Run & Shutdown ───────────────────────────────────────

    async def run(self) -> None:
        """Startet die Bridge mit allen Verbindungen parallel.

        Ohne Audio-Geraet laeuft nur core + rvs (reiner Relay-Modus).
        """
        self.running = True

        tasks = [
            # connect_to_core entfaellt — Bridge ruft jetzt aria-brain ueber
            # HTTP (siehe send_to_core). Keine persistente WS-Verbindung mehr.
            asyncio.create_task(self.connect_to_rvs()),
            # Interner HTTP-Listener — empfaengt Trigger-Feuer-Pushes vom Brain.
            asyncio.create_task(self._serve_internal_http()),
        ]

        if self.audio_available:
            tasks.append(asyncio.create_task(self.audio_loop()))
        else:
            logger.info("Audio-Loop deaktiviert — kein Audio-Geraet")

        try:
            await asyncio.gather(*tasks)
        except asyncio.CancelledError:
            logger.info("Bridge-Tasks abgebrochen")

    def shutdown(self) -> None:
        """Faehrt die Bridge sauber herunter."""
        logger.info("Bridge wird heruntergefahren...")
        self.running = False


# ── File-Based Liveness Watchdog ─────────────────────────────
#
# Separater OS-Thread (NICHT asyncio) — schreibt periodisch eine
# Liveness-Datei mit aktuellem Timestamp und prüft ob der asyncio-Loop
# noch lebt. Wenn ueber LIVENESS_SELFKILL_SEC keine erfolgreiche Heart-
# beat-Bestätigung vom RVS kam, killt der Watchdog den ganzen Prozess
# (os._exit). Docker restart-Policy startet neu. Last-Resort fuer den
# Fall dass weder TCP-Keepalive noch der asyncio-Heartbeat-Watchdog
# greifen — z.B. wenn der event loop selbst korrumpiert ist.

LIVENESS_FILE = Path("/shared/health/bridge_alive")
LIVENESS_CHECK_INTERVAL_SEC = 15
LIVENESS_SELFKILL_SEC = 180  # 3 min — alle anderen Watchdogs (TCP-Keepalive
                              # ~1 min, asyncio-Watchdog 60s) sollten vorher
                              # greifen. Wenn nicht, ist der Prozess wirklich
                              # kaputt.


def _liveness_watchdog(bridge: "ARIABridge") -> None:
    try:
        LIVENESS_FILE.parent.mkdir(parents=True, exist_ok=True)
    except Exception:
        pass
    while True:
        time.sleep(LIVENESS_CHECK_INTERVAL_SEC)
        # 1) Timestamp schreiben — externe Watcher koennen das pollen
        try:
            LIVENESS_FILE.write_text(str(int(time.time())))
        except Exception:
            pass
        # 2) Letzten heartbeat checken (wird vom asyncio-Loop gesetzt). Wenn
        # zu lange stale → Self-Kill. Docker-restart-Policy uebernimmt.
        last_ok = getattr(bridge, "_last_heartbeat_ok", None)
        if last_ok is None:
            continue  # noch keine RVS-Verbindung gewesen, fair, kein Kill
        stale = time.monotonic() - last_ok
        if stale > LIVENESS_SELFKILL_SEC:
            sys.stderr.write(
                f"[liveness] heartbeat {int(stale)}s stale — Self-Kill "
                f"(Docker restart_policy uebernimmt)\n"
            )
            sys.stderr.flush()
            os._exit(1)


# ── Hauptprogramm ────────────────────────────────────────────


def main() -> None:
    """Startet die ARIA Voice Bridge."""
    bridge = ARIABridge()

    # Signal-Handler fuer sauberes Herunterfahren
    def handle_signal(signum: int, _frame: object) -> None:
        sig_name = signal.Signals(signum).name
        logger.info("Signal %s empfangen — fahre herunter", sig_name)
        bridge.shutdown()

    signal.signal(signal.SIGTERM, handle_signal)
    signal.signal(signal.SIGINT, handle_signal)

    # Initialisierung (synchron — bevor die Event-Loop startet)
    try:
        bridge.initialize()
    except Exception:
        logger.exception("Initialisierung fehlgeschlagen")
        sys.exit(1)

    # Liveness-Watchdog als daemon-Thread starten (immune gegen asyncio-Hangs)
    threading.Thread(target=_liveness_watchdog, args=(bridge,),
                     daemon=True, name="liveness-watchdog").start()
    logger.info("[liveness] Watchdog-Thread gestartet (selfkill nach %ds Heartbeat-Staleness)",
                LIVENESS_SELFKILL_SEC)

    # Event-Loop starten
    try:
        asyncio.run(bridge.run())
    except KeyboardInterrupt:
        logger.info("Keyboard Interrupt — Bridge beendet")
    finally:
        logger.info("ARIA Voice Bridge beendet")


if __name__ == "__main__":
    main()
