Upload 8 files

audio_buffer_manager.py

@@ -0,0 +1,296 @@
+"""
+Audio Buffer Manager for Flare
+==============================
+Manages audio buffering, silence detection, and chunk processing
+"""
+import asyncio
+from typing import Dict, Optional, List, Tuple
+from collections import deque
+from datetime import datetime
+import base64
+import numpy as np
+from dataclasses import dataclass
+import traceback
+
+from event_bus import EventBus, Event, EventType
+from logger import log_info, log_error, log_debug, log_warning
+
+
+@dataclass
+class AudioChunk:
+    """Audio chunk with metadata"""
+    data: bytes
+    timestamp: datetime
+    chunk_index: int
+    is_speech: bool = True
+    energy_level: float = 0.0
+
+
+class SilenceDetector:
+    """Detect silence in audio stream"""
+    
+    def __init__(self, 
+                 threshold_ms: int = 2000,
+                 energy_threshold: float = 0.01,
+                 sample_rate: int = 16000):
+        self.threshold_ms = threshold_ms
+        self.energy_threshold = energy_threshold
+        self.sample_rate = sample_rate
+        self.silence_start: Optional[datetime] = None
+        
+    def detect_silence(self, audio_chunk: bytes) -> Tuple[bool, int]:
+        """
+        Detect if chunk is silence and return duration
+        Returns: (is_silence, silence_duration_ms)
+        """
+        try:
+            # Handle empty or invalid chunk
+            if not audio_chunk or len(audio_chunk) < 2:
+                return True, 0
+                
+            # Ensure even number of bytes for 16-bit audio
+            if len(audio_chunk) % 2 != 0:
+                audio_chunk = audio_chunk[:-1]
+                
+            # Convert to numpy array
+            audio_data = np.frombuffer(audio_chunk, dtype=np.int16)
+            
+            if len(audio_data) == 0:
+                return True, 0
+                
+            # Calculate RMS energy
+            rms = np.sqrt(np.mean(audio_data.astype(float) ** 2))
+            normalized_rms = rms / 32768.0  # Normalize for 16-bit audio
+            
+            is_silence = normalized_rms < self.energy_threshold
+            
+            # Track silence duration
+            now = datetime.utcnow()
+            if is_silence:
+                if self.silence_start is None:
+                    self.silence_start = now
+                duration_ms = int((now - self.silence_start).total_seconds() * 1000)
+            else:
+                self.silence_start = None
+                duration_ms = 0
+                
+            return is_silence, duration_ms
+            
+        except Exception as e:
+            log_warning(f"Silence detection error: {e}")
+            return False, 0
+            
+    def reset(self):
+        """Reset silence detection state"""
+        self.silence_start = None
+
+
+class AudioBuffer:
+    """Manage audio chunks for a session"""
+    
+    def __init__(self, 
+                 session_id: str,
+                 max_chunks: int = 1000,
+                 chunk_size_bytes: int = 4096):
+        self.session_id = session_id
+        self.max_chunks = max_chunks
+        self.chunk_size_bytes = chunk_size_bytes
+        self.chunks: deque[AudioChunk] = deque(maxlen=max_chunks)
+        self.chunk_counter = 0
+        self.total_bytes = 0
+        self.lock = asyncio.Lock()
+        
+    async def add_chunk(self, audio_data: bytes, timestamp: Optional[datetime] = None) -> AudioChunk:
+        """Add audio chunk to buffer"""
+        async with self.lock:
+            if timestamp is None:
+                timestamp = datetime.utcnow()
+                
+            chunk = AudioChunk(
+                data=audio_data,
+                timestamp=timestamp,
+                chunk_index=self.chunk_counter
+            )
+            
+            self.chunks.append(chunk)
+            self.chunk_counter += 1
+            self.total_bytes += len(audio_data)
+            
+            return chunk
+            
+    async def get_recent_audio(self, duration_ms: int = 5000) -> bytes:
+        """Get recent audio data"""
+        async with self.lock:
+            cutoff_time = datetime.utcnow()
+            audio_parts = []
+            
+            # Iterate backwards through chunks
+            for chunk in reversed(self.chunks):
+                time_diff = (cutoff_time - chunk.timestamp).total_seconds() * 1000
+                if time_diff > duration_ms:
+                    break
+                audio_parts.append(chunk.data)
+                
+            # Reverse to maintain chronological order
+            audio_parts.reverse()
+            return b''.join(audio_parts)
+            
+    async def clear(self):
+        """Clear buffer"""
+        async with self.lock:
+            self.chunks.clear()
+            self.chunk_counter = 0
+            self.total_bytes = 0
+            
+    def get_stats(self) -> Dict[str, Any]:
+        """Get buffer statistics"""
+        return {
+            "chunks": len(self.chunks),
+            "total_bytes": self.total_bytes,
+            "chunk_counter": self.chunk_counter,
+            "oldest_chunk": self.chunks[0].timestamp if self.chunks else None,
+            "newest_chunk": self.chunks[-1].timestamp if self.chunks else None
+        }
+
+
+class AudioBufferManager:
+    """Manage audio buffers for all sessions"""
+    
+    def __init__(self, event_bus: EventBus):
+        self.event_bus = event_bus
+        self.session_buffers: Dict[str, AudioBuffer] = {}
+        self.silence_detectors: Dict[str, SilenceDetector] = {}
+        self._setup_event_handlers()
+        
+    def _setup_event_handlers(self):
+        """Subscribe to audio events"""
+        self.event_bus.subscribe(EventType.SESSION_STARTED, self._handle_session_started)
+        self.event_bus.subscribe(EventType.SESSION_ENDED, self._handle_session_ended)
+        self.event_bus.subscribe(EventType.AUDIO_CHUNK_RECEIVED, self._handle_audio_chunk)
+        
+    async def _handle_session_started(self, event: Event):
+        """Initialize buffer for new session"""
+        session_id = event.session_id
+        config = event.data
+        
+        # Create audio buffer
+        self.session_buffers[session_id] = AudioBuffer(
+            session_id=session_id,
+            max_chunks=config.get("max_chunks", 1000),
+            chunk_size_bytes=config.get("chunk_size", 4096)
+        )
+        
+        # Create silence detector
+        self.silence_detectors[session_id] = SilenceDetector(
+            threshold_ms=config.get("silence_threshold_ms", 2000),
+            energy_threshold=config.get("energy_threshold", 0.01),
+            sample_rate=config.get("sample_rate", 16000)
+        )
+        
+        log_info(f"📦 Audio buffer initialized", session_id=session_id)
+        
+    async def _handle_session_ended(self, event: Event):
+        """Cleanup session buffers"""
+        session_id = event.session_id
+        
+        # Clear and remove buffer
+        if session_id in self.session_buffers:
+            await self.session_buffers[session_id].clear()
+            del self.session_buffers[session_id]
+            
+        # Remove silence detector
+        if session_id in self.silence_detectors:
+            del self.silence_detectors[session_id]
+            
+        log_info(f"📦 Audio buffer cleaned up", session_id=session_id)
+        
+    async def _handle_audio_chunk(self, event: Event):
+        """Process incoming audio chunk"""
+        session_id = event.session_id
+        
+        buffer = self.session_buffers.get(session_id)
+        detector = self.silence_detectors.get(session_id)
+        
+        if not buffer or not detector:
+            log_warning(f"⚠️ No buffer for session", session_id=session_id)
+            return
+            
+        try:
+            # Decode audio data
+            audio_data = base64.b64decode(event.data.get("audio_data", ""))
+            
+            # Add to buffer
+            chunk = await buffer.add_chunk(audio_data)
+            
+            # Detect silence
+            is_silence, silence_duration = detector.detect_silence(audio_data)
+            
+            # Update chunk metadata
+            chunk.is_speech = not is_silence
+            chunk.energy_level = 1.0 - (silence_duration / detector.threshold_ms)
+            
+            # Check for end of speech
+            if silence_duration > detector.threshold_ms:
+                log_info(
+                    f"🔇 Speech ended (silence: {silence_duration}ms)",
+                    session_id=session_id
+                )
+                
+                # Get complete audio
+                complete_audio = await buffer.get_recent_audio()
+                
+                # Publish speech ended event
+                await self.event_bus.publish(Event(
+                    type=EventType.STT_RESULT,
+                    session_id=session_id,
+                    data={
+                        "audio_data": base64.b64encode(complete_audio).decode(),
+                        "is_final": True,
+                        "silence_triggered": True
+                    },
+                    priority=5
+                ))
+                
+                # Reset for next utterance
+                await self.reset_buffer(session_id)
+                
+            # Log periodically
+            if chunk.chunk_index % 100 == 0:
+                stats = buffer.get_stats()
+                log_debug(
+                    f"📊 Buffer stats",
+                    session_id=session_id,
+                    **stats
+                )
+                
+        except Exception as e:
+            log_error(
+                f"❌ Error processing audio chunk",
+                session_id=session_id,
+                error=str(e),
+                traceback=traceback.format_exc()
+            )
+            
+    async def get_buffer(self, session_id: str) -> Optional[AudioBuffer]:
+        """Get buffer for session"""
+        return self.session_buffers.get(session_id)
+        
+    async def reset_buffer(self, session_id: str):
+        """Reset buffer for new utterance"""
+        buffer = self.session_buffers.get(session_id)
+        detector = self.silence_detectors.get(session_id)
+        
+        if buffer:
+            await buffer.clear()
+            
+        if detector:
+            detector.reset()
+            
+        log_debug(f"🔄 Audio buffer reset", session_id=session_id)
+        
+    def get_all_stats(self) -> Dict[str, Dict[str, Any]]:
+        """Get statistics for all buffers"""
+        stats = {}
+        for session_id, buffer in self.session_buffers.items():
+            stats[session_id] = buffer.get_stats()
+        return stats

event_bus.py

