Trixy7/trixy_core/conversation/manager.py

# -*- coding: utf-8 -*-
"""
Conversation Manager - Verwaltet Dialog-Sessions.
"""

from dataclasses import dataclass, field
from datetime import datetime
from typing import TYPE_CHECKING, Callable, Iterator
import asyncio
import threading

from trixy_core.service import IService, ServicePriority, ServiceGroup
from trixy_core.conversation.session import (
    ConversationSession,
    SessionConfig,
    SessionState,
)
from trixy_core.utils.debug import perror

if TYPE_CHECKING:
    from trixy_core.application import BaseApplication


@dataclass
class ManagerConfig:
    """Konfiguration für den ConversationManager."""

    # Session-Defaults
    default_session_config: SessionConfig = field(default_factory=SessionConfig)

    # Limits
    max_active_sessions: int = 100
    max_sessions_per_satellite: int = 1

    # Cleanup
    cleanup_interval_seconds: float = 30.0
    keep_completed_seconds: float = 300.0  # 5 Minuten behalten

    # Timeouts
    arbitration_window_seconds: float = 1.0


class ConversationManager(IService):
    """
    Verwaltet Conversation-Sessions.

    Verantwortlich für:
    - Session-Erstellung und -Verwaltung
    - Session-Lookup nach ID oder Satellite
    - Timeout-Handling und Cleanup
    - Session-Events
    """

    PRIORITY = ServicePriority.MANAGER
    GROUP = ServiceGroup.CONVERSATION
    DEPENDENCIES = ["event_manager"]

    def __init__(
        self,
        application: "BaseApplication",
        config: ManagerConfig | None = None,
    ):
        """
        Initialisiert den Manager.

        Args:
            application: Anwendungs-Instanz
            config: Manager-Konfiguration
        """
        super().__init__(application)

        self._config = config or ManagerConfig()
        self._lock = threading.RLock()

        # Sessions
        self._sessions: dict[str, ConversationSession] = {}
        self._sessions_by_satellite: dict[str, list[str]] = {}

        # Cleanup
        self._cleanup_task: asyncio.Task | None = None

        # Callbacks
        self._on_session_start: list[Callable[[ConversationSession], None]] = []
        self._on_session_complete: list[Callable[[ConversationSession], None]] = []
        self._on_follow_up: list[Callable[[ConversationSession, str], None]] = []

    def _emit_event(self, event_name: str, data: dict) -> None:
        """Emittiert Event async-safe aus sync-Kontext."""
        coro = self.application.event_manager.emit(event_name, data)
        try:
            loop = asyncio.get_running_loop()
            task = loop.create_task(coro)
            task.add_done_callback(
                lambda t: perror(f"Event-Fehler '{event_name}': {t.exception()}")
                if t.exception() else None
            )
        except RuntimeError:
            # Kein laufender Event-Loop
            pass

    @property
    def config(self) -> ManagerConfig:
        """Gibt Konfiguration zurück."""
        return self._config

    @property
    def session_count(self) -> int:
        """Anzahl aller Sessions."""
        with self._lock:
            return len(self._sessions)

    @property
    def active_session_count(self) -> int:
        """Anzahl aktiver Sessions."""
        with self._lock:
            return sum(1 for s in self._sessions.values() if s.is_active)

    async def on_start(self) -> None:
        """Service-Start."""
        # Starte Cleanup-Task
        self._cleanup_task = asyncio.create_task(self._cleanup_loop())
        self.logger.info("ConversationManager gestartet")

    async def on_stop(self) -> None:
        """Service-Stop."""
        # Stoppe Cleanup
        if self._cleanup_task:
            self._cleanup_task.cancel()
            try:
                await self._cleanup_task
            except asyncio.CancelledError:
                pass

        # Beende alle aktiven Sessions
        with self._lock:
            for session in self._sessions.values():
                if session.is_active:
                    session.cancel("Manager wird gestoppt")

    # === Session-Erstellung ===

    def create_session(
        self,
        satellite_id: str | None = None,
        wakeword_type: str | None = None,
        session_id: str | None = None,
        config: SessionConfig | None = None,
    ) -> ConversationSession:
        """
        Erstellt eine neue Conversation-Session.

        Args:
            satellite_id: ID des auslösenden Satellites
            wakeword_type: Typ des erkannten Wakewords
            session_id: Session-ID (optional, wird generiert)
            config: Session-Konfiguration (optional)

        Returns:
            Neue Session

        Raises:
            RuntimeError: Bei Limit-Überschreitung
        """
        with self._lock:
            # Prüfe Limits
            if self.active_session_count >= self._config.max_active_sessions:
                raise RuntimeError("Maximale aktive Sessions erreicht")

            if satellite_id:
                satellite_sessions = self._sessions_by_satellite.get(satellite_id, [])
                active_count = sum(
                    1 for sid in satellite_sessions
                    if sid in self._sessions and self._sessions[sid].is_active
                )
                if active_count >= self._config.max_sessions_per_satellite:
                    raise RuntimeError(
                        f"Satellite {satellite_id} hat bereits aktive Session"
                    )

            # Erstelle Session
            session = ConversationSession(
                session_id=session_id,
                satellite_id=satellite_id,
                wakeword_type=wakeword_type,
                config=config or self._config.default_session_config,
            )

            # Registriere
            self._sessions[session.session_id] = session

            if satellite_id:
                if satellite_id not in self._sessions_by_satellite:
                    self._sessions_by_satellite[satellite_id] = []
                self._sessions_by_satellite[satellite_id].append(session.session_id)

        # Callbacks
        for callback in self._on_session_start:
            try:
                callback(session)
            except Exception:
                pass

        # Event emittieren (async-safe aus sync-Kontext)
        self._emit_event("conversation_started", {
            "session_id": session.session_id,
            "satellite_id": satellite_id,
            "wakeword_type": wakeword_type,
        })

        self.logger.info(
            f"Session erstellt: {session.session_id} "
            f"(Satellite: {satellite_id}, Wakeword: {wakeword_type})"
        )

        return session

    # === Session-Lookup ===

    def get_session(self, session_id: str) -> ConversationSession | None:
        """
        Gibt Session nach ID zurück.

        Args:
            session_id: Session-ID

        Returns:
            Session oder None
        """
        with self._lock:
            return self._sessions.get(session_id)

    def get_active_session(self, satellite_id: str) -> ConversationSession | None:
        """
        Gibt aktive Session für Satellite zurück.

        Args:
            satellite_id: Satellite-ID

        Returns:
            Aktive Session oder None
        """
        with self._lock:
            session_ids = self._sessions_by_satellite.get(satellite_id, [])

            for sid in session_ids:
                session = self._sessions.get(sid)
                if session and session.is_active:
                    return session

            return None

    def get_sessions(
        self,
        satellite_id: str | None = None,
        state: SessionState | None = None,
        active_only: bool = False,
    ) -> list[ConversationSession]:
        """
        Gibt gefilterte Sessions zurück.

        Args:
            satellite_id: Filter nach Satellite
            state: Filter nach Zustand
            active_only: Nur aktive Sessions

        Returns:
            Liste von Sessions
        """
        with self._lock:
            sessions = list(self._sessions.values())

        # Filter
        if satellite_id:
            sessions = [s for s in sessions if s.satellite_id == satellite_id]

        if state:
            sessions = [s for s in sessions if s.state == state]

        if active_only:
            sessions = [s for s in sessions if s.is_active]

        return sessions

    def iterate_active(self) -> Iterator[ConversationSession]:
        """Iteriert über aktive Sessions."""
        with self._lock:
            sessions = [s for s in self._sessions.values() if s.is_active]

        for session in sessions:
            yield session

    # === Session-Lifecycle ===

    def start_session(self, session_id: str) -> bool:
        """
        Startet eine Session.

        Args:
            session_id: Session-ID

        Returns:
            True wenn gestartet
        """
        session = self.get_session(session_id)
        if not session:
            return False

        session.start()
        return True

    def complete_session(
        self,
        session_id: str,
        reason: str | None = None,
    ) -> bool:
        """
        Schließt eine Session ab.

        Args:
            session_id: Session-ID
            reason: Abschlussgrund

        Returns:
            True wenn abgeschlossen
        """
        session = self.get_session(session_id)
        if not session:
            return False

        if session.is_completed:
            return False

        session.complete(reason)
        self._on_session_completed(session)

        return True

    def cancel_session(
        self,
        session_id: str,
        reason: str | None = None,
    ) -> bool:
        """
        Bricht eine Session ab.

        Args:
            session_id: Session-ID
            reason: Abbruchgrund

        Returns:
            True wenn abgebrochen
        """
        session = self.get_session(session_id)
        if not session:
            return False

        if session.is_completed:
            return False

        session.cancel(reason)
        self._on_session_completed(session)

        return True

    def _on_session_completed(self, session: ConversationSession) -> None:
        """Intern: Session wurde abgeschlossen."""
        # Callbacks
        for callback in self._on_session_complete:
            try:
                callback(session)
            except Exception:
                pass

        # Event emittieren (async-safe aus sync-Kontext)
        self._emit_event("conversation_ended", {
            "session_id": session.session_id,
            "satellite_id": session.satellite_id,
            "state": session.state.value,
            "duration_seconds": session.duration_seconds,
            "turn_count": session.turn_count,
        })

        self.logger.info(
            f"Session beendet: {session.session_id} "
            f"(State: {session.state.value}, Dauer: {session.duration_seconds:.1f}s)"
        )

    # === Follow-Up ===

    def request_follow_up(
        self,
        session_id: str,
        question: str,
        audio_data: bytes | None = None,
    ) -> bool:
        """
        Stellt eine Rückfrage in einer Session.

        Args:
            session_id: Session-ID
            question: Rückfrage-Text
            audio_data: TTS-Audio (optional)

        Returns:
            True wenn Rückfrage gestellt
        """
        session = self.get_session(session_id)
        if not session or not session.is_active:
            return False

        # Prüfe Limit
        if session.follow_up_count >= session.config.max_follow_ups:
            self.logger.warning(
                f"Max Follow-Ups erreicht für Session {session_id}"
            )
            return False

        # Füge Follow-Up Turn hinzu
        session.add_assistant_turn(
            content=question,
            audio_data=audio_data,
            is_follow_up=True,
        )

        # Callbacks
        for callback in self._on_follow_up:
            try:
                callback(session, question)
            except Exception:
                pass

        # Event emittieren (async-safe aus sync-Kontext)
        self._emit_event("conversation_follow_up", {
            "session_id": session_id,
            "satellite_id": session.satellite_id,
            "question": question,
            "follow_up_count": session.follow_up_count,
        })

        self.logger.info(
            f"Follow-Up #{session.follow_up_count} für Session {session_id}"
        )

        return True

    # === Cleanup ===

    async def _cleanup_loop(self) -> None:
        """Cleanup-Loop für Timeouts und alte Sessions."""
        while True:
            try:
                await asyncio.sleep(self._config.cleanup_interval_seconds)
                await self._do_cleanup()
            except asyncio.CancelledError:
                break
            except Exception as e:
                self.logger.error(f"Cleanup-Fehler: {e}")

    async def _do_cleanup(self) -> None:
        """Führt Cleanup durch."""
        now = datetime.now()
        to_remove = []

        with self._lock:
            for session_id, session in self._sessions.items():
                # Prüfe Timeout für aktive Sessions
                if session.is_active:
                    session.check_timeout()

                    if session.is_completed:
                        self._on_session_completed(session)

                # Entferne alte abgeschlossene Sessions
                elif session.completed_at:
                    age = (now - session.completed_at).total_seconds()
                    if age > self._config.keep_completed_seconds:
                        to_remove.append(session_id)

        # Entferne Sessions außerhalb des Locks
        for session_id in to_remove:
            await self.remove_session(session_id)

        if to_remove:
            self.logger.debug(f"Cleanup: {len(to_remove)} Sessions entfernt")

    async def remove_session(self, session_id: str) -> bool:
        """
        Entfernt eine Session.

        Args:
            session_id: Session-ID

        Returns:
            True wenn entfernt
        """
        with self._lock:
            session = self._sessions.pop(session_id, None)

            if not session:
                return False

            # Aus Satellite-Index entfernen
            if session.satellite_id:
                sat_sessions = self._sessions_by_satellite.get(session.satellite_id, [])
                if session_id in sat_sessions:
                    sat_sessions.remove(session_id)

        return True

    # === Callbacks ===

    def on_session_start(
        self,
        callback: Callable[[ConversationSession], None],
    ) -> None:
        """Registriert Callback für Session-Start."""
        self._on_session_start.append(callback)

    def on_session_complete(
        self,
        callback: Callable[[ConversationSession], None],
    ) -> None:
        """Registriert Callback für Session-Abschluss."""
        self._on_session_complete.append(callback)

    def on_follow_up(
        self,
        callback: Callable[[ConversationSession, str], None],
    ) -> None:
        """Registriert Callback für Rückfragen."""
        self._on_follow_up.append(callback)

    # === Stats ===

    def get_stats(self) -> dict:
        """Gibt Statistiken zurück."""
        with self._lock:
            sessions = list(self._sessions.values())

        active = [s for s in sessions if s.is_active]
        completed = [s for s in sessions if s.is_completed]

        return {
            "total_sessions": len(sessions),
            "active_sessions": len(active),
            "completed_sessions": len(completed),
            "satellites_with_sessions": len(self._sessions_by_satellite),
            "states": {
                state.value: sum(1 for s in sessions if s.state == state)
                for state in SessionState
            },
        }

    def get_summary(self) -> dict:
        """Gibt Zusammenfassung zurück."""
        stats = self.get_stats()

        with self._lock:
            sessions = list(self._sessions.values())

        # Durchschnittliche Dauer
        completed = [s for s in sessions if s.is_completed]
        avg_duration = (
            sum(s.duration_seconds for s in completed) / len(completed)
            if completed else 0.0
        )

        # Durchschnittliche Turns
        avg_turns = (
            sum(s.turn_count for s in sessions) / len(sessions)
            if sessions else 0.0
        )

        return {
            **stats,
            "avg_duration_seconds": avg_duration,
            "avg_turns": avg_turns,
        }