Fix ray module import error and add interactive consultation UI

- Make ray import optional in agent/ray_executor.py with graceful fallback
- Add try/except guards in agent/pipeline_engine.py for ray_executor usage
- Add interactive consultation tab to app.py with chat interface
- Users can now have back-and-forth conversation before running pipeline
- Consultation summary can be extracted and used in full pipeline

agent/pipeline_engine.py

@@ -194,10 +194,15 @@ class OvergrowthPipeline:
         self.rca_analyzer = RootCauseAnalyzer(self.incident_db)
         self.test_generator = RegressionTestGenerator()
         
-        # Ray distributed execution
-        from agent.ray_executor import RayExecutor
-        self.ray_executor = RayExecutor()
-        self.parallel_mode = False  # Enable for fleet operations
+        # Ray distributed execution (optional)
+        try:
+            from agent.ray_executor import RayExecutor
+            self.ray_executor = RayExecutor()
+            self.parallel_mode = False  # Enable for fleet operations
+        except (ImportError, NotImplementedError) as e:
+            logger.warning(f"Ray executor not available: {e}")
+            self.ray_executor = None
+            self.parallel_mode = False
     
     def stage0_preflight(self, model: NetworkModel) -> Dict[str, Any]:
         """
@@ -326,8 +331,8 @@ class OvergrowthPipeline:
         
         Uses parallel execution when parallel_mode=True and >10 devices
         """
-        # Use parallel execution for large fleets
-        if self.parallel_mode and len(model.devices) > 10:
+        # Use parallel execution for large fleets (only if ray_executor available)
+        if self.parallel_mode and self.ray_executor and len(model.devices) > 10:
             return self._parallel_config_generation(model)
         
         configs = {}
@@ -1150,6 +1155,10 @@ Be specific and practical. Use RFC1918 addressing. Consider scalability and secu
         Args:
             ray_address: Ray cluster address (None for local mode)
         """
+        if not self.ray_executor:
+            logger.error("Ray executor not available - cannot enable parallel mode")
+            return
+        
         self.parallel_mode = True
         if ray_address:
             self.ray_executor.ray_address = ray_address
@@ -1163,7 +1172,8 @@ Be specific and practical. Use RFC1918 addressing. Consider scalability and secu
     def disable_parallel_mode(self):
         """Disable parallel execution mode"""
         self.parallel_mode = False
-        self.ray_executor.shutdown()
+        if self.ray_executor:
+            self.ray_executor.shutdown()
         logger.info("Parallel mode disabled")
     
     def parallel_deploy_fleet(self, model: NetworkModel, 
@@ -1186,6 +1196,12 @@ Be specific and practical. Use RFC1918 addressing. Consider scalability and secu
             logger.warning("Parallel mode not enabled - enabling automatically")
             self.enable_parallel_mode()
         
+        if not self.ray_executor:
+            return {
+                'status': 'error',
+                'message': 'Ray executor not available - cannot perform parallel deployment'
+            }
+        
         # Generate configs for all devices
         configs = self._generate_configs_for_batfish(model)
         

agent/ray_executor.py

@@ -10,14 +10,22 @@ Enables parallel execution of:
 Works locally (single machine) or on Ray clusters with zero code changes.
 """
 
-import ray
-from ray.util.queue import Queue as RayQueue
-import time
 import logging
 from typing import List, Dict, Any, Optional, Callable, Tuple
 from dataclasses import dataclass, field
 from enum import Enum
 import asyncio
+
+# Optional ray import - gracefully degrade if not available
+try:
+    import ray
+    from ray.util.queue import Queue as RayQueue
+    RAY_AVAILABLE = True
+except ImportError:
+    RAY_AVAILABLE = False
+    logging.warning("Ray not installed - distributed execution disabled. Install with: pip install ray")
+
+import time
 from datetime import datetime
 
 logger = logging.getLogger(__name__)
@@ -98,513 +106,568 @@ class ExecutionProgress:
         }
 
 
