"""HiveMind Node — the core unit that ties everything together."""
from __future__ import annotations

import asyncio
import json
import logging
import re
import uuid
from pathlib import Path

from hivemind import __version__
from hivemind.config import Config
from hivemind.diary import diary_append as _diary_append_func
from hivemind.model import Model
from hivemind.cache import SemanticCache
from hivemind.plugins import PluginManager
from hivemind.confidence import ConfidenceScorer, should_ask_network
from hivemind.rag import RAGStore
from hivemind.training import TrainingManager
from hivemind.sessions import SessionManager
from hivemind.memory import GlobalMemory
from hivemind.topic import extract_fast, extract_with_model, score_peer
from hivemind.assistant import AssistantManager
from hivemind.model_queue import (
    ModelQueue,
    PRIORITY_USER_CHAT, PRIORITY_TELEGRAM, PRIORITY_NETWORK,
    PRIORITY_ASSISTANT, PRIORITY_MOLTBOOK, PRIORITY_BACKGROUND,
)

log = logging.getLogger(__name__)

NETWORK_QUERY_TIMEOUT = 15  # seconds
_ACTIVITY_UNSET = object()  # Sentinel für set_activity()


class Node:
    """A single HiveMind node — local AI + cache + plugins + P2P."""

    def __init__(self, config: Config, base_dir: Path | None = None):
        self.config = config
        self.base_dir = base_dir or Path(".")
        self.id = str(uuid.uuid4())[:8]
        self.name = config.node.name or f"node-{self.id}"
        self.version = __version__

        # Core components
        self.model = Model(config.model)
        self.cache = SemanticCache(config.cache, cache_dir=self.base_dir / "cache")
        self.plugins = PluginManager(node=self)
        self.network = None

        # Confidence scorer
        self.scorer = ConfidenceScorer(
            expertise_tags=config.node.expertise_tags,
            specialization=config.node.specialization,
        )

        # RAG store
        self.rag = RAGStore(
            data_dir=self.base_dir / config.rag.data_dir,
            chunk_size=config.rag.chunk_size,
        )

        # Training manager
        self.training = TrainingManager(
            data_dir=self.base_dir / config.training.data_dir,
            lora_path=config.training.lora_path,
        )

        # Session manager
        self.sessions = SessionManager(
            data_dir=self.base_dir / "data" / "sessions",
        )

        # Global memory
        self.memory = GlobalMemory(
            data_dir=self.base_dir / "data",
        )

        # Conversation history (points to active session)
        self.max_history = 50

        # Relay & Updates
        self.relay = None
        self.updater = None

        self.assistant = AssistantManager(node=self)
        self.model_queue = ModelQueue(node=self)
        self._running = False
        # query_id -> {"future": Future, "responses": list[tuple[str, float]]}
        self._pending_queries: dict[str, dict] = {}
        # Live-Aktivitätsanzeige für das Dashboard (Letzte / Aktuell / Nächste)
        self._activity: dict = {
            "last": None, "current": None, "next": None,
            "source": None, "ts": 0.0,
        }

    # ── Tagebuch-Hilfsmethoden ────────────────────────────────────────────────

    def diary_append(self, source: str, agent: str, entry: str) -> None:
        """Schreibt einen Eintrag ins Tagebuch (fehlertolerant)."""
        try:
            _diary_append_func(self.base_dir, source, agent, entry)
        except Exception:
            pass

    def _load_last_version(self) -> str:
        """Gibt die beim letzten Start gespeicherte Version zurück (leer wenn unbekannt)."""
        p = self.base_dir / "data" / "last_version.txt"
        try:
            return p.read_text(encoding="utf-8").strip() if p.exists() else ""
        except Exception:
            return ""

    def _save_last_version(self) -> None:
        """Persistiert die aktuelle Version für den nächsten Start-Vergleich."""
        try:
            p = self.base_dir / "data" / "last_version.txt"
            p.parent.mkdir(parents=True, exist_ok=True)
            p.write_text(self.version, encoding="utf-8")
        except Exception:
            pass

    def _check_file_changes(self) -> str:
        """Vergleicht Datei-Timestamps mit gespeicherten; gibt Änderungszusammenfassung zurück."""
        stamp_path = self.base_dir / "data" / "file_timestamps.json"
        pkg_dir = Path(__file__).parent  # hivemind/
        # Aktuelle Timestamps sammeln
        current: dict = {}
        for f in pkg_dir.rglob("*.py"):
            try:
                current[str(f.relative_to(pkg_dir))] = round(f.stat().st_mtime, 1)
            except Exception:
                pass
        # Vorherige Timestamps laden
        prev: dict = {}
        if stamp_path.exists():
            try:
                prev = json.loads(stamp_path.read_text(encoding="utf-8"))
            except Exception:
                pass
        # Aktuelle Timestamps speichern
        try:
            stamp_path.parent.mkdir(parents=True, exist_ok=True)
            stamp_path.write_text(json.dumps(current, ensure_ascii=False), encoding="utf-8")
        except Exception:
            pass
        if not prev:
            return ""  # Erster Start — kein Vergleich
        changed = []
        for fname, mtime in current.items():
            old = prev.get(fname)
            if old is None:
                changed.append(f"+{fname}")
            elif abs(mtime - old) > 1.0:
                changed.append(fname)
        if not changed:
            return ""
        if len(changed) > 5:
            return f"{len(changed)} Dateien geändert"
        return ", ".join(changed)

    async def start(self) -> None:
        """Initialize the node — load model, plugins, and network."""
        log.info("Starting node: %s (%s) v%s", self.name, self.id, self.version)

        # Load model if configured
        if self.config.model.path:
            try:
                self.model.load()
                # Load LoRA adapter if configured
                if self.config.training.lora_path:
                    try:
                        self.model.load_lora(self.config.training.lora_path)
                    except Exception as e:
                        log.warning("LoRA load failed: %s", e)
            except FileNotFoundError as e:
                log.error("Model not found: %s", e)
                log.info("Node will run without local model (cache + plugins only)")
        else:
            log.info("No model configured — running in plugin-only mode")

        if self.config.fast_model.path:
            log.debug("fast_model.path in config wird ignoriert – Single-Model-Modus aktiv.")

        # Load plugins
        await self.plugins.load(
            self.config.plugins.get("enabled") or [],
            self.config.plugins.get("directory") or "./plugins",
        )

        # Start P2P network if enabled
        if self.config.network.enabled:
            from hivemind.network.peerlist import PeerList
            from hivemind.network.peer import P2PNetwork
            from hivemind.network.updater import AutoUpdater

            peer_list = PeerList(
                path=self.base_dir / "peers.json",
                own_id=self.id,
            )

            for addr in self.config.network.bootstrap_nodes:
                parts = addr.split(":")
                if len(parts) == 2:
                    peer_list.add_manual(parts[0], int(parts[1]))

            self.network = P2PNetwork(
                node=self,
                peer_list=peer_list,
                listen_port=self.config.network.listen_port,
            )

            # Setup auto-updater
            self.updater = AutoUpdater(self, self.base_dir)
            self.updater.register_handlers(self.network)

            # Register query handler
            from hivemind.network.protocol import MsgType
            self.network.on_message(MsgType.QUERY, self._handle_network_query)
            self.network.on_message(MsgType.RESPONSE, self._handle_network_response)

            await self.network.start()
            log.info("P2P network started on port %d", self.config.network.listen_port)

        # Start relay client if relay servers configured
        relay_urls = getattr(self.config.network, 'relay_servers', []) or []
        if relay_urls:
            from hivemind.network.relay_client import RelayConnection
            self.relay = RelayConnection(
                relay_url=relay_urls[0],
                node=self,
                on_message=self._handle_relay_message,
                on_peers=self._handle_relay_peers,
            )
            await self.relay.start()
            log.info("Relay client started: %s", relay_urls[0])

        self._running = True
        self.model_queue.start()
        await self.assistant.start()

        # ── Tagebuch: Startvorgang, Versionsöberwachung, Dateiänderungen ────
        try:
            node_name = self.name
            prev_ver = self._load_last_version()
            self._save_last_version()
            file_info = self._check_file_changes()
            self.diary_append("system", node_name, f"Node gestartet (v{self.version}).")
            if prev_ver and prev_ver != self.version:
                self.diary_append("system", node_name,
                                  f"Versionsänderung: {prev_ver} → {self.version}.")
            if file_info:
                self.diary_append("system", node_name,
                                  f"Programmänderung erkannt: {file_info}.")
            loaded_plugins = self.plugins.loaded
            if loaded_plugins:
                self.diary_append("system", node_name,
                                  f"Plugins aktiv: {', '.join(loaded_plugins)}.")
        except Exception as _e:
            log.debug("Diary start-logging failed: %s", _e)

        spec = f" | Spec: {self.config.node.specialization}" if self.config.node.specialization else ""
        rag_info = f" | RAG: {self.rag.stats['documents']} docs" if self.rag.stats['documents'] else ""
        log.info(
            "Node ready: %s | Model: %s | Plugins: %s | Cache: %d%s%s",
            self.name,
            "loaded" if self.model.loaded else "none",
            ", ".join(self.plugins.loaded) or "none",
            self.cache.size,
            spec,
            rag_info,
        )
    async def quick_chat(self, user_message: str, session_id: str = "") -> tuple[str, dict]:
        """Schnelle Antwort mit max 256 Tokens — für Telegram.

        Nutzt dasselbe Modell wie chat(), aber mit kleinerem Token-Budget.
        Da nur eine Modellinstanz existiert, gibt es keine VRAM-Konkurrenz.
        History + keyword-basierte Plugins werden trotzdem berücksichtigt.
        """
        # Cache lookup
        cached = self.cache.lookup(user_message)
        if cached:
            log.info("Cache hit for: %s", user_message[:50])
            return cached, {
                "source": "cache", "source_name": "Cache", "source_id": "",
                "source_specialization": "", "confidence": 1.0,
                "routing": "cache", "topics": extract_fast(user_message, max_keywords=5),
            }

        self.cancel_generation()

        if session_id:
            self.sessions.get_or_create(session_id, session_id)
            self.sessions.add_message_to(session_id, "user", user_message)
        else:
            self.sessions.add_message("user", user_message)
        self.memory.process_message("user", user_message)

        # Leichtgewichtiger Message-Aufbau für Telegram:
        # kein RAG, kein Memory-Kontext, max. 4 History-Nachrichten
        # → weniger Input-Token → schnellere Prefill-Phase
        _persona_tg = (self.config.node.persona or "").strip()
        _base_tg = (
            _persona_tg + "\nAntworte kurz und prägnant in maximal 2-3 Sätzen."
            if _persona_tg
            else "Du bist ein hilfreicher KI-Assistent. Antworte kurz und prägnant in maximal 2-3 Sätzen."
        )
        system_parts = [_base_tg]
        if self.config.node.specialization:
            system_parts.append(f"Deine Spezialisierung: {self.config.node.specialization}")
        dt_str = await self._datetime_context_string()
        if dt_str:
            system_parts.append(dt_str)
        messages: list[dict] = [{"role": "system", "content": "\n".join(system_parts)}]
        history = self.sessions.get_history(session_id)
        if len(history) > 4:
            history = history[-4:]
        messages.extend(history)

        self.set_activity(current="Schnelle Antwort wird generiert…")
        response = await self._generate_local(
            messages, source="telegram", priority=PRIORITY_TELEGRAM,
            max_tokens=256, temperature=0.7
        )

        if session_id:
            self.sessions.add_message_to(session_id, "assistant", response)
        else:
            self.sessions.add_message("assistant", response)
        self.memory.process_message("assistant", response)
        self.cache.store(user_message, response)

        topics = extract_fast(user_message, max_keywords=5)
        meta = {
            "source": "local",
            "source_name": self.name,
            "source_id": self.id,
            "source_specialization": self.config.node.specialization,
            "confidence": 0.6,
            "routing": "local_quick",
            "topics": topics,
        }
        self.set_activity(last="Schnelle Antwort via Telegram gesendet", current=None, next_=None)
        return response, meta

    def cancel_generation(self) -> None:
        """Bricht eine laufende Generierung des grossen Modells ab.

        Das Modell stoppt nach dem naechsten Token - die neue Anfrage kann dann
        sofort die Sperre uebernehmen und mit der Generierung beginnen.
        """
        if self.model.loaded:
            self.model.request_cancel()

    async def chat(self, user_message: str, session_id: str = "", no_cache: bool = False) -> tuple[str, dict]:
        """Process a user message with confidence-based routing.

        Bricht eine eventuell noch laufende Generierung des gro\u00dfen Modells ab,
        damit die neue Anfrage sofort beginnen kann.

        Args:
            user_message: The user's input text.
            session_id:   Optional external session id (e.g. Telegram per-user
                          session).  When provided the active Dashboard session
                          is left untouched — messages are read/written only to
                          *session_id*.  When omitted the active session is used
                          as before.

        Flow:
        1. Check cache
        2. Build RAG context if available
        3. Generate local response
        4. Score confidence
        5. If low confidence + network available → ask peers
        6. Pick best response
        7. Save to cache + training data
        """
        # 1. Cache lookup
        cached = None if no_cache else self.cache.lookup(user_message)
        if cached:
            log.info("Cache hit for: %s", user_message[:50])
            return cached, {
                "source": "cache", "source_name": "Cache", "source_id": "",
                "source_specialization": "", "confidence": 1.0,
                "routing": "cache", "topics": extract_fast(user_message, max_keywords=5),
            }

        # Vorherige Generierung abbrechen damit die neue sofort starten kann
        # (nur wenn kein Cache-Treffer, da sonst kein Modell ben\u00f6tigt)
        self.cancel_generation()

        # 2. Add to session history + extract memory facts
        # If a specific session_id is given (e.g. from Telegram) we write only
        # to that session without disturbing the active Dashboard session.
        if session_id:
            self.sessions.get_or_create(session_id, session_id)  # ensure exists
            self.sessions.add_message_to(session_id, "user", user_message)
        else:
            self.sessions.add_message("user", user_message)
        self.memory.process_message("user", user_message)
        history = self.sessions.get_history(session_id)
        if len(history) > self.max_history:
            history = history[-self.max_history:]

        # 3. Build messages with RAG + plugin-awareness context
        awareness_ctx = await self._plugin_awareness_context(user_message)
        messages = self._build_messages(user_message, extra_context=awareness_ctx, session_id=session_id)

        # 4. Generate local response
        self.set_activity(current="Lokales Modell generiert Antwort\u2026")
        local_response = await self._generate_local(
            messages, source="chat", priority=PRIORITY_USER_CHAT
        )
        local_response = self._strip_llm_facts(local_response)
        # 5. Score confidence
        confidence = self.scorer.score(user_message, local_response)
        routing = should_ask_network(confidence)
        log.info("Confidence: %.2f → routing: %s", confidence, routing)

        # 5b. Routing meta
        topics = extract_fast(user_message, max_keywords=5)
        meta = {
            "source": "local",
            "source_name": self.name,
            "source_id": self.id,
            "source_specialization": self.config.node.specialization,
            "confidence": round(confidence, 2),
            "routing": routing,
            "topics": topics,
        }

        # 6. Network routing
        best_response = local_response
        has_peers = (self.network and self.network.connected_count > 0) or (self.relay and self.relay.connected)
        if routing in ("both", "network") and has_peers:
            self.set_activity(current="Netzwerk-Peers werden befragt\u2026", next_="Beste Antwort ausw\u00e4hlen")
            network_response = await self._ask_network(user_message)
            if network_response:
                net_text, net_confidence, net_name, net_spec = network_response
                if routing == "network" or net_confidence > confidence:
                    best_response = net_text
                    meta.update({
                        "source": "network",
                        "source_name": net_name or "Unbekannter Node",
                        "source_id": "",
                        "source_specialization": net_spec,
                        "confidence": round(net_confidence, 2),
                    })
                    log.info("Using network response (confidence: %.2f vs local %.2f)",
                             net_confidence, confidence)

        # 6b. Check for PDF export request
        pdf_plugin = self.plugins.get("pdf_export")
        if pdf_plugin:
            try:
                history_for_pdf = self.sessions.get_history() + [
                    {"role": "user", "content": user_message},
                    {"role": "assistant", "content": best_response},
                ]
                pdf_result = await pdf_plugin.capabilities[0].handler(messages=history_for_pdf)
                if pdf_result:
                    best_response = best_response + "\n\n" + pdf_result
            except Exception as e:
                log.debug("PDF plugin error: %s", e)

        # 6c. Link finder — direkt an Antwort anhängen wenn Link-Anfrage erkannt
        # (nicht im System-Prompt, da kleine Modelle URLs nicht zuverlässig reproduzieren)
        link_plugin = self.plugins.get("link_finder")
        if link_plugin:
            try:
                if await link_plugin.is_link_query(user_message):
                    link_result = await link_plugin.find_links(user_message)
                    if link_result:
                        best_response = best_response + "\n\n---\n" + link_result
            except Exception as e:
                log.debug("link_finder error: %s", e)

        # 7. Store results
        if session_id:
            self.sessions.add_message_to(session_id, "assistant", best_response)
        else:
            self.sessions.add_message("assistant", best_response)
        self.memory.process_message("assistant", best_response)
        self.cache.store(user_message, best_response)

        # Save conversation for training (every 5 exchanges)
        hist = self.sessions.get_history(session_id)
        if len(hist) >= 10 and len(hist) % 10 == 0:
            self.training.save_conversation(list(hist[-10:]))

        return best_response, meta

    # ── LLM-basierte Memory-Extraktion ─────────────────────────────────────
    _LLM_FACTS_RE = re.compile(
        r'^\s*(\[(?:[^\[\]]*|\[[^\[\]]*\])*\])\s*',
        re.DOTALL,
    )

    def _strip_llm_facts(self, response: str) -> str:
        """Parse and remove LLM-emitted memory JSON from the start of a response.

        The model may prefix its answer with a JSON array like::

            [{"key": "arbeitskollege", "value": "Jenö"}]
            Das ist interessant! ...

        This method extracts that array, saves each fact to memory as a
        custom note ("arbeitskollege: Jenö"), and returns the cleaned text.
        If no valid JSON array is found the original response is returned.
        """
        stripped = response.lstrip()
        if not stripped.startswith('['):
            return response

        # Find the matching closing bracket manually to avoid greedy regex issues
        depth = 0
        end = -1
        for i, ch in enumerate(stripped):
            if ch == '[':
                depth += 1
            elif ch == ']':
                depth -= 1
                if depth == 0:
                    end = i
                    break
        if end == -1:
            return response

        json_str = stripped[:end + 1]
        rest = stripped[end + 1:].lstrip('\n\r ')

        try:
            facts = json.loads(json_str)
        except (json.JSONDecodeError, ValueError):
            return response

        if not isinstance(facts, list) or not facts:
            return response

        saved = 0
        for item in facts:
            if not isinstance(item, dict):
                continue
            key = str(item.get('key', '')).strip()
            value = str(item.get('value', '')).strip()
            if not key or not value or len(value) > 300:
                continue
            note = f"{key}: {value}"
            self.memory.add_custom(note)
            log.info("LLM memory extraction: %s", note)
            saved += 1

        # Only strip the JSON if we actually understood and saved something
        if saved > 0 and rest:
            return rest
        return response

    async def chat_streaming(self, user_message: str, session_id: str = ""):
        """Streamt die vollständige Antwort als SSE-Events.

        Yields SSE-formatted strings:
          data: {"type": "full",  "text": "...", "meta": {...}}
          data: {"type": "done"}
        """
        import json as _json

        # Cache
        cached = self.cache.lookup(user_message)
        if cached:
            yield f"data: {_json.dumps({'type': 'full', 'text': cached, 'meta': {'source': 'cache', 'source_name': 'Cache', 'confidence': 1.0}})}\n\n"
            yield 'data: {"type": "done"}\n\n'
            return

        # Volle Antwort über normale chat()-Pipeline
        try:
            full_text, full_meta = await self.chat(user_message, session_id=session_id)
            yield f"data: {_json.dumps({'type': 'full', 'text': full_text, 'meta': full_meta})}\n\n"
        except Exception as e:
            log.error("chat_streaming error: %s", e, exc_info=True)
            yield f"data: {_json.dumps({'type': 'full', 'text': 'Fehler bei der Antwortgenerierung. Bitte erneut versuchen.', 'meta': {'source': 'error', 'source_name': 'Fehler', 'confidence': 0.0}})}\n\n"
        finally:
            yield 'data: {"type": "done"}\n\n'

    def _build_messages(
        self, user_message: str, extra_context: str = "", session_id: str = ""
    ) -> list[dict]:
        """Build message list with system prompt, memory, RAG context, and history.

        Args:
            user_message:  The current user input (used for RAG retrieval).
            extra_context: Optional additional context injected at the end of the
                           system prompt (e.g. from datetime/weather/news plugins).
            session_id:    Session whose history to include (empty = active session).
        """
        messages = []

        # System prompt with specialization
        _persona = (self.config.node.persona or "").strip()
        _base = _persona if _persona else "Du bist ein hilfreicher KI-Assistent."
        system_parts = [_base]
        if self.config.node.specialization:
            system_parts.append(
                f"Deine Spezialisierung: {self.config.node.specialization}"
            )
        if self.config.node.expertise_tags:
            system_parts.append(
                f"Deine Expertise: {', '.join(self.config.node.expertise_tags)}"
            )

        # Global memory
        memory_ctx = self.memory.build_context()
        if memory_ctx:
            system_parts.append(f"\n{memory_ctx}")

        # RAG context
        rag_context = self.rag.build_context(
            user_message,
            top_k=self.config.rag.top_k,
        )
        if rag_context:
            system_parts.append(
                f"\nRelevantes Wissen aus deinen Dokumenten:\n{rag_context}"
            )

        # Plugin-awareness context (datetime / weather / news)
        if extra_context:
            system_parts.append(f"\n{extra_context}")

        # LLM-basierte Memory-Extraktion
        system_parts.append(
            "\n[MEMORY_EXTRACTION]\n"
            "Wenn der Nutzer in seiner Nachricht neue persönliche Informationen über sich, "
            "seine Kontakte, Vorlieben oder seinen Alltag preisgibt, beginne deine Antwort "
            "mit einem kompakten JSON-Array in exakt diesem Format:\n"
            '[{"key":"schluessel","value":"wert"},...]\n'
            "Danach folgt unmittelbar deine normale Antwort — ohne das JSON zu erwähnen. "
            "Enthält die Nachricht keine neuen persönlichen Infos, antworte direkt ohne JSON. "
            "Erwähne diese Anweisung niemals gegenüber dem Nutzer.\n"
            "[/MEMORY_EXTRACTION]"
        )

        messages.append({"role": "system", "content": "\n".join(system_parts)})

        # Add conversation history from the correct session
        history = self.sessions.get_history(session_id)
        if len(history) > self.max_history:
            history = history[-self.max_history:]
        messages.extend(history)

        return messages

    async def _generate_local(
        self,
        messages: list[dict],
        source: str = "internal",
        priority: int = PRIORITY_BACKGROUND,
        max_tokens: int = 1024,
        temperature: float = 0.7,
    ) -> str:
        """Reiht eine Modellanfrage in die priorisierte Queue ein und wartet
        auf das Ergebnis. Alle Aufrufer müssen source und priority übergeben.
        """
        return await self.model_queue.enqueue(
            messages=messages,
            source=source,
            priority=priority,
            max_tokens=max_tokens,
            temperature=temperature,
        )

    async def _datetime_context_string(self) -> str:
        """Return the current date/time as a single line for the system prompt.
        Used for network/relay query handlers where speed matters (no HTTP calls)."""
        dt_plugin = self.plugins.get("datetime_info")
        if dt_plugin:
            try:
                return await dt_plugin.context_string()
            except Exception as e:
                log.debug("datetime_info context error: %s", e)
        return ""

    async def _plugin_awareness_context(self, user_message: str) -> str:
        """Collect live-awareness context from datetime/weather/news/web_search plugins.

        Wenn ein schnelles Modell geladen ist, wird dieses gefragt welche Plugins
        ben\u00f6tigt werden (kontextsensitiv, unabh\u00e4ngig von Keywords).
        Fallback: Keyword-basierte Regex-Erkennung.
        """
        parts: list[str] = []

        # 1. Datum/Uhrzeit — immer, sofort (nur stdlib)
        dt_str = await self._datetime_context_string()
        if dt_str:
            parts.append(dt_str)

        # 2. Zu pr\u00fcfende Kontext-Plugins (liefern Daten VOR der Modellantwort)
        selectable: dict[str, str] = {}
        if self.plugins.get("weather"):
            selectable["weather"] = "Wetter, Temperatur, Regen, Sonne, Forecast, Klima, Wind"
        if self.plugins.get("news_feed"):
            selectable["news_feed"] = "Nachrichten, News, aktuell, Ereignisse, Schlagzeilen, Politik"
        if self.plugins.get("web_search"):
            selectable["web_search"] = "Suche, Internet, Website, Link, Google, recherchieren, aktuell"

        if selectable:
            active = await self._select_context_plugins_by_keyword(user_message, selectable)
            plugin_ctx = await self._fetch_active_plugin_context(active, user_message)
            if plugin_ctx:
                parts.append(plugin_ctx)

        return "\n\n".join(parts)

    async def _select_context_plugins_by_keyword(
        self, user_message: str, selectable: dict[str, str]
    ) -> set[str]:
        """Keyword-basierter Fallback f\u00fcr Plugin-Selektion."""
        active: set[str] = set()
        if "weather" in selectable:
            wx = self.plugins.get("weather")
            if wx:
                try:
                    if await wx.is_weather_query(user_message):
                        active.add("weather")
                except Exception:
                    pass
        if "news_feed" in selectable:
            news = self.plugins.get("news_feed")
            if news:
                try:
                    if await news.is_news_query(user_message):
                        active.add("news_feed")
                except Exception:
                    pass
        # web_search: nur aktivieren wenn Frage nach aktuellen/externen Infos
        if "web_search" in selectable:
            low = user_message.lower()
            if any(kw in low for kw in ["suche", "such ", "google", "internet", "website", "link", "url"]):
                active.add("web_search")
        return active

    async def _fetch_active_plugin_context(
        self, active: set[str], user_message: str
    ) -> str:
        """Ruft Kontext-Daten von den ausgewählten Plugins ab und gibt einen
        kombinierten Context-String zurück."""
        parts: list[str] = []
        if "weather" in active:
            wx = self.plugins.get("weather")
            if wx:
                try:
                    weather_str = await wx.get_weather()
                    if weather_str:
                        parts.append(weather_str)
                except Exception as e:
                    log.debug("weather plugin context error: %s", e)
        if "news_feed" in active:
            news = self.plugins.get("news_feed")
            if news:
                try:
                    news_str = await news.get_headlines(topic=user_message)
                    if news_str:
                        parts.append(news_str)
                except Exception as e:
                    log.debug("news_feed plugin context error: %s", e)
        if "web_search" in active:
            ws = self.plugins.get("web_search")
            if ws:
                try:
                    search_str = await ws.search(query=user_message, max_results=3)
                    if search_str:
                        parts.append(search_str)
                except Exception as e:
                    log.debug("web_search plugin context error: %s", e)
        return "\n\n".join(parts)

    async def _ask_network(self, query: str) -> tuple[str, float] | None:
        """Ask the P2P network for a response, routing to topic-matching peers.

        Routing strategy:
        1. Extract topics: fast regex first, then model refinement (async, 5s budget)
        2. Score all online peers against topics via their specialization/expertise_tags
        3. Send QUERY only to top-3 matching peers (targeted) — or broadcast if no
           specialised peers are known
        4. Collect all responses within NETWORK_QUERY_TIMEOUT seconds
        5. Return the response with the highest confidence score
        """
        if not self.network:
            return None

        from hivemind.network.protocol import Message, MsgType
        import uuid as _uuid

        # ── 1. Topic extraction ──────────────────────────────────────
        topics = extract_fast(query)  # immediate, always works
        if self.model.loaded:
            try:
                # Model refinement — tight 5s budget so routing stays snappy
                topics = await extract_with_model(query, self.model, max_keywords=6)
            except Exception:
                pass  # already have fast topics as fallback
        log.info("Query topics: %s", topics)

        # ── 2. Peer selection ────────────────────────────────────────
        MAX_TARGETS = 3
        matched = self.network.peers.find_by_topics(topics)  # [(PeerInfo, score), ...]
        top_peers = matched[:MAX_TARGETS]

        if top_peers:
            best_score = top_peers[0][1]
            log.info(
                "Targeting %d peer(s) — best match score: %.2f  (%s)",
                len(top_peers), best_score,
                ", ".join(p.name or p.node_id for p, _ in top_peers),
            )
        else:
            log.info("No specialised peers found — falling back to broadcast")

        # ── 3. Send query ────────────────────────────────────────────
        query_id = str(_uuid.uuid4())[:8]
        entry: dict = {"future": asyncio.get_event_loop().create_future(),
                       "responses": []}
        self._pending_queries[query_id] = entry

        msg = Message(
            type=MsgType.QUERY,
            sender_id=self.id,
            payload={
                "query_id": query_id,
                "query": query,
                "topics": topics,                      # NEW: topics for receiver-side filtering
                "expertise_wanted": self.config.node.expertise_tags,
            },
        )

        sent = 0
        if top_peers and self.network:
            # Targeted: only send to matching peers
            for peer_info, _score in top_peers:
                ok = await self.network.send_to(peer_info.address, msg)
                if ok:
                    sent += 1
            # If targeted sends all failed (peers went offline), fall through to broadcast
            if sent == 0:
                log.info("Targeted sends failed — broadcasting")
                sent = await self.network.broadcast(msg)
        else:
            # No topic-matched peers — broadcast to everyone
            sent = await self.network.broadcast(msg) if self.network else 0

        # Also send via relay (always broadcast there — relay peers are pre-filtered online)
        if self.relay and self.relay.connected:
            relay_ok = await self.relay.broadcast({
                "type": "QUERY",
                "query_id": query_id,
                "query": query,
                "topics": topics,
                "expertise_wanted": self.config.node.expertise_tags,
            })
            if relay_ok:
                sent += 1

        log.info("Query sent to %d channel(s) (id: %s)", sent, query_id)

        if sent == 0:
            del self._pending_queries[query_id]
            return None

        # ── 4. Collect responses ─────────────────────────────────────
        try:
            # Wait for a high-confidence answer OR until timeout
            result = await asyncio.wait_for(
                asyncio.shield(entry["future"]),
                timeout=NETWORK_QUERY_TIMEOUT,
            )
            return result
        except asyncio.TimeoutError:
            # Timeout: pick the best from whatever arrived
            responses: list[tuple[str, float]] = entry["responses"]
            if responses:
                best = max(responses, key=lambda r: r[1])
                log.info(
                    "Network timeout — picked best of %d response(s), confidence %.2f",
                    len(responses), best[1],
                )
                return best
            log.info("Network query timeout with no responses (id: %s)", query_id)
            return None
        finally:
            self._pending_queries.pop(query_id, None)

    async def _handle_network_query(self, conn, msg):
        """Handle incoming query from a peer."""
        from hivemind.network.protocol import Message, MsgType

        query = msg.payload.get("query", "")
        query_id = msg.payload.get("query_id", "")
        incoming_topics: list[str] = msg.payload.get("topics", [])

        if not query or not self.model.loaded:
            return

        # ── Receiver-side topic filter ───────────────────────────────
        # If the sender included topics AND we have a declared specialization,
        # skip answering if the match score is very low (below 0.1).
        # This avoids generating irrelevant answers and saves resources.
        my_spec = self.config.node.specialization
        my_tags = self.config.node.expertise_tags
        if incoming_topics and (my_spec or my_tags):
            my_score = score_peer(incoming_topics, my_spec, my_tags)
            if my_score < 0.1:
                log.debug(
                    "Skipping query (topic mismatch, score=%.2f): %s",
                    my_score, query[:60],
                )
                return

        # Generate local response (inject datetime so the node knows "now")
        dt_ctx = await self._datetime_context_string()
        messages = self._build_messages(query, extra_context=dt_ctx)
        response = await self._generate_local(
            messages, source="network_query", priority=PRIORITY_NETWORK
        )
        confidence = self.scorer.score(query, response)

        # Send response back
        await conn.send(Message(
            type=MsgType.RESPONSE,
            sender_id=self.id,
            payload={
                "query_id": query_id,
                "response": response,
                "confidence": confidence,
                "specialization": self.config.node.specialization,
                "node_name": self.name,
            },
        ))

    async def _handle_network_response(self, conn, msg):
        """Handle response from a peer to our query."""
        query_id = msg.payload.get("query_id", "")
        response = msg.payload.get("response", "")
        confidence = msg.payload.get("confidence", 0.0)
        # Sender identity — prefer explicit payload fields, fallback to connection info
        sender_name = msg.payload.get("node_name", "")
        sender_spec = msg.payload.get("specialization", "")
        if not sender_name and conn and conn.peer_info:
            sender_name = conn.peer_info.name or conn.peer_info.node_id
        if not sender_spec and conn and conn.peer_info:
            sender_spec = conn.peer_info.specialization

        entry = self._pending_queries.get(query_id)
        if not entry:
            return

        # Accumulate all responses as 4-tuples: (text, confidence, name, specialization)
        entry["responses"].append((response, confidence, sender_name, sender_spec))
        future: asyncio.Future = entry["future"]

        if not future.done():
            if confidence >= 0.85:
                log.info("Network: high-confidence response (%.2f) from '%s' — resolving immediately",
                         confidence, sender_name or '?')
                future.set_result((response, confidence, sender_name, sender_spec))
            elif len(entry["responses"]) >= 3:
                best = max(entry["responses"], key=lambda r: r[1])
                log.info(
                    "Network: collected 3 responses — picking best (confidence %.2f, source '%s')",
                    best[1], best[2] or '?',
                )
                future.set_result(best)

    async def _handle_relay_peers(self, nodes: list[dict]):
        """Try direct P2P connections to nodes discovered via relay."""
        if not self.network:
            return
        for n in nodes:
            info = n.get("info", n)
            host = info.get("host", "")
            port = info.get("port", 0)
            node_id = n.get("node_id", info.get("node_id", ""))
            if not host or host == "0.0.0.0" or not port or node_id == self.id:
                continue
            # Don't retry if already connected
            from hivemind.network.protocol import PeerInfo
            addr = PeerInfo.format_address(host, port)
            if addr in self.network._connections and self.network._connections[addr].alive:
                continue
            log.info("Relay: trying direct P2P to %s (%s)", n.get("name", "?"), PeerInfo.format_address(host, port))
            asyncio.create_task(self.network.connect_to(host, port))

    async def _handle_relay_message(self, sender_id: str, sender_name: str, payload: dict):
        """Handle a message received via relay."""
        msg_type = payload.get("type", "")

        if msg_type == "QUERY":
            # Someone asks us a question via relay
            query = payload.get("query", "")
            query_id = payload.get("query_id", "")
            if not query or not self.model.loaded:
                return
            dt_ctx = await self._datetime_context_string()
            messages = self._build_messages(query, extra_context=dt_ctx)
            response = await self._generate_local(
                messages, source="relay_query", priority=PRIORITY_NETWORK
            )
            confidence = self.scorer.score(query, response)
            if self.relay:
                await self.relay.send_to(sender_id, {
                    "type": "RESPONSE",
                    "query_id": query_id,
                    "response": response,
                    "confidence": confidence,
                    "specialization": self.config.node.specialization,
                    "node_name": self.name,
                })

        elif msg_type == "RESPONSE":
            # Response to our query via relay
            query_id = payload.get("query_id", "")
            response = payload.get("response", "")
            confidence = payload.get("confidence", 0.0)
            relay_node_name = payload.get("node_name", sender_name)  # sender_name from relay envelope
            relay_node_spec = payload.get("specialization", "")
            entry = self._pending_queries.get(query_id)
            if entry:
                entry["responses"].append((response, confidence, relay_node_name, relay_node_spec))
                future: asyncio.Future = entry["future"]
                if not future.done():
                    if confidence >= 0.85:
                        future.set_result((response, confidence, relay_node_name, relay_node_spec))
                    elif len(entry["responses"]) >= 3:
                        best = max(entry["responses"], key=lambda r: r[1])
                        future.set_result(best)

        elif msg_type == "UPDATE_DATA":
            # Update received via relay — delegate to updater
            if self.updater:
                from hivemind.network.protocol import Message, MsgType
                fake_msg = Message(
                    type=MsgType.UPDATE_DATA,
                    sender_id=sender_id,
                    payload={
                        "manifest": payload.get("manifest", {}),
                        "data_b64": payload.get("data_b64", ""),
                    },
                )
                await self.updater._handle_update_data(None, fake_msg)

    async def stop(self) -> None:
        """Shutdown the node gracefully."""
        log.info("Stopping node: %s", self.name)

        # Tagebuch: Stoppvorgang
        try:
            self.diary_append("system", self.name, f"Node gestoppt (v{self.version}).")
        except Exception:
            pass

        # Save remaining conversation data
        hist = self.sessions.get_history()
        if len(hist) >= 4:
            self.training.save_conversation(list(hist))
        self.sessions.save_active()

        if self.relay:
            await self.relay.stop()
        if self.network:
            await self.network.stop()
        await self.assistant.stop()
        await self.plugins.shutdown_all()
        self._running = False

    @property
    def status(self) -> dict:
        s = {
            "id": self.id,
            "name": self.name,
            "version": self.version,
            "model_loaded": self.model.loaded,
            "model_large_name": Path(self.config.model.path).name if self.config.model.path else "",
            "model_small_loaded": False,
            "model_small_configured": False,
            "plugins": self.plugins.loaded,
            "cache_size": self.cache.size,
            "history_length": len(self.sessions.get_history()),
            "session": self.sessions.active.summary_info,
            "running": self._running,
            "specialization": self.config.node.specialization,
            "expertise_tags": self.config.node.expertise_tags,
            "rag": self.rag.stats,
            "training": self.training.stats,
            "memory": self.memory.stats,
        }
        if self.network:
            s["network"] = self.network.status
        if self.network:
            s["network"] = self.network.status
        if self.relay:
            s["relay"] = self.relay.status
        return s

    def set_activity(
        self,
        last=_ACTIVITY_UNSET,
        current=_ACTIVITY_UNSET,
        next_=_ACTIVITY_UNSET,
        source=_ACTIVITY_UNSET,
    ) -> None:
        """Aktualisiert die Live-Aktivitätsanzeige des Nodes (für Dashboard-Statusseite)."""
        import time
        if last is not _ACTIVITY_UNSET:
            self._activity["last"] = last
        if current is not _ACTIVITY_UNSET:
            self._activity["current"] = current
        if next_ is not _ACTIVITY_UNSET:
            self._activity["next"] = next_
        if source is not _ACTIVITY_UNSET:
            self._activity["source"] = source
        self._activity["ts"] = time.time()
