# -*- 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, }