""" Conversation Context Management This module implements context management and preservation for conversations in the Trixy voice assistant system. It provides sophisticated context tracking, scope management, and retention policies to maintain conversation continuity across multiple turns and sessions. Key Features: - Multi-scope context management (session, global, plugin-specific) - Context entry versioning and history - Retention policies for context cleanup - Thread-safe context operations - Context persistence and restoration - Integration with conversation sessions - Performance optimization for large contexts Classes: - ConversationContext: Main context management class - ContextEntry: Individual context data entry - ContextScope: Context scope enumeration - ContextRetentionPolicy: Context cleanup policies """ import json import threading import time from datetime import datetime, timezone, timedelta from dataclasses import dataclass, field from enum import Enum from typing import Any, Dict, List, Optional, Set, Union, Callable, Iterator from pathlib import Path import weakref import pickle import gzip from collections import defaultdict, OrderedDict def pprint(message: str) -> None: """Conversation context logging function.""" print(f"[CONVERSATION_CONTEXT] {message}") class ContextScope(Enum): """Scopes for context data.""" SESSION = "session" # Scoped to current conversation session SPEAKER = "speaker" # Scoped to specific speaker across sessions SATELLITE = "satellite" # Scoped to specific satellite device ROOM = "room" # Scoped to room/location GLOBAL = "global" # Global context across all conversations PLUGIN = "plugin" # Plugin-specific context TEMPORARY = "temporary" # Temporary context with short lifetime class ContextRetentionPolicy(Enum): """Retention policies for context cleanup.""" IMMEDIATE = "immediate" # Delete immediately when scope ends SESSION_END = "session_end" # Keep until session ends TIME_BASED = "time_based" # Keep for specified time duration TURN_BASED = "turn_based" # Keep for specified number of turns MANUAL = "manual" # Manual cleanup only PERSISTENT = "persistent" # Keep indefinitely @dataclass class ContextEntry: """ Represents a single context entry with metadata. Context entries store data with versioning, timestamps, and retention policies to enable sophisticated context management. """ key: str value: Any scope: ContextScope created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) accessed_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) # Retention and lifecycle retention_policy: ContextRetentionPolicy = ContextRetentionPolicy.SESSION_END expires_at: Optional[datetime] = None max_turns: Optional[int] = None turn_count: int = 0 # Versioning version: int = 1 previous_values: List[Any] = field(default_factory=list) # Metadata source: str = "system" # Who created this context tags: Set[str] = field(default_factory=set) priority: int = 0 # Higher priority entries are preserved longer metadata: Dict[str, Any] = field(default_factory=dict) def is_expired(self) -> bool: """Check if the context entry has expired.""" now = datetime.now(timezone.utc) # Check time-based expiration if self.expires_at and now > self.expires_at: return True # Check turn-based expiration if self.max_turns and self.turn_count >= self.max_turns: return True return False def update_value(self, new_value: Any, source: str = "system") -> None: """Update the context value with versioning.""" # Store previous value if len(self.previous_values) >= 10: # Keep last 10 versions self.previous_values.pop(0) self.previous_values.append(self.value) # Update current value self.value = new_value self.updated_at = datetime.now(timezone.utc) self.version += 1 self.source = source def access(self) -> Any: """Access the value and update access timestamp.""" self.accessed_at = datetime.now(timezone.utc) return self.value def increment_turn(self) -> None: """Increment the turn counter.""" self.turn_count += 1 def add_tag(self, tag: str) -> None: """Add a tag to the context entry.""" self.tags.add(tag) def remove_tag(self, tag: str) -> None: """Remove a tag from the context entry.""" self.tags.discard(tag) def has_tag(self, tag: str) -> bool: """Check if context entry has a specific tag.""" return tag in self.tags def to_dict(self) -> Dict[str, Any]: """Convert context entry to dictionary representation.""" return { "key": self.key, "value": self.value, "scope": self.scope.value, "created_at": self.created_at.isoformat(), "updated_at": self.updated_at.isoformat(), "accessed_at": self.accessed_at.isoformat(), "retention_policy": self.retention_policy.value, "expires_at": self.expires_at.isoformat() if self.expires_at else None, "max_turns": self.max_turns, "turn_count": self.turn_count, "version": self.version, "previous_values": self.previous_values, "source": self.source, "tags": list(self.tags), "priority": self.priority, "metadata": self.metadata } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'ContextEntry': """Create ContextEntry from dictionary.""" expires_at = None if data.get("expires_at"): expires_at = datetime.fromisoformat(data["expires_at"]) return cls( key=data["key"], value=data["value"], scope=ContextScope(data["scope"]), created_at=datetime.fromisoformat(data["created_at"]), updated_at=datetime.fromisoformat(data["updated_at"]), accessed_at=datetime.fromisoformat(data["accessed_at"]), retention_policy=ContextRetentionPolicy(data.get("retention_policy", "session_end")), expires_at=expires_at, max_turns=data.get("max_turns"), turn_count=data.get("turn_count", 0), version=data.get("version", 1), previous_values=data.get("previous_values", []), source=data.get("source", "system"), tags=set(data.get("tags", [])), priority=data.get("priority", 0), metadata=data.get("metadata", {}) ) class ConversationContextError(Exception): """Base exception for conversation context errors.""" pass class ConversationContext: """ Manages conversation context with multi-scope support and retention policies. This class provides sophisticated context management for conversations, supporting different scopes (session, speaker, global, etc.) and retention policies to automatically clean up old context data. Features: - Multi-scope context management - Automatic context cleanup based on retention policies - Context versioning and history - Thread-safe operations - Context persistence and restoration - Performance optimization for large contexts - Integration with conversation sessions The context system supports the CLAUDE.md workflow by maintaining state across conversation turns and enabling plugins to preserve context for multi-turn interactions. """ def __init__( self, conversation_id: str, speaker_id: Optional[str] = None, satellite_id: Optional[str] = None, room_id: Optional[str] = None, max_entries_per_scope: int = 1000, cleanup_interval_seconds: float = 60.0, enable_persistence: bool = False, persistence_path: Optional[Path] = None ): """ Initialize conversation context manager. Args: conversation_id: Current conversation ID speaker_id: Speaker identifier for speaker-scoped context satellite_id: Satellite identifier for satellite-scoped context room_id: Room identifier for room-scoped context max_entries_per_scope: Maximum entries per scope (for memory management) cleanup_interval_seconds: Interval for automatic cleanup enable_persistence: Enable context persistence to disk persistence_path: Path for persisting context data """ # Core identifiers self._conversation_id = conversation_id self._speaker_id = speaker_id self._satellite_id = satellite_id self._room_id = room_id # Configuration self._max_entries_per_scope = max_entries_per_scope self._cleanup_interval = cleanup_interval_seconds self._enable_persistence = enable_persistence self._persistence_path = persistence_path or Path(f"./context/{conversation_id}") # Context storage organized by scope self._contexts: Dict[ContextScope, Dict[str, ContextEntry]] = { scope: OrderedDict() for scope in ContextScope } # Thread safety self._lock = threading.RLock() # Cleanup management self._last_cleanup = time.time() self._cleanup_stats = defaultdict(int) # Performance tracking self._access_count = 0 self._hit_count = 0 self._miss_count = 0 # Session tracking self._session_start = datetime.now(timezone.utc) self._turn_number = 0 pprint(f"Context manager initialized for conversation {conversation_id}") # Load persisted context if enabled if self._enable_persistence: self._load_persistent_context() # Core Context Operations def set( self, key: str, value: Any, scope: ContextScope = ContextScope.SESSION, retention_policy: ContextRetentionPolicy = ContextRetentionPolicy.SESSION_END, expires_in_seconds: Optional[float] = None, max_turns: Optional[int] = None, source: str = "system", tags: Optional[Set[str]] = None, priority: int = 0, metadata: Optional[Dict[str, Any]] = None ) -> None: """ Set a context value with specified scope and retention policy. Args: key: Context key value: Context value scope: Context scope retention_policy: How long to retain the context expires_in_seconds: Expiration time in seconds max_turns: Maximum turns to keep context source: Source that created the context tags: Optional tags for the context entry priority: Priority level (higher = more important) metadata: Additional metadata """ with self._lock: # Calculate expiration time expires_at = None if expires_in_seconds: expires_at = datetime.now(timezone.utc) + timedelta(seconds=expires_in_seconds) elif retention_policy == ContextRetentionPolicy.TIME_BASED: # Default to 1 hour for time-based retention expires_at = datetime.now(timezone.utc) + timedelta(hours=1) # Get or create context entry context_dict = self._contexts[scope] if key in context_dict: # Update existing entry entry = context_dict[key] entry.update_value(value, source) if expires_at: entry.expires_at = expires_at if max_turns: entry.max_turns = max_turns if tags: entry.tags.update(tags) if metadata: entry.metadata.update(metadata) else: # Create new entry entry = ContextEntry( key=key, value=value, scope=scope, retention_policy=retention_policy, expires_at=expires_at, max_turns=max_turns, source=source, tags=tags or set(), priority=priority, metadata=metadata or {} ) # Add to context context_dict[key] = entry # Move to end (LRU) context_dict.move_to_end(key) # Check if scope is over capacity if len(context_dict) > self._max_entries_per_scope: self._cleanup_scope_capacity(scope) # Persist if enabled if self._enable_persistence and scope in [ContextScope.SPEAKER, ContextScope.GLOBAL, ContextScope.PLUGIN]: self._persist_context_entry(entry) def get( self, key: str, scope: ContextScope = ContextScope.SESSION, default: Any = None, mark_accessed: bool = True ) -> Any: """ Get a context value from the specified scope. Args: key: Context key scope: Context scope to search default: Default value if not found mark_accessed: Whether to mark as accessed (for LRU) Returns: Context value or default """ with self._lock: self._access_count += 1 context_dict = self._contexts[scope] if key in context_dict: entry = context_dict[key] # Check if expired if entry.is_expired(): del context_dict[key] self._miss_count += 1 return default # Mark as accessed if mark_accessed: value = entry.access() context_dict.move_to_end(key) # LRU else: value = entry.value self._hit_count += 1 return value self._miss_count += 1 return default def get_multi_scope( self, key: str, scopes: List[ContextScope], default: Any = None, mark_accessed: bool = True ) -> Any: """ Get a context value searching multiple scopes in order. Args: key: Context key scopes: List of scopes to search in order default: Default value if not found mark_accessed: Whether to mark as accessed Returns: First found context value or default """ for scope in scopes: value = self.get(key, scope, None, mark_accessed) if value is not None: return value return default def has(self, key: str, scope: ContextScope = ContextScope.SESSION) -> bool: """ Check if a context key exists in the specified scope. Args: key: Context key scope: Context scope Returns: True if key exists and is not expired """ with self._lock: context_dict = self._contexts[scope] if key not in context_dict: return False entry = context_dict[key] if entry.is_expired(): del context_dict[key] return False return True def delete(self, key: str, scope: ContextScope = ContextScope.SESSION) -> bool: """ Delete a context entry. Args: key: Context key scope: Context scope Returns: True if entry was deleted """ with self._lock: context_dict = self._contexts[scope] if key in context_dict: del context_dict[key] # Remove from persistence if enabled if self._enable_persistence and scope in [ContextScope.SPEAKER, ContextScope.GLOBAL, ContextScope.PLUGIN]: self._remove_persistent_entry(key, scope) return True return False def update( self, data: Dict[str, Any], scope: ContextScope = ContextScope.SESSION, source: str = "system" ) -> None: """ Update multiple context values at once. Args: data: Dictionary of key-value pairs scope: Context scope source: Source identifier """ for key, value in data.items(): self.set(key, value, scope=scope, source=source) def clear_scope(self, scope: ContextScope) -> int: """ Clear all entries from a specific scope. Args: scope: Scope to clear Returns: Number of entries cleared """ with self._lock: context_dict = self._contexts[scope] count = len(context_dict) context_dict.clear() pprint(f"Cleared {count} entries from scope {scope.value}") return count def clear_all(self) -> int: """ Clear all context entries. Returns: Total number of entries cleared """ total_cleared = 0 for scope in ContextScope: total_cleared += self.clear_scope(scope) pprint(f"Cleared all context entries: {total_cleared} total") return total_cleared # Advanced Query Operations def search_by_tag( self, tag: str, scope: Optional[ContextScope] = None ) -> List[ContextEntry]: """ Search for context entries by tag. Args: tag: Tag to search for scope: Optional scope to limit search Returns: List of matching context entries """ with self._lock: results = [] scopes_to_search = [scope] if scope else list(ContextScope) for search_scope in scopes_to_search: context_dict = self._contexts[search_scope] for entry in context_dict.values(): if not entry.is_expired() and entry.has_tag(tag): results.append(entry) return results def search_by_source( self, source: str, scope: Optional[ContextScope] = None ) -> List[ContextEntry]: """ Search for context entries by source. Args: source: Source identifier scope: Optional scope to limit search Returns: List of matching context entries """ with self._lock: results = [] scopes_to_search = [scope] if scope else list(ContextScope) for search_scope in scopes_to_search: context_dict = self._contexts[search_scope] for entry in context_dict.values(): if not entry.is_expired() and entry.source == source: results.append(entry) return results def get_entries_by_scope(self, scope: ContextScope) -> List[ContextEntry]: """ Get all entries from a specific scope. Args: scope: Context scope Returns: List of context entries """ with self._lock: context_dict = self._contexts[scope] return [entry for entry in context_dict.values() if not entry.is_expired()] def get_all_entries(self) -> Dict[ContextScope, List[ContextEntry]]: """ Get all context entries organized by scope. Returns: Dictionary mapping scopes to entry lists """ with self._lock: result = {} for scope in ContextScope: result[scope] = self.get_entries_by_scope(scope) return result # Session and Turn Management def new_turn(self) -> None: """Mark the start of a new conversation turn.""" with self._lock: self._turn_number += 1 # Increment turn counter for all entries for scope_dict in self._contexts.values(): for entry in scope_dict.values(): entry.increment_turn() # Trigger cleanup if needed self._maybe_cleanup() def end_session(self) -> None: """Mark the end of the conversation session.""" with self._lock: # Clean up session-scoped context self.clear_scope(ContextScope.SESSION) # Clean up temporary context self.clear_scope(ContextScope.TEMPORARY) # Clean up entries with SESSION_END retention policy for scope_dict in self._contexts.values(): expired_keys = [] for key, entry in scope_dict.items(): if entry.retention_policy == ContextRetentionPolicy.SESSION_END: expired_keys.append(key) for key in expired_keys: del scope_dict[key] pprint(f"Session ended for conversation {self._conversation_id}") # Cleanup and Maintenance def _maybe_cleanup(self) -> None: """Perform cleanup if interval has passed.""" now = time.time() if now - self._last_cleanup > self._cleanup_interval: self.cleanup_expired() self._last_cleanup = now def cleanup_expired(self) -> int: """ Clean up expired context entries. Returns: Number of entries cleaned up """ with self._lock: total_cleaned = 0 for scope, context_dict in self._contexts.items(): expired_keys = [] for key, entry in context_dict.items(): if entry.is_expired(): expired_keys.append(key) for key in expired_keys: del context_dict[key] total_cleaned += 1 self._cleanup_stats[scope.value] += len(expired_keys) if total_cleaned > 0: pprint(f"Cleaned up {total_cleaned} expired context entries") return total_cleaned def _cleanup_scope_capacity(self, scope: ContextScope) -> None: """Clean up a scope that's over capacity using LRU.""" context_dict = self._contexts[scope] # Remove oldest entries (LRU) until under capacity while len(context_dict) > self._max_entries_per_scope: # Remove from the beginning (oldest) oldest_key = next(iter(context_dict)) del context_dict[oldest_key] self._cleanup_stats[f"{scope.value}_capacity"] += 1 def force_cleanup( self, max_age_hours: float = 24.0, max_entries_per_scope: Optional[int] = None ) -> int: """ Force cleanup of old context entries. Args: max_age_hours: Maximum age in hours max_entries_per_scope: Override max entries per scope Returns: Number of entries cleaned up """ with self._lock: cutoff_time = datetime.now(timezone.utc) - timedelta(hours=max_age_hours) max_entries = max_entries_per_scope or self._max_entries_per_scope total_cleaned = 0 for scope, context_dict in self._contexts.items(): # Remove old entries old_keys = [] for key, entry in context_dict.items(): if (entry.created_at < cutoff_time and entry.retention_policy != ContextRetentionPolicy.PERSISTENT): old_keys.append(key) for key in old_keys: del context_dict[key] total_cleaned += 1 # Remove excess entries (LRU) while len(context_dict) > max_entries: oldest_key = next(iter(context_dict)) del context_dict[oldest_key] total_cleaned += 1 pprint(f"Force cleanup removed {total_cleaned} context entries") return total_cleaned # Persistence Operations def _persist_context_entry(self, entry: ContextEntry) -> None: """Persist a single context entry to disk.""" if not self._enable_persistence: return try: # Create persistence directory self._persistence_path.mkdir(parents=True, exist_ok=True) # Create scope-specific file scope_file = self._persistence_path / f"{entry.scope.value}.json" # Load existing data data = {} if scope_file.exists(): with open(scope_file, 'r', encoding='utf-8') as f: data = json.load(f) # Update with new entry data[entry.key] = entry.to_dict() # Save back to file with open(scope_file, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) except Exception as e: pprint(f"Error persisting context entry: {e}") def _load_persistent_context(self) -> None: """Load persistent context from disk.""" if not self._enable_persistence or not self._persistence_path.exists(): return try: loaded_count = 0 for scope_file in self._persistence_path.glob("*.json"): scope_name = scope_file.stem try: scope = ContextScope(scope_name) except ValueError: continue # Skip unknown scopes with open(scope_file, 'r', encoding='utf-8') as f: data = json.load(f) for key, entry_data in data.items(): try: entry = ContextEntry.from_dict(entry_data) # Only load non-expired entries if not entry.is_expired(): self._contexts[scope][key] = entry loaded_count += 1 except Exception as e: pprint(f"Error loading context entry {key}: {e}") if loaded_count > 0: pprint(f"Loaded {loaded_count} persistent context entries") except Exception as e: pprint(f"Error loading persistent context: {e}") def _remove_persistent_entry(self, key: str, scope: ContextScope) -> None: """Remove a persistent context entry.""" if not self._enable_persistence: return try: scope_file = self._persistence_path / f"{scope.value}.json" if scope_file.exists(): with open(scope_file, 'r', encoding='utf-8') as f: data = json.load(f) if key in data: del data[key] with open(scope_file, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) except Exception as e: pprint(f"Error removing persistent context entry: {e}") def save_to_file(self, file_path: Path) -> None: """ Save all context data to a file. Args: file_path: Path to save the context data """ try: data = { "conversation_id": self._conversation_id, "speaker_id": self._speaker_id, "satellite_id": self._satellite_id, "room_id": self._room_id, "session_start": self._session_start.isoformat(), "turn_number": self._turn_number, "contexts": { scope.value: { key: entry.to_dict() for key, entry in context_dict.items() if not entry.is_expired() } for scope, context_dict in self._contexts.items() } } # Compress if file is large if json.dumps(data).encode('utf-8').__len__() > 1024 * 1024: # 1MB with gzip.open(f"{file_path}.gz", 'wt', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) else: with open(file_path, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) pprint(f"Context saved to {file_path}") except Exception as e: raise ConversationContextError(f"Failed to save context: {e}") @classmethod def load_from_file(cls, file_path: Path) -> 'ConversationContext': """ Load context data from a file. Args: file_path: Path to load from Returns: ConversationContext: Loaded context manager """ try: # Try compressed format first if file_path.with_suffix('.gz').exists(): with gzip.open(file_path.with_suffix('.gz'), 'rt', encoding='utf-8') as f: data = json.load(f) else: with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) # Create context manager context = cls( conversation_id=data["conversation_id"], speaker_id=data.get("speaker_id"), satellite_id=data.get("satellite_id"), room_id=data.get("room_id") ) # Restore state context._session_start = datetime.fromisoformat(data["session_start"]) context._turn_number = data.get("turn_number", 0) # Restore context entries for scope_name, scope_data in data.get("contexts", {}).items(): try: scope = ContextScope(scope_name) for key, entry_data in scope_data.items(): entry = ContextEntry.from_dict(entry_data) if not entry.is_expired(): context._contexts[scope][key] = entry except ValueError: continue # Skip unknown scopes pprint(f"Context loaded from {file_path}") return context except Exception as e: raise ConversationContextError(f"Failed to load context: {e}") # Analytics and Monitoring def get_stats(self) -> Dict[str, Any]: """ Get context statistics and performance metrics. Returns: Dictionary containing context statistics """ with self._lock: total_entries = sum(len(context_dict) for context_dict in self._contexts.values()) scope_stats = {} for scope, context_dict in self._contexts.items(): scope_stats[scope.value] = { "count": len(context_dict), "oldest": min((e.created_at for e in context_dict.values()), default=None), "newest": max((e.created_at for e in context_dict.values()), default=None), "total_versions": sum(e.version for e in context_dict.values()), "expired_count": sum(1 for e in context_dict.values() if e.is_expired()) } hit_rate = (self._hit_count / max(1, self._access_count)) * 100 return { "conversation_id": self._conversation_id, "total_entries": total_entries, "total_accesses": self._access_count, "cache_hits": self._hit_count, "cache_misses": self._miss_count, "hit_rate_percent": round(hit_rate, 2), "session_duration_seconds": (datetime.now(timezone.utc) - self._session_start).total_seconds(), "turn_number": self._turn_number, "scope_stats": scope_stats, "cleanup_stats": dict(self._cleanup_stats), "last_cleanup": self._last_cleanup } def __str__(self) -> str: """String representation of the context manager.""" total_entries = sum(len(context_dict) for context_dict in self._contexts.values()) return f"ConversationContext(conversation={self._conversation_id}, entries={total_entries})" def __repr__(self) -> str: """Detailed representation of the context manager.""" return (f"ConversationContext(conversation_id='{self._conversation_id}', " f"speaker_id='{self._speaker_id}', satellite_id='{self._satellite_id}', " f"room_id='{self._room_id}', turn_number={self._turn_number})") # Factory function def create_conversation_context( conversation_id: str, speaker_id: Optional[str] = None, satellite_id: Optional[str] = None, room_id: Optional[str] = None, **kwargs ) -> ConversationContext: """ Create a ConversationContext with default configuration. Args: conversation_id: Conversation identifier speaker_id: Speaker identifier satellite_id: Satellite identifier room_id: Room identifier **kwargs: Additional context parameters Returns: ConversationContext: Configured context manager """ return ConversationContext( conversation_id=conversation_id, speaker_id=speaker_id, satellite_id=satellite_id, room_id=room_id, **kwargs ) # Module exports __all__ = [ "ConversationContext", "ConversationContextError", "ContextScope", "ContextEntry", "ContextRetentionPolicy", "create_conversation_context" ]