3cloud-backend/worker/workerClient.py

#Worker.py
import functools
from logger import logger
from worker_tasks.error_types import SchedulingError
from worker_tasks.ovs_bridge_scanner import OVSBridgeScannerTask
from worker_tasks.pci_device_scanner import PCIDeviceScannerTask

from vncsession import VNCSession
from worker_tasks.container import ContainerTask
from worker_tasks.dns import DnsTask
from worker_tasks.file_presence import FilePresenceTask
from worker_tasks.ping import PingTask
from worker_tasks.ping import PingTask
from worker_tasks.report import ReportTask
from worker_tasks.ovs_sdn import OVS_SDN
from settings import settings # Import global settings
import json
import asyncio
import socketio
import threading
import time
class WorkerClient:
    def __init__(self, event_queue=None, docker_monitor=None, libvirt_monitor=None):
        """Initializes the WorkerClient using global settings."""
        try:
            self.worker_id = settings.get_value("WORKER_ID")
            self.worker_secret = settings.get_value("WORKER_SECRET")
            self.server_url = settings.get_value("WEBSOCKET_SERVER_URL")
            self.socketio_path = settings.get_value("WEBSOCKET_SERVER_PATH", "/ws").lstrip("/")
            self.vnc_host = settings.get_value("VNC_HOST", "127.0.0.1")
            self.debug_socketio = settings.get_value("DEBUG_SOCKETIO", False)
            self.enable_websocket_ping_debug = settings.get_value("ENABLE_WEBSOCKET_PING_DEBUG", False)
            # Store monitors for starting after join
            self.docker_monitor = docker_monitor
            self.libvirt_monitor = libvirt_monitor
            # Libvirt config is now also potentially in settings, or keep defaults here
            self.libvirt_config = {
                 "default_volume_path": settings.get_value("DEFAULT_VOLUME_PATH", "/tmp")
            }
        except KeyError as e:
            logger.error(f"Failed to initialize WorkerClient: Missing required setting: {e}")
            raise ValueError(f"Missing required setting for WorkerClient: {e}") from e
        
        self.event_queue = event_queue
        self.running = True
        self.join_attempts = 0
        self.max_join_attempts = 5
        self.monitor_threads = []  # Changed from monitor_processes to monitor_threads
        self.monitors_started = False
        self.first_join_completed = False  # Track if this is the first join
        self.reconnection_in_progress = False  # Track if we're in a reconnection cycle
        

        if self.debug_socketio:
            self.sio = socketio.AsyncClient(
                logger=logger,
                engineio_logger=logger,
                reconnection=False  # Disable automatic reconnection - we handle it manually
            )
        else:
            self.sio = socketio.AsyncClient(
                logger=False,
                engineio_logger=False,
                reconnection=False  # Disable automatic reconnection - we handle it manually
            )
        self.joined_server = False

        # Bind events
        self.sio.on("connect", self.on_connect)
        self.sio.on("connect_error", self.on_connect_error)
        self.sio.on("disconnect", self.on_disconnect)
        self.sio.on("task", self.handle_task)
        self.sio.on("join_accept", self.on_join_accept)
        self.sio.on("join_reject", self.on_join_reject)
        self.sio.on("message", self.on_message)
        # self.sio.on("*", self.debug_all_events)
        
        
        self.sio.on("container-log",self.logs_retrieve)
        
        
        self.log_stream_threads = {}
        self.sio.on("start_container_log_stream", self.logs_stream_start)
        self.sio.on("stop_container_log_stream", self.logs_stream_stop)

        # VM console log streaming (mirrors the container log stream pattern)
        self.vm_log_stream_threads = {}
        self.sio.on("start_vm_log_stream", self.vm_logs_stream_start)
        self.sio.on("stop_vm_log_stream", self.vm_logs_stream_stop)
                
        self.terminal_sessions = {}
        self.sio.on("start_terminal", self.start_terminal_session)
        self.sio.on("terminal_input", self.terminal_input)
        self.sio.on("stop_terminal", self.stop_terminal_session)



        self.vnc_sessions = {}
        self.sio.on("start_vnc_stream_on_worker", self.start_vnc_stream)
        self.sio.on("stop_vnc_stream_on_worker", self.stop_vnc_stream)
        self.sio.on("vnc_frame_from_novnc", self.vnc_frame_from_novnc)  # Add this line

        self.sio.on("worker_ping", self.handle_worker_ping)

        # OVS Bridge Scanner
        try:
            self.ovs_scanner = OVSBridgeScannerTask(logger)
        except ValueError as e:
            logger.error(f"Failed to initialize OVS scanner: {e}")
            self.ovs_scanner = None

        # PCI Device Scanner
        try:
            self.pci_scanner = PCIDeviceScannerTask(logger)
        except ValueError as e:
            logger.error(f"Failed to initialize PCI scanner: {e}")
            self.pci_scanner = None
        
        # DNS Task Handler
        try:
            import docker
            docker_client = docker.from_env()
            self.dns_task = DnsTask(logger, docker_client, settings.get_value("WORKER_ID"))
        except Exception as e:
            logger.error(f"Failed to initialize DNS task handler: {e}")
            self.dns_task = None

    def start_monitors(self):
        """Start monitor threads if monitors are available."""
        if self.docker_monitor:
            try:
                docker_thread = threading.Thread(
                    target=self.docker_monitor.run,
                    daemon=True,
                    name="DockerMonitor"
                )
                docker_thread.start()
                self.monitor_threads.append(docker_thread)
                logger.info("Started Docker monitor thread")
            except Exception as e:
                logger.error(f"Failed to start Docker monitor thread: {e}")
        if self.libvirt_monitor:
            try:
                libvirt_thread = threading.Thread(
                    target=self.libvirt_monitor.run,
                    daemon=True,
                    name="LibvirtMonitor"
                )
                libvirt_thread.start()
                self.monitor_threads.append(libvirt_thread)
                logger.info("Started Libvirt monitor thread")
            except Exception as e:
                logger.error(f"Failed to start Libvirt monitor thread: {e}")

    def stop_monitors(self):
        """Stop all monitor threads and monitors."""
        # Set stop flags first to signal threads to exit
        if self.docker_monitor:
            logger.info("Stopping Docker Monitor")
            try:
                self.docker_monitor.stop()
            except Exception as e:
                logger.error(f"Error stopping Docker monitor: {e}")
        if self.libvirt_monitor:
            logger.info("Stopping Libvirt Monitor")
            try:
                self.libvirt_monitor.stop()
            except Exception as e:
                logger.error(f"Error stopping Libvirt monitor: {e}")
        
        # Wait for threads to finish gracefully
        for thread in self.monitor_threads:
            if thread.is_alive():
                logger.info(f"Waiting for thread {thread.name} to finish")
                thread.join(timeout=5)
                if thread.is_alive():
                    logger.warning(f"Thread {thread.name} did not finish within timeout")
        
        self.monitor_threads.clear()

    async def handle_worker_ping(self, data):
        """
        Respond to a liveness ping from the WebSocket server.
        Emits a 'worker_pong' event with the same ping_id and this worker's ID.
        """
        try:
            ping_id = data["ping_id"]
            worker_id = self.worker_id
            
            # Diagnostic: Log transport state during ping only if debug flag is enabled
            if self.enable_websocket_ping_debug:
                try:
                    transport = self.sio.transport()
                    logger.debug(f"[PING] Received ping {ping_id[:8]}... | Transport: {transport} | connected={self.sio.connected}")
                except:
                    logger.debug(f"[PING] Received ping {ping_id[:8]}... | connected={self.sio.connected}")
            
            await self.sio.emit("worker_pong", {"ping_id": ping_id,"worker_id": worker_id})
        except Exception as e:
            logger.error(f"Error handling worker_ping: {e}")

    async def on_connect(self):
        # Enhanced diagnostic logging
        try:
            current_transport = self.sio.transport()
            logger.info(f"✓ CONNECTED - Transport: {current_transport} | sio.connected={self.sio.connected}")
        except Exception as e:
            logger.info(f"✓ CONNECTED (transport unknown: {e}) | sio.connected={self.sio.connected}")
        await self.send_join_request()

    async def on_connect_error(self, data):
        logger.error(f"Connection failed: {data}")
        self.joined_server = False

    async def on_disconnect(self, *args):
        # Enhanced disconnect logging
        logger.warning(f"✗ DISCONNECTED | joined={self.joined_server} | monitors={self.monitors_started} | Args: {args}")
        self.joined_server = False
        self.monitors_started = False
        self.reconnection_in_progress = True
        self.stop_monitors()
        # Start reconnection process
        asyncio.create_task(self.reconnect())

    async def on_message(self, data):
        logger.info(f"Message received: {data}")

    async def on_join_accept(self, data):
        # Enhanced join logging
        try:
            transport = self.sio.transport()
            logger.info(f"✓ JOIN ACCEPTED | Transport: {transport} | first_join={self.first_join_completed}")
        except:
            logger.info(f"✓ JOIN ACCEPTED | first_join={self.first_join_completed}")
        
        self.joined_server = True

        # Stop existing monitors if already started
        if self.monitors_started:
            self.stop_monitors()

        # Start new monitor processes
        self.start_monitors()
        self.monitors_started = True
        
        # Start OVS bridge scanner if available
        if self.ovs_scanner:
            # Initial scan
            self.ovs_scanner.execute()

        # Start PCI device scanner if available
        if self.pci_scanner:
            # Initial scan
            self.pci_scanner.execute()

        self.first_join_completed = True

    async def on_join_reject(self, data):
        logger.error("Join rejected.")
        self.joined_server = False

        if self.join_attempts < self.max_join_attempts:
            delay = 2 ** (self.join_attempts - 1)  # Exponential backoff
            logger.info(f"Retrying join in {delay} seconds (attempt {self.join_attempts}/{self.max_join_attempts})")
            asyncio.create_task(self.retry_join_after_delay(delay))
        else:
            logger.error(f"Max join attempts ({self.max_join_attempts}) reached. Exiting.")
            exit()

    async def retry_join_after_delay(self, delay):
        await asyncio.sleep(delay)
        if not self.sio.connected:
            try:
                await self.sio.connect(self.server_url, socketio_path=self.socketio_path)
            except Exception as e:
                logger.error(f"Failed to reconnect: {e}")
                return
        await self.send_join_request()

    async def reconnect(self):
        """Handle reconnection to the server after disconnection."""
        logger.warning(f"[RECONNECT] Starting reconnection process | sio.connected={self.sio.connected}")
        reconnect_delay = 1  # Start with 1 second delay
        max_reconnect_delay = 30  # Maximum delay of 30 seconds
        
        while self.reconnection_in_progress:
            try:
                logger.info(f"[RECONNECT] Attempt #{int(reconnect_delay)} | delay: {reconnect_delay}s | sio.connected={self.sio.connected}")
                await asyncio.sleep(reconnect_delay)
                
                # Try to connect
                if not self.sio.connected:
                    logger.info(f"[RECONNECT] Calling sio.connect({self.server_url})")
                    await self.sio.connect(self.server_url, socketio_path=self.socketio_path)
                else:
                    logger.warning(f"[RECONNECT] Already connected! sio.connected={self.sio.connected}")
                
                # If we get here, we've successfully reconnected
                logger.info(f"[RECONNECT] ✓ Success | sio.connected={self.sio.connected}")
                self.reconnection_in_progress = False
                # The on_connect handler will send the join request
                break
            except Exception as e:
                logger.error(f"[RECONNECT] ✗ Failed: {e}")
                # Increase delay exponentially, up to max_reconnect_delay
                reconnect_delay = min(reconnect_delay * 2, max_reconnect_delay)
                continue
    
    
    async def send_join_request(self):
        """Notify the server about this worker (Initial join)."""
        logger.info("Sending join request")
        if not self.joined_server:
            self.join_attempts += 1
            await self.sio.emit("join_request", {"worker_id": self.worker_id, "worker_secret": self.worker_secret})
            logger.info(f"Worker {self.worker_id} asked to join the server (attempt {self.join_attempts}).")
    
    async def handle_task(self, data):
        """Handle incoming tasks from the server."""
        if not self.joined_server:
            logger.error("Not joined to server. ignoring task")
            return

        task_type = data["type"]
        task_id = data["task_id"]
        job_details = json.loads(data["job_details"])
        task_worker_id = data["worker_id"]

        if self.docker_monitor and task_type.startswith("container-"):
            job_details["docker_monitor"] = self.docker_monitor
            logger.debug("Added docker_monitor to job_details for container task")

        logger.info(f"Received task {task_id} of type '{task_type}' with job_details: {job_details}")

        loop = asyncio.get_event_loop()
        result = None

        try:
            # ----------------------- Task dispatch ------------------------ #
            if task_type == "report":
                result = await loop.run_in_executor(None, functools.partial(ReportTask("", logger).Execute))
            elif task_type == "ping":
                result = await loop.run_in_executor(None, functools.partial(PingTask(job_details, logger).Execute))
            elif task_type == "sdn-update":
                result = await loop.run_in_executor(None, functools.partial(OVS_SDN(job_details, logger).Execute))
            elif task_type == "file_presence":
                result = await loop.run_in_executor(None, functools.partial(FilePresenceTask(job_details, logger).Execute))
            elif task_type == "pod-update":
                result = await loop.run_in_executor(None, functools.partial(ContainerTask(logger,self.docker_monitor).handle_pod_update_with_reconciliation, job_details))
            elif task_type == "virtual-machine-create":
                from worker_tasks.libvirt import LibvirtVirtualMachineTask
                result = await loop.run_in_executor(None, functools.partial(LibvirtVirtualMachineTask(job_details, logger).execute))
            elif task_type == "virtual-machine-delete":
                from worker_tasks.libvirt import LibvirtVirtualMachineTask
                result = await loop.run_in_executor(None, functools.partial(LibvirtVirtualMachineTask(job_details, logger).execute))
            elif task_type == "container-reconcile":
                result = await loop.run_in_executor(None, functools.partial(ContainerTask(logger,self.docker_monitor).reconcile_all_containers))
            elif task_type == "reconcile_and_delete":
                result = await loop.run_in_executor(None, functools.partial(ContainerTask(logger,self.docker_monitor).handle_reconcile_and_delete, job_details))
            elif task_type == "dns-update":
                if self.dns_task:
                    result = await loop.run_in_executor(None, functools.partial(self.dns_task.handle_dns_update, job_details))
                else:
                    result = {"success": False, "error": "DNS task handler not initialized"}
            else:
                raise ValueError(f"Unknown task type: {task_type}")

            # --------------------- ACK back to server --------------------- #
            await self.send_task_result(task_id, result, task_worker_id)

            # ---------------- Emit launch-failure events ------------------ #
            if task_type == "pod-update" and result and result.get("launch_failures"):
                for fail in result["launch_failures"]:
                    failed_id  = fail["id"]
                    error_info = fail["error"]

                    event_data=({
                        "source": "docker",
                        "event_type": "launch_failed",
                        "container_id": failed_id,
                        "system_container_id": failed_id,
                        "container_name": failed_id,
                        "timestamp": time.time(),
                        "error": error_info
                    })

                    # Put event in the queue for the worker client
                    self.event_queue.put(event_data)
                    logger.debug(f"Added Docker event to queue: launch_failed for container {failed_id}.")


        except SchedulingError as e:
            logger.error(f"Error processing task {task_id} [{e.error_type.value}]: {e}")
            await self.send_task_result(task_id, e.to_result(), task_worker_id)
        except Exception as e:
            logger.error(f"Error processing task {task_id}: {e}")
            await self.send_task_result(task_id, {"success": False, "response": str(e), "error_type": "UNKNOWN"}, task_worker_id)

    async def send_task_result(self, task_id, result, worker_id):
        """Send task result back to the server, waiting for reconnection if needed."""
        for attempt in range(10):
            if self.sio.connected:
                await self.sio.emit("ack", {"task_id": task_id, "worker_id": worker_id, "result": result})
                logger.debug(f"Sent result for task {task_id}: {result}")
                return
            logger.warning(f"[RESULT] Socket not connected, waiting to send task {task_id} result (attempt {attempt + 1}/10)")
            await asyncio.sleep(2 ** attempt if attempt < 4 else 16)
        logger.error(f"[RESULT] Failed to send result for task {task_id} after retries — socket never reconnected")

    async def send_event(self, event_data):
        """Send monitoring event to the server, handling libvirt and docker events separately."""
        if not self.joined_server:
            return

        source = event_data.get('source', 'unknown')
        event_type = event_data.get('event_type', 'event')

        # Construct the base event payload
        payload = {
            "worker_id": self.worker_id,
            "type": f"{source}_{event_type}",
            "details": event_data
        }

        # Handle libvirt events
        if source == "libvirt":
            payload["details"] = {
                "system_vm_id": event_data.get("system_vm_id"),
                "libvirt_vm_id": event_data.get("libvirt_vm_id"),
                "timestamp": event_data.get("timestamp"),
            }

            # Emit the event to the server
            await self.sio.emit("libvirt_event", payload)
            logger.debug(f"Sent libvirt event: {event_type}\n{payload}")


        # Handle docker events
        elif source == "docker":
            payload["details"] = {
                "container_id": event_data.get("container_id"),
                "system_container_id": event_data.get("system_container_id"),
                "container_name": event_data.get("container_name"),
                "timestamp": event_data.get("timestamp"),
                "status": event_data.get("details", {}).get("status"),
                "image": event_data.get("details", {}).get("from"),
                "action": event_data.get("details", {}).get("Action"),
                **({"error": event_data.get("error")} if event_data.get("error") else {})
            }
            
            # Include pull_progress data for pull_progress events
            if event_type == "pull_progress" and "pull_progress" in event_data:
                logger.debug("Sengin pull progreesss info")
                payload["details"]["pull_progress"] = event_data["pull_progress"]

            # Emit the event to the server
            await self.sio.emit("docker_event", payload)
            logger.debug(f"Sent {source} event: {event_type}\n{payload}")

    async def process_event_queue(self):
        """Process events from the queue and send them to the server."""
        if not self.event_queue:
            return
            
        # Dictionary to store the most recent event for each container and status combination
        pending_events = {}
        
        # Timer to ensure events are sent even if no new events arrive
        last_send_time = time.time()
        send_interval = 1.0  # Send events at least every second
        
        while self.running:
            try:
                # Non-blocking check for events
                events_processed = 0
                while not self.event_queue.empty() and events_processed < 100:  # Process up to 100 events at once
                    event = self.event_queue.get(block=False)
                    events_processed += 1
                    
                    # Create a key for deduplication based on container and event type
                    container_id = event.get('system_container_id') or event.get('container_id')
                    event_type = event.get('event_type')
                    source = event.get('source')
                    
                    if container_id and event_type and source:
                        # Create a unique key for this container and event type combination
                        key = f"{source}:{container_id}:{event_type}"
                        
                        # Store the most recent event for this key
                        pending_events[key] = event
                        logger.debug(f"Added/updated pending event: {key}")
                    else:
                        # If we can't create a key, send the event immediately
                        logger.debug(f"Processing event immediately (no key): {event}")
                        await self.send_event(event)
                
                # Send pending events if we have any or if enough time has passed
                current_time = time.time()
                if pending_events and (len(pending_events) >= 10 or (current_time - last_send_time) >= send_interval):
                    logger.debug(f"Sending {len(pending_events)} pending events")
                    for key, event in list(pending_events.items()):
                        logger.debug(f"Processing event from queue: {event}")
                        await self.send_event(event)
                        del pending_events[key]
                    last_send_time = current_time
                
                # Small delay to prevent CPU thrashing
                await asyncio.sleep(0.01)
            except Exception as e:
                logger.error(f"Error in event processing: {e}")
    
    async def start(self):
        """Worker connects to the API server and processes tasks."""
        try:
            logger.info(f"[START] Worker {self.worker_id} connecting to {self.server_url}...")
            await self.sio.connect(self.server_url, socketio_path=self.socketio_path)
            
            logger.info(f"[START] Initial connection established | sio.connected={self.sio.connected}")
                       
            # Start the event queue processor if we have an event queue
            if self.event_queue:
                event_processor = asyncio.create_task(self.process_event_queue())
                await self.sio.wait()
                event_processor.cancel()

            else:
                await self.sio.wait()

        except Exception as e:
            logger.error(f"[START] Error: {e}")
            await self.sio.disconnect()

    def stop(self):
        """Stop the worker client."""
        logger.info("Stopping worker client...")
        self.running = False
        self.monitors_started = False
        try:
            self.stop_monitors()
        except Exception as e:
            logger.error(f"Error stopping monitors: {e}")
        logger.info("Worker client stopped.")



    async def logs_retrieve(self,data):
        logger.info(f"Handling container-log {data}")
        
        container_name = data.get("container_name")
        lines = int(data.get("lines", 100))
        request_id = data.get("request_id")
        worker_id = data.get("worker_id")

        if worker_id != self.worker_id:
            logger.debug(f"Ignoring container-log request for worker {worker_id}")
            return

        logger.info(f"Handling container-log for {container_name}, lines={lines}")

        log_data=ContainerTask(logger,self.docker_monitor).handle_container_log_request(data)
        
        log_data["worker_id"] = self.worker_id
        log_data["request_id"] = request_id

        await self.sio.emit("container_log_response", log_data)
        logger.debug(f"Sent container-log-response for {container_name}")

    async def logs_stream_start(self, data):
        request_id = data["request_id"]
        container_name = data["container_name"]
        worker_id = data.get("worker_id")
        logger.debug(f"Starting stream with {data}")

        if worker_id != self.worker_id:
            logger.debug(f"Ignoring stream request for worker {worker_id}")
            return

        if request_id in self.log_stream_threads:
            logger.warning(f"Log stream already running for request {request_id}")
            return

        # Set cancellation flag
        self.log_stream_threads[request_id] = True

        container_task = ContainerTask(logger,self.docker_monitor)

        def cancel_flag():
            return not self.log_stream_threads.get(request_id, False)

        def on_log(log_line):
            asyncio.run(self.sio.emit("container-log-stream", {
                "request_id": request_id,
                "worker_id": self.worker_id,
                "logs": log_line
            }))

        def on_error(error_msg):
            asyncio.run(self.sio.emit("container-log-stream", {
                "request_id": request_id,
                "worker_id": self.worker_id,
                "logs": f"Stream error: {error_msg}"
            }))

        thread = threading.Thread(
            target=container_task.stream_logs,
            args=(container_name, on_log, on_error, cancel_flag),
            daemon=True
        )
        thread.start()

        logger.info(f"Started log stream thread for request {request_id}")

    async def logs_stream_stop(self, data):
        request_id = data.get("request_id")
        if request_id in self.log_stream_threads:
            self.log_stream_threads[request_id] = False
            logger.info(f"Stopped log stream for request {request_id}")
        else:
            logger.warning(f"No active log stream found for request {request_id}")


    async def vm_logs_stream_start(self, data):
        """Start streaming a VM's serial console log to the WebSocket server."""
        request_id = data.get("request_id")
        vm_id = data.get("vm_id")
        worker_id = data.get("worker_id")

        if worker_id != self.worker_id:
            logger.debug(f"Ignoring vm log stream request for worker {worker_id}")
            return

        if request_id in self.vm_log_stream_threads:
            logger.warning(f"VM log stream already running for request {request_id}")
            return

        logger.info(f"[{request_id}] Starting VM log stream for VM '{vm_id}'")
        self.vm_log_stream_threads[request_id] = True

        from worker_tasks.libvirt import LibvirtVirtualMachineTask

        task = LibvirtVirtualMachineTask.__new__(LibvirtVirtualMachineTask)
        task.logger = logger

        loop = asyncio.get_event_loop()

        def cancel_flag():
            return not self.vm_log_stream_threads.get(request_id, False)

        def on_log(line):
            asyncio.run_coroutine_threadsafe(
                self.sio.emit("vm-log-stream", {
                    "request_id": request_id,
                    "worker_id": self.worker_id,
                    "log": line
                }),
                loop
            )

        def on_error(msg):
            asyncio.run_coroutine_threadsafe(
                self.sio.emit("vm-log-stream", {
                    "request_id": request_id,
                    "worker_id": self.worker_id,
                    "log": f"[stream error] {msg}"
                }),
                loop
            )

        thread = threading.Thread(
            target=task.stream_vm_logs,
            args=(vm_id, on_log, on_error, cancel_flag),
            daemon=True
        )
        thread.start()
        logger.info(f"[{request_id}] VM log stream thread started for '{vm_id}'")

    async def vm_logs_stream_stop(self, data):
        request_id = data.get("request_id")
        if request_id in self.vm_log_stream_threads:
            self.vm_log_stream_threads[request_id] = False
            logger.info(f"[{request_id}] Stopped VM log stream")
        else:
            logger.warning(f"[{request_id}] No active VM log stream found")


    async def start_terminal_session(self, data):
        request_id = data["request_id"]
        container_name = data["container_name"]
        command = data.get("command", "/bin/sh")
        logger.debug(f"Starting terminal session with {data}")

        if request_id in self.terminal_sessions:
            logger.warning(f"Terminal session already exists for {request_id}")
            return

        container_task = ContainerTask(logger,self.docker_monitor)
        session = container_task.start_terminal(logger, container_name, command)

        def cancel_flag():
            return request_id not in self.terminal_sessions

        def on_output(output):
            asyncio.run(self.sio.emit("terminal_output", {
                "request_id": request_id,
                "data": output
            }))

        self.terminal_sessions[request_id] = session

        threading.Thread(
            target=session.stream_output,
            args=(logger, on_output, cancel_flag),
            daemon=True
        ).start()

        logger.info(f"Started terminal session for {request_id}")

    async def terminal_input(self, data):
        logger.debug(f"Terminal input {data}")
        request_id = data["request_id"]
        session = self.terminal_sessions.get(request_id)
        if session:
            session.write(data["data"])

    async def stop_terminal_session(self, data):
        logger.debug(f"Stopping terminal session with {data}")
        request_id = data["request_id"]
        session = self.terminal_sessions.pop(request_id, None)
        if session:
            session.close()
            logger.info(f"Closed terminal session for {request_id}")



    async def start_vnc_stream(self, data):
        logger.debug(f"start vnc {data}")
        request_id = data["vnc_request_id"]
        virtual_machine_id = data.get("virtual_machine_id")
        vnc_host = self.vnc_host

        if not virtual_machine_id:
            logger.error(f"[VNC] Rejecting session {request_id}: no virtual_machine_id provided")
            await self.sio.emit("vnc_error", {"vnc_request_id": request_id, "error": "no virtual_machine_id"})
            return

        if not self.libvirt_monitor or not self.libvirt_monitor.conn:
            logger.error(f"[VNC] Rejecting session {request_id}: libvirt not available")
            await self.sio.emit("vnc_error", {"vnc_request_id": request_id, "error": "libvirt unavailable"})
            return

        try:
            dom = self.libvirt_monitor.conn.lookupByName(virtual_machine_id)
            vnc_port = self.libvirt_monitor.get_vnc_port(dom)
        except Exception as e:
            logger.error(f"[VNC] Rejecting session {request_id}: VM {virtual_machine_id} not found in libvirt: {e}")
            await self.sio.emit("vnc_error", {"vnc_request_id": request_id, "error": "VM not found in libvirt"})
            return

        if not vnc_port or vnc_port <= 0:
            logger.error(f"[VNC] Rejecting session {request_id}: VM {virtual_machine_id} has no VNC port configured")
            await self.sio.emit("vnc_error", {"vnc_request_id": request_id, "error": "VM has no VNC port"})
            return

        logger.info(f"[VNC] Starting session {request_id} on port {vnc_port}")

        if request_id in self.vnc_sessions:
            logger.warning(f"[VNC] Session already running for {request_id}")
            return

        def cancel_flag():
            return request_id not in self.vnc_sessions

        def on_output(vnc_data):
            asyncio.run(self.sio.emit("vnc_frame_from_worker", {
                "vnc_request_id": request_id,
                "data": vnc_data
            }))

        session = VNCSession(logger, request_id, vnc_host, vnc_port, on_output, cancel_flag)
        self.vnc_sessions[request_id] = session

        thread = threading.Thread(target=session.start, daemon=True)
        thread.start()
        logger.info(f"[VNC] Session thread started for {request_id}")

    async def vnc_frame_from_novnc(self, data):
        request_id = data["vnc_request_id"]
        session = self.vnc_sessions.get(request_id)
        if session:
            session.write(data["data"])

    async def stop_vnc_stream(self, data):
        request_id = data["vnc_request_id"]
        session = self.vnc_sessions.pop(request_id, None)
        if session:
            session.close()
            logger.info(f"[VNC] Stopped session for {request_id}")