@@ -0,0 +1,410 @@
+"""
+Event Bus Implementation for Flare
+==================================
+Provides async event publishing and subscription mechanism
+"""
+import asyncio
+from typing import Dict, List, Callable, Any, Optional
+from enum import Enum
+from dataclasses import dataclass
+from datetime import datetime
+import traceback
+from collections import defaultdict
+import sys
+
+from logger import log_info, log_error, log_debug, log_warning
+
+
+class EventType(Enum):
+    """All event types in the system"""
+    # Lifecycle events
+    SESSION_STARTED = "session_started"
+    SESSION_ENDED = "session_ended"
+    
+    # STT events
+    STT_STARTED = "stt_started"
+    STT_STOPPED = "stt_stopped"
+    STT_RESULT = "stt_result"
+    STT_ERROR = "stt_error"
+    STT_READY = "stt_ready"
+    
+    # TTS events
+    TTS_STARTED = "tts_started"
+    TTS_CHUNK_READY = "tts_chunk_ready"
+    TTS_COMPLETED = "tts_completed"
+    TTS_ERROR = "tts_error"
+    
+    # Audio events
+    AUDIO_PLAYBACK_STARTED = "audio_playback_started"
+    AUDIO_PLAYBACK_COMPLETED = "audio_playback_completed"
+    AUDIO_BUFFER_LOW = "audio_buffer_low"
+    AUDIO_CHUNK_RECEIVED = "audio_chunk_received"
+    
+    # LLM events
+    LLM_PROCESSING_STARTED = "llm_processing_started"
+    LLM_RESPONSE_READY = "llm_response_ready"
+    LLM_ERROR = "llm_error"
+    
+    # Error events
+    CRITICAL_ERROR = "critical_error"
+    RECOVERABLE_ERROR = "recoverable_error"
+    
+    # State events
+    STATE_TRANSITION = "state_transition"
+    STATE_ROLLBACK = "state_rollback"
+    
+    # WebSocket events
+    WEBSOCKET_CONNECTED = "websocket_connected"
+    WEBSOCKET_DISCONNECTED = "websocket_disconnected"
+    WEBSOCKET_MESSAGE = "websocket_message"
+    WEBSOCKET_ERROR = "websocket_error"
+
+
+@dataclass
+class Event:
+    """Event data structure"""
+    type: EventType
+    session_id: str
+    data: Dict[str, Any]
+    timestamp: datetime = None
+    priority: int = 0  # Higher priority = processed first
+    
+    def __post_init__(self):
+        if self.timestamp is None:
+            self.timestamp = datetime.utcnow()
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary for serialization"""
+        return {
+            "type": self.type.value,
+            "session_id": self.session_id,
+            "data": self.data,
+            "timestamp": self.timestamp.isoformat(),
+            "priority": self.priority
+        }
+
+
+class EventBus:
+    """Central event bus for component communication with session isolation"""
+    
+    def __init__(self):
+        self._subscribers: Dict[EventType, List[Callable]] = defaultdict(list)
+        self._session_handlers: Dict[str, Dict[EventType, List[Callable]]] = defaultdict(lambda: defaultdict(list))
+        
+        # Session-specific queues for parallel processing
+        self._session_queues: Dict[str, asyncio.PriorityQueue] = {}
+        self._session_processors: Dict[str, asyncio.Task] = {}
+        
+        # Global queue for non-session events
+        self._global_queue: asyncio.PriorityQueue = asyncio.PriorityQueue()
+        self._global_processor: Optional[asyncio.Task] = None
+        
+        self._running = False
+        self._event_history: List[Event] = []
+        self._max_history_size = 1000
+        
+    async def start(self):
+        """Start the event processor"""
+        if self._running:
+            log_warning("EventBus already running")
+            return
+            
+        self._running = True
+        
+        # Start global processor
+        self._global_processor = asyncio.create_task(self._process_global_events())
+        
+        log_info("✅ EventBus started")
+        
+    async def stop(self):
+        """Stop the event processor"""
+        self._running = False
+        
+        # Stop all session processors
+        for session_id, task in list(self._session_processors.items()):
+            task.cancel()
+            try:
+                await asyncio.wait_for(task, timeout=2.0)
+            except (asyncio.TimeoutError, asyncio.CancelledError):
+                pass
+                
+        # Stop global processor
+        if self._global_processor:
+            await self._global_queue.put((999, None))  # Sentinel
+            try:
+                await asyncio.wait_for(self._global_processor, timeout=5.0)
+            except asyncio.TimeoutError:
+                log_warning("EventBus global processor timeout, cancelling")
+                self._global_processor.cancel()
+                
+        log_info("✅ EventBus stopped")
+        
+    async def publish(self, event: Event):
+        """Publish an event to the bus"""
+        if not self._running:
+            log_error("EventBus not running, cannot publish event", event_type=event.type.value)
+            return
+            
+        # Add to history
+        self._event_history.append(event)
+        if len(self._event_history) > self._max_history_size:
+            self._event_history.pop(0)
+            
+        # Route to appropriate queue
+        if event.session_id:
+            # Ensure session queue exists
+            if event.session_id not in self._session_queues:
+                await self._create_session_processor(event.session_id)
+                
+            # Add to session queue
+            queue = self._session_queues[event.session_id]
+            await queue.put((-event.priority, event))
+        else:
+            # Add to global queue
+            await self._global_queue.put((-event.priority, event))
+            
+        log_debug(
+            f"📤 Event published",
+            event_type=event.type.value,
+            session_id=event.session_id,
+            priority=event.priority
+        )
+        
+    async def _create_session_processor(self, session_id: str):
+        """Create a processor for session-specific events"""
+        if session_id in self._session_processors:
+            return
+            
+        # Create queue
+        self._session_queues[session_id] = asyncio.PriorityQueue()
+        
+        # Create processor task
+        task = asyncio.create_task(self._process_session_events(session_id))
+        self._session_processors[session_id] = task
+        
+        log_debug(f"📌 Created session processor", session_id=session_id)
+        
+    async def _process_session_events(self, session_id: str):
+        """Process events for a specific session"""
+        queue = self._session_queues[session_id]
+        log_info(f"🔄 Session event processor started", session_id=session_id)
+        
+        while self._running:
+            try:
+                # Wait for event with timeout
+                priority, event = await asyncio.wait_for(
+                    queue.get(),
+                    timeout=60.0  # Longer timeout for sessions
+                )
+                
+                # Check for session cleanup
+                if event is None:
+                    break
+                    
+                # Process the event
+                await self._dispatch_event(event)
+                
+            except asyncio.TimeoutError:
+                # Check if session is still active
+                if session_id not in self._session_handlers:
+                    log_info(f"Session inactive, stopping processor", session_id=session_id)
+                    break
+                continue
+            except Exception as e:
+                log_error(
+                    f"❌ Error processing session event",
+                    session_id=session_id,
+                    error=str(e),
+                    traceback=traceback.format_exc()
+                )
+                
+        # Cleanup
+        self._session_queues.pop(session_id, None)
+        self._session_processors.pop(session_id, None)
+        log_info(f"🔄 Session event processor stopped", session_id=session_id)
+        
+    async def _process_global_events(self):
+        """Process global events (no session_id)"""
+        log_info("🔄 Global event processor started")
+        
+        while self._running:
+            try:
+                priority, event = await asyncio.wait_for(
+                    self._global_queue.get(),
+                    timeout=1.0
+                )
+                
+                if event is None:  # Sentinel
+                    break
+                    
+                await self._dispatch_event(event)
+                
+            except asyncio.TimeoutError:
+                continue
+            except Exception as e:
+                log_error(
+                    "❌ Error processing global event",
+                    error=str(e),
+                    traceback=traceback.format_exc()
+                )
+                
+        log_info("🔄 Global event processor stopped")
+        
+    def subscribe(self, event_type: EventType, handler: Callable):
+        """Subscribe to an event type globally"""
+        self._subscribers[event_type].append(handler)
+        log_debug(f"📌 Global subscription added", event_type=event_type.value)
+        
+    def subscribe_session(self, session_id: str, event_type: EventType, handler: Callable):
+        """Subscribe to an event type for a specific session"""
+        self._session_handlers[session_id][event_type].append(handler)
+        log_debug(
+            f"📌 Session subscription added",
+            event_type=event_type.value,
+            session_id=session_id
+        )
+        
+    def unsubscribe(self, event_type: EventType, handler: Callable):
+        """Unsubscribe from an event type"""
+        if handler in self._subscribers[event_type]:
+            self._subscribers[event_type].remove(handler)
+            log_debug(f"📌 Global subscription removed", event_type=event_type.value)
+            
+    def unsubscribe_session(self, session_id: str, event_type: EventType = None):
+        """Unsubscribe session handlers"""
+        if event_type:
+            # Remove specific event type for session
+            if session_id in self._session_handlers and event_type in self._session_handlers[session_id]:
+                del self._session_handlers[session_id][event_type]
+        else:
+            # Remove all handlers for session
+            if session_id in self._session_handlers:
+                del self._session_handlers[session_id]
+                log_debug(f"📌 All session subscriptions removed", session_id=session_id)
+
+        
+    async def _dispatch_event(self, event: Event):
+        """Dispatch event to all subscribers"""
+        try:
+            handlers = []
+            
+            # Get global handlers
+            if event.type in self._subscribers:
+                handlers.extend(self._subscribers[event.type])
+                
+            # Get session-specific handlers
+            if event.session_id in self._session_handlers:
+                if event.type in self._session_handlers[event.session_id]:
+                    handlers.extend(self._session_handlers[event.session_id][event.type])
+                    
+            if not handlers:
+                log_debug(
+                    f"📭 No handlers for event",
+                    event_type=event.type.value,
+                    session_id=event.session_id
+                )
+                return
+                
+            log_debug(
+                f"📨 Dispatching event to {len(handlers)} handlers",
+                event_type=event.type.value,
+                session_id=event.session_id
+            )
+            
+            # Call all handlers concurrently
+            tasks = []
+            for handler in handlers:
+                if asyncio.iscoroutinefunction(handler):
+                    task = asyncio.create_task(handler(event))
+                else:
+                    # Wrap sync handler in async
+                    task = asyncio.create_task(asyncio.to_thread(handler, event))
+                tasks.append(task)
+                
+            # Wait for all handlers to complete
+            results = await asyncio.gather(*tasks, return_exceptions=True)
+            
+            # Log any exceptions
+            for i, result in enumerate(results):
+                if isinstance(result, Exception):
+                    log_error(
+                        f"❌ Handler error",
+                        handler=handlers[i].__name__,
+                        event_type=event.type.value,
+                        error=str(result),
+                        traceback=traceback.format_exception(type(result), result, result.__traceback__)
+                    )
+                    
+        except Exception as e:
+            log_error(
+                f"❌ Error dispatching event",
+                event_type=event.type.value,
+                error=str(e),
+                traceback=traceback.format_exc()
+            )
+            
+    def get_event_history(self, session_id: Optional[str] = None, event_type: Optional[EventType] = None) -> List[Event]:
+        """Get event history with optional filters"""
+        history = self._event_history
+        
+        if session_id:
+            history = [e for e in history if e.session_id == session_id]
+            
+        if event_type:
+            history = [e for e in history if e.type == event_type]
+            
+        return history
+        
+    def clear_session_data(self, session_id: str):
+        """Clear all session-related data and stop processor"""
+        # Remove session handlers
+        self.unsubscribe_session(session_id)
+        
+        # Stop session processor
+        if session_id in self._session_processors:
+            task = self._session_processors[session_id]
+            task.cancel()
+            
+        # Clear queues
+        self._session_queues.pop(session_id, None)
+        self._session_processors.pop(session_id, None)
+        
+        # Remove session events from history
+        self._event_history = [e for e in self._event_history if e.session_id != session_id]
+        
+        log_debug(f"🧹 Session data cleared", session_id=session_id)
+
+
+# Global event bus instance
+event_bus = EventBus()
+
+
+# Helper functions for common event publishing patterns
+async def publish_error(session_id: str, error_type: str, error_message: str, details: Dict[str, Any] = None):
+    """Helper to publish error events"""
+    event = Event(
+        type=EventType.RECOVERABLE_ERROR if error_type != "critical" else EventType.CRITICAL_ERROR,
+        session_id=session_id,
+        data={
+            "error_type": error_type,
+            "message": error_message,
+            "details": details or {}
+        },
+        priority=10  # High priority for errors
+    )
+    await event_bus.publish(event)
+    
+
+async def publish_state_transition(session_id: str, from_state: str, to_state: str, reason: str = None):
+    """Helper to publish state transition events"""
+    event = Event(
+        type=EventType.STATE_TRANSITION,
+        session_id=session_id,
+        data={
+            "from_state": from_state,
+            "to_state": to_state,
+            "reason": reason
+        },
+        priority=5  # Medium priority for state changes
+    )
+    await event_bus.publish(event)

llm_manager.py

@@ -0,0 +1,689 @@
+"""
+LLM Manager for Flare
+====================
+Manages LLM interactions per session with stateless approach
+"""
+import asyncio
+from typing import Dict, Optional, Any, List
+from datetime import datetime
+import traceback
+from dataclasses import dataclass, field
+import json
+
+from event_bus import EventBus, Event, EventType, publish_error
+from resource_manager import ResourceManager, ResourceType
+from session import Session
+from llm_factory import LLMFactory
+from llm_interface import LLMInterface
+from prompt_builder import build_intent_prompt, build_parameter_prompt
+from logger import log_info, log_error, log_debug, log_warning
+from config_provider import ConfigProvider
+
+
+@dataclass
+class LLMJob:
+    """LLM processing job"""
+    job_id: str
+    session_id: str
+    input_text: str
+    job_type: str  # "intent_detection", "parameter_collection", "response_generation"
+    created_at: datetime = field(default_factory=datetime.utcnow)
+    completed_at: Optional[datetime] = None
+    response_text: Optional[str] = None
+    detected_intent: Optional[str] = None
+    error: Optional[str] = None
+    metadata: Dict[str, Any] = field(default_factory=dict)
+    
+    def complete(self, response_text: str, intent: Optional[str] = None):
+        """Mark job as completed"""
+        self.response_text = response_text
+        self.detected_intent = intent
+        self.completed_at = datetime.utcnow()
+        
+    def fail(self, error: str):
+        """Mark job as failed"""
+        self.error = error
+        self.completed_at = datetime.utcnow()
+
+
+@dataclass
+class LLMSession:
+    """LLM session wrapper"""
+    session_id: str
+    session: Session
+    llm_instance: LLMInterface
+    active_job: Optional[LLMJob] = None
+    job_history: List[LLMJob] = field(default_factory=list)
+    created_at: datetime = field(default_factory=datetime.utcnow)
+    last_activity: datetime = field(default_factory=datetime.utcnow)
+    total_jobs = 0
+    total_tokens = 0
+    
+    def update_activity(self):
+        """Update last activity timestamp"""
+        self.last_activity = datetime.utcnow()
+
+
+class LLMManager:
+    """Manages LLM interactions with stateless approach"""
+    
+    def __init__(self, event_bus: EventBus, resource_manager: ResourceManager):
+        self.event_bus = event_bus
+        self.resource_manager = resource_manager
+        self.llm_sessions: Dict[str, LLMSession] = {}
+        self.config = ConfigProvider.get()
+        self._setup_event_handlers()
+        self._setup_resource_pool()
+        
+    def _setup_event_handlers(self):
+        """Subscribe to LLM-related events"""
+        self.event_bus.subscribe(EventType.LLM_PROCESSING_STARTED, self._handle_llm_processing)
+        self.event_bus.subscribe(EventType.SESSION_ENDED, self._handle_session_ended)
+        
+    def _setup_resource_pool(self):
+        """Setup LLM instance pool"""
+        self.resource_manager.register_pool(
+            resource_type=ResourceType.LLM_CONTEXT,
+            factory=self._create_llm_instance,
+            max_idle=2,  # Lower pool size for LLM
+            max_age_seconds=900  # 15 minutes
+        )
+        
+    async def _create_llm_instance(self) -> LLMInterface:
+        """Factory for creating LLM instances"""
+        try:
+            llm_instance = LLMFactory.create_provider()
+            if not llm_instance:
+                raise ValueError("Failed to create LLM instance")
+                
+            log_debug("🤖 Created new LLM instance")
+            return llm_instance
+            
+        except Exception as e:
+            log_error(f"❌ Failed to create LLM instance", error=str(e))
+            raise
+            
+    async def _handle_llm_processing(self, event: Event):
+        """Handle LLM processing request"""
+        session_id = event.session_id
+        input_text = event.data.get("text", "")
+        
+        if not input_text:
+            log_warning(f"⚠️ Empty text for LLM", session_id=session_id)
+            return
+            
+        try:
+            log_info(
+                f"🤖 Starting LLM processing",
+                session_id=session_id,
+                text_length=len(input_text)
+            )
+            
+            # Get or create LLM session
+            llm_session = await self._get_or_create_session(session_id)
+            if not llm_session:
+                raise ValueError("Failed to create LLM session")
+                
+            # Determine job type based on session state
+            job_type = self._determine_job_type(llm_session.session)
+            
+            # Create job
+            job_id = f"{session_id}_{llm_session.total_jobs}"
+            job = LLMJob(
+                job_id=job_id,
+                session_id=session_id,
+                input_text=input_text,
+                job_type=job_type,
+                metadata={
+                    "session_state": llm_session.session.state,
+                    "current_intent": llm_session.session.current_intent
+                }
+            )
+            
+            llm_session.active_job = job
+            llm_session.total_jobs += 1
+            llm_session.update_activity()
+            
+            # Process based on job type
+            if job_type == "intent_detection":
+                await self._process_intent_detection(llm_session, job)
+            elif job_type == "parameter_collection":
+                await self._process_parameter_collection(llm_session, job)
+            else:
+                await self._process_response_generation(llm_session, job)
+                
+        except Exception as e:
+            log_error(
+                f"❌ Failed to process LLM request",
+                session_id=session_id,
+                error=str(e),
+                traceback=traceback.format_exc()
+            )
+            
+            # Publish error event
+            await publish_error(
+                session_id=session_id,
+                error_type="llm_error",
+                error_message=f"LLM processing failed: {str(e)}"
+            )
+            
+    async def _get_or_create_session(self, session_id: str) -> Optional[LLMSession]:
+        """Get or create LLM session"""
+        if session_id in self.llm_sessions:
+            return self.llm_sessions[session_id]
+            
+        # Get session from store
+        from session import session_store
+        session = session_store.get_session(session_id)
+        if not session:
+            log_error(f"❌ Session not found", session_id=session_id)
+            return None
+            
+        # Acquire LLM instance from pool
+        resource_id = f"llm_{session_id}"
+        llm_instance = await self.resource_manager.acquire(
+            resource_id=resource_id,
+            session_id=session_id,
+            resource_type=ResourceType.LLM_CONTEXT,
+            cleanup_callback=self._cleanup_llm_instance
+        )
+        
+        # Create LLM session
+        llm_session = LLMSession(
+            session_id=session_id,
+            session=session,
+            llm_instance=llm_instance
+        )
+        
+        self.llm_sessions[session_id] = llm_session
+        return llm_session
+        
+    def _determine_job_type(self, session: Session) -> str:
+        """Determine job type based on session state"""
+        if session.state == "idle":
+            return "intent_detection"
+        elif session.state == "collect_params":
+            return "parameter_collection"
+        else:
+            return "response_generation"
+            
+    async def _process_intent_detection(self, llm_session: LLMSession, job: LLMJob):
+        """Process intent detection"""
+        try:
+            session = llm_session.session
+            
+            # Get project and version config
+            project = next((p for p in self.config.projects if p.name == session.project_name), None)
+            if not project:
+                raise ValueError(f"Project not found: {session.project_name}")
+                
+            version = session.get_version_config()
+            if not version:
+                raise ValueError("Version config not found")
+                
+            # Build intent detection prompt
+            prompt = build_intent_prompt(
+                version=version,
+                conversation=session.chat_history,
+                project_locale=project.default_locale
+            )
+            
+            log_debug(
+                f"📝 Intent detection prompt built",
+                session_id=job.session_id,
+                prompt_length=len(prompt)
+            )
+            
+            # Call LLM
+            response = await llm_session.llm_instance.generate(
+                system_prompt=prompt,
+                user_input=job.input_text,
+                context=session.chat_history[-10:]  # Last 10 messages
+            )
+            
+            # Parse intent
+            intent_name, response_text = self._parse_intent_response(response)
+            
+            if intent_name:
+                # Find intent config
+                intent_config = next((i for i in version.intents if i.name == intent_name), None)
+                
+                if intent_config:
+                    # Update session
+                    session.current_intent = intent_name
+                    session.set_intent_config(intent_config)
+                    session.state = "collect_params"
+                    
+                    log_info(
+                        f"🎯 Intent detected",
+                        session_id=job.session_id,
+                        intent=intent_name
+                    )
+                    
+                    # Check if we need to collect parameters
+                    missing_params = [
+                        p.name for p in intent_config.parameters
+                        if p.required and p.variable_name not in session.variables
+                    ]
+                    
+                    if not missing_params:
+                        # All parameters ready, execute action
+                        await self._execute_intent_action(llm_session, intent_config)
+                        return
+                    else:
+                        # Need to collect parameters
+                        await self._request_parameter_collection(llm_session, intent_config, missing_params)
+                        return
+                        
+            # No intent detected, use response as is
+            response_text = self._clean_response(response)
+            job.complete(response_text, intent_name)
+            
+            # Publish response
+            await self._publish_response(job)
+            
+        except Exception as e:
+            job.fail(str(e))
+            raise
+            
+    async def _process_parameter_collection(self, llm_session: LLMSession, job: LLMJob):
+        """Process parameter collection"""
+        try:
+            session = llm_session.session
+            intent_config = session.get_intent_config()
+            
+            if not intent_config:
+                raise ValueError("No intent config in session")
+                
+            # Extract parameters from user input
+            extracted_params = await self._extract_parameters(
+                llm_session,
+                job.input_text,
+                intent_config,
+                session.variables
+            )
+            
+            # Update session variables
+            for param_name, param_value in extracted_params.items():
+                param_config = next(
+                    (p for p in intent_config.parameters if p.name == param_name),
+                    None
+                )
+                if param_config:
+                    session.variables[param_config.variable_name] = str(param_value)
+                    
+            # Check what parameters are still missing
+            missing_params = [
+                p.name for p in intent_config.parameters
+                if p.required and p.variable_name not in session.variables
+            ]
+            
+            if not missing_params:
+                # All parameters collected, execute action
+                await self._execute_intent_action(llm_session, intent_config)
+            else:
+                # Still need more parameters
+                await self._request_parameter_collection(llm_session, intent_config, missing_params)
+                
+        except Exception as e:
+            job.fail(str(e))
+            raise
+            
+    async def _process_response_generation(self, llm_session: LLMSession, job: LLMJob):
+        """Process general response generation"""
+        try:
+            session = llm_session.session
+            
+            # Get version config
+            version = session.get_version_config()
+            if not version:
+                raise ValueError("Version config not found")
+                
+            # Use general prompt
+            prompt = version.general_prompt
+            
+            # Generate response
+            response = await llm_session.llm_instance.generate(
+                system_prompt=prompt,
+                user_input=job.input_text,
+                context=session.chat_history[-10:]
+            )
+            
+            response_text = self._clean_response(response)
+            job.complete(response_text)
+            
+            # Publish response
+            await self._publish_response(job)
+            
+        except Exception as e:
+            job.fail(str(e))
+            raise
+            
+    async def _extract_parameters(self, 
+                                 llm_session: LLMSession,
+                                 user_input: str,
+                                 intent_config: Any,
+                                 existing_params: Dict[str, str]) -> Dict[str, Any]:
+        """Extract parameters from user input"""
+        # Build extraction prompt
+        param_info = []
+        for param in intent_config.parameters:
+            if param.variable_name not in existing_params:
+                param_info.append({
+                    'name': param.name,
+                    'type': param.type,
+                    'required': param.required,
+                    'extraction_prompt': param.extraction_prompt
+                })
+                
+        prompt = f"""
+        Extract parameters from user message: "{user_input}"
+        
+        Expected parameters:
+        {json.dumps(param_info, ensure_ascii=False)}
+        
+        Return as JSON object with parameter names as keys.
+        """
+        
+        # Call LLM
+        response = await llm_session.llm_instance.generate(
+            system_prompt=prompt,
+            user_input=user_input,
+            context=[]
+        )
+        
+        # Parse JSON response
+        try:
+            # Look for JSON block in response
+            import re
+            json_match = re.search(r'```json\s*(.*?)\s*```', response, re.DOTALL)
+            if not json_match:
+                json_match = re.search(r'\{[^}]+\}', response)
+                
+            if json_match:
+                json_str = json_match.group(1) if '```' in response else json_match.group(0)
+                return json.loads(json_str)
+        except:
+            pass
+            
+        return {}
+        
+    async def _request_parameter_collection(self, 
+                                          llm_session: LLMSession,
+                                          intent_config: Any,
+                                          missing_params: List[str]):
+        """Request parameter collection from user"""
+        session = llm_session.session
+        
+        # Get project config
+        project = next((p for p in self.config.projects if p.name == session.project_name), None)
+        if not project:
+            return
+            
+        version = session.get_version_config()
+        if not version:
+            return
+            
+        # Get parameter collection config
+        collection_config = self.config.global_config.llm_provider.settings.get("parameter_collection_config", {})
+        max_params = collection_config.get("max_params_per_question", 2)
+        
+        # Decide which parameters to ask
+        params_to_ask = missing_params[:max_params]
+        
+        # Build parameter collection prompt
+        prompt = build_parameter_prompt(
+            version=version,
+            intent_config=intent_config,
+            chat_history=session.chat_history,
+            collected_params=session.variables,
+            missing_params=missing_params,
+            params_to_ask=params_to_ask,
+            max_params=max_params,
+            project_locale=project.default_locale,
+            unanswered_params=session.unanswered_parameters
+        )
+        
+        # Generate question
+        response = await llm_session.llm_instance.generate(
+            system_prompt=prompt,
+            user_input="",
+            context=session.chat_history[-5:]
+        )
+        
+        response_text = self._clean_response(response)
+        
+        # Create a job for the response
+        job = LLMJob(
+            job_id=f"{session.session_id}_param_request",
+            session_id=session.session_id,
+            input_text="",
+            job_type="parameter_request",
+            response_text=response_text
+        )
+        
+        await self._publish_response(job)
+        
+    async def _execute_intent_action(self, llm_session: LLMSession, intent_config: Any):
+        """Execute intent action (API call)"""
+        session = llm_session.session
+        
+        try:
+            # Get API config
+            api_name = intent_config.action
+            api_config = self.config.get_api(api_name)
+            
+            if not api_config:
+                raise ValueError(f"API config not found: {api_name}")
+                
+            log_info(
+                f"📡 Executing intent action",
+                session_id=session.session_id,
+                api_name=api_name,
+                variables=session.variables
+            )
+            
+            # Execute API call
+            from api_executor import call_api
+            response = call_api(api_config, session)
+            api_json = response.json()
+            
+            log_info(f"✅ API response received", session_id=session.session_id)
+            
+            # Humanize response if prompt exists
+            if api_config.response_prompt:
+                prompt = api_config.response_prompt.replace(
+                    "{{api_response}}", 
+                    json.dumps(api_json, ensure_ascii=False)
+                )
+                
+                human_response = await llm_session.llm_instance.generate(
+                    system_prompt=prompt,
+                    user_input=json.dumps(api_json),
+                    context=[]
+                )
+                
+                response_text = self._clean_response(human_response)
+            else:
+                response_text = f"İşlem tamamlandı: {api_json}"
+                
+            # Reset session flow
+            session.reset_flow()
+            
+            # Create job for response
+            job = LLMJob(
+                job_id=f"{session.session_id}_action_result",
+                session_id=session.session_id,
+                input_text="",
+                job_type="action_result",
+                response_text=response_text
+            )
+            
+            await self._publish_response(job)
+            
+        except Exception as e:
+            log_error(
+                f"❌ API execution failed",
+                session_id=session.session_id,
+                error=str(e)
+            )
+            
+            # Reset flow
+            session.reset_flow()
+            
+            # Send error response
+            error_response = self._get_user_friendly_error("api_error", {"api_name": api_name})
+            
+            job = LLMJob(
+                job_id=f"{session.session_id}_error",
+                session_id=session.session_id,
+                input_text="",
+                job_type="error",
+                response_text=error_response
+            )
+            
+            await self._publish_response(job)
+            
+    async def _publish_response(self, job: LLMJob):
+        """Publish LLM response"""
+        # Update job history
+        llm_session = self.llm_sessions.get(job.session_id)
+        if llm_session:
+            llm_session.job_history.append(job)
+            # Keep only last 20 jobs
+            if len(llm_session.job_history) > 20:
+                llm_session.job_history.pop(0)
+                
+        # Publish event
+        await self.event_bus.publish(Event(
+            type=EventType.LLM_RESPONSE_READY,
+            session_id=job.session_id,
+            data={
+                "text": job.response_text,
+                "intent": job.detected_intent,
+                "job_type": job.job_type
+            }
+        ))
+        
+        log_info(
+            f"✅ LLM response published",
+            session_id=job.session_id,
+            response_length=len(job.response_text) if job.response_text else 0
+        )
+        
+    def _parse_intent_response(self, response: str) -> tuple[str, str]:
+        """Parse intent from LLM response"""
+        import re
+        
+        # Look for intent pattern
+        match = re.search(r"#DETECTED_INTENT:\s*([A-Za-z0-9_-]+)", response)
+        if not match:
+            return "", response
+            
+        intent_name = match.group(1)
+        
+        # Remove 'assistant' suffix if exists
+        if intent_name.endswith("assistant"):
+            intent_name = intent_name[:-9]
+            
+        # Get remaining text after intent
+        remaining_text = response[match.end():]
+        
+        return intent_name, remaining_text
+        
+    def _clean_response(self, response: str) -> str:
+        """Clean LLM response"""
+        # Remove everything after the first logical assistant block or intent tag
+        for stop in ["#DETECTED_INTENT", "⚠️", "\nassistant", "assistant\n", "assistant"]:
+            idx = response.find(stop)
+            if idx != -1:
+                response = response[:idx]
+                
+        # Normalize common greetings
+        import re
+        response = re.sub(r"Hoş[\s-]?geldin(iz)?", "Hoş geldiniz", response, flags=re.IGNORECASE)
+        
+        return response.strip()
+        
+    def _get_user_friendly_error(self, error_type: str, context: dict = None) -> str:
+        """Get user-friendly error messages"""
+        error_messages = {
+            "session_not_found": "Oturumunuz bulunamadı. Lütfen yeni bir konuşma başlatın.",
+            "project_not_found": "Proje konfigürasyonu bulunamadı. Lütfen yönetici ile iletişime geçin.",
+            "version_not_found": "Proje versiyonu bulunamadı. Lütfen geçerli bir versiyon seçin.",
+            "intent_not_found": "Üzgünüm, ne yapmak istediğinizi anlayamadım. Lütfen daha açık bir şekilde belirtir misiniz?",
+            "api_timeout": "İşlem zaman aşımına uğradı. Lütfen tekrar deneyin.",
+            "api_error": "İşlem sırasında bir hata oluştu. Lütfen daha sonra tekrar deneyin.",
+            "parameter_validation": "Girdiğiniz bilgide bir hata var. Lütfen kontrol edip tekrar deneyin.",
+            "llm_error": "Sistem yanıt veremedi. Lütfen biraz sonra tekrar deneyin.",
+            "llm_timeout": "Sistem meşgul. Lütfen birkaç saniye bekleyip tekrar deneyin.",
+            "session_expired": "Oturumunuz zaman aşımına uğradı. Lütfen yeni bir konuşma başlatın.",
+            "rate_limit": "Çok fazla istek gönderdiniz. Lütfen biraz bekleyin.",
+            "internal_error": "Beklenmeyen bir hata oluştu. Lütfen yönetici ile iletişime geçin."
+        }
+        
+        message = error_messages.get(error_type, error_messages["internal_error"])
+        
+        # Add context if available
+        if context:
+            if error_type == "api_error" and "api_name" in context:
+                message = f"{context['api_name']} servisi için {message}"
+                
+        return message
+        
+    async def _handle_session_ended(self, event: Event):
+        """Clean up LLM resources when session ends"""
+        session_id = event.session_id
+        await self._cleanup_session(session_id)
+        
+    async def _cleanup_session(self, session_id: str):
+        """Clean up LLM session"""
+        llm_session = self.llm_sessions.pop(session_id, None)
+        if not llm_session:
+            return
+            
+        try:
+            # Release resource
+            resource_id = f"llm_{session_id}"
+            await self.resource_manager.release(resource_id, delay_seconds=180)  # 3 minutes
+            
+            log_info(
+                f"🧹 LLM session cleaned up",
+                session_id=session_id,
+                total_jobs=llm_session.total_jobs,
+                job_history_size=len(llm_session.job_history)
+            )
+            
+        except Exception as e:
+            log_error(
+                f"❌ Error cleaning up LLM session",
+                session_id=session_id,
+                error=str(e)
+            )
+            
+    async def _cleanup_llm_instance(self, llm_instance: LLMInterface):
+        """Cleanup callback for LLM instance"""
+        try:
+            # LLM instances typically don't need special cleanup
+            log_debug("🧹 LLM instance cleaned up")
+            
+        except Exception as e:
+            log_error(f"❌ Error cleaning up LLM instance", error=str(e))
+            
+    def get_stats(self) -> Dict[str, Any]:
+        """Get LLM manager statistics"""
+        session_stats = {}
+        for session_id, llm_session in self.llm_sessions.items():
+            session_stats[session_id] = {
+                "active_job": llm_session.active_job.job_id if llm_session.active_job else None,
+                "total_jobs": llm_session.total_jobs,
+                "job_history_size": len(llm_session.job_history),
+                "uptime_seconds": (datetime.utcnow() - llm_session.created_at).total_seconds(),
+                "last_activity": llm_session.last_activity.isoformat()
+            }
+            
+        return {
+            "active_sessions": len(self.llm_sessions),
+            "total_active_jobs": sum(1 for s in self.llm_sessions.values() if s.active_job),
+            "sessions": session_stats
+        }

resource_manager.py

@@ -0,0 +1,401 @@
+"""
+Resource Manager for Flare
+==========================
+Manages lifecycle of all session resources
+"""
+import asyncio
+from typing import Dict, Any, Optional, Callable, Set
+from datetime import datetime, timedelta
+from dataclasses import dataclass, field
+import traceback
+from enum import Enum
+
+from event_bus import EventBus, Event, EventType
+from logger import log_info, log_error, log_debug, log_warning
+
+
+class ResourceType(Enum):
+    """Types of resources managed"""
+    STT_INSTANCE = "stt_instance"
+    TTS_INSTANCE = "tts_instance"
+    LLM_CONTEXT = "llm_context"
+    AUDIO_BUFFER = "audio_buffer"
+    WEBSOCKET = "websocket"
+    GENERIC = "generic"
+
+
+@dataclass
+class Resource:
+    """Resource wrapper with metadata"""
+    resource_id: str
+    resource_type: ResourceType
+    session_id: str
+    instance: Any
+    created_at: datetime = field(default_factory=datetime.utcnow)
+    last_accessed: datetime = field(default_factory=datetime.utcnow)
+    disposal_task: Optional[asyncio.Task] = None
+    cleanup_callback: Optional[Callable] = None
+    
+    def touch(self):
+        """Update last accessed time"""
+        self.last_accessed = datetime.utcnow()
+        
+    async def cleanup(self):
+        """Cleanup the resource"""
+        try:
+            if self.cleanup_callback:
+                if asyncio.iscoroutinefunction(self.cleanup_callback):
+                    await self.cleanup_callback(self.instance)
+                else:
+                    await asyncio.to_thread(self.cleanup_callback, self.instance)
+                    
+            log_debug(
+                f"🧹 Resource cleaned up",
+                resource_id=self.resource_id,
+                resource_type=self.resource_type.value
+            )
+        except Exception as e:
+            log_error(
+                f"❌ Error cleaning up resource",
+                resource_id=self.resource_id,
+                error=str(e)
+            )
+
+
+class ResourcePool:
+    """Pool for reusable resources"""
+    
+    def __init__(self, 
+                 resource_type: ResourceType,
+                 factory: Callable,
+                 max_idle: int = 10,
+                 max_age_seconds: int = 300):
+        self.resource_type = resource_type
+        self.factory = factory
+        self.max_idle = max_idle
+        self.max_age_seconds = max_age_seconds
+        self.idle_resources: List[Resource] = []
+        self.lock = asyncio.Lock()
+        
+    async def acquire(self, session_id: str) -> Any:
+        """Get resource from pool or create new"""
+        async with self.lock:
+            # Try to get from pool
+            now = datetime.utcnow()
+            while self.idle_resources:
+                resource = self.idle_resources.pop(0)
+                age = (now - resource.created_at).total_seconds()
+                
+                if age < self.max_age_seconds:
+                    # Reuse this resource
+                    resource.session_id = session_id
+                    resource.touch()
+                    log_debug(
+                        f"♻️ Reused pooled resource",
+                        resource_type=self.resource_type.value,
+                        age_seconds=age
+                    )
+                    return resource.instance
+                else:
+                    # Too old, cleanup
+                    await resource.cleanup()
+                    
+        # Create new resource
+        if asyncio.iscoroutinefunction(self.factory):
+            instance = await self.factory()
+        else:
+            instance = await asyncio.to_thread(self.factory)
+            
+        log_debug(
+            f"🏗️ Created new resource",
+            resource_type=self.resource_type.value
+        )
+        return instance
+        
+    async def release(self, resource: Resource):
+        """Return resource to pool"""
+        async with self.lock:
+            if len(self.idle_resources) < self.max_idle:
+                resource.session_id = ""  # Clear session
+                self.idle_resources.append(resource)
+                log_debug(
+                    f"📥 Resource returned to pool",
+                    resource_type=self.resource_type.value,
+                    pool_size=len(self.idle_resources)
+                )
+            else:
+                # Pool full, cleanup
+                await resource.cleanup()
+                
+    async def cleanup_old(self):
+        """Cleanup old resources in pool"""
+        async with self.lock:
+            now = datetime.utcnow()
+            active_resources = []
+            
+            for resource in self.idle_resources:
+                age = (now - resource.created_at).total_seconds()
+                if age < self.max_age_seconds:
+                    active_resources.append(resource)
+                else:
+                    await resource.cleanup()
+                    
+            self.idle_resources = active_resources
+
+
+class ResourceManager:
+    """Manages all resources with lifecycle and pooling"""
+    
+    def __init__(self, event_bus: EventBus):
+        self.event_bus = event_bus
+        self.resources: Dict[str, Resource] = {}
+        self.session_resources: Dict[str, Set[str]] = {}
+        self.pools: Dict[ResourceType, ResourcePool] = {}
+        self.disposal_delay_seconds = 60  # Default disposal delay
+        self._cleanup_task: Optional[asyncio.Task] = None
+        self._running = False
+        self._setup_event_handlers()
+        
+    def _setup_event_handlers(self):
+        """Subscribe to lifecycle events"""
+        self.event_bus.subscribe(EventType.SESSION_STARTED, self._handle_session_started)
+        self.event_bus.subscribe(EventType.SESSION_ENDED, self._handle_session_ended)
+        
+    async def start(self):
+        """Start resource manager"""
+        if self._running:
+            return
+            
+        self._running = True
+        self._cleanup_task = asyncio.create_task(self._periodic_cleanup())
+        log_info("✅ Resource manager started")
+        
+    async def stop(self):
+        """Stop resource manager"""
+        self._running = False
+        
+        if self._cleanup_task:
+            self._cleanup_task.cancel()
+            try:
+                await self._cleanup_task
+            except asyncio.CancelledError:
+                pass
+                
+        # Cleanup all resources
+        for resource_id in list(self.resources.keys()):
+            await self.release(resource_id, immediate=True)
+            
+        log_info("✅ Resource manager stopped")
+        
+    def register_pool(self, 
+                     resource_type: ResourceType,
+                     factory: Callable,
+                     max_idle: int = 10,
+                     max_age_seconds: int = 300):
+        """Register a resource pool"""
+        self.pools[resource_type] = ResourcePool(
+            resource_type=resource_type,
+            factory=factory,
+            max_idle=max_idle,
+            max_age_seconds=max_age_seconds
+        )
+        log_info(
+            f"📊 Resource pool registered",
+            resource_type=resource_type.value,
+            max_idle=max_idle
+        )
+        
+    async def acquire(self,
+                     resource_id: str,
+                     session_id: str,
+                     resource_type: ResourceType,
+                     factory: Optional[Callable] = None,
+                     cleanup_callback: Optional[Callable] = None) -> Any:
+        """Acquire a resource"""
+        
+        # Check if already exists
+        if resource_id in self.resources:
+            resource = self.resources[resource_id]
+            resource.touch()
+            
+            # Cancel any pending disposal
+            if resource.disposal_task:
+                resource.disposal_task.cancel()
+                resource.disposal_task = None
+                
+            return resource.instance
+            
+        # Try to get from pool
+        instance = None
+        if resource_type in self.pools:
+            instance = await self.pools[resource_type].acquire(session_id)
+        elif factory:
+            # Create new resource
+            if asyncio.iscoroutinefunction(factory):
+                instance = await factory()
+            else:
+                instance = await asyncio.to_thread(factory)
+        else:
+            raise ValueError(f"No factory or pool for resource type: {resource_type}")
+            
+        # Create resource wrapper
+        resource = Resource(
+            resource_id=resource_id,
+            resource_type=resource_type,
+            session_id=session_id,
+            instance=instance,
+            cleanup_callback=cleanup_callback
+        )
+        
+        # Track resource
+        self.resources[resource_id] = resource
+        
+        if session_id not in self.session_resources:
+            self.session_resources[session_id] = set()
+        self.session_resources[session_id].add(resource_id)
+        
+        log_info(
+            f"📌 Resource acquired",
+            resource_id=resource_id,
+            resource_type=resource_type.value,
+            session_id=session_id
+        )
+        
+        return instance
+        
+    async def release(self, 
+                     resource_id: str,
+                     delay_seconds: Optional[int] = None,
+                     immediate: bool = False):
+        """Release a resource with optional delay"""
+        
+        resource = self.resources.get(resource_id)
+        if not resource:
+            return
+            
+        if immediate:
+            # Immediate cleanup
+            await self._dispose_resource(resource_id)
+        else:
+            # Schedule disposal
+            delay = delay_seconds or self.disposal_delay_seconds
+            resource.disposal_task = asyncio.create_task(
+                self._delayed_disposal(resource_id, delay)
+            )
+            
+            log_debug(
+                f"⏱️ Resource disposal scheduled",
+                resource_id=resource_id,
+                delay_seconds=delay
+            )
+            
+    async def _delayed_disposal(self, resource_id: str, delay_seconds: int):
+        """Dispose resource after delay"""
+        try:
+            await asyncio.sleep(delay_seconds)
+            await self._dispose_resource(resource_id)
+        except asyncio.CancelledError:
+            log_debug(f"🚫 Disposal cancelled", resource_id=resource_id)
+            
+    async def _dispose_resource(self, resource_id: str):
+        """Actually dispose of a resource"""
+        resource = self.resources.pop(resource_id, None)
+        if not resource:
+            return
+            
+        # Remove from session tracking
+        if resource.session_id in self.session_resources:
+            self.session_resources[resource.session_id].discard(resource_id)
+            
+        # Return to pool or cleanup
+        if resource.resource_type in self.pools:
+            await self.pools[resource.resource_type].release(resource)
+        else:
+            await resource.cleanup()
+            
+        log_info(
+            f"♻️ Resource disposed",
+            resource_id=resource_id,
+            resource_type=resource.resource_type.value
+        )
+        
+    async def release_session_resources(self, session_id: str):
+        """Release all resources for a session"""
+        resource_ids = self.session_resources.get(session_id, set()).copy()
+        
+        for resource_id in resource_ids:
+            await self.release(resource_id, immediate=True)
+            
+        # Remove session tracking
+        self.session_resources.pop(session_id, None)
+        
+        log_info(
+            f"🧹 Session resources released",
+            session_id=session_id,
+            count=len(resource_ids)
+        )
+        
+    async def _handle_session_started(self, event: Event):
+        """Initialize session resource tracking"""
+        session_id = event.session_id
+        self.session_resources[session_id] = set()
+        
+    async def _handle_session_ended(self, event: Event):
+        """Cleanup session resources"""
+        session_id = event.session_id
+        await self.release_session_resources(session_id)
+        
+    async def _periodic_cleanup(self):
+        """Periodic cleanup of old resources"""
+        while self._running:
+            try:
+                await asyncio.sleep(60)  # Check every minute
+                
+                # Cleanup old pooled resources
+                for pool in self.pools.values():
+                    await pool.cleanup_old()
+                    
+                # Check for orphaned resources
+                now = datetime.utcnow()
+                for resource_id, resource in list(self.resources.items()):
+                    age = (now - resource.last_accessed).total_seconds()
+                    
+                    # If not accessed for 5 minutes and no disposal scheduled
+                    if age > 300 and not resource.disposal_task:
+                        log_warning(
+                            f"⚠️ Orphaned resource detected",
+                            resource_id=resource_id,
+                            age_seconds=age
+                        )
+                        await self.release(resource_id, delay_seconds=30)
+                        
+            except Exception as e:
+                log_error(
+                    f"❌ Error in periodic cleanup",
+                    error=str(e),
+                    traceback=traceback.format_exc()
+                )
+                
+    def get_stats(self) -> Dict[str, Any]:
+        """Get resource manager statistics"""
+        pool_stats = {}
+        for resource_type, pool in self.pools.items():
+            pool_stats[resource_type.value] = {
+                "idle_count": len(pool.idle_resources),
+                "max_idle": pool.max_idle
+            }
+            
+        return {
+            "active_resources": len(self.resources),
+            "sessions": len(self.session_resources),
+            "pools": pool_stats,
+            "total_resources_by_type": self._count_by_type()
+        }
+        
+    def _count_by_type(self) -> Dict[str, int]:
+        """Count resources by type"""
+        counts = {}
+        for resource in self.resources.values():
+            type_name = resource.resource_type.value
+            counts[type_name] = counts.get(type_name, 0) + 1
+        return counts

state_orchestrator.py

@@ -0,0 +1,511 @@
+"""
+State Orchestrator for Flare Realtime Chat
+==========================================
+Central state machine and flow control
+"""
+import asyncio
+from typing import Dict, Optional, Set, Any
+from enum import Enum
+from datetime import datetime
+import traceback
+from dataclasses import dataclass, field
+
+from event_bus import EventBus, Event, EventType, publish_state_transition, publish_error
+from session import Session
+from logger import log_info, log_error, log_debug, log_warning
+
+
+class ConversationState(Enum):
+    """Conversation states"""
+    IDLE = "idle"
+    INITIALIZING = "initializing"
+    PREPARING_WELCOME = "preparing_welcome"
+    PLAYING_WELCOME = "playing_welcome"
+    LISTENING = "listening"
+    PROCESSING_SPEECH = "processing_speech"
+    PREPARING_RESPONSE = "preparing_response"
+    PLAYING_RESPONSE = "playing_response"
+    ERROR = "error"
+    ENDED = "ended"
+
+
+@dataclass
+class SessionContext:
+    """Context for a conversation session"""
+    session_id: str
+    session: Session
+    state: ConversationState = ConversationState.IDLE
+    stt_instance: Optional[Any] = None
+    tts_instance: Optional[Any] = None
+    llm_context: Optional[Any] = None
+    audio_buffer: Optional[Any] = None
+    websocket_connection: Optional[Any] = None
+    created_at: datetime = field(default_factory=datetime.utcnow)
+    last_activity: datetime = field(default_factory=datetime.utcnow)
+    metadata: Dict[str, Any] = field(default_factory=dict)
+    
+    def update_activity(self):
+        """Update last activity timestamp"""
+        self.last_activity = datetime.utcnow()
+        
+    async def cleanup(self):
+        """Cleanup all session resources"""
+        # Cleanup will be implemented by resource managers
+        log_debug(f"🧹 Cleaning up session context", session_id=self.session_id)
+
+
+class StateOrchestrator:
+    """Central state machine for conversation flow"""
+    
+    # Valid state transitions
+    VALID_TRANSITIONS = {
+        ConversationState.IDLE: {ConversationState.INITIALIZING},
+        ConversationState.INITIALIZING: {ConversationState.PREPARING_WELCOME, ConversationState.LISTENING},
+        ConversationState.PREPARING_WELCOME: {ConversationState.PLAYING_WELCOME, ConversationState.ERROR},
+        ConversationState.PLAYING_WELCOME: {ConversationState.LISTENING, ConversationState.ERROR},
+        ConversationState.LISTENING: {ConversationState.PROCESSING_SPEECH, ConversationState.ERROR, ConversationState.ENDED},
+        ConversationState.PROCESSING_SPEECH: {ConversationState.PREPARING_RESPONSE, ConversationState.ERROR},
+        ConversationState.PREPARING_RESPONSE: {ConversationState.PLAYING_RESPONSE, ConversationState.ERROR},
+        ConversationState.PLAYING_RESPONSE: {ConversationState.LISTENING, ConversationState.ERROR},
+        ConversationState.ERROR: {ConversationState.LISTENING, ConversationState.ENDED},
+        ConversationState.ENDED: set()  # No transitions from ENDED
+    }
+    
+    def __init__(self, event_bus: EventBus):
+        self.event_bus = event_bus
+        self.sessions: Dict[str, SessionContext] = {}
+        self._setup_event_handlers()
+        
+    def _setup_event_handlers(self):
+        """Subscribe to relevant events"""
+        # Session lifecycle
+        self.event_bus.subscribe(EventType.SESSION_STARTED, self._handle_session_started)
+        self.event_bus.subscribe(EventType.SESSION_ENDED, self._handle_session_ended)
+        
+        # STT events
+        self.event_bus.subscribe(EventType.STT_READY, self._handle_stt_ready)
+        self.event_bus.subscribe(EventType.STT_RESULT, self._handle_stt_result)
+        self.event_bus.subscribe(EventType.STT_ERROR, self._handle_stt_error)
+        
+        # TTS events
+        self.event_bus.subscribe(EventType.TTS_COMPLETED, self._handle_tts_completed)
+        self.event_bus.subscribe(EventType.TTS_ERROR, self._handle_tts_error)
+        
+        # Audio events
+        self.event_bus.subscribe(EventType.AUDIO_PLAYBACK_COMPLETED, self._handle_audio_playback_completed)
+        
+        # LLM events
+        self.event_bus.subscribe(EventType.LLM_RESPONSE_READY, self._handle_llm_response_ready)
+        self.event_bus.subscribe(EventType.LLM_ERROR, self._handle_llm_error)
+        
+        # Error events
+        self.event_bus.subscribe(EventType.CRITICAL_ERROR, self._handle_critical_error)
+        
+    async def _handle_session_started(self, event: Event):
+        """Handle session start"""
+        session_id = event.session_id
+        session_data = event.data
+        
+        log_info(f"🎬 Session started", session_id=session_id)
+        
+        # Create session context
+        context = SessionContext(
+            session_id=session_id,
+            session=session_data.get("session"),
+            metadata={
+                "has_welcome": session_data.get("has_welcome", False),
+                "welcome_text": session_data.get("welcome_text", "")
+            }
+        )
+        
+        self.sessions[session_id] = context
+        
+        # Transition to INITIALIZING
+        await self.transition_to(session_id, ConversationState.INITIALIZING)
+        
+        # Check if welcome prompt exists
+        if session_data.get("has_welcome"):
+            await self.transition_to(session_id, ConversationState.PREPARING_WELCOME)
+            
+            # Request TTS for welcome message
+            await self.event_bus.publish(Event(
+                type=EventType.TTS_STARTED,
+                session_id=session_id,
+                data={
+                    "text": session_data.get("welcome_text", ""),
+                    "is_welcome": True
+                }
+            ))
+        else:
+            # No welcome, go straight to listening
+            await self.transition_to(session_id, ConversationState.LISTENING)
+            
+            # Request STT start
+            await self.event_bus.publish(Event(
+                type=EventType.STT_STARTED,
+                session_id=session_id,
+                data={}
+            ))
+            
+    async def _handle_session_ended(self, event: Event):
+        """Handle session end"""
+        session_id = event.session_id
+        
+        log_info(f"🏁 Session ended", session_id=session_id)
+        
+        # Get context for cleanup
+        context = self.sessions.get(session_id)
+        
+        # Transition to ended
+        await self.transition_to(session_id, ConversationState.ENDED)
+        
+        # Stop all components
+        await self.event_bus.publish(Event(
+            type=EventType.STT_STOPPED,
+            session_id=session_id,
+            data={"reason": "session_ended"}
+        ))
+        
+        # Cleanup session context
+        if context:
+            await context.cleanup()
+            
+        # Remove session
+        self.sessions.pop(session_id, None)
+        
+        # Clear event bus session data
+        self.event_bus.clear_session_data(session_id)
+        
+    async def _handle_stt_ready(self, event: Event):
+        """Handle STT ready signal"""
+        session_id = event.session_id
+        current_state = self.get_state(session_id)
+        
+        log_debug(f"🎤 STT ready", session_id=session_id, current_state=current_state)
+        
+        # Only process if we're expecting STT to be ready
+        if current_state in [ConversationState.LISTENING, ConversationState.PLAYING_WELCOME]:
+            # STT is ready, we're already in the right state
+            pass
+            
+    async def _handle_stt_result(self, event: Event):
+        """Handle STT transcription result"""
+        session_id = event.session_id
+        current_state = self.get_state(session_id)
+        
+        if current_state != ConversationState.LISTENING:
+            log_warning(
+                f"⚠️ STT result in unexpected state",
+                session_id=session_id,
+                state=current_state
+            )
+            return
+            
+        result_data = event.data
+        is_final = result_data.get("is_final", False)
+        
+        if is_final:
+            text = result_data.get("text", "")
+            log_info(f"💬 Final transcription: '{text}'", session_id=session_id)
+            
+            # Stop STT
+            await self.event_bus.publish(Event(
+                type=EventType.STT_STOPPED,
+                session_id=session_id,
+                data={"reason": "final_result"}
+            ))
+            
+            # Transition to processing
+            await self.transition_to(session_id, ConversationState.PROCESSING_SPEECH)
+            
+            # Send to LLM
+            await self.event_bus.publish(Event(
+                type=EventType.LLM_PROCESSING_STARTED,
+                session_id=session_id,
+                data={"text": text}
+            ))
+            
+    async def _handle_llm_response_ready(self, event: Event):
+        """Handle LLM response"""
+        session_id = event.session_id
+        current_state = self.get_state(session_id)
+        
+        if current_state != ConversationState.PROCESSING_SPEECH:
+            log_warning(
+                f"⚠️ LLM response in unexpected state",
+                session_id=session_id,
+                state=current_state
+            )
+            return
+            
+        response_text = event.data.get("text", "")
+        log_info(f"🤖 LLM response ready", session_id=session_id, length=len(response_text))
+        
+        # Transition to preparing response
+        await self.transition_to(session_id, ConversationState.PREPARING_RESPONSE)
+        
+        # Request TTS
+        await self.event_bus.publish(Event(
+            type=EventType.TTS_STARTED,
+            session_id=session_id,
+            data={"text": response_text}
+        ))
+        
+    async def _handle_tts_completed(self, event: Event):
+        """Handle TTS completion"""
+        session_id = event.session_id
+        current_state = self.get_state(session_id)
+        
+        log_info(f"🔊 TTS completed", session_id=session_id, state=current_state)
+        
+        if current_state == ConversationState.PREPARING_WELCOME:
+            await self.transition_to(session_id, ConversationState.PLAYING_WELCOME)
+        elif current_state == ConversationState.PREPARING_RESPONSE:
+            await self.transition_to(session_id, ConversationState.PLAYING_RESPONSE)
+            
+    async def _handle_audio_playback_completed(self, event: Event):
+        """Handle audio playback completion"""
+        session_id = event.session_id
+        current_state = self.get_state(session_id)
+        
+        log_info(f"🎵 Audio playback completed", session_id=session_id, state=current_state)
+        
+        if current_state in [ConversationState.PLAYING_WELCOME, ConversationState.PLAYING_RESPONSE]:
+            # Transition back to listening
+            await self.transition_to(session_id, ConversationState.LISTENING)
+            
+            # Start STT
+            await self.event_bus.publish(Event(
+                type=EventType.STT_STARTED,
+                session_id=session_id,
+                data={}
+            ))
+            
+    async def _handle_stt_error(self, event: Event):
+        """Handle STT errors"""
+        session_id = event.session_id
+        error_data = event.data
+        
+        log_error(
+            f"❌ STT error",
+            session_id=session_id,
+            error=error_data.get("message")
+        )
+        
+        # Try to recover by transitioning back to listening
+        current_state = self.get_state(session_id)
+        if current_state != ConversationState.ENDED:
+            await self.transition_to(session_id, ConversationState.ERROR)
+            
+            # Try recovery after delay
+            await asyncio.sleep(2.0)
+            
+            if self.get_state(session_id) == ConversationState.ERROR:
+                await self.transition_to(session_id, ConversationState.LISTENING)
+                
+                # Restart STT
+                await self.event_bus.publish(Event(
+                    type=EventType.STT_STARTED,
+                    session_id=session_id,
+                    data={"retry": True}
+                ))
+                
+    async def _handle_tts_error(self, event: Event):
+        """Handle TTS errors"""
+        session_id = event.session_id
+        error_data = event.data
+        
+        log_error(
+            f"❌ TTS error",
+            session_id=session_id,
+            error=error_data.get("message")
+        )
+        
+        # Skip TTS and go to listening
+        current_state = self.get_state(session_id)
+        if current_state in [ConversationState.PREPARING_WELCOME, ConversationState.PREPARING_RESPONSE]:
+            await self.transition_to(session_id, ConversationState.LISTENING)
+            
+            # Start STT
+            await self.event_bus.publish(Event(
+                type=EventType.STT_STARTED,
+                session_id=session_id,
+                data={}
+            ))
+            
+    async def _handle_llm_error(self, event: Event):
+        """Handle LLM errors"""
+        session_id = event.session_id
+        error_data = event.data
+        
+        log_error(
+            f"❌ LLM error",
+            session_id=session_id,
+            error=error_data.get("message")
+        )
+        
+        # Go back to listening
+        await self.transition_to(session_id, ConversationState.LISTENING)
+        
+        # Start STT
+        await self.event_bus.publish(Event(
+            type=EventType.STT_STARTED,
+            session_id=session_id,
+            data={}
+        ))
+        
+    async def _handle_critical_error(self, event: Event):
+        """Handle critical errors"""
+        session_id = event.session_id
+        error_data = event.data
+        
+        log_error(
+            f"💥 Critical error",
+            session_id=session_id,
+            error=error_data.get("message")
+        )
+        
+        # End session
+        await self.transition_to(session_id, ConversationState.ENDED)
+        
+        # Publish session end event
+        await self.event_bus.publish(Event(
+            type=EventType.SESSION_ENDED,
+            session_id=session_id,
+            data={"reason": "critical_error"}
+        ))
+        
+    async def transition_to(self, session_id: str, new_state: ConversationState):
+        """Transition to a new state"""
+        current_state = self.get_state(session_id)
+        
+        if current_state is None:
+            log_warning(f"⚠️ Session not found for transition", session_id=session_id)
+            return
+            
+        # Check if transition is valid
+        if new_state not in self.VALID_TRANSITIONS.get(current_state, set()):
+            log_error(
+                f"❌ Invalid state transition",
+                session_id=session_id,
+                from_state=current_state.value,
+                to_state=new_state.value
+            )
+            
+            await publish_error(
+                session_id=session_id,
+                error_type="invalid_transition",
+                error_message=f"Cannot transition from {current_state.value} to {new_state.value}"
+            )
+            return
+            
+        # Update state
+        self.sessions[session_id] = new_state
+        
+        log_info(
+            f"🔄 State transition",
+            session_id=session_id,
+            from_state=current_state.value,
+            to_state=new_state.value
+        )
+        
+        # Publish state transition event
+        await publish_state_transition(
+            session_id=session_id,
+            from_state=current_state.value,
+            to_state=new_state.value
+        )
+        
+    def get_state(self, session_id: str) -> Optional[ConversationState]:
+        """Get current state for a session"""
+        return self.sessions.get(session_id)
+        
+    def get_session_data(self, session_id: str) -> Optional[Dict[str, Any]]:
+        """Get session data"""
+        return self.session_data.get(session_id)
+        
+    async def handle_error_recovery(self, session_id: str, error_type: str):
+        """Handle error recovery strategies"""
+        context = self.sessions.get(session_id)
+        
+        if not context or context.state == ConversationState.ENDED:
+            return
+            
+        log_info(
+            f"🔧 Attempting error recovery",
+            session_id=session_id,
+            error_type=error_type,
+            current_state=context.state.value
+        )
+        
+        # Update activity
+        context.update_activity()
+        
+        # Define recovery strategies
+        recovery_strategies = {
+            "stt_error": self._recover_from_stt_error,
+            "tts_error": self._recover_from_tts_error,
+            "llm_error": self._recover_from_llm_error,
+            "websocket_error": self._recover_from_websocket_error
+        }
+        
+        strategy = recovery_strategies.get(error_type)
+        if strategy:
+            await strategy(session_id)
+        else:
+            # Default recovery: go to error state then back to listening
+            await self.transition_to(session_id, ConversationState.ERROR)
+            await asyncio.sleep(1.0)
+            await self.transition_to(session_id, ConversationState.LISTENING)
+            
+    async def _recover_from_stt_error(self, session_id: str):
+        """Recover from STT error"""
+        # Stop STT, wait, restart
+        await self.event_bus.publish(Event(
+            type=EventType.STT_STOPPED,
+            session_id=session_id,
+            data={"reason": "error_recovery"}
+        ))
+        
+        await asyncio.sleep(2.0)
+        
+        await self.transition_to(session_id, ConversationState.LISTENING)
+        
+        await self.event_bus.publish(Event(
+            type=EventType.STT_STARTED,
+            session_id=session_id,
+            data={"retry": True}
+        ))
+        
+    async def _recover_from_tts_error(self, session_id: str):
+        """Recover from TTS error"""
+        # Skip TTS, go directly to listening
+        await self.transition_to(session_id, ConversationState.LISTENING)
+        
+        await self.event_bus.publish(Event(
+            type=EventType.STT_STARTED,
+            session_id=session_id,
+            data={}
+        ))
+        
+    async def _recover_from_llm_error(self, session_id: str):
+        """Recover from LLM error"""
+        # Go back to listening
+        await self.transition_to(session_id, ConversationState.LISTENING)
+        
+        await self.event_bus.publish(Event(
+            type=EventType.STT_STARTED,
+            session_id=session_id,
+            data={}
+        ))
+        
+    async def _recover_from_websocket_error(self, session_id: str):
+        """Recover from WebSocket error"""
+        # End session cleanly
+        await self.transition_to(session_id, ConversationState.ENDED)
+        
+        await self.event_bus.publish(Event(
+            type=EventType.SESSION_ENDED,
+            session_id=session_id,
+            data={"reason": "websocket_error"}
+        ))

stt_lifecycle_manager.py

@@ -0,0 +1,366 @@
+"""
+STT Lifecycle Manager for Flare
+===============================
+Manages STT instances lifecycle per session
+"""
+import asyncio
+from typing import Dict, Optional, Any
+from datetime import datetime
+import traceback
+import base64
+
+from event_bus import EventBus, Event, EventType, publish_error
+from resource_manager import ResourceManager, ResourceType
+from stt_factory import STTFactory
+from stt_interface import STTInterface, STTConfig, TranscriptionResult
+from logger import log_info, log_error, log_debug, log_warning
+
+
+class STTSession:
+    """STT session wrapper"""
+    
+    def __init__(self, session_id: str, stt_instance: STTInterface):
+        self.session_id = session_id
+        self.stt_instance = stt_instance
+        self.is_streaming = False
+        self.config: Optional[STTConfig] = None
+        self.created_at = datetime.utcnow()
+        self.last_activity = datetime.utcnow()
+        self.total_chunks = 0
+        self.total_bytes = 0
+        
+    def update_activity(self):
+        """Update last activity timestamp"""
+        self.last_activity = datetime.utcnow()
+
+
+class STTLifecycleManager:
+    """Manages STT instances lifecycle"""
+    
+    def __init__(self, event_bus: EventBus, resource_manager: ResourceManager):
+        self.event_bus = event_bus
+        self.resource_manager = resource_manager
+        self.stt_sessions: Dict[str, STTSession] = {}
+        self._setup_event_handlers()
+        self._setup_resource_pool()
+        
+    def _setup_event_handlers(self):
+        """Subscribe to STT-related events"""
+        self.event_bus.subscribe(EventType.STT_STARTED, self._handle_stt_start)
+        self.event_bus.subscribe(EventType.STT_STOPPED, self._handle_stt_stop)
+        self.event_bus.subscribe(EventType.AUDIO_CHUNK_RECEIVED, self._handle_audio_chunk)
+        self.event_bus.subscribe(EventType.SESSION_ENDED, self._handle_session_ended)
+        
+    def _setup_resource_pool(self):
+        """Setup STT instance pool"""
+        self.resource_manager.register_pool(
+            resource_type=ResourceType.STT_INSTANCE,
+            factory=self._create_stt_instance,
+            max_idle=5,
+            max_age_seconds=300  # 5 minutes
+        )
+        
+    async def _create_stt_instance(self) -> STTInterface:
+        """Factory for creating STT instances"""
+        try:
+            stt_instance = STTFactory.create_provider()
+            if not stt_instance:
+                raise ValueError("Failed to create STT instance")
+                
+            log_debug("🎤 Created new STT instance")
+            return stt_instance
+            
+        except Exception as e:
+            log_error(f"❌ Failed to create STT instance", error=str(e))
+            raise
+            
+    async def _handle_stt_start(self, event: Event):
+        """Handle STT start request"""
+        session_id = event.session_id
+        config_data = event.data
+        
+        try:
+            log_info(f"🎤 Starting STT", session_id=session_id)
+            
+            # Check if already exists
+            if session_id in self.stt_sessions:
+                stt_session = self.stt_sessions[session_id]
+                if stt_session.is_streaming:
+                    log_warning(f"⚠️ STT already streaming", session_id=session_id)
+                    return
+            else:
+                # Acquire STT instance from pool
+                resource_id = f"stt_{session_id}"
+                stt_instance = await self.resource_manager.acquire(
+                    resource_id=resource_id,
+                    session_id=session_id,
+                    resource_type=ResourceType.STT_INSTANCE,
+                    cleanup_callback=self._cleanup_stt_instance
+                )
+                
+                # Create session wrapper
+                stt_session = STTSession(session_id, stt_instance)
+                self.stt_sessions[session_id] = stt_session
+                
+            # Get session locale from state orchestrator
+            locale = config_data.get("locale", "tr")
+            
+            # Build STT config
+            stt_config = STTConfig(
+                language=self._get_language_code(locale),
+                sample_rate=config_data.get("sample_rate", 16000),
+                encoding=config_data.get("encoding", "WEBM_OPUS"),
+                enable_punctuation=config_data.get("enable_punctuation", True),
+                enable_word_timestamps=False,
+                model=config_data.get("model", "latest_long"),
+                use_enhanced=config_data.get("use_enhanced", True),
+                single_utterance=False,  # Continuous listening
+                interim_results=config_data.get("interim_results", True),
+                vad_enabled=config_data.get("vad_enabled", True),
+                speech_timeout_ms=config_data.get("speech_timeout_ms", 2000),
+                noise_reduction_enabled=config_data.get("noise_reduction_enabled", True),
+                noise_reduction_level=config_data.get("noise_reduction_level", 2)
+            )
+            
+            stt_session.config = stt_config
+            
+            # Start streaming
+            await stt_session.stt_instance.start_streaming(stt_config)
+            stt_session.is_streaming = True
+            stt_session.update_activity()
+            
+            log_info(f"✅ STT started", session_id=session_id, language=stt_config.language)
+            
+            # Notify STT is ready
+            await self.event_bus.publish(Event(
+                type=EventType.STT_READY,
+                session_id=session_id,
+                data={"language": stt_config.language}
+            ))
+            
+        except Exception as e:
+            log_error(
+                f"❌ Failed to start STT",
+                session_id=session_id,
+                error=str(e),
+                traceback=traceback.format_exc()
+            )
+            
+            # Clean up on error
+            if session_id in self.stt_sessions:
+                await self._cleanup_session(session_id)
+                
+            # Publish error event
+            await publish_error(
+                session_id=session_id,
+                error_type="stt_error",
+                error_message=f"Failed to start STT: {str(e)}"
+            )
+            
+    async def _handle_stt_stop(self, event: Event):
+        """Handle STT stop request"""
+        session_id = event.session_id
+        reason = event.data.get("reason", "unknown")
+        
+        log_info(f"🛑 Stopping STT", session_id=session_id, reason=reason)
+        
+        stt_session = self.stt_sessions.get(session_id)
+        if not stt_session:
+            log_warning(f"⚠️ No STT session found", session_id=session_id)
+            return
+            
+        try:
+            if stt_session.is_streaming:
+                # Stop streaming
+                final_result = await stt_session.stt_instance.stop_streaming()
+                stt_session.is_streaming = False
+                
+                # If we got a final result, publish it
+                if final_result and final_result.text:
+                    await self.event_bus.publish(Event(
+                        type=EventType.STT_RESULT,
+                        session_id=session_id,
+                        data={
+                            "text": final_result.text,
+                            "is_final": True,
+                            "confidence": final_result.confidence
+                        }
+                    ))
+                    
+            # Don't remove session immediately - might restart
+            stt_session.update_activity()
+            
+            log_info(f"✅ STT stopped", session_id=session_id)
+            
+        except Exception as e:
+            log_error(
+                f"❌ Error stopping STT",
+                session_id=session_id,
+                error=str(e)
+            )
+            
+    async def _handle_audio_chunk(self, event: Event):
+        """Process audio chunk through STT"""
+        session_id = event.session_id
+        
+        stt_session = self.stt_sessions.get(session_id)
+        if not stt_session or not stt_session.is_streaming:
+            # STT not ready, ignore chunk
+            return
+            
+        try:
+            # Decode audio data
+            audio_data = base64.b64decode(event.data.get("audio_data", ""))
+            
+            # Update stats
+            stt_session.total_chunks += 1
+            stt_session.total_bytes += len(audio_data)
+            stt_session.update_activity()
+            
+            # Stream to STT
+            async for result in stt_session.stt_instance.stream_audio(audio_data):
+                # Publish transcription results
+                await self.event_bus.publish(Event(
+                    type=EventType.STT_RESULT,
+                    session_id=session_id,
+                    data={
+                        "text": result.text,
+                        "is_final": result.is_final,
+                        "confidence": result.confidence,
+                        "timestamp": result.timestamp
+                    }
+                ))
+                
+                # Log final results
+                if result.is_final:
+                    log_info(
+                        f"📝 STT final result",
+                        session_id=session_id,
+                        text=result.text[:50] + "..." if len(result.text) > 50 else result.text,
+                        confidence=result.confidence
+                    )
+                    
+            # Log progress periodically
+            if stt_session.total_chunks % 100 == 0:
+                log_debug(
+                    f"📊 STT progress",
+                    session_id=session_id,
+                    chunks=stt_session.total_chunks,
+                    bytes=stt_session.total_bytes
+                )
+                
+        except Exception as e:
+            log_error(
+                f"❌ Error processing audio chunk",
+                session_id=session_id,
+                error=str(e)
+            )
+            
+            # Check if it's a recoverable error
+            if "stream duration" in str(e) or "timeout" in str(e).lower():
+                # STT timeout, restart needed
+                await publish_error(
+                    session_id=session_id,
+                    error_type="stt_timeout",
+                    error_message="STT stream timeout, restart needed"
+                )
+            else:
+                # Other STT error
+                await publish_error(
+                    session_id=session_id,
+                    error_type="stt_error",
+                    error_message=str(e)
+                )
+                
+    async def _handle_session_ended(self, event: Event):
+        """Clean up STT resources when session ends"""
+        session_id = event.session_id
+        await self._cleanup_session(session_id)
+        
+    async def _cleanup_session(self, session_id: str):
+        """Clean up STT session"""
+        stt_session = self.stt_sessions.pop(session_id, None)
+        if not stt_session:
+            return
+            
+        try:
+            # Stop streaming if active
+            if stt_session.is_streaming:
+                await stt_session.stt_instance.stop_streaming()
+                
+            # Release resource
+            resource_id = f"stt_{session_id}"
+            await self.resource_manager.release(resource_id, delay_seconds=60)
+            
+            log_info(
+                f"🧹 STT session cleaned up",
+                session_id=session_id,
+                total_chunks=stt_session.total_chunks,
+                total_bytes=stt_session.total_bytes
+            )
+            
+        except Exception as e:
+            log_error(
+                f"❌ Error cleaning up STT session",
+                session_id=session_id,
+                error=str(e)
+            )
+            
+    async def _cleanup_stt_instance(self, stt_instance: STTInterface):
+        """Cleanup callback for STT instance"""
+        try:
+            # Ensure streaming is stopped
+            if hasattr(stt_instance, 'is_streaming') and stt_instance.is_streaming:
+                await stt_instance.stop_streaming()
+                
+            log_debug("🧹 STT instance cleaned up")
+            
+        except Exception as e:
+            log_error(f"❌ Error cleaning up STT instance", error=str(e))
+            
+    def _get_language_code(self, locale: str) -> str:
+        """Convert locale to STT language code"""
+        # Map common locales to STT language codes
+        locale_map = {
+            "tr": "tr-TR",
+            "en": "en-US",
+            "de": "de-DE",
+            "fr": "fr-FR",
+            "es": "es-ES",
+            "it": "it-IT",
+            "pt": "pt-BR",
+            "ru": "ru-RU",
+            "ja": "ja-JP",
+            "ko": "ko-KR",
+            "zh": "zh-CN",
+            "ar": "ar-SA"
+        }
+        
+        # Check direct match
+        if locale in locale_map:
+            return locale_map[locale]
+            
+        # Check if it's already a full code
+        if "-" in locale and len(locale) == 5:
+            return locale
+            
+        # Default to locale-LOCALE format
+        return f"{locale}-{locale.upper()}"
+        
+    def get_stats(self) -> Dict[str, Any]:
+        """Get STT manager statistics"""
+        session_stats = {}
+        for session_id, stt_session in self.stt_sessions.items():
+            session_stats[session_id] = {
+                "is_streaming": stt_session.is_streaming,
+                "total_chunks": stt_session.total_chunks,
+                "total_bytes": stt_session.total_bytes,
+                "uptime_seconds": (datetime.utcnow() - stt_session.created_at).total_seconds(),
+                "last_activity": stt_session.last_activity.isoformat()
+            }
+            
+        return {
+            "active_sessions": len(self.stt_sessions),
+            "streaming_sessions": sum(1 for s in self.stt_sessions.values() if s.is_streaming),
+            "sessions": session_stats
+        }

tts_lifecycle_manager.py

@@ -0,0 +1,377 @@
+"""
+TTS Lifecycle Manager for Flare
+===============================
+Manages TTS instances lifecycle per session
+"""
+import asyncio
+from typing import Dict, Optional, Any, List
+from datetime import datetime
+import traceback
+import base64
+
+from event_bus import EventBus, Event, EventType, publish_error
+from resource_manager import ResourceManager, ResourceType
+from tts_factory import TTSFactory
+from tts_interface import TTSInterface
+from tts_preprocessor import TTSPreprocessor
+from logger import log_info, log_error, log_debug, log_warning
+
+
+class TTSJob:
+    """TTS synthesis job"""
+    
+    def __init__(self, job_id: str, session_id: str, text: str, is_welcome: bool = False):
+        self.job_id = job_id
+        self.session_id = session_id
+        self.text = text
+        self.is_welcome = is_welcome
+        self.created_at = datetime.utcnow()
+        self.completed_at: Optional[datetime] = None
+        self.audio_data: Optional[bytes] = None
+        self.error: Optional[str] = None
+        self.chunks_sent = 0
+        
+    def complete(self, audio_data: bytes):
+        """Mark job as completed"""
+        self.audio_data = audio_data
+        self.completed_at = datetime.utcnow()
+        
+    def fail(self, error: str):
+        """Mark job as failed"""
+        self.error = error
+        self.completed_at = datetime.utcnow()
+
+
+class TTSSession:
+    """TTS session wrapper"""
+    
+    def __init__(self, session_id: str, tts_instance: TTSInterface):
+        self.session_id = session_id
+        self.tts_instance = tts_instance
+        self.preprocessor: Optional[TTSPreprocessor] = None
+        self.active_jobs: Dict[str, TTSJob] = {}
+        self.completed_jobs: List[TTSJob] = []
+        self.created_at = datetime.utcnow()
+        self.last_activity = datetime.utcnow()
+        self.total_jobs = 0
+        self.total_chars = 0
+        
+    def update_activity(self):
+        """Update last activity timestamp"""
+        self.last_activity = datetime.utcnow()
+
+
+class TTSLifecycleManager:
+    """Manages TTS instances lifecycle"""
+    
+    def __init__(self, event_bus: EventBus, resource_manager: ResourceManager):
+        self.event_bus = event_bus
+        self.resource_manager = resource_manager
+        self.tts_sessions: Dict[str, TTSSession] = {}
+        self.chunk_size = 16384  # 16KB chunks for base64
+        self._setup_event_handlers()
+        self._setup_resource_pool()
+        
+    def _setup_event_handlers(self):
+        """Subscribe to TTS-related events"""
+        self.event_bus.subscribe(EventType.TTS_STARTED, self._handle_tts_start)
+        self.event_bus.subscribe(EventType.SESSION_ENDED, self._handle_session_ended)
+        
+    def _setup_resource_pool(self):
+        """Setup TTS instance pool"""
+        self.resource_manager.register_pool(
+            resource_type=ResourceType.TTS_INSTANCE,
+            factory=self._create_tts_instance,
+            max_idle=3,
+            max_age_seconds=600  # 10 minutes
+        )
+        
+    async def _create_tts_instance(self) -> Optional[TTSInterface]:
+        """Factory for creating TTS instances"""
+        try:
+            tts_instance = TTSFactory.create_provider()
+            if not tts_instance:
+                log_warning("⚠️ No TTS provider configured")
+                return None
+                
+            log_debug("🔊 Created new TTS instance")
+            return tts_instance
+            
+        except Exception as e:
+            log_error(f"❌ Failed to create TTS instance", error=str(e))
+            return None
+            
+    async def _handle_tts_start(self, event: Event):
+        """Handle TTS synthesis request"""
+        session_id = event.session_id
+        text = event.data.get("text", "")
+        is_welcome = event.data.get("is_welcome", False)
+        
+        if not text:
+            log_warning(f"⚠️ Empty text for TTS", session_id=session_id)
+            return
+            
+        try:
+            log_info(
+                f"🔊 Starting TTS",
+                session_id=session_id,
+                text_length=len(text),
+                is_welcome=is_welcome
+            )
+            
+            # Get or create session
+            if session_id not in self.tts_sessions:
+                # Acquire TTS instance from pool
+                resource_id = f"tts_{session_id}"
+                tts_instance = await self.resource_manager.acquire(
+                    resource_id=resource_id,
+                    session_id=session_id,
+                    resource_type=ResourceType.TTS_INSTANCE,
+                    cleanup_callback=self._cleanup_tts_instance
+                )
+                
+                if not tts_instance:
+                    # No TTS available
+                    await self._handle_no_tts(session_id, text, is_welcome)
+                    return
+                    
+                # Create session
+                tts_session = TTSSession(session_id, tts_instance)
+                
+                # Get locale from event data or default
+                locale = event.data.get("locale", "tr")
+                tts_session.preprocessor = TTSPreprocessor(language=locale)
+                
+                self.tts_sessions[session_id] = tts_session
+            else:
+                tts_session = self.tts_sessions[session_id]
+                
+            # Create job
+            job_id = f"{session_id}_{tts_session.total_jobs}"
+            job = TTSJob(job_id, session_id, text, is_welcome)
+            tts_session.active_jobs[job_id] = job
+            tts_session.total_jobs += 1
+            tts_session.total_chars += len(text)
+            tts_session.update_activity()
+            
+            # Process TTS
+            await self._process_tts_job(tts_session, job)
+            
+        except Exception as e:
+            log_error(
+                f"❌ Failed to start TTS",
+                session_id=session_id,
+                error=str(e),
+                traceback=traceback.format_exc()
+            )
+            
+            # Publish error event
+            await publish_error(
+                session_id=session_id,
+                error_type="tts_error",
+                error_message=f"Failed to synthesize speech: {str(e)}"
+            )
+            
+    async def _process_tts_job(self, tts_session: TTSSession, job: TTSJob):
+        """Process a TTS job"""
+        try:
+            # Preprocess text
+            processed_text = tts_session.preprocessor.preprocess(
+                job.text,
+                tts_session.tts_instance.get_preprocessing_flags()
+            )
+            
+            log_debug(
+                f"📝 TTS preprocessed",
+                session_id=job.session_id,
+                original_length=len(job.text),
+                processed_length=len(processed_text)
+            )
+            
+            # Synthesize audio
+            audio_data = await tts_session.tts_instance.synthesize(processed_text)
+            
+            if not audio_data:
+                raise ValueError("TTS returned empty audio data")
+                
+            job.complete(audio_data)
+            
+            log_info(
+                f"✅ TTS synthesis complete",
+                session_id=job.session_id,
+                audio_size=len(audio_data),
+                duration_ms=(datetime.utcnow() - job.created_at).total_seconds() * 1000
+            )
+            
+            # Stream audio chunks
+            await self._stream_audio_chunks(tts_session, job)
+            
+            # Move to completed
+            tts_session.active_jobs.pop(job.job_id, None)
+            tts_session.completed_jobs.append(job)
+            
+            # Keep only last 10 completed jobs
+            if len(tts_session.completed_jobs) > 10:
+                tts_session.completed_jobs.pop(0)
+                
+        except Exception as e:
+            job.fail(str(e))
+            
+            # Handle specific TTS errors
+            error_message = str(e)
+            if "quota" in error_message.lower() or "limit" in error_message.lower():
+                log_error(f"❌ TTS quota exceeded", session_id=job.session_id)
+                await publish_error(
+                    session_id=job.session_id,
+                    error_type="tts_quota_exceeded",
+                    error_message="TTS service quota exceeded"
+                )
+            else:
+                log_error(
+                    f"❌ TTS synthesis failed",
+                    session_id=job.session_id,
+                    error=error_message
+                )
+                await publish_error(
+                    session_id=job.session_id,
+                    error_type="tts_error",
+                    error_message=error_message
+                )
+                
+    async def _stream_audio_chunks(self, tts_session: TTSSession, job: TTSJob):
+        """Stream audio data as chunks"""
+        if not job.audio_data:
+            return
+            
+        # Convert to base64
+        audio_base64 = base64.b64encode(job.audio_data).decode('utf-8')
+        total_length = len(audio_base64)
+        total_chunks = (total_length + self.chunk_size - 1) // self.chunk_size
+        
+        log_debug(
+            f"📤 Streaming TTS audio",
+            session_id=job.session_id,
+            total_size=len(job.audio_data),
+            base64_size=total_length,
+            chunks=total_chunks
+        )
+        
+        # Stream chunks
+        for i in range(0, total_length, self.chunk_size):
+            chunk = audio_base64[i:i + self.chunk_size]
+            chunk_index = i // self.chunk_size
+            is_last = chunk_index == total_chunks - 1
+            
+            await self.event_bus.publish(Event(
+                type=EventType.TTS_CHUNK_READY,
+                session_id=job.session_id,
+                data={
+                    "audio_data": chunk,
+                    "chunk_index": chunk_index,
+                    "total_chunks": total_chunks,
+                    "is_last": is_last,
+                    "mime_type": "audio/mpeg",
+                    "is_welcome": job.is_welcome
+                },
+                priority=8  # Higher priority for audio chunks
+            ))
+            
+            job.chunks_sent += 1
+            
+            # Small delay between chunks to prevent overwhelming
+            await asyncio.sleep(0.01)
+            
+        # Notify completion
+        await self.event_bus.publish(Event(
+            type=EventType.TTS_COMPLETED,
+            session_id=job.session_id,
+            data={
+                "job_id": job.job_id,
+                "total_chunks": total_chunks,
+                "is_welcome": job.is_welcome
+            }
+        ))
+        
+        log_info(
+            f"✅ TTS streaming complete",
+            session_id=job.session_id,
+            chunks_sent=job.chunks_sent
+        )
+        
+    async def _handle_no_tts(self, session_id: str, text: str, is_welcome: bool):
+        """Handle case when TTS is not available"""
+        log_warning(f"⚠️ No TTS available, skipping audio generation", session_id=session_id)
+        
+        # Just notify completion without audio
+        await self.event_bus.publish(Event(
+            type=EventType.TTS_COMPLETED,
+            session_id=session_id,
+            data={
+                "no_audio": True,
+                "text": text,
+                "is_welcome": is_welcome
+            }
+        ))
+        
+    async def _handle_session_ended(self, event: Event):
+        """Clean up TTS resources when session ends"""
+        session_id = event.session_id
+        await self._cleanup_session(session_id)
+        
+    async def _cleanup_session(self, session_id: str):
+        """Clean up TTS session"""
+        tts_session = self.tts_sessions.pop(session_id, None)
+        if not tts_session:
+            return
+            
+        try:
+            # Cancel any active jobs
+            for job in tts_session.active_jobs.values():
+                if not job.completed_at:
+                    job.fail("Session ended")
+                    
+            # Release resource
+            resource_id = f"tts_{session_id}"
+            await self.resource_manager.release(resource_id, delay_seconds=120)
+            
+            log_info(
+                f"🧹 TTS session cleaned up",
+                session_id=session_id,
+                total_jobs=tts_session.total_jobs,
+                total_chars=tts_session.total_chars
+            )
+            
+        except Exception as e:
+            log_error(
+                f"❌ Error cleaning up TTS session",
+                session_id=session_id,
+                error=str(e)
+            )
+            
+    async def _cleanup_tts_instance(self, tts_instance: TTSInterface):
+        """Cleanup callback for TTS instance"""
+        try:
+            # TTS instances typically don't need special cleanup
+            log_debug("🧹 TTS instance cleaned up")
+            
+        except Exception as e:
+            log_error(f"❌ Error cleaning up TTS instance", error=str(e))
+            
+    def get_stats(self) -> Dict[str, Any]:
+        """Get TTS manager statistics"""
+        session_stats = {}
+        for session_id, tts_session in self.tts_sessions.items():
+            session_stats[session_id] = {
+                "active_jobs": len(tts_session.active_jobs),
+                "completed_jobs": len(tts_session.completed_jobs),
+                "total_jobs": tts_session.total_jobs,
+                "total_chars": tts_session.total_chars,
+                "uptime_seconds": (datetime.utcnow() - tts_session.created_at).total_seconds(),
+                "last_activity": tts_session.last_activity.isoformat()
+            }
+            
+        return {
+            "active_sessions": len(self.tts_sessions),
+            "total_active_jobs": sum(len(s.active_jobs) for s in self.tts_sessions.values()),
+            "sessions": session_stats
+        }

websocket_manager.py

@@ -0,0 +1,408 @@
+"""
+WebSocket Manager for Flare
+===========================
+Manages WebSocket connections and message routing
+"""
+import asyncio
+from typing import Dict, Optional, Set
+from fastapi import WebSocket, WebSocketDisconnect
+import json
+from datetime import datetime
+import traceback
+
+from event_bus import EventBus, Event, EventType
+from logger import log_info, log_error, log_debug, log_warning
+
+
+class WebSocketConnection:
+    """Wrapper for WebSocket connection with metadata"""
+    
+    def __init__(self, websocket: WebSocket, session_id: str):
+        self.websocket = websocket
+        self.session_id = session_id
+        self.connected_at = datetime.utcnow()
+        self.last_activity = datetime.utcnow()
+        self.is_active = True
+        
+    async def send_json(self, data: dict):
+        """Send JSON data to client"""
+        try:
+            if self.is_active:
+                await self.websocket.send_json(data)
+                self.last_activity = datetime.utcnow()
+        except Exception as e:
+            log_error(
+                f"❌ Failed to send message",
+                session_id=self.session_id,
+                error=str(e)
+            )
+            self.is_active = False
+            raise
+            
+    async def receive_json(self) -> dict:
+        """Receive JSON data from client"""
+        try:
+            data = await self.websocket.receive_json()
+            self.last_activity = datetime.utcnow()
+            return data
+        except WebSocketDisconnect:
+            self.is_active = False
+            raise
+        except Exception as e:
+            log_error(
+                f"❌ Failed to receive message",
+                session_id=self.session_id,
+                error=str(e)
+            )
+            self.is_active = False
+            raise
+            
+    async def close(self):
+        """Close the connection"""
+        try:
+            self.is_active = False
+            await self.websocket.close()
+        except:
+            pass
+
+
+class WebSocketManager:
+    """Manages WebSocket connections and routing"""
+    
+    def __init__(self, event_bus: EventBus):
+        self.event_bus = event_bus
+        self.connections: Dict[str, WebSocketConnection] = {}
+        self.message_queues: Dict[str, asyncio.Queue] = {}
+        self._setup_event_handlers()
+        
+    def _setup_event_handlers(self):
+        """Subscribe to events that need to be sent to clients"""
+        # State events
+        self.event_bus.subscribe(EventType.STATE_TRANSITION, self._handle_state_transition)
+        
+        # STT events
+        self.event_bus.subscribe(EventType.STT_READY, self._handle_stt_ready)
+        self.event_bus.subscribe(EventType.STT_RESULT, self._handle_stt_result)
+        
+        # TTS events
+        self.event_bus.subscribe(EventType.TTS_CHUNK_READY, self._handle_tts_chunk)
+        self.event_bus.subscribe(EventType.TTS_COMPLETED, self._handle_tts_completed)
+        
+        # LLM events
+        self.event_bus.subscribe(EventType.LLM_RESPONSE_READY, self._handle_llm_response)
+        
+        # Error events
+        self.event_bus.subscribe(EventType.RECOVERABLE_ERROR, self._handle_error)
+        self.event_bus.subscribe(EventType.CRITICAL_ERROR, self._handle_error)
+        
+    async def connect(self, websocket: WebSocket, session_id: str):
+        """Accept new WebSocket connection"""
+        await websocket.accept()
+        
+        # Check for existing connection
+        if session_id in self.connections:
+            log_warning(
+                f"⚠️ Existing connection for session, closing old one",
+                session_id=session_id
+            )
+            await self.disconnect(session_id)
+            
+        # Create connection wrapper
+        connection = WebSocketConnection(websocket, session_id)
+        self.connections[session_id] = connection
+        
+        # Create message queue
+        self.message_queues[session_id] = asyncio.Queue()
+        
+        log_info(
+            f"✅ WebSocket connected",
+            session_id=session_id,
+            total_connections=len(self.connections)
+        )
+        
+        # Publish connection event
+        await self.event_bus.publish(Event(
+            type=EventType.WEBSOCKET_CONNECTED,
+            session_id=session_id,
+            data={}
+        ))
+        
+    async def disconnect(self, session_id: str):
+        """Disconnect WebSocket connection"""
+        connection = self.connections.get(session_id)
+        if connection:
+            await connection.close()
+            del self.connections[session_id]
+            
+        # Remove message queue
+        if session_id in self.message_queues:
+            del self.message_queues[session_id]
+            
+        log_info(
+            f"🔌 WebSocket disconnected",
+            session_id=session_id,
+            total_connections=len(self.connections)
+        )
+        
+        # Publish disconnection event
+        await self.event_bus.publish(Event(
+            type=EventType.WEBSOCKET_DISCONNECTED,
+            session_id=session_id,
+            data={}
+        ))
+        
+    async def handle_connection(self, websocket: WebSocket, session_id: str):
+        """Handle WebSocket connection lifecycle"""
+        try:
+            # Connect
+            await self.connect(websocket, session_id)
+            
+            # Create tasks for bidirectional communication
+            receive_task = asyncio.create_task(self._receive_messages(session_id))
+            send_task = asyncio.create_task(self._send_messages(session_id))
+            
+            # Wait for either task to complete
+            done, pending = await asyncio.wait(
+                [receive_task, send_task],
+                return_when=asyncio.FIRST_COMPLETED
+            )
+            
+            # Cancel pending tasks
+            for task in pending:
+                task.cancel()
+                try:
+                    await task
+                except asyncio.CancelledError:
+                    pass
+                    
+        except WebSocketDisconnect:
+            log_info(f"WebSocket disconnected normally", session_id=session_id)
+        except Exception as e:
+            log_error(
+                f"❌ WebSocket error",
+                session_id=session_id,
+                error=str(e),
+                traceback=traceback.format_exc()
+            )
+            
+            # Publish error event
+            await self.event_bus.publish(Event(
+                type=EventType.WEBSOCKET_ERROR,
+                session_id=session_id,
+                data={
+                    "error_type": "websocket_error",
+                    "message": str(e)
+                }
+            ))
+        finally:
+            # Ensure disconnection
+            await self.disconnect(session_id)
+            
+    async def _receive_messages(self, session_id: str):
+        """Receive messages from client"""
+        connection = self.connections.get(session_id)
+        if not connection:
+            return
+            
+        try:
+            while connection.is_active:
+                # Receive message
+                message = await connection.receive_json()
+                
+                log_debug(
+                    f"📨 Received message",
+                    session_id=session_id,
+                    message_type=message.get("type")
+                )
+                
+                # Route message based on type
+                await self._route_client_message(session_id, message)
+                
+        except WebSocketDisconnect:
+            log_info(f"Client disconnected", session_id=session_id)
+        except Exception as e:
+            log_error(
+                f"❌ Error receiving messages",
+                session_id=session_id,
+                error=str(e)
+            )
+            raise
+            
+    async def _send_messages(self, session_id: str):
+        """Send queued messages to client"""
+        connection = self.connections.get(session_id)
+        queue = self.message_queues.get(session_id)
+        
+        if not connection or not queue:
+            return
+            
+        try:
+            while connection.is_active:
+                # Wait for message with timeout
+                try:
+                    message = await asyncio.wait_for(queue.get(), timeout=30.0)
+                    
+                    # Send to client
+                    await connection.send_json(message)
+                    
+                    log_debug(
+                        f"📤 Sent message",
+                        session_id=session_id,
+                        message_type=message.get("type")
+                    )
+                    
+                except asyncio.TimeoutError:
+                    # Send ping to keep connection alive
+                    await connection.send_json({"type": "ping"})
+                    
+        except Exception as e:
+            log_error(
+                f"❌ Error sending messages",
+                session_id=session_id,
+                error=str(e)
+            )
+            raise
+            
+    async def _route_client_message(self, session_id: str, message: dict):
+        """Route message from client to appropriate handler"""
+        message_type = message.get("type")
+        
+        if message_type == "audio_chunk":
+            # Audio data from client
+            await self.event_bus.publish(Event(
+                type=EventType.AUDIO_CHUNK_RECEIVED,
+                session_id=session_id,
+                data={
+                    "audio_data": message.get("data"),
+                    "timestamp": message.get("timestamp")
+                }
+            ))
+            
+        elif message_type == "control":
+            # Control messages
+            action = message.get("action")
+            
+            if action == "start_session":
+                await self.event_bus.publish(Event(
+                    type=EventType.SESSION_STARTED,
+                    session_id=session_id,
+                    data=message.get("config", {})
+                ))
+                
+            elif action == "end_session":
+                await self.event_bus.publish(Event(
+                    type=EventType.SESSION_ENDED,
+                    session_id=session_id,
+                    data={"reason": "user_request"}
+                ))
+                
+            elif action == "audio_ended":
+                await self.event_bus.publish(Event(
+                    type=EventType.AUDIO_PLAYBACK_COMPLETED,
+                    session_id=session_id,
+                    data={}
+                ))
+                
+        elif message_type == "ping":
+            # Respond to ping
+            await self.send_message(session_id, {"type": "pong"})
+            
+        else:
+            log_warning(
+                f"⚠️ Unknown message type",
+                session_id=session_id,
+                message_type=message_type
+            )
+            
+    async def send_message(self, session_id: str, message: dict):
+        """Queue message for sending to client"""
+        queue = self.message_queues.get(session_id)
+        if queue:
+            await queue.put(message)
+        else:
+            log_warning(
+                f"⚠️ No queue for session",
+                session_id=session_id
+            )
+            
+    async def broadcast_to_session(self, session_id: str, message: dict):
+        """Send message immediately (bypass queue)"""
+        connection = self.connections.get(session_id)
+        if connection and connection.is_active:
+            await connection.send_json(message)
+            
+    # Event handlers for sending messages to clients
+    
+    async def _handle_state_transition(self, event: Event):
+        """Send state transition to client"""
+        await self.send_message(event.session_id, {
+            "type": "state_change",
+            "from": event.data.get("from_state"),
+            "to": event.data.get("to_state")
+        })
+        
+    async def _handle_stt_ready(self, event: Event):
+        """Send STT ready signal to client"""
+        await self.send_message(event.session_id, {
+            "type": "stt_ready",
+            "message": "STT is ready to receive audio"
+        })
+        
+    async def _handle_stt_result(self, event: Event):
+        """Send STT result to client"""
+        await self.send_message(event.session_id, {
+            "type": "transcription",
+            "text": event.data.get("text", ""),
+            "is_final": event.data.get("is_final", False),
+            "confidence": event.data.get("confidence", 0.0)
+        })
+        
+    async def _handle_tts_chunk(self, event: Event):
+        """Send TTS audio chunk to client"""
+        await self.send_message(event.session_id, {
+            "type": "tts_audio",
+            "data": event.data.get("audio_data"),
+            "chunk_index": event.data.get("chunk_index"),
+            "total_chunks": event.data.get("total_chunks"),
+            "is_last": event.data.get("is_last", False),
+            "mime_type": event.data.get("mime_type", "audio/mpeg")
+        })
+        
+    async def _handle_tts_completed(self, event: Event):
+        """Notify client that TTS is complete"""
+        # Client knows from is_last flag in chunks
+        pass
+        
+    async def _handle_llm_response(self, event: Event):
+        """Send LLM response to client"""
+        await self.send_message(event.session_id, {
+            "type": "assistant_response",
+            "text": event.data.get("text", ""),
+            "is_welcome": event.data.get("is_welcome", False)
+        })
+        
+    async def _handle_error(self, event: Event):
+        """Send error to client"""
+        error_type = event.data.get("error_type", "unknown")
+        message = event.data.get("message", "An error occurred")
+        
+        await self.send_message(event.session_id, {
+            "type": "error",
+            "error_type": error_type,
+            "message": message,
+            "details": event.data.get("details", {})
+        })
+        
+    def get_connection_count(self) -> int:
+        """Get number of active connections"""
+        return len(self.connections)
+        
+    def get_session_connections(self) -> Set[str]:
+        """Get all active session IDs"""
+        return set(self.connections.keys())
+        
+    async def close_all_connections(self):
+        """Close all active connections"""
+        session_ids = list(self.connections.keys())
+        for session_id in session_ids:
+            await self.disconnect(session_id)