""" Conversation Session Management This module implements individual conversation session tracking for the Trixy voice assistant system. Each ConversationSession represents a single multi-turn conversation between a user and the system, maintaining conversation history, state, and metadata. Key Features: - Individual conversation session tracking - Turn-based conversation history - Speaker and satellite information tracking - Conversation timing and metrics - Thread-safe operations - Conversation state transitions - Context preservation across turns - Plugin interaction support Classes: - ConversationSession: Main session class - ConversationTurn: Individual conversation turn - ConversationRole: Speaker role enumeration - ConversationSessionState: Session state enumeration """ import json import threading import time import uuid from datetime import datetime, timezone, timedelta from dataclasses import dataclass, field from enum import Enum from typing import Any, Dict, List, Optional, Tuple, Union, Callable from pathlib import Path import weakref from ..events.event_data import SatelliteInfo, SpeakerInfo def pprint(message: str) -> None: """Conversation session logging function.""" print(f"[CONVERSATION_SESSION] {message}") class ConversationRole(Enum): """Roles in a conversation.""" USER = "user" ASSISTANT = "assistant" SYSTEM = "system" PLUGIN = "plugin" class ConversationSessionState(Enum): """States of a conversation session.""" INITIALIZING = "initializing" ACTIVE = "active" WAITING_FOR_INPUT = "waiting_for_input" PROCESSING = "processing" WAITING_FOR_PLUGIN = "waiting_for_plugin" COMPLETED = "completed" TIMED_OUT = "timed_out" ERROR = "error" CANCELLED = "cancelled" @dataclass class ConversationTurn: """ Represents a single turn in a conversation. A turn consists of input from a user and optional response from the system. This tracks the complete interaction cycle for analytics and context. """ turn_id: str = field(default_factory=lambda: str(uuid.uuid4())) turn_number: int = 0 role: ConversationRole = ConversationRole.USER timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) # Input data input_type: str = "" # "audio", "text", "intent", "wakeword" input_text: Optional[str] = None input_audio_data: Optional[bytes] = None input_confidence: float = 0.0 # Processing data intent: Optional[str] = None entities: Dict[str, Any] = field(default_factory=dict) processing_time_ms: float = 0.0 # Response data response_text: Optional[str] = None response_audio_data: Optional[bytes] = None response_voice_settings: Dict[str, Any] = field(default_factory=dict) # Metadata speaker_info: Optional[SpeakerInfo] = None satellite_info: Optional[SatelliteInfo] = None plugin_name: Optional[str] = None error_message: Optional[str] = None metadata: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: """Convert turn to dictionary representation.""" return { "turn_id": self.turn_id, "turn_number": self.turn_number, "role": self.role.value, "timestamp": self.timestamp.isoformat(), "input_type": self.input_type, "input_text": self.input_text, "input_confidence": self.input_confidence, "intent": self.intent, "entities": self.entities, "processing_time_ms": self.processing_time_ms, "response_text": self.response_text, "response_voice_settings": self.response_voice_settings, "speaker_info": self.speaker_info.to_dict() if self.speaker_info else None, "satellite_info": self.satellite_info.to_dict() if self.satellite_info else None, "plugin_name": self.plugin_name, "error_message": self.error_message, "metadata": self.metadata } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'ConversationTurn': """Create ConversationTurn from dictionary.""" # Parse datetime timestamp = datetime.fromisoformat(data.get("timestamp", datetime.now(timezone.utc).isoformat())) # Parse speaker info speaker_info = None if data.get("speaker_info"): speaker_info = SpeakerInfo(**data["speaker_info"]) # Parse satellite info satellite_info = None if data.get("satellite_info"): satellite_info = SatelliteInfo(**data["satellite_info"]) return cls( turn_id=data.get("turn_id", str(uuid.uuid4())), turn_number=data.get("turn_number", 0), role=ConversationRole(data.get("role", "user")), timestamp=timestamp, input_type=data.get("input_type", ""), input_text=data.get("input_text"), input_confidence=data.get("input_confidence", 0.0), intent=data.get("intent"), entities=data.get("entities", {}), processing_time_ms=data.get("processing_time_ms", 0.0), response_text=data.get("response_text"), response_voice_settings=data.get("response_voice_settings", {}), speaker_info=speaker_info, satellite_info=satellite_info, plugin_name=data.get("plugin_name"), error_message=data.get("error_message"), metadata=data.get("metadata", {}) ) class ConversationSessionError(Exception): """Base exception for conversation session errors.""" pass class ConversationSession: """ Represents an individual conversation session between a user and the Trixy system. A conversation session tracks: - Conversation metadata (ID, participants, timing) - Turn-by-turn conversation history - Current conversation state and context - Plugin interactions and multi-turn workflows - Performance metrics and analytics The session supports the conversation workflow specified in CLAUDE.md: 1. Wakeword detection starts conversation session 2. Raw audio input with conversation_id 3. Text processing with same conversation_id 4. Intent handling with conversation tracking 5. Multi-turn conversations where plugins ask questions and wait for responses Thread-safe for multi-satellite environments. """ def __init__( self, conversation_id: str, satellite_info: SatelliteInfo, speaker_info: Optional[SpeakerInfo] = None, trigger_event: str = "wakeword_received", timeout_seconds: float = 300.0, # 5 minutes default max_turns: int = 50, application_ref: Optional[weakref.ref] = None ): """ Initialize a conversation session. Args: conversation_id: Unique conversation identifier satellite_info: Information about the satellite device speaker_info: Information about the detected speaker trigger_event: Event that triggered the conversation timeout_seconds: Conversation timeout in seconds max_turns: Maximum number of conversation turns application_ref: Weak reference to application container """ # Core identifiers self._conversation_id = conversation_id self._trigger_event = trigger_event # Participant information self._satellite_info = satellite_info self._speaker_info = speaker_info # Timing and limits self._start_time = datetime.now(timezone.utc) self._last_activity = self._start_time self._end_time: Optional[datetime] = None self._timeout_seconds = timeout_seconds self._max_turns = max_turns # State management self._state = ConversationSessionState.INITIALIZING self._state_lock = threading.RLock() # Conversation history self._turns: List[ConversationTurn] = [] self._turns_lock = threading.RLock() self._current_turn_number = 0 # Context and metadata self._context_data: Dict[str, Any] = {} self._metadata: Dict[str, Any] = { "created_at": self._start_time.isoformat(), "trigger_event": trigger_event, "version": "1.0.0" } # Plugin interaction self._waiting_for_plugin: Optional[str] = None self._plugin_question: Optional[str] = None self._plugin_callback: Optional[Callable] = None self._plugin_timeout: Optional[datetime] = None # Application integration self._application_ref = application_ref # Analytics self._total_processing_time = 0.0 self._error_count = 0 self._successful_turns = 0 pprint(f"Conversation session created: {conversation_id}") self._set_state(ConversationSessionState.ACTIVE) # Properties @property def conversation_id(self) -> str: """Get the conversation ID.""" return self._conversation_id @property def state(self) -> ConversationSessionState: """Get the current conversation state.""" with self._state_lock: return self._state @property def satellite_info(self) -> SatelliteInfo: """Get satellite information.""" return self._satellite_info @property def speaker_info(self) -> Optional[SpeakerInfo]: """Get speaker information.""" return self._speaker_info @property def start_time(self) -> datetime: """Get conversation start time.""" return self._start_time @property def last_activity(self) -> datetime: """Get last activity time.""" return self._last_activity @property def duration_seconds(self) -> float: """Get conversation duration in seconds.""" end_time = self._end_time or datetime.now(timezone.utc) return (end_time - self._start_time).total_seconds() @property def turn_count(self) -> int: """Get number of conversation turns.""" with self._turns_lock: return len(self._turns) @property def is_active(self) -> bool: """Check if conversation is active.""" with self._state_lock: return self._state in [ ConversationSessionState.ACTIVE, ConversationSessionState.WAITING_FOR_INPUT, ConversationSessionState.PROCESSING, ConversationSessionState.WAITING_FOR_PLUGIN ] @property def is_timed_out(self) -> bool: """Check if conversation has timed out.""" if not self.is_active: return False now = datetime.now(timezone.utc) # Check overall timeout if (now - self._start_time).total_seconds() > self._timeout_seconds: return True # Check plugin timeout if (self._plugin_timeout and self._state == ConversationSessionState.WAITING_FOR_PLUGIN and now > self._plugin_timeout): return True return False # State Management def _set_state(self, new_state: ConversationSessionState) -> None: """Set conversation state thread-safely.""" with self._state_lock: old_state = self._state self._state = new_state self._last_activity = datetime.now(timezone.utc) if old_state != new_state: pprint(f"Conversation {self._conversation_id} state: {old_state.value} -> {new_state.value}") # Handle state transitions if new_state in [ConversationSessionState.COMPLETED, ConversationSessionState.TIMED_OUT, ConversationSessionState.ERROR, ConversationSessionState.CANCELLED]: self._end_time = self._last_activity def _update_activity(self) -> None: """Update last activity timestamp.""" self._last_activity = datetime.now(timezone.utc) # Turn Management def add_turn( self, role: ConversationRole, input_type: str = "", input_text: Optional[str] = None, input_audio_data: Optional[bytes] = None, input_confidence: float = 0.0, speaker_info: Optional[SpeakerInfo] = None, satellite_info: Optional[SatelliteInfo] = None, metadata: Optional[Dict[str, Any]] = None ) -> ConversationTurn: """ Add a new turn to the conversation. Args: role: Role of the speaker (user, assistant, plugin) input_type: Type of input (audio, text, intent, wakeword) input_text: Text content of the input input_audio_data: Audio data if applicable input_confidence: Confidence score for the input speaker_info: Speaker information (uses session default if None) satellite_info: Satellite information (uses session default if None) metadata: Additional metadata for the turn Returns: ConversationTurn: The created turn Raises: ConversationSessionError: If session is not in valid state or limits exceeded """ if not self.is_active: raise ConversationSessionError(f"Cannot add turn to conversation in state: {self._state.value}") with self._turns_lock: if len(self._turns) >= self._max_turns: raise ConversationSessionError(f"Maximum turns ({self._max_turns}) exceeded") # Increment turn number self._current_turn_number += 1 # Create turn turn = ConversationTurn( turn_number=self._current_turn_number, role=role, input_type=input_type, input_text=input_text, input_audio_data=input_audio_data, input_confidence=input_confidence, speaker_info=speaker_info or self._speaker_info, satellite_info=satellite_info or self._satellite_info, metadata=metadata or {} ) # Add to history self._turns.append(turn) self._update_activity() pprint(f"Added turn {self._current_turn_number} to conversation {self._conversation_id}") return turn def get_current_turn(self) -> Optional[ConversationTurn]: """Get the current (most recent) turn.""" with self._turns_lock: return self._turns[-1] if self._turns else None def get_turn_by_number(self, turn_number: int) -> Optional[ConversationTurn]: """Get a turn by its number.""" with self._turns_lock: for turn in self._turns: if turn.turn_number == turn_number: return turn return None def get_turns(self, role: Optional[ConversationRole] = None) -> List[ConversationTurn]: """ Get conversation turns, optionally filtered by role. Args: role: Optional role filter Returns: List of conversation turns """ with self._turns_lock: if role is None: return self._turns.copy() return [turn for turn in self._turns if turn.role == role] def update_turn_response( self, turn_id: str, response_text: Optional[str] = None, response_audio_data: Optional[bytes] = None, response_voice_settings: Optional[Dict[str, Any]] = None, processing_time_ms: float = 0.0, plugin_name: Optional[str] = None ) -> bool: """ Update a turn with response information. Args: turn_id: ID of the turn to update response_text: Response text response_audio_data: Response audio data response_voice_settings: Voice settings used processing_time_ms: Processing time in milliseconds plugin_name: Name of plugin that generated response Returns: bool: True if turn was found and updated """ with self._turns_lock: for turn in self._turns: if turn.turn_id == turn_id: if response_text is not None: turn.response_text = response_text if response_audio_data is not None: turn.response_audio_data = response_audio_data if response_voice_settings is not None: turn.response_voice_settings = response_voice_settings if processing_time_ms > 0: turn.processing_time_ms = processing_time_ms self._total_processing_time += processing_time_ms if plugin_name is not None: turn.plugin_name = plugin_name self._update_activity() return True return False def update_turn_intent( self, turn_id: str, intent: str, entities: Optional[Dict[str, Any]] = None, confidence: float = 0.0 ) -> bool: """ Update a turn with intent information. Args: turn_id: ID of the turn to update intent: Detected intent entities: Extracted entities confidence: Intent confidence score Returns: bool: True if turn was found and updated """ with self._turns_lock: for turn in self._turns: if turn.turn_id == turn_id: turn.intent = intent turn.entities = entities or {} turn.input_confidence = max(turn.input_confidence, confidence) self._update_activity() return True return False def mark_turn_error(self, turn_id: str, error_message: str) -> bool: """ Mark a turn as having an error. Args: turn_id: ID of the turn to update error_message: Error description Returns: bool: True if turn was found and updated """ with self._turns_lock: for turn in self._turns: if turn.turn_id == turn_id: turn.error_message = error_message self._error_count += 1 self._update_activity() return True return False # Plugin Interaction def ask_question( self, question: str, plugin_name: str, callback: Optional[Callable] = None, timeout_seconds: float = 30.0, voice_settings: Optional[Dict[str, Any]] = None ) -> ConversationTurn: """ Ask a question and wait for user response (plugin interaction). Args: question: Question to ask the user plugin_name: Name of the plugin asking the question callback: Optional callback function for the response timeout_seconds: Timeout for the response voice_settings: Voice settings for TTS Returns: ConversationTurn: The question turn Raises: ConversationSessionError: If session is not in valid state """ if not self.is_active: raise ConversationSessionError(f"Cannot ask question in state: {self._state.value}") # Add question turn turn = self.add_turn( role=ConversationRole.PLUGIN, input_type="question", input_text=question, metadata={ "plugin_name": plugin_name, "is_question": True, "timeout_seconds": timeout_seconds } ) # Update response info self.update_turn_response( turn.turn_id, response_text=question, response_voice_settings=voice_settings or {}, plugin_name=plugin_name ) # Set waiting state self._waiting_for_plugin = plugin_name self._plugin_question = question self._plugin_callback = callback self._plugin_timeout = datetime.now(timezone.utc) + timedelta(seconds=timeout_seconds) self._set_state(ConversationSessionState.WAITING_FOR_PLUGIN) pprint(f"Plugin {plugin_name} asked question in conversation {self._conversation_id}: {question}") return turn def answer_question( self, answer: str, confidence: float = 1.0, speaker_info: Optional[SpeakerInfo] = None ) -> Optional[ConversationTurn]: """ Provide an answer to a plugin question. Args: answer: User's answer confidence: Confidence in the answer speaker_info: Speaker information Returns: ConversationTurn: The answer turn, or None if not waiting for answer """ if self._state != ConversationSessionState.WAITING_FOR_PLUGIN: pprint(f"Not waiting for plugin response in conversation {self._conversation_id}") return None # Add answer turn turn = self.add_turn( role=ConversationRole.USER, input_type="answer", input_text=answer, input_confidence=confidence, speaker_info=speaker_info, metadata={ "answering_plugin": self._waiting_for_plugin, "original_question": self._plugin_question, "is_answer": True } ) # Call plugin callback if available if self._plugin_callback: try: self._plugin_callback(self._conversation_id, answer, turn) except Exception as e: pprint(f"Error in plugin callback: {e}") self.mark_turn_error(turn.turn_id, f"Plugin callback error: {e}") # Clear plugin state self._waiting_for_plugin = None self._plugin_question = None self._plugin_callback = None self._plugin_timeout = None self._set_state(ConversationSessionState.ACTIVE) pprint(f"Received answer in conversation {self._conversation_id}: {answer}") return turn def cancel_plugin_question(self, reason: str = "timeout") -> None: """ Cancel a pending plugin question. Args: reason: Reason for cancellation """ if self._state == ConversationSessionState.WAITING_FOR_PLUGIN: pprint(f"Cancelling plugin question in conversation {self._conversation_id}: {reason}") # Clear plugin state self._waiting_for_plugin = None self._plugin_question = None self._plugin_callback = None self._plugin_timeout = None if reason == "timeout": self._set_state(ConversationSessionState.TIMED_OUT) else: self._set_state(ConversationSessionState.ACTIVE) # Context Management def set_context(self, key: str, value: Any) -> None: """Set a context value.""" self._context_data[key] = value self._update_activity() def get_context(self, key: str, default: Any = None) -> Any: """Get a context value.""" return self._context_data.get(key, default) def update_context(self, context_data: Dict[str, Any]) -> None: """Update multiple context values.""" self._context_data.update(context_data) self._update_activity() def clear_context(self) -> None: """Clear all context data.""" self._context_data.clear() self._update_activity() def get_all_context(self) -> Dict[str, Any]: """Get all context data.""" return self._context_data.copy() # Session Control def complete(self, final_response: Optional[str] = None) -> None: """ Complete the conversation session. Args: final_response: Optional final response to the user """ if final_response: self.add_turn( role=ConversationRole.ASSISTANT, input_type="completion", input_text=final_response, metadata={"is_final_response": True} ) self._successful_turns = len([t for t in self._turns if not t.error_message]) self._set_state(ConversationSessionState.COMPLETED) pprint(f"Conversation {self._conversation_id} completed with {self.turn_count} turns") def cancel(self, reason: str = "user_cancelled") -> None: """ Cancel the conversation session. Args: reason: Reason for cancellation """ self._metadata["cancellation_reason"] = reason self._set_state(ConversationSessionState.CANCELLED) pprint(f"Conversation {self._conversation_id} cancelled: {reason}") def timeout(self) -> None: """Mark conversation as timed out.""" self._set_state(ConversationSessionState.TIMED_OUT) pprint(f"Conversation {self._conversation_id} timed out after {self.duration_seconds:.1f}s") def error(self, error_message: str) -> None: """ Mark conversation as having an error. Args: error_message: Error description """ self._metadata["error_message"] = error_message self._error_count += 1 self._set_state(ConversationSessionState.ERROR) pprint(f"Conversation {self._conversation_id} error: {error_message}") # Analytics and Export def get_analytics(self) -> Dict[str, Any]: """ Get conversation analytics and metrics. Returns: Dictionary containing analytics data """ return { "conversation_id": self._conversation_id, "duration_seconds": self.duration_seconds, "turn_count": self.turn_count, "successful_turns": self._successful_turns, "error_count": self._error_count, "total_processing_time_ms": self._total_processing_time, "average_processing_time_ms": ( self._total_processing_time / max(1, self.turn_count) ), "state": self._state.value, "trigger_event": self._trigger_event, "speaker_id": self._speaker_info.speaker_id if self._speaker_info else None, "satellite_id": self._satellite_info.satellite_id, "satellite_room": self._satellite_info.room_id, "start_time": self._start_time.isoformat(), "end_time": self._end_time.isoformat() if self._end_time else None, "last_activity": self._last_activity.isoformat(), "context_keys": list(self._context_data.keys()), "metadata": self._metadata.copy() } def to_dict(self) -> Dict[str, Any]: """ Convert conversation session to dictionary representation. Returns: Dictionary containing complete session data """ with self._turns_lock: return { "conversation_id": self._conversation_id, "state": self._state.value, "trigger_event": self._trigger_event, "start_time": self._start_time.isoformat(), "end_time": self._end_time.isoformat() if self._end_time else None, "last_activity": self._last_activity.isoformat(), "timeout_seconds": self._timeout_seconds, "max_turns": self._max_turns, "speaker_info": self._speaker_info.to_dict() if self._speaker_info else None, "satellite_info": self._satellite_info.to_dict(), "turns": [turn.to_dict() for turn in self._turns], "context_data": self._context_data.copy(), "metadata": self._metadata.copy(), "analytics": self.get_analytics() } def save_to_file(self, file_path: Path) -> None: """ Save conversation session to JSON file. Args: file_path: Path to save the file """ try: with open(file_path, 'w', encoding='utf-8') as f: json.dump(self.to_dict(), f, indent=2, ensure_ascii=False) pprint(f"Conversation {self._conversation_id} saved to {file_path}") except Exception as e: raise ConversationSessionError(f"Failed to save conversation: {e}") @classmethod def load_from_file(cls, file_path: Path, application_ref: Optional[weakref.ref] = None) -> 'ConversationSession': """ Load conversation session from JSON file. Args: file_path: Path to load from application_ref: Application reference Returns: ConversationSession: Loaded session """ try: with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) # Parse satellite info satellite_info = SatelliteInfo(**data["satellite_info"]) # Parse speaker info speaker_info = None if data.get("speaker_info"): speaker_info = SpeakerInfo(**data["speaker_info"]) # Create session session = cls( conversation_id=data["conversation_id"], satellite_info=satellite_info, speaker_info=speaker_info, trigger_event=data.get("trigger_event", "wakeword_received"), timeout_seconds=data.get("timeout_seconds", 300.0), max_turns=data.get("max_turns", 50), application_ref=application_ref ) # Restore state session._state = ConversationSessionState(data["state"]) session._start_time = datetime.fromisoformat(data["start_time"]) if data.get("end_time"): session._end_time = datetime.fromisoformat(data["end_time"]) session._last_activity = datetime.fromisoformat(data["last_activity"]) session._context_data = data.get("context_data", {}) session._metadata = data.get("metadata", {}) # Restore turns for turn_data in data.get("turns", []): turn = ConversationTurn.from_dict(turn_data) session._turns.append(turn) session._current_turn_number = max(session._current_turn_number, turn.turn_number) pprint(f"Conversation {session.conversation_id} loaded from {file_path}") return session except Exception as e: raise ConversationSessionError(f"Failed to load conversation: {e}") def __str__(self) -> str: """String representation of the conversation session.""" return (f"ConversationSession(id={self._conversation_id}, " f"state={self._state.value}, turns={self.turn_count}, " f"duration={self.duration_seconds:.1f}s)") def __repr__(self) -> str: """Detailed representation of the conversation session.""" return (f"ConversationSession(conversation_id='{self._conversation_id}', " f"state={self._state.value}, turn_count={self.turn_count}, " f"speaker='{self._speaker_info.speaker_name if self._speaker_info else None}', " f"satellite='{self._satellite_info.alias}', " f"duration={self.duration_seconds:.1f}s)") # Factory function def create_conversation_session( conversation_id: str, satellite_info: SatelliteInfo, speaker_info: Optional[SpeakerInfo] = None, **kwargs ) -> ConversationSession: """ Create a ConversationSession with default configuration. Args: conversation_id: Unique conversation identifier satellite_info: Satellite device information speaker_info: Speaker information **kwargs: Additional session parameters Returns: ConversationSession: Configured session """ return ConversationSession( conversation_id=conversation_id, satellite_info=satellite_info, speaker_info=speaker_info, **kwargs ) # Module exports __all__ = [ "ConversationSession", "ConversationSessionError", "ConversationSessionState", "ConversationTurn", "ConversationRole", "create_conversation_session" ]