dispatcher.py

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from typing import List, Dict, Any, Optional
import asyncio
import json
import time
import os
from dataclasses import dataclass, asdict
from enum import Enum
import uuid
import aiohttp
import logging
from collections import defaultdict, deque
from datetime import datetime

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Analytics and monitoring
class SystemAnalytics:
    def __init__(self):
        self.start_time = time.time()
        self.total_connections = 0
        self.active_connections = 0
        self.total_interactions = 0
        self.ip_addresses = defaultdict(int)  # IP -> connection count
        self.session_durations = deque(maxlen=100)  # Last 100 session durations
        self.waiting_times = deque(maxlen=100)  # Last 100 waiting times
        self.users_bypassed_queue = 0  # Users who got GPU immediately
        self.users_waited_in_queue = 0  # Users who had to wait
        self.gpu_utilization_samples = deque(maxlen=100)  # GPU utilization over time
        self.queue_size_samples = deque(maxlen=100)  # Queue size over time
        self.log_file = None
        self._init_log_file()
    
    def _init_log_file(self):
        """Initialize the system log file"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        log_filename = f"system_analytics_{timestamp}.log"
        self.log_file = log_filename
        self._write_log("="*80)
        self._write_log("NEURAL OS MULTI-GPU SYSTEM ANALYTICS")
        self._write_log("="*80)
        self._write_log(f"System started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        self._write_log("")
    
    def _write_log(self, message):
        """Write message to log file and console"""
        timestamp = datetime.now().strftime("%H:%M:%S")
        log_message = f"[{timestamp}] {message}"
        print(log_message)
        with open(self.log_file, "a") as f:
            f.write(log_message + "\n")
    
    def log_new_connection(self, client_id: str, ip: str):
        """Log new connection"""
        self.total_connections += 1
        self.active_connections += 1
        self.ip_addresses[ip] += 1
        
        unique_ips = len(self.ip_addresses)
        self._write_log(f"🔗 NEW CONNECTION: {client_id} from {ip}")
        self._write_log(f"   📊 Total connections: {self.total_connections} | Active: {self.active_connections} | Unique IPs: {unique_ips}")
    
    def log_connection_closed(self, client_id: str, duration: float, interactions: int, reason: str = "normal"):
        """Log connection closed"""
        self.active_connections -= 1
        self.total_interactions += interactions
        self.session_durations.append(duration)
        
        avg_duration = sum(self.session_durations) / len(self.session_durations) if self.session_durations else 0
        
        self._write_log(f"🚪 CONNECTION CLOSED: {client_id}")
        self._write_log(f"   ⏱️  Duration: {duration:.1f}s | Interactions: {interactions} | Reason: {reason}")
        self._write_log(f"   📊 Active connections: {self.active_connections} | Avg session duration: {avg_duration:.1f}s")
    
    def log_queue_bypass(self, client_id: str):
        """Log when user bypasses queue (gets GPU immediately)"""
        self.users_bypassed_queue += 1
        bypass_rate = (self.users_bypassed_queue / self.total_connections) * 100 if self.total_connections > 0 else 0
        self._write_log(f"⚡ QUEUE BYPASS: {client_id} got GPU immediately")
        self._write_log(f"   📊 Bypass rate: {bypass_rate:.1f}% ({self.users_bypassed_queue}/{self.total_connections})")
    
    def log_queue_wait(self, client_id: str, wait_time: float, queue_position: int):
        """Log when user had to wait in queue"""
        self.users_waited_in_queue += 1
        self.waiting_times.append(wait_time)
        
        avg_wait = sum(self.waiting_times) / len(self.waiting_times) if self.waiting_times else 0
        wait_rate = (self.users_waited_in_queue / self.total_connections) * 100 if self.total_connections > 0 else 0
        
        self._write_log(f"⏳ QUEUE WAIT: {client_id} waited {wait_time:.1f}s (was #{queue_position})")
        self._write_log(f"   📊 Wait rate: {wait_rate:.1f}% | Avg wait time: {avg_wait:.1f}s")
    
    def log_gpu_status(self, total_gpus: int, active_gpus: int, available_gpus: int):
        """Log GPU utilization"""
        utilization = (active_gpus / total_gpus) * 100 if total_gpus > 0 else 0
        self.gpu_utilization_samples.append(utilization)
        
        avg_utilization = sum(self.gpu_utilization_samples) / len(self.gpu_utilization_samples) if self.gpu_utilization_samples else 0
        
        self._write_log(f"🖥️  GPU STATUS: {active_gpus}/{total_gpus} in use ({utilization:.1f}% utilization)")
        self._write_log(f"   📊 Available: {available_gpus} | Avg utilization: {avg_utilization:.1f}%")
    
    def log_worker_registered(self, worker_id: str, gpu_id: int, endpoint: str):
        """Log when a worker registers"""
        self._write_log(f"⚙️  WORKER REGISTERED: {worker_id} (GPU {gpu_id}) at {endpoint}")
    
    def log_worker_disconnected(self, worker_id: str, gpu_id: int):
        """Log when a worker disconnects"""
        self._write_log(f"⚙️  WORKER DISCONNECTED: {worker_id} (GPU {gpu_id})")
    
    def log_no_workers_available(self, queue_size: int):
        """Log critical situation when no workers are available"""
        self._write_log(f"⚠️  CRITICAL: No GPU workers available! {queue_size} users waiting")
        self._write_log("   Please check worker processes and GPU availability")
    
    def log_queue_status(self, queue_size: int, estimated_wait: float):
        """Log queue status"""
        self.queue_size_samples.append(queue_size)
        
        avg_queue_size = sum(self.queue_size_samples) / len(self.queue_size_samples) if self.queue_size_samples else 0
        
        if queue_size > 0:
            self._write_log(f"📝 QUEUE STATUS: {queue_size} users waiting | Est. wait: {estimated_wait:.1f}s")
            self._write_log(f"   📊 Avg queue size: {avg_queue_size:.1f}")
    
    def log_periodic_summary(self):
        """Log periodic system summary"""
        uptime = time.time() - self.start_time
        uptime_hours = uptime / 3600
        
        unique_ips = len(self.ip_addresses)
        avg_duration = sum(self.session_durations) / len(self.session_durations) if self.session_durations else 0
        avg_wait = sum(self.waiting_times) / len(self.waiting_times) if self.waiting_times else 0
        avg_utilization = sum(self.gpu_utilization_samples) / len(self.gpu_utilization_samples) if self.gpu_utilization_samples else 0
        avg_queue_size = sum(self.queue_size_samples) / len(self.queue_size_samples) if self.queue_size_samples else 0
        
        bypass_rate = (self.users_bypassed_queue / self.total_connections) * 100 if self.total_connections > 0 else 0
        
        self._write_log("")
        self._write_log("="*60)
        self._write_log("📊 SYSTEM SUMMARY")
        self._write_log("="*60)
        self._write_log(f"⏱️  Uptime: {uptime_hours:.1f} hours")
        self._write_log(f"🔗 Connections: {self.total_connections} total | {self.active_connections} active | {unique_ips} unique IPs")
        self._write_log(f"💬 Total interactions: {self.total_interactions}")
        self._write_log(f"⚡ Queue bypass rate: {bypass_rate:.1f}% ({self.users_bypassed_queue}/{self.total_connections})")
        self._write_log(f"⏳ Avg waiting time: {avg_wait:.1f}s")
        self._write_log(f"📝 Avg queue size: {avg_queue_size:.1f}")
        self._write_log(f"🖥️  Avg GPU utilization: {avg_utilization:.1f}%")
        self._write_log(f"⏱️  Avg session duration: {avg_duration:.1f}s")
        self._write_log("")
        self._write_log("🌍 TOP IP ADDRESSES:")
        for ip, count in sorted(self.ip_addresses.items(), key=lambda x: x[1], reverse=True)[:10]:
            self._write_log(f"   {ip}: {count} connections")
        self._write_log("="*60)
        self._write_log("")

# Initialize analytics
analytics = SystemAnalytics()

class SessionStatus(Enum):
    QUEUED = "queued"
    ACTIVE = "active"
    COMPLETED = "completed"
    TIMEOUT = "timeout"

@dataclass
class UserSession:
    session_id: str
    client_id: str
    websocket: WebSocket
    created_at: float
    status: SessionStatus
    worker_id: Optional[str] = None
    last_activity: Optional[float] = None
    max_session_time: Optional[float] = None
    user_has_interacted: bool = False
    ip_address: Optional[str] = None
    interaction_count: int = 0
    queue_start_time: Optional[float] = None

@dataclass
class WorkerInfo:
    worker_id: str
    gpu_id: int
    endpoint: str
    is_available: bool
    current_session: Optional[str] = None
    last_ping: float = 0

class SessionManager:
    def __init__(self):
        self.sessions: Dict[str, UserSession] = {}
        self.workers: Dict[str, WorkerInfo] = {}
        self.session_queue: List[str] = []
        self.active_sessions: Dict[str, str] = {}  # session_id -> worker_id
        
        # Configuration
        self.IDLE_TIMEOUT = 20.0  # When no queue
        self.QUEUE_WARNING_TIME = 10.0
        self.MAX_SESSION_TIME_WITH_QUEUE = 60.0  # When there's a queue
        self.QUEUE_SESSION_WARNING_TIME = 45.0  # 15 seconds before timeout
        self.GRACE_PERIOD = 10.0

    async def register_worker(self, worker_id: str, gpu_id: int, endpoint: str):
        """Register a new worker"""
        self.workers[worker_id] = WorkerInfo(
            worker_id=worker_id,
            gpu_id=gpu_id,
            endpoint=endpoint,
            is_available=True,
            last_ping=time.time()
        )
        logger.info(f"Registered worker {worker_id} on GPU {gpu_id} at {endpoint}")
        
        # Log worker registration
        analytics.log_worker_registered(worker_id, gpu_id, endpoint)
        
        # Log GPU status
        total_gpus = len(self.workers)
        active_gpus = len([w for w in self.workers.values() if not w.is_available])
        available_gpus = total_gpus - active_gpus
        analytics.log_gpu_status(total_gpus, active_gpus, available_gpus)

    async def get_available_worker(self) -> Optional[WorkerInfo]:
        """Get an available worker"""
        for worker in self.workers.values():
            if worker.is_available and time.time() - worker.last_ping < 30:  # Worker ping timeout
                return worker
        return None

    async def add_session_to_queue(self, session: UserSession):
        """Add a session to the queue"""
        self.sessions[session.session_id] = session
        self.session_queue.append(session.session_id)
        session.status = SessionStatus.QUEUED
        session.queue_start_time = time.time()
        logger.info(f"Added session {session.session_id} to queue. Queue size: {len(self.session_queue)}")

    async def process_queue(self):
        """Process the session queue"""
        while self.session_queue:
            session_id = self.session_queue[0]
            session = self.sessions.get(session_id)
            
            if not session or session.status != SessionStatus.QUEUED:
                self.session_queue.pop(0)
                continue
                
            worker = await self.get_available_worker()
            if not worker:
                # Log critical situation if no workers are available
                if len(self.workers) == 0:
                    analytics.log_no_workers_available(len(self.session_queue))
                break  # No available workers
                
            # Calculate wait time
            wait_time = time.time() - session.queue_start_time if session.queue_start_time else 0
            queue_position = self.session_queue.index(session_id) + 1
            
            # Assign session to worker
            self.session_queue.pop(0)
            session.status = SessionStatus.ACTIVE
            session.worker_id = worker.worker_id
            session.last_activity = time.time()
            
            # Set session time limit based on queue status
            if len(self.session_queue) > 0:
                session.max_session_time = self.MAX_SESSION_TIME_WITH_QUEUE
            
            worker.is_available = False
            worker.current_session = session_id
            self.active_sessions[session_id] = worker.worker_id
            
            logger.info(f"Assigned session {session_id} to worker {worker.worker_id}")
            
            # Log analytics
            if wait_time > 0:
                analytics.log_queue_wait(session.client_id, wait_time, queue_position)
            else:
                analytics.log_queue_bypass(session.client_id)
            
            # Log GPU status
            total_gpus = len(self.workers)
            active_gpus = len([w for w in self.workers.values() if not w.is_available])
            available_gpus = total_gpus - active_gpus
            analytics.log_gpu_status(total_gpus, active_gpus, available_gpus)
            
            # Initialize session on worker with client_id for logging
            try:
                async with aiohttp.ClientSession() as client_session:
                    await client_session.post(f"{worker.endpoint}/init_session", json={
                        "session_id": session_id,
                        "client_id": session.client_id
                    })
            except Exception as e:
                logger.error(f"Failed to initialize session on worker {worker.worker_id}: {e}")
            
            # Notify user that their session is starting
            await self.notify_session_start(session)
            
            # Start session monitoring
            asyncio.create_task(self.monitor_active_session(session_id))

    async def notify_session_start(self, session: UserSession):
        """Notify user that their session is starting"""
        try:
            await session.websocket.send_json({
                "type": "session_start",
                "worker_id": session.worker_id,
                "max_session_time": session.max_session_time
            })
        except Exception as e:
            logger.error(f"Failed to notify session start for {session.session_id}: {e}")

    async def monitor_active_session(self, session_id: str):
        """Monitor an active session for timeouts"""
        session = self.sessions.get(session_id)
        if not session:
            return
            
        try:
            while session.status == SessionStatus.ACTIVE:
                current_time = time.time()
                
                # Check if session has exceeded time limit
                if session.max_session_time:
                    elapsed = current_time - session.last_activity if session.last_activity else 0
                    remaining = session.max_session_time - elapsed
                    
                    # Send warning at 15 seconds before timeout
                    if remaining <= 15 and remaining > 10:
                        await session.websocket.send_json({
                            "type": "session_warning",
                            "time_remaining": remaining,
                            "queue_size": len(self.session_queue)
                        })
                    
                    # Grace period handling
                    elif remaining <= 10 and remaining > 0:
                        # Check if queue is empty - if so, extend session
                        if len(self.session_queue) == 0:
                            session.max_session_time = None  # Remove time limit
                            await session.websocket.send_json({
                                "type": "time_limit_removed",
                                "reason": "queue_empty"
                            })
                        else:
                            await session.websocket.send_json({
                                "type": "grace_period",
                                "time_remaining": remaining,
                                "queue_size": len(self.session_queue)
                            })
                    
                    # Timeout
                    elif remaining <= 0:
                        await self.end_session(session_id, SessionStatus.TIMEOUT)
                        return
                
                # Check idle timeout when no queue
                elif not session.max_session_time and session.last_activity:
                    idle_time = current_time - session.last_activity
                    if idle_time >= self.IDLE_TIMEOUT:
                        await self.end_session(session_id, SessionStatus.TIMEOUT)
                        return
                    elif idle_time >= self.QUEUE_WARNING_TIME:
                        await session.websocket.send_json({
                            "type": "idle_warning",
                            "time_remaining": self.IDLE_TIMEOUT - idle_time
                        })
                
                await asyncio.sleep(1)  # Check every second
                
        except Exception as e:
            logger.error(f"Error monitoring session {session_id}: {e}")
            await self.end_session(session_id, SessionStatus.COMPLETED)

    async def end_session(self, session_id: str, status: SessionStatus):
        """End a session and free up the worker"""
        session = self.sessions.get(session_id)
        if not session:
            return
            
        session.status = status
        
        # Calculate session duration
        duration = time.time() - session.created_at
        
        # Log analytics
        reason = "timeout" if status == SessionStatus.TIMEOUT else "normal"
        analytics.log_connection_closed(session.client_id, duration, session.interaction_count, reason)
        
        # Free up the worker
        if session.worker_id and session.worker_id in self.workers:
            worker = self.workers[session.worker_id]
            worker.is_available = True
            worker.current_session = None
            
            # Log GPU status
            total_gpus = len(self.workers)
            active_gpus = len([w for w in self.workers.values() if not w.is_available])
            available_gpus = total_gpus - active_gpus
            analytics.log_gpu_status(total_gpus, active_gpus, available_gpus)
            
            # Notify worker to clean up
            try:
                async with aiohttp.ClientSession() as client_session:
                    await client_session.post(f"{worker.endpoint}/end_session", 
                                            json={"session_id": session_id})
            except Exception as e:
                logger.error(f"Failed to notify worker {worker.worker_id} of session end: {e}")
        
        # Remove from active sessions
        if session_id in self.active_sessions:
            del self.active_sessions[session_id]
            
        logger.info(f"Ended session {session_id} with status {status}")
        
        # Process next in queue
        asyncio.create_task(self.process_queue())

    async def update_queue_info(self):
        """Send queue information to waiting users"""
        for i, session_id in enumerate(self.session_queue):
            session = self.sessions.get(session_id)
            if session and session.status == SessionStatus.QUEUED:
                try:
                    # Calculate dynamic estimated wait time
                    estimated_wait = self._calculate_dynamic_wait_time(i + 1)
                    
                    await session.websocket.send_json({
                        "type": "queue_update",
                        "position": i + 1,
                        "total_waiting": len(self.session_queue),
                        "estimated_wait_seconds": estimated_wait,
                        "active_sessions": len(self.active_sessions),
                        "available_workers": len([w for w in self.workers.values() if w.is_available])
                    })
                except Exception as e:
                    logger.error(f"Failed to send queue update to session {session_id}: {e}")
        
        # Log queue status if there's a queue
        if self.session_queue:
            estimated_wait = self._calculate_dynamic_wait_time(1)
            analytics.log_queue_status(len(self.session_queue), estimated_wait)

    def _calculate_dynamic_wait_time(self, position_in_queue: int) -> float:
        """Calculate dynamic estimated wait time based on current session progress"""
        current_time = time.time()
        available_workers = len([w for w in self.workers.values() if w.is_available])
        
        # If there are available workers, no wait time
        if available_workers > 0:
            return 0
        
        # Calculate remaining time for active sessions
        min_remaining_time = float('inf')
        active_session_times = []
        
        for session_id in self.active_sessions:
            session = self.sessions.get(session_id)
            if session and session.last_activity:
                if session.max_session_time:
                    # Session has time limit (queue exists)
                    elapsed = current_time - session.last_activity
                    remaining = session.max_session_time - elapsed
                    remaining = max(0, remaining)  # Don't go negative
                else:
                    # No time limit, estimate based on average usage
                    elapsed = current_time - session.last_activity
                    # Assume sessions without time limits will run for average of 45 seconds more
                    remaining = max(45 - elapsed, 10)  # Minimum 10 seconds
                
                active_session_times.append(remaining)
                min_remaining_time = min(min_remaining_time, remaining)
        
        # If no active sessions found, use default
        if not active_session_times:
            min_remaining_time = 30.0
        
        # Calculate estimated wait time based on position
        num_workers = len(self.workers)
        if num_workers == 0:
            return 999  # No workers available
        
        if position_in_queue <= num_workers:
            # User will get a worker as soon as current sessions end
            return min_remaining_time
        else:
            # User needs to wait for multiple session cycles
            cycles_to_wait = (position_in_queue - 1) // num_workers
            remaining_in_current_cycle = (position_in_queue - 1) % num_workers + 1
            
            # Time for complete cycles (use average session time)
            avg_session_time = self.MAX_SESSION_TIME_WITH_QUEUE if len(self.session_queue) > 0 else 45.0
            full_cycles_time = cycles_to_wait * avg_session_time
            
            # Time for current partial cycle
            if remaining_in_current_cycle <= len(active_session_times):
                # Sort session times to get when the Nth worker will be free
                sorted_times = sorted(active_session_times)
                current_cycle_time = sorted_times[remaining_in_current_cycle - 1]
            else:
                current_cycle_time = min_remaining_time
            
            return full_cycles_time + current_cycle_time

    async def handle_user_activity(self, session_id: str):
        """Update user activity timestamp"""
        session = self.sessions.get(session_id)
        if session:
            session.last_activity = time.time()
            session.interaction_count += 1
            if not session.user_has_interacted:
                session.user_has_interacted = True
                logger.info(f"User started interacting in session {session_id}")

    async def _forward_to_worker(self, worker: WorkerInfo, session_id: str, data: dict):
        """Forward input to worker asynchronously"""
        try:
            async with aiohttp.ClientSession() as client_session:
                async with client_session.post(
                    f"{worker.endpoint}/process_input",
                    json={
                        "session_id": session_id,
                        "data": data
                    }
                ) as response:
                    if response.status != 200:
                        logger.error(f"Worker returned status {response.status}")
                        # Optionally handle worker errors here
        except Exception as e:
            logger.error(f"Error forwarding to worker {worker.worker_id}: {e}")

# Global session manager
session_manager = SessionManager()

app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")

# Global connection counter like in main.py
connection_counter = 0

@app.get("/")
async def get():
    return HTMLResponse(open("static/index.html").read())

@app.post("/register_worker")
async def register_worker(worker_info: dict):
    """Endpoint for workers to register themselves"""
    await session_manager.register_worker(
        worker_info["worker_id"],
        worker_info["gpu_id"], 
        worker_info["endpoint"]
    )
    return {"status": "registered"}

@app.post("/worker_ping")
async def worker_ping(worker_info: dict):
    """Endpoint for workers to ping their availability"""
    worker_id = worker_info["worker_id"]
    if worker_id in session_manager.workers:
        session_manager.workers[worker_id].last_ping = time.time()
        session_manager.workers[worker_id].is_available = worker_info.get("is_available", True)
    return {"status": "ok"}

@app.post("/worker_result")
async def worker_result(result_data: dict):
    """Endpoint for workers to send back processing results"""
    session_id = result_data.get("session_id")
    worker_id = result_data.get("worker_id")
    result = result_data.get("result")
    
    if not session_id or not result:
        raise HTTPException(status_code=400, detail="Missing session_id or result")
    
    # Find the session and send result to the WebSocket
    session = session_manager.sessions.get(session_id)
    if session and session.status == SessionStatus.ACTIVE:
        try:
            await session.websocket.send_json(result)
            logger.info(f"Sent result to session {session_id}")
        except Exception as e:
            logger.error(f"Failed to send result to session {session_id}: {e}")
    else:
        logger.warning(f"Could not find active session {session_id} for result")
    
    return {"status": "ok"}

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    global connection_counter
    await websocket.accept()
    
    # Extract client IP address
    client_ip = "unknown"
    if websocket.client and hasattr(websocket.client, 'host'):
        client_ip = websocket.client.host
    elif hasattr(websocket, 'headers'):
        # Try to get real IP from headers (for proxy setups)
        client_ip = websocket.headers.get('x-forwarded-for', 
                     websocket.headers.get('x-real-ip', 
                     websocket.headers.get('cf-connecting-ip', 'unknown')))
        if ',' in client_ip:
            client_ip = client_ip.split(',')[0].strip()
    
    # Create session with connection counter like in main.py
    connection_counter += 1
    session_id = str(uuid.uuid4())
    client_id = f"{int(time.time())}_{connection_counter}"
    
    session = UserSession(
        session_id=session_id,
        client_id=client_id,
        websocket=websocket,
        created_at=time.time(),
        status=SessionStatus.QUEUED,
        ip_address=client_ip
    )
    
    logger.info(f"New WebSocket connection: {client_id} from {client_ip}")
    
    # Log new connection analytics
    analytics.log_new_connection(client_id, client_ip)
    
    try:
        # Add to queue
        await session_manager.add_session_to_queue(session)
        
        # Try to process queue immediately
        await session_manager.process_queue()
        
        # Send initial queue status
        if session.status == SessionStatus.QUEUED:
            await session_manager.update_queue_info()
        
        # Main message loop
        while True:
            try:
                data = await websocket.receive_json()
                
                # Update activity only for real user inputs, not auto inputs
                if not data.get("is_auto_input", False):
                    await session_manager.handle_user_activity(session_id)
                
                # Handle different message types
                if data.get("type") == "heartbeat":
                    await websocket.send_json({"type": "heartbeat_response"})
                    continue
                
                # If session is active, forward to worker
                if session.status == SessionStatus.ACTIVE and session.worker_id:
                    worker = session_manager.workers.get(session.worker_id)
                    if worker:
                        try:
                            # Forward message to worker (don't wait for response for regular inputs)
                            # The worker will send results back asynchronously via /worker_result
                            asyncio.create_task(session_manager._forward_to_worker(worker, session_id, data))
                        except Exception as e:
                            logger.error(f"Error forwarding to worker: {e}")
                
                # Handle control messages (these need synchronous responses)
                elif data.get("type") in ["reset", "update_sampling_steps", "update_use_rnn", "get_settings"]:
                    if session.status == SessionStatus.ACTIVE and session.worker_id:
                        worker = session_manager.workers.get(session.worker_id)
                        if worker:
                            try:
                                async with aiohttp.ClientSession() as client_session:
                                    async with client_session.post(
                                        f"{worker.endpoint}/process_input",
                                        json={
                                            "session_id": session_id,
                                            "data": data
                                        }
                                    ) as response:
                                        if response.status == 200:
                                            result = await response.json()
                                            await websocket.send_json(result)
                                        else:
                                            logger.error(f"Worker returned status {response.status}")
                            except Exception as e:
                                logger.error(f"Error forwarding control message: {e}")
                    else:
                        # Send appropriate response for queued users
                        await websocket.send_json({
                            "type": "error",
                            "message": "Session not active yet. Please wait in queue."
                        })
                        
            except asyncio.TimeoutError:
                logger.info("WebSocket connection timed out")
                break
            except WebSocketDisconnect:
                logger.info(f"WebSocket disconnected: {client_id}")
                break
                
    except Exception as e:
        logger.error(f"Error in WebSocket connection {client_id}: {e}")
        import traceback
        traceback.print_exc()
    
    finally:
        # Clean up session
        if session_id in session_manager.sessions:
            await session_manager.end_session(session_id, SessionStatus.COMPLETED)
            del session_manager.sessions[session_id]
        
        logger.info(f"WebSocket connection closed: {client_id}")

# Background task to periodically update queue info
async def periodic_queue_update():
    while True:
        try:
            await session_manager.update_queue_info()
            await asyncio.sleep(2)  # Update every 2 seconds for responsive experience
        except Exception as e:
            logger.error(f"Error in periodic queue update: {e}")

# Background task to periodically log analytics summary
async def periodic_analytics_summary():
    while True:
        try:
            await asyncio.sleep(300)  # Log summary every 5 minutes
            analytics.log_periodic_summary()
        except Exception as e:
            logger.error(f"Error in periodic analytics summary: {e}")

# Background task to check worker health
async def periodic_worker_health_check():
    while True:
        try:
            await asyncio.sleep(60)  # Check every minute
            current_time = time.time()
            disconnected_workers = []
            
            for worker_id, worker in list(session_manager.workers.items()):
                if current_time - worker.last_ping > 30:  # 30 second timeout
                    disconnected_workers.append((worker_id, worker.gpu_id))
            
            for worker_id, gpu_id in disconnected_workers:
                analytics.log_worker_disconnected(worker_id, gpu_id)
                del session_manager.workers[worker_id]
                logger.warning(f"Removed disconnected worker {worker_id} (GPU {gpu_id})")
                
            if disconnected_workers:
                # Log updated GPU status
                total_gpus = len(session_manager.workers)
                active_gpus = len([w for w in session_manager.workers.values() if not w.is_available])
                available_gpus = total_gpus - active_gpus
                analytics.log_gpu_status(total_gpus, active_gpus, available_gpus)
                
        except Exception as e:
            logger.error(f"Error in periodic worker health check: {e}")

@app.on_event("startup")
async def startup_event():
    # Start background tasks
    asyncio.create_task(periodic_queue_update())
    asyncio.create_task(periodic_analytics_summary())
    asyncio.create_task(periodic_worker_health_check())
    
    # Log initial system status
    analytics._write_log("🚀 System initialized and ready to accept connections")
    analytics._write_log("   Waiting for GPU workers to register...")

@app.on_event("shutdown")
async def shutdown_event():
    # Log final system summary
    analytics._write_log("")
    analytics._write_log("🛑 System shutting down...")
    analytics.log_periodic_summary()
    analytics._write_log("System shutdown complete.")

if __name__ == "__main__":
    import uvicorn
    import argparse
    
    parser = argparse.ArgumentParser(description="Dispatcher for Neural OS")
    parser.add_argument("--port", type=int, default=8000, help="Port to run the dispatcher on")
    args = parser.parse_args()
    
    uvicorn.run(app, host="0.0.0.0", port=args.port)