-@ray.remote
-class ProgressTracker:
-    """Actor for tracking execution progress across distributed workers"""
-    
-    def __init__(self, total_devices: int):
-        self.progress = ExecutionProgress(total_devices=total_devices)
-        self.results: List[TaskResult] = []
-    
-    def update_status(self, device_id: str, status: TaskStatus):
-        """Update device status"""
-        if status == TaskStatus.RUNNING:
-            self.progress.running += 1
-            self.progress.pending -= 1
-        elif status == TaskStatus.SUCCESS:
-            self.progress.running -= 1
-            self.progress.completed += 1
-        elif status == TaskStatus.FAILED:
-            self.progress.running -= 1
-            self.progress.failed += 1
-    
-    def add_result(self, result: TaskResult):
-        """Add task result"""
-        self.results.append(result)
-    
-    def get_progress(self) -> Dict[str, Any]:
-        """Get current progress"""
-        return self.progress.to_dict()
-    
-    def get_results(self) -> List[TaskResult]:
-        """Get all results"""
-        return self.results
-    
-    def get_failed_devices(self) -> List[str]:
-        """Get list of failed device IDs"""
-        return [r.device_id for r in self.results if r.status == TaskStatus.FAILED]
+# Only define ray-dependent classes if ray is available
+if RAY_AVAILABLE:
+    @ray.remote
+    class ProgressTracker:
+        """Actor for tracking execution progress across distributed workers"""
+        
+        def __init__(self, total_devices: int):
+            self.progress = ExecutionProgress(total_devices=total_devices)
+            self.results: List[TaskResult] = []
+        
+        def update_status(self, device_id: str, status: TaskStatus):
+            """Update device status"""
+            if status == TaskStatus.RUNNING:
+                self.progress.running += 1
+                self.progress.pending -= 1
+            elif status == TaskStatus.SUCCESS:
+                self.progress.running -= 1
+                self.progress.completed += 1
+            elif status == TaskStatus.FAILED:
+                self.progress.running -= 1
+                self.progress.failed += 1
+        
+        def add_result(self, result: TaskResult):
+            """Add task result"""
+            self.results.append(result)
+        
+        def get_progress(self) -> Dict[str, Any]:
+            """Get current progress"""
+            return self.progress.to_dict()
+        
+        def get_results(self) -> List[TaskResult]:
+            """Get all results"""
+            return self.results
+        
+        def get_failed_devices(self) -> List[str]:
+            """Get list of failed device IDs"""
+            return [r.device_id for r in self.results if r.status == TaskStatus.FAILED]
 
 
-@ray.remote
-def generate_device_config(device_id: str, device_data: Dict[str, Any], 
-                          template_fn: Callable, progress_tracker: Any) -> TaskResult:
-    """
-    Ray remote function for parallel config generation.
-    
-    Args:
-        device_id: Unique device identifier
-        device_data: Device parameters (hostname, ip, role, etc.)
-        template_fn: Function to generate config from device data
-        progress_tracker: Progress tracking actor
-    
-    Returns:
-        TaskResult with generated config or error
-    """
-    start_time = time.time()
-    
-    try:
-        # Update status to running
-        ray.get(progress_tracker.update_status.remote(device_id, TaskStatus.RUNNING))
-        
-        # Generate config
-        config = template_fn(device_data)
-        
-        duration = time.time() - start_time
-        result = TaskResult(
-            device_id=device_id,
-            status=TaskStatus.SUCCESS,
-            result=config,
-            duration_seconds=duration
-        )
-        
-        # Update status to success
-        ray.get(progress_tracker.update_status.remote(device_id, TaskStatus.SUCCESS))
-        ray.get(progress_tracker.add_result.remote(result))
-        
-        return result
-        
-    except Exception as e:
-        duration = time.time() - start_time
-        result = TaskResult(
-            device_id=device_id,
-            status=TaskStatus.FAILED,
-            error=str(e),
-            duration_seconds=duration
-        )
-        
-        ray.get(progress_tracker.update_status.remote(device_id, TaskStatus.FAILED))
-        ray.get(progress_tracker.add_result.remote(result))
-        
-        return result
-
-
-@ray.remote
-def analyze_device_config(device_id: str, config: str, 
-                         batfish_client: Any, progress_tracker: Any) -> TaskResult:
-    """
-    Ray remote function for parallel Batfish analysis.
-    
-    Args:
-        device_id: Unique device identifier
-        config: Device configuration to analyze
-        batfish_client: Batfish client instance
-        progress_tracker: Progress tracking actor
-    
-    Returns:
-        TaskResult with analysis results or error
-    """
-    start_time = time.time()
-    
-    try:
-        ray.get(progress_tracker.update_status.remote(device_id, TaskStatus.RUNNING))
-        
-        # Run Batfish analysis
-        analysis = batfish_client.analyze_configs({device_id: config})
-        
-        duration = time.time() - start_time
-        result = TaskResult(
-            device_id=device_id,
-            status=TaskStatus.SUCCESS,
-            result=analysis,
-            duration_seconds=duration
-        )
-        
-        ray.get(progress_tracker.update_status.remote(device_id, TaskStatus.SUCCESS))
-        ray.get(progress_tracker.add_result.remote(result))
-        
-        return result
+    @ray.remote
+    def generate_device_config(device_id: str, device_data: Dict[str, Any], 
+                              template_fn: Callable, progress_tracker: Any) -> TaskResult:
+        """
+        Ray remote function for parallel config generation.
         
-    except Exception as e:
-        duration = time.time() - start_time
-        result = TaskResult(
-            device_id=device_id,
-            status=TaskStatus.FAILED,
-            error=str(e),
-            duration_seconds=duration
-        )
+        Args:
+            device_id: Unique device identifier
+            device_data: Device parameters (hostname, ip, role, etc.)
+            template_fn: Function to generate config from device data
+            progress_tracker: Progress tracking actor
         
-        ray.get(progress_tracker.update_status.remote(device_id, TaskStatus.FAILED))
-        ray.get(progress_tracker.add_result.remote(result))
+        Returns:
+            TaskResult with generated config or error
+        """
+        start_time = time.time()
         
-        return result
-
-
-@ray.remote
-def deploy_to_device(device_id: str, config: str, 
-                    gns3_client: Any, progress_tracker: Any,
-                    max_retries: int = 3) -> TaskResult:
-    """
-    Ray remote function for parallel device deployment.
-    
-    Args:
-        device_id: Unique device identifier
-        config: Configuration to deploy
-        gns3_client: GNS3 client instance
-        progress_tracker: Progress tracking actor
-        max_retries: Maximum retry attempts on failure
-    
-    Returns:
-        TaskResult with deployment status or error
-    """
-    start_time = time.time()
-    retry_count = 0
-    
-    while retry_count <= max_retries:
         try:
-            if retry_count > 0:
-                ray.get(progress_tracker.update_status.remote(device_id, TaskStatus.RETRYING))
-            else:
-                ray.get(progress_tracker.update_status.remote(device_id, TaskStatus.RUNNING))
+            # Update status to running
+            ray.get(progress_tracker.update_status.remote(device_id, TaskStatus.RUNNING))
             
-            # Deploy config to device
-            deployment_result = gns3_client.apply_config(device_id, config)
+            # Generate config
+            config = template_fn(device_data)
             
             duration = time.time() - start_time
             result = TaskResult(
                 device_id=device_id,
                 status=TaskStatus.SUCCESS,
-                result=deployment_result,
-                duration_seconds=duration,
-                retry_count=retry_count
+                result=config,
+                duration_seconds=duration
             )
             
+            # Update status to success
             ray.get(progress_tracker.update_status.remote(device_id, TaskStatus.SUCCESS))
             ray.get(progress_tracker.add_result.remote(result))
             
             return result
             
         except Exception as e:
-            retry_count += 1
-            if retry_count > max_retries:
-                duration = time.time() - start_time
-                result = TaskResult(
-                    device_id=device_id,
-                    status=TaskStatus.FAILED,
-                    error=f"Failed after {retry_count} retries: {str(e)}",
-                    duration_seconds=duration,
-                    retry_count=retry_count - 1
-                )
-                
-                ray.get(progress_tracker.update_status.remote(device_id, TaskStatus.FAILED))
-                ray.get(progress_tracker.add_result.remote(result))
-                
-                return result
+            duration = time.time() - start_time
+            result = TaskResult(
+                device_id=device_id,
+                status=TaskStatus.FAILED,
+                error=str(e),
+                duration_seconds=duration
+            )
+            
+            ray.get(progress_tracker.update_status.remote(device_id, TaskStatus.FAILED))
+            ray.get(progress_tracker.add_result.remote(result))
             
-            # Exponential backoff
-            time.sleep(2 ** retry_count)
+            return result
 
 
-class RayExecutor:
-    """
-    Distributed execution engine for hyperscale network automation.
-    
-    Provides parallel execution of config generation, analysis, and deployment
-    across thousands of devices using Ray's distributed computing framework.
-    """
-    
-    def __init__(self, ray_address: Optional[str] = None, num_cpus: Optional[int] = None):
+    @ray.remote
+    def analyze_device_config(device_id: str, config: str, 
+                             batfish_client: Any, progress_tracker: Any) -> TaskResult:
         """
-        Initialize Ray executor.
+        Ray remote function for parallel Batfish analysis.
         
         Args:
-            ray_address: Ray cluster address (None for local mode)
-            num_cpus: Number of CPUs to use (None for auto-detect)
+            device_id: Unique device identifier
+            config: Device configuration to analyze
+            batfish_client: Batfish client instance
+            progress_tracker: Progress tracking actor
+        
+        Returns:
+            TaskResult with analysis results or error
         """
-        self.ray_address = ray_address
-        self.num_cpus = num_cpus
-        self.initialized = False
-        self._progress_tracker = None
-    
-    def initialize(self):
-        """Initialize Ray runtime"""
-        if self.initialized:
-            return
+        start_time = time.time()
         
         try:
-            # Check if Ray is already initialized
-            if ray.is_initialized():
-                logger.info("Ray already initialized")
-            else:
-                # Initialize Ray
-                if self.ray_address:
-                    # Connect to existing cluster
-                    ray.init(address=self.ray_address)
-                    logger.info(f"Connected to Ray cluster at {self.ray_address}")
-                else:
-                    # Start local Ray instance
-                    init_kwargs = {}
-                    if self.num_cpus:
-                        init_kwargs['num_cpus'] = self.num_cpus
-                    
-                    ray.init(**init_kwargs)
-                    logger.info(f"Started local Ray instance with {ray.available_resources().get('CPU', 0)} CPUs")
+            ray.get(progress_tracker.update_status.remote(device_id, TaskStatus.RUNNING))
             
-            self.initialized = True
+            # Run Batfish analysis
+            analysis = batfish_client.analyze_configs({device_id: config})
             
-        except Exception as e:
-            logger.error(f"Failed to initialize Ray: {e}")
-            raise
-    
-    def shutdown(self):
-        """Shutdown Ray runtime"""
-        if self.initialized and ray.is_initialized():
-            ray.shutdown()
-            self.initialized = False
-            logger.info("Ray shutdown complete")
-    
-    def parallel_config_generation(self, devices: List[Dict[str, Any]], 
-                                   template_fn: Callable,
-                                   batch_size: int = 100) -> Tuple[List[TaskResult], ExecutionProgress]:
-        """
-        Generate configs for multiple devices in parallel.
-        
-        Args:
-            devices: List of device data dicts
-            template_fn: Function to generate config from device data
-            batch_size: Number of devices to process in each batch
-        
-        Returns:
-            Tuple of (results, final_progress)
-        """
-        self.initialize()
-        
-        # Create progress tracker
-        progress_tracker = ProgressTracker.remote(total_devices=len(devices))
-        
-        # Initialize pending count
-        ray.get(progress_tracker.update_status.remote("_init_", TaskStatus.PENDING))
-        for _ in range(len(devices) - 1):
-            ray.get(progress_tracker.update_status.remote("_init_", TaskStatus.PENDING))
-        
-        # Launch parallel tasks
-        futures = []
-        for device in devices:
-            future = generate_device_config.remote(
-                device_id=device['device_id'],
-                device_data=device,
-                template_fn=template_fn,
-                progress_tracker=progress_tracker
+            duration = time.time() - start_time
+            result = TaskResult(
+                device_id=device_id,
+                status=TaskStatus.SUCCESS,
+                result=analysis,
+                duration_seconds=duration
             )
-            futures.append(future)
             
-            # Process in batches to avoid overwhelming the cluster
-            if len(futures) >= batch_size:
-                ray.get(futures)
-                futures = []
-        
-        # Wait for remaining tasks
-        if futures:
-            ray.get(futures)
-        
-        # Get final results
-        results = ray.get(progress_tracker.get_results.remote())
-        final_progress = ray.get(progress_tracker.get_progress.remote())
-        
-        return results, final_progress
-    
-    def parallel_batfish_analysis(self, configs: Dict[str, str],
-                                  batfish_client: Any,
-                                  batch_size: int = 50) -> Tuple[List[TaskResult], ExecutionProgress]:
-        """
-        Analyze configs in parallel using Batfish.
-        
-        Args:
-            configs: Dict mapping device_id to config string
-            batfish_client: Batfish client instance
-            batch_size: Number of configs to analyze in each batch
-        
-        Returns:
-            Tuple of (results, final_progress)
-        """
-        self.initialize()
-        
-        progress_tracker = ProgressTracker.remote(total_devices=len(configs))
-        
-        # Initialize pending count
-        for _ in range(len(configs)):
-            ray.get(progress_tracker.update_status.remote("_init_", TaskStatus.PENDING))
-        
-        # Launch parallel analysis tasks
-        futures = []
-        for device_id, config in configs.items():
-            future = analyze_device_config.remote(
+            ray.get(progress_tracker.update_status.remote(device_id, TaskStatus.SUCCESS))
+            ray.get(progress_tracker.add_result.remote(result))
+            
+            return result
+            
+        except Exception as e:
+            duration = time.time() - start_time
+            result = TaskResult(
                 device_id=device_id,
-                config=config,
-                batfish_client=batfish_client,
-                progress_tracker=progress_tracker
+                status=TaskStatus.FAILED,
+                error=str(e),
+                duration_seconds=duration
             )
-            futures.append(future)
             
-            if len(futures) >= batch_size:
-                ray.get(futures)
-                futures = []
-        
-        if futures:
-            ray.get(futures)
-        
-        results = ray.get(progress_tracker.get_results.remote())
-        final_progress = ray.get(progress_tracker.get_progress.remote())
-        
-        return results, final_progress
-    
-    def parallel_deployment(self, deployments: Dict[str, str],
-                           gns3_client: Any,
-                           batch_size: int = 20,
-                           max_retries: int = 3) -> Tuple[List[TaskResult], ExecutionProgress]:
+            ray.get(progress_tracker.update_status.remote(device_id, TaskStatus.FAILED))
+            ray.get(progress_tracker.add_result.remote(result))
+            
+            return result
+
+
+    @ray.remote
+    def deploy_to_device(device_id: str, config: str, 
+                        gns3_client: Any, progress_tracker: Any,
+                        max_retries: int = 3) -> TaskResult:
         """
-        Deploy configs to multiple devices in parallel.
+        Ray remote function for parallel device deployment.
         
         Args:
-            deployments: Dict mapping device_id to config string
+            device_id: Unique device identifier
+            config: Configuration to deploy
             gns3_client: GNS3 client instance
-            batch_size: Number of devices to deploy to simultaneously
-            max_retries: Maximum retry attempts per device
+            progress_tracker: Progress tracking actor
+            max_retries: Maximum retry attempts on failure
         
         Returns:
-            Tuple of (results, final_progress)
+            TaskResult with deployment status or error
         """
-        self.initialize()
-        
-        progress_tracker = ProgressTracker.remote(total_devices=len(deployments))
-        
-        # Initialize pending count
-        for _ in range(len(deployments)):
-            ray.get(progress_tracker.update_status.remote("_init_", TaskStatus.PENDING))
+        start_time = time.time()
+        retry_count = 0
         
-        # Launch parallel deployment tasks
-        futures = []
-        for device_id, config in deployments.items():
-            future = deploy_to_device.remote(
-                device_id=device_id,
-                config=config,
-                gns3_client=gns3_client,
-                progress_tracker=progress_tracker,
-                max_retries=max_retries
-            )
-            futures.append(future)
-            
-            # Deploy in smaller batches to avoid overwhelming network
-            if len(futures) >= batch_size:
-                ray.get(futures)
-                futures = []
-        
-        if futures:
-            ray.get(futures)
-        
-        results = ray.get(progress_tracker.get_results.remote())
-        final_progress = ray.get(progress_tracker.get_progress.remote())
-        
-        return results, final_progress
-    
-    def get_cluster_resources(self) -> Dict[str, Any]:
-        """Get available cluster resources"""
-        self.initialize()
-        return {
-            'available': ray.available_resources(),
-            'total': ray.cluster_resources()
-        }
-    
-    def staggered_rollout(self, deployments: Dict[str, str],
-                         gns3_client: Any,
-                         stages: List[float] = [0.01, 0.1, 0.5, 1.0],
-                         validation_fn: Optional[Callable] = None) -> Tuple[List[TaskResult], ExecutionProgress]:
+        while retry_count <= max_retries:
+            try:
+                if retry_count > 0:
+                    ray.get(progress_tracker.update_status.remote(device_id, TaskStatus.RETRYING))
+                else:
+                    ray.get(progress_tracker.update_status.remote(device_id, TaskStatus.RUNNING))
+                
+                # Deploy config to device
+                deployment_result = gns3_client.apply_config(device_id, config)
+                
+                duration = time.time() - start_time
+                result = TaskResult(
+                    device_id=device_id,
+                    status=TaskStatus.SUCCESS,
+                    result=deployment_result,
+                    duration_seconds=duration,
+                    retry_count=retry_count
+                )
+                
+                ray.get(progress_tracker.update_status.remote(device_id, TaskStatus.SUCCESS))
+                ray.get(progress_tracker.add_result.remote(result))
+                
+                return result
+                
+            except Exception as e:
+                retry_count += 1
+                if retry_count > max_retries:
+                    duration = time.time() - start_time
+                    result = TaskResult(
+                        device_id=device_id,
+                        status=TaskStatus.FAILED,
+                        error=f"Failed after {retry_count} retries: {str(e)}",
+                        duration_seconds=duration,
+                        retry_count=retry_count - 1
+                    )
+                    
+                    ray.get(progress_tracker.update_status.remote(device_id, TaskStatus.FAILED))
+                    ray.get(progress_tracker.add_result.remote(result))
+                    
+                    return result
+                
+                # Exponential backoff
+                time.sleep(2 ** retry_count)
+
+
+    class RayExecutor:
         """
-        Deploy to devices in stages with validation between stages.
-        
-        Implements canary deployment pattern:
-        - Stage 1: 1% of fleet
-        - Stage 2: 10% of fleet
-        - Stage 3: 50% of fleet
-        - Stage 4: 100% of fleet
+        Distributed execution engine for hyperscale network automation.
         
-        Args:
-            deployments: Dict mapping device_id to config
-            gns3_client: GNS3 client instance
-            stages: List of percentages for each stage (0.0 to 1.0)
-            validation_fn: Optional function to validate stage success
-        
-        Returns:
-            Tuple of (results, final_progress)
+        Provides parallel execution of config generation, analysis, and deployment
+        across thousands of devices using Ray's distributed computing framework.
         """
-        self.initialize()
         
-        device_ids = list(deployments.keys())
-        total_devices = len(device_ids)
-        all_results = []
+        def __init__(self, ray_address: Optional[str] = None, num_cpus: Optional[int] = None):
+            """
+            Initialize Ray executor.
+            
+            Args:
+                ray_address: Ray cluster address (None for local mode)
+                num_cpus: Number of CPUs to use (None for auto-detect)
+            """
+            self.ray_address = ray_address
+            self.num_cpus = num_cpus
+            self.initialized = False
+            self._progress_tracker = NoneNone
         
-        current_index = 0
+        def initialize(self):
+            """Initialize Ray runtime"""
+            if self.initialized:
+                return
+            
+            try:
+                # Check if Ray is already initialized
+                if ray.is_initialized():
+                    logger.info("Ray already initialized")
+                else:
+                    # Initialize Ray
+                    if self.ray_address:
+                        # Connect to existing cluster
+                        ray.init(address=self.ray_address)
+                        logger.info(f"Connected to Ray cluster at {self.ray_address}")
+                    else:
+                        # Start local Ray instance
+                        init_kwargs = {}
+                        if self.num_cpus:
+                            init_kwargs['num_cpus'] = self.num_cpus
+                        
+                        ray.init(**init_kwargs)
+                        logger.info(f"Started local Ray instance with {ray.available_resources().get('CPU', 0)} CPUs")
+                
+                self.initialized = True
+                
+            except Exception as e:
+                logger.error(f"Failed to initialize Ray: {e}")
+                raise
+        
+        def shutdown(self):
+            """Shutdown Ray runtime"""
+            if self.initialized and ray.is_initialized():
+                ray.shutdown()
+                self.initialized = False
+                logger.info("Ray shutdown complete")
+        
+        def parallel_config_generation(self, devices: List[Dict[str, Any]], 
+                                       template_fn: Callable,
+                                       batch_size: int = 100) -> Tuple[List[TaskResult], ExecutionProgress]:
+            """
+            Generate configs for multiple devices in parallel.
+            
+            Args:
+                devices: List of device data dicts
+                template_fn: Function to generate config from device data
+                batch_size: Number of devices to process in each batch
+            
+            Returns:
+                Tuple of (results, final_progress)
+            """
+            self.initialize()
+            
+            # Create progress tracker
+            progress_tracker = ProgressTracker.remote(total_devices=len(devices))
+            
+            # Initialize pending count
+            ray.get(progress_tracker.update_status.remote("_init_", TaskStatus.PENDING))
+            for _ in range(len(devices) - 1):
+                ray.get(progress_tracker.update_status.remote("_init_", TaskStatus.PENDING))
+            
+            # Launch parallel tasks
+            futures = []
+            for device in devices:
+                future = generate_device_config.remote(
+                    device_id=device['device_id'],
+                    device_data=device,
+                    template_fn=template_fn,
+                    progress_tracker=progress_tracker
+                )
+                futures.append(future)
+                
+                # Process in batches to avoid overwhelming the cluster
+                if len(futures) >= batch_size:
+                    ray.get(futures)
+                    futures = []
+            
+            # Wait for remaining tasks
+            if futures:
+                ray.get(futures)
+            
+            # Get final results
+            results = ray.get(progress_tracker.get_results.remote())
+            final_progress = ray.get(progress_tracker.get_progress.remote())
+            
+            return results, final_progress
         
-        for stage_pct in stages:
-            stage_count = int(total_devices * stage_pct) - current_index
-            if stage_count <= 0:
-                continue
+        def parallel_batfish_analysis(self, configs: Dict[str, str],
+                                      batfish_client: Any,
+                                      batch_size: int = 50) -> Tuple[List[TaskResult], ExecutionProgress]:
+            """
+            Analyze configs in parallel using Batfish.
             
-            stage_devices = device_ids[current_index:current_index + stage_count]
-            stage_deployments = {did: deployments[did] for did in stage_devices}
+            Args:
+                configs: Dict mapping device_id to config string
+                batfish_client: Batfish client instance
+                batch_size: Number of configs to analyze in each batch
             
-            logger.info(f"Starting stage {stage_pct*100}%: deploying to {len(stage_devices)} devices")
+            Returns:
+                Tuple of (results, final_progress)
+            """
+            self.initialize()
             
-            # Deploy this stage
-            results, progress = self.parallel_deployment(
-                deployments=stage_deployments,
-                gns3_client=gns3_client,
-                batch_size=min(20, len(stage_devices))
-            )
+            progress_tracker = ProgressTracker.remote(total_devices=len(configs))
             
-            all_results.extend(results)
+            # Initialize pending count
+            for _ in range(len(configs)):
+                ray.get(progress_tracker.update_status.remote("_init_", TaskStatus.PENDING))
             
-            # Check for failures
-            failed_count = sum(1 for r in results if r.status == TaskStatus.FAILED)
-            failure_rate = failed_count / len(results) if results else 0
+            # Launch parallel analysis tasks
+            futures = []
+            for device_id, config in configs.items():
+                future = analyze_device_config.remote(
+                    device_id=device_id,
+                    config=config,
+                    batfish_client=batfish_client,
+                    progress_tracker=progress_tracker
+                )
+                futures.append(future)
+                
+                if len(futures) >= batch_size:
+                    ray.get(futures)
+                    futures = []
+            
+            if futures:
+                ray.get(futures)
+            
+            results = ray.get(progress_tracker.get_results.remote())
+            final_progress = ray.get(progress_tracker.get_progress.remote())
+            
+            return results, final_progressress
+        
+        def parallel_deployment(self, deployments: Dict[str, str],
+                               gns3_client: Any,
+                               batch_size: int = 20,
+                               max_retries: int = 3) -> Tuple[List[TaskResult], ExecutionProgress]:
+            """
+            Deploy configs to multiple devices in parallel.
+            
+            Args:
+                deployments: Dict mapping device_id to config string
+                gns3_client: GNS3 client instance
+                batch_size: Number of devices to deploy to simultaneously
+                max_retries: Maximum retry attempts per device
             
-            if failure_rate > 0.1:  # More than 10% failure rate
-                logger.error(f"Stage failed with {failure_rate*100}% failure rate. Stopping rollout.")
-                # Return partial results
-                final_progress = ExecutionProgress(
-                    total_devices=total_devices,
-                    completed=sum(1 for r in all_results if r.status == TaskStatus.SUCCESS),
-                    failed=sum(1 for r in all_results if r.status == TaskStatus.FAILED)
+            Returns:
+                Tuple of (results, final_progress)
+            """
+            self.initialize()
+            
+            progress_tracker = ProgressTracker.remote(total_devices=len(deployments))
+            
+            # Initialize pending count
+            for _ in range(len(deployments)):
+                ray.get(progress_tracker.update_status.remote("_init_", TaskStatus.PENDING))
+            
+            # Launch parallel deployment tasks
+            futures = []
+            for device_id, config in deployments.items():
+                future = deploy_to_device.remote(
+                    device_id=device_id,
+                    config=config,
+                    gns3_client=gns3_client,
+                    progress_tracker=progress_tracker,
+                    max_retries=max_retries
                 )
-                return all_results, final_progress.to_dict()
+                futures.append(future)
+                
+                # Deploy in smaller batches to avoid overwhelming network
+                if len(futures) >= batch_size:
+                    ray.get(futures)
+                    futures = []
             
-            # Run validation if provided
-            if validation_fn:
-                try:
-                    if not validation_fn(stage_devices, results):
-                        logger.error("Stage validation failed. Stopping rollout.")
-                        final_progress = ExecutionProgress(
-                            total_devices=total_devices,
-                            completed=sum(1 for r in all_results if r.status == TaskStatus.SUCCESS),
-                            failed=sum(1 for r in all_results if r.status == TaskStatus.FAILED)
-                        )
-                        return all_results, final_progress.to_dict()
-                except Exception as e:
-                    logger.error(f"Stage validation error: {e}. Stopping rollout.")
+            if futures:
+                ray.get(futures)
+            
+            results = ray.get(progress_tracker.get_results.remote())
+            final_progress = ray.get(progress_tracker.get_progress.remote())
+            
+            return results, final_progress
+        
+        def get_cluster_resources(self) -> Dict[str, Any]:
+            """Get available cluster resources"""
+            self.initialize()
+            return {
+                'available': ray.available_resources(),
+                'total': ray.cluster_resources()
+            }
+        
+        def staggered_rollout(self, deployments: Dict[str, str],
+                             gns3_client: Any,
+                             stages: List[float] = [0.01, 0.1, 0.5, 1.0],
+                             validation_fn: Optional[Callable] = None) -> Tuple[List[TaskResult], ExecutionProgress]:
+            """
+            Deploy to devices in stages with validation between stages.
+            
+            Implements canary deployment pattern:
+            - Stage 1: 1% of fleet
+            - Stage 2: 10% of fleet
+            - Stage 3: 50% of fleet
+            - Stage 4: 100% of fleet
+            
+            Args:
+                deployments: Dict mapping device_id to config
+                gns3_client: GNS3 client instance
+                stages: List of percentages for each stage (0.0 to 1.0)
+                validation_fn: Optional function to validate stage success
+            
+            Returns:
+                Tuple of (results, final_progress)
+            """
+            self.initialize()
+            
+            device_ids = list(deployments.keys())
+            total_devices = len(device_ids)
+            all_results = []
+            
+            current_index = 0
+            
+            for stage_pct in stages:
+                stage_count = int(total_devices * stage_pct) - current_index
+                if stage_count <= 0:
+                    continue
+                
+                stage_devices = device_ids[current_index:current_index + stage_count]
+                stage_deployments = {did: deployments[did] for did in stage_devices}
+                
+                logger.info(f"Starting stage {stage_pct*100}%: deploying to {len(stage_devices)} devices")
+                
+                # Deploy this stage
+                results, progress = self.parallel_deployment(
+                    deployments=stage_deployments,
+                    gns3_client=gns3_client,
+                    batch_size=min(20, len(stage_devices))
+                )
+                
+                all_results.extend(results)
+                
+                # Check for failures
+                failed_count = sum(1 for r in results if r.status == TaskStatus.FAILED)
+                failure_rate = failed_count / len(results) if results else 0
+                
+                if failure_rate > 0.1:  # More than 10% failure rate
+                    logger.error(f"Stage failed with {failure_rate*100}% failure rate. Stopping rollout.")
+                    # Return partial results
                     final_progress = ExecutionProgress(
                         total_devices=total_devices,
                         completed=sum(1 for r in all_results if r.status == TaskStatus.SUCCESS),
                         failed=sum(1 for r in all_results if r.status == TaskStatus.FAILED)
                     )
                     return all_results, final_progress.to_dict()
+                
+                # Run validation if provided
+                if validation_fn:
+                    try:
+                        if not validation_fn(stage_devices, results):
+                            logger.error("Stage validation failed. Stopping rollout.")
+                            final_progress = ExecutionProgress(
+                                total_devices=total_devices,
+                                completed=sum(1 for r in all_results if r.status == TaskStatus.SUCCESS),
+                                failed=sum(1 for r in all_results if r.status == TaskStatus.FAILED)
+                            )
+                            return all_results, final_progress.to_dict()
+                    except Exception as e:
+                        logger.error(f"Stage validation error: {e}. Stopping rollout.")
+                        final_progress = ExecutionProgress(
+                            total_devices=total_devices,
+                            completed=sum(1 for r in all_results if r.status == TaskStatus.SUCCESS),
+                            failed=sum(1 for r in all_results if r.status == TaskStatus.FAILED)
+                        )
+                        return all_results, final_progress.to_dict()
+                
+                logger.info(f"Stage {stage_pct*100}% completed successfully")
+                current_index += stage_count
             
-            logger.info(f"Stage {stage_pct*100}% completed successfully")
-            current_index += stage_count
+            # Create final progress
+            final_progress = ExecutionProgress(
+                total_devices=total_devices,
+                completed=sum(1 for r in all_results if r.status == TaskStatus.SUCCESS),
+                failed=sum(1 for r in all_results if r.status == TaskStatus.FAILED)
+            )
+            
+            return all_results, final_progress.to_dict()
+
+else:
+    # Fallback executor when ray is not available
+    class RayExecutor:
+        """Fallback executor without distributed capabilities"""
+        
+        def __init__(self, ray_address: Optional[str] = None, num_cpus: Optional[int] = None):
+            logging.warning("Ray not available - using fallback sequential executor")
+            self.ray_address = ray_address
+            self.num_cpus = num_cpus
+            self.initialized = False
+        
+        def initialize(self):
+            """No-op initialization for fallback"""
+            self.initialized = True
+        
+        def shutdown(self):
+            """No-op shutdown for fallback"""
+            self.initialized = False
+        
+        def parallel_config_generation(self, devices: List[Dict[str, Any]], 
+                                       template_fn: Callable,
+                                       batch_size: int = 100) -> Tuple[List[TaskResult], ExecutionProgress]:
+            """Sequential fallback for config generation"""
+            raise NotImplementedError(
+                "Distributed execution requires ray. Install with: pip install ray"
+            )
         
-        # Create final progress
-        final_progress = ExecutionProgress(
-            total_devices=total_devices,
-            completed=sum(1 for r in all_results if r.status == TaskStatus.SUCCESS),
-            failed=sum(1 for r in all_results if r.status == TaskStatus.FAILED)
-        )
+        def parallel_batfish_analysis(self, configs: Dict[str, str],
+                                      batfish_client: Any,
+                                      batch_size: int = 50) -> Tuple[List[TaskResult], ExecutionProgress]:
+            """Sequential fallback for analysis"""
+            raise NotImplementedError(
+                "Distributed execution requires ray. Install with: pip install ray"
+            )
         
-        return all_results, final_progress.to_dict()
+        def parallel_deployment(self, deployments: Dict[str, str],
+                               gns3_client: Any,
+                               batch_size: int = 20,
+                               max_retries: int = 3) -> Tuple[List[TaskResult], ExecutionProgress]:
+            """Sequential fallback for deployment"""
+            raise NotImplementedError(
+                "Distributed execution requires ray. Install with: pip install ray"
+            )
+        
+        def staged_rollout(self, deployments: Dict[str, str],
+                          gns3_client: Any,
+                          stages: List[float] = None,
+                          validation_fn: Optional[Callable] = None) -> Tuple[List[TaskResult], Dict[str, Any]]:
+            """Sequential fallback for staged rollout"""
+            raise NotImplementedError(
+                "Distributed execution requires ray. Install with: pip install ray"
+            )

app.py

@@ -237,6 +237,162 @@ def build_ui():
         
         gr.Markdown("---")
 
+        # ===== INTERACTIVE CONSULTATION TAB =====
+        gr.Markdown("""### 💬 Interactive Consultation
+        **Have a conversation before running the full pipeline**
+        
+        Not sure what you need? Start here! The AI will ask clarifying questions to understand your network requirements.
+        """)
+        
+        with gr.Row(elem_classes=["og-main-row"]):
+            with gr.Column(elem_classes=["og-panel"]):
+                gr.Markdown("#### 🤝 Consultation Chat")
+                consultation_chat = gr.Chatbot(
+                    label="AI Network Consultant",
+                    value=[],
+                    height=400,
+                    show_label=False,
+                )
+                consultation_input = gr.Textbox(
+                    label="Your message",
+                    placeholder="Hi, I need help designing a network for my small business...",
+                    lines=2,
+                )
+                with gr.Row():
+                    send_consultation_btn = gr.Button("📤 Send", variant="primary")
+                    start_new_consultation_btn = gr.Button("🔄 New Consultation", variant="secondary")
+                    use_consultation_btn = gr.Button("✅ Use This for Pipeline", variant="primary")
+                
+                consultation_summary = gr.Textbox(
+                    label="Consultation Summary (will be used in pipeline)",
+                    lines=5,
+                    interactive=False,
+                )
+        
+        # Consultation state
+        consultation_state = gr.State(value={"consultant": None, "is_complete": False})
+        
+        def start_consultation():
+            """Initialize a new consultation"""
+            return {
+                "consultant": None,
+                "is_complete": False
+            }, [], ""
+        
+        def send_consultation_message(user_message, state):
+            """Send a message in the consultation"""
+            if not user_message.strip():
+                # Get current chat history from state
+                consultant = state.get("consultant")
+                if consultant and hasattr(consultant, 'conversation_history'):
+                    # Convert conversation history to chatbot format
+                    chatbot_messages = []
+                    for msg in consultant.conversation_history:
+                        if msg.role == "user" and not msg.content.startswith("Initial request:"):
+                            chatbot_messages.append((msg.content, None))
+                        elif msg.role == "assistant" and chatbot_messages:
+                            # Pair with previous user message
+                            if chatbot_messages[-1][1] is None:
+                                chatbot_messages[-1] = (chatbot_messages[-1][0], msg.content)
+                            else:
+                                chatbot_messages.append((None, msg.content))
+                    return state, chatbot_messages, ""
+                return state, [], ""
+            
+            from agent.consultation import NetworkConsultant
+            
+            # Get or create consultant
+            consultant = state.get("consultant")
+            first_message = consultant is None
+            
+            if first_message:
+                consultant = NetworkConsultant()
+                state["consultant"] = consultant
+                # Start consultation with user's first message
+                is_complete, response, intent_data = consultant.start_consultation(user_message)
+            else:
+                # Continue existing consultation
+                is_complete, response, intent_data = consultant.continue_consultation(user_message)
+            
+            # Update completion state
+            state["is_complete"] = is_complete
+            if is_complete and intent_data:
+                state["intent_data"] = intent_data
+            
+            # Build chatbot display from conversation history
+            chatbot_messages = []
+            for msg in consultant.conversation_history:
+                # Skip system messages and "Initial request:" prefix
+                if msg.role == "system":
+                    continue
+                if msg.role == "user":
+                    content = msg.content
+                    if content.startswith("Initial request: "):
+                        content = content[len("Initial request: "):]
+                    chatbot_messages.append((content, None))
+                elif msg.role == "assistant":
+                    # Pair with the last user message
+                    if chatbot_messages and chatbot_messages[-1][1] is None:
+                        chatbot_messages[-1] = (chatbot_messages[-1][0], msg.content)
+                    else:
+                        # Standalone assistant message (shouldn't happen normally)
+                        chatbot_messages.append((None, msg.content))
+            
+            return state, chatbot_messages, ""
+        
+        def use_consultation_summary(state):
+            """Extract the consultation summary for use in the pipeline"""
+            consultant = state.get("consultant")
+            if not consultant or not consultant.conversation_history:
+                return "No consultation history yet. Start a conversation first!"
+            
+            # If consultation is complete, use the structured intent
+            if state.get("is_complete") and state.get("intent_data"):
+                intent = state["intent_data"]
+                summary = "# Consultation Summary (Complete)\n\n"
+                for key, value in intent.items():
+                    summary += f"**{key}:** {value}\n\n"
+                return summary
+            
+            # Otherwise, combine all messages into a summary
+            summary = "# Consultation Summary (In Progress)\n\n"
+            for msg in consultant.conversation_history:
+                if msg.role == "system":
+                    continue
+                prefix = "**You:** " if msg.role == "user" else "**AI:** "
+                content = msg.content
+                if content.startswith("Initial request: "):
+                    content = content[len("Initial request: "):]
+                summary += f"{prefix}{content}\n\n"
+            
+            return summary
+        
+        # Wire up consultation handlers
+        start_new_consultation_btn.click(
+            fn=start_consultation,
+            outputs=[consultation_state, consultation_chat, consultation_input]
+        )
+        
+        send_consultation_btn.click(
+            fn=send_consultation_message,
+            inputs=[consultation_input, consultation_state],
+            outputs=[consultation_state, consultation_chat, consultation_input]
+        )
+        
+        consultation_input.submit(
+            fn=send_consultation_message,
+            inputs=[consultation_input, consultation_state],
+            outputs=[consultation_state, consultation_chat, consultation_input]
+        )
+        
+        use_consultation_btn.click(
+            fn=use_consultation_summary,
+            inputs=[consultation_state],
+            outputs=[consultation_summary]
+        )
+        
+        gr.Markdown("---")
+
         # Main Pipeline UI
         gr.Markdown("""### 🌿 Network Automation Pipeline
         **From Consultation → Production**