""" Overgrowth Network Automation Pipeline From consultation to production-ready network Pipeline Stages: 1. Consultation - Natural language intent capture 2. Source of Truth - Generate/update network data model 3. Diagram - Visual representation 4. Bill of Materials - Hardware/software shopping list 5. Setup Guide - Human deployment instructions (physical + OOB) 6. Autonomous Deploy - AI agents configure everything 7. Observability - Monitoring, topology discovery, telemetry 8. Validation - Verify and maintain intended state """ from typing import Dict, List, Optional, Any from dataclasses import dataclass, asdict from pathlib import Path import json import yaml import logging from .netbox_client import NetBoxClient logger = logging.getLogger(__name__) @dataclass class NetworkIntent: """Captured from consultation phase""" description: str business_requirements: List[str] constraints: List[str] timeline: Optional[str] = None budget: Optional[str] = None @dataclass class Device: """Network device in source of truth""" name: str role: str # core, distribution, access, edge, etc. model: str vendor: str mgmt_ip: str location: str interfaces: List[Dict[str, Any]] configs: Optional[Dict[str, Any]] = None @dataclass class NetworkModel: """Single Source of Truth for the network""" name: str version: str intent: NetworkIntent devices: List[Device] vlans: List[Dict[str, Any]] subnets: List[Dict[str, Any]] routing: Dict[str, Any] services: List[str] # DHCP, DNS, NTP, etc. def to_dict(self): return asdict(self) def to_yaml(self) -> str: return yaml.dump(self.to_dict(), default_flow_style=False) @classmethod def from_yaml(cls, yaml_str: str): data = yaml.safe_load(yaml_str) return cls(**data) @dataclass class BillOfMaterials: """Hardware and software requirements""" network_name: str devices: List[Dict[str, Any]] # quantity, model, purpose, vendor, estimated_cost cables: List[Dict[str, Any]] # type, length, quantity accessories: List[Dict[str, Any]] # racks, power, console cables, etc. software_licenses: List[Dict[str, Any]] total_estimated_cost: float procurement_links: List[str] def to_shopping_list(self) -> str: """Generate human-readable shopping list""" lines = [] lines.append(f"# Bill of Materials: {self.network_name}\n") lines.append("## Network Devices") for item in self.devices: lines.append(f"- [{item['quantity']}x] {item['model']} - {item['purpose']}") lines.append(f" Vendor: {item['vendor']} | Est. Cost: ${item['estimated_cost']}") if item.get('link'): lines.append(f" Link: {item['link']}") lines.append("\n## Cabling") for item in self.cables: lines.append(f"- [{item['quantity']}x] {item['type']} ({item['length']})") if 'estimated_cost' in item: lines.append(f" Est. Cost: ${item['estimated_cost']:.2f}") lines.append("\n## Accessories") for item in self.accessories: lines.append(f"- {item['name']} - {item['purpose']}") if 'estimated_cost' in item: lines.append(f" Est. Cost: ${item['estimated_cost']:.2f}") lines.append("\n## Software Licenses") for item in self.software_licenses: lines.append(f"- {item['name']} ({item['license_type']})") lines.append(f"\n## Total Estimated Cost: ${self.total_estimated_cost:,.2f}") if self.procurement_links: lines.append("\n## Procurement Links") for link in self.procurement_links: lines.append(f"- {link}") return "\n".join(lines) @dataclass class SetupGuide: """Human deployment instructions""" network_name: str phases: List[Dict[str, Any]] oob_network_config: Dict[str, Any] safety_checklist: List[str] rollback_plan: List[str] def to_markdown(self) -> str: """Generate deployment guide""" lines = [] lines.append(f"# Network Deployment Guide: {self.network_name}\n") lines.append("## Safety Checklist") for item in self.safety_checklist: lines.append(f"- [ ] {item}") lines.append("\n## Deployment Phases\n") for i, phase in enumerate(self.phases, 1): lines.append(f"### Phase {i}: {phase['name']}") lines.append(f"**Duration:** {phase.get('duration', 'TBD')}") lines.append(f"**Prerequisites:** {', '.join(phase.get('prerequisites', []))}") lines.append("\n**Steps:**") for step in phase['steps']: lines.append(f"- [ ] {step}") lines.append("") lines.append("## Out-of-Band Management Network") lines.append("```yaml") lines.append(yaml.dump(self.oob_network_config, default_flow_style=False)) lines.append("```") lines.append("\n## Rollback Plan") for step in self.rollback_plan: lines.append(f"- {step}") return "\n".join(lines) class OvergrowthPipeline: """ Main pipeline orchestrator Manages the flow from consultation to production """ def __init__(self, workspace_dir: Path = Path("./infra"), use_netbox: bool = True): self.workspace_dir = workspace_dir self.workspace_dir.mkdir(exist_ok=True) # Pipeline state storage self.state_file = workspace_dir / "pipeline_state.json" self.sot_file = workspace_dir / "network_model.yaml" self.bom_file = workspace_dir / "bill_of_materials.json" self.setup_guide_file = workspace_dir / "setup_guide.md" # NetBox integration self.use_netbox = use_netbox if use_netbox: self.netbox = NetBoxClient() if not self.netbox.mock_mode: logger.info("Using NetBox as Source of Truth backend") else: logger.warning("NetBox not available - falling back to YAML files") self.use_netbox = False # Batfish integration from agent.batfish_client import BatfishClient self.batfish = BatfishClient(use_batfish=True) # SuzieQ integration from agent.suzieq_client import SuzieQClient self.suzieq = SuzieQClient(use_suzieq=True) # Incident learning system from agent.incident_learning import IncidentDatabase, RootCauseAnalyzer, RegressionTestGenerator self.incident_db = IncidentDatabase() self.rca_analyzer = RootCauseAnalyzer(self.incident_db) self.test_generator = RegressionTestGenerator() # 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]: """ Stage 0: Pre-flight validation (BEFORE deployment) Schema validation, policy checks, and eventually Batfish analysis """ logger.info("Stage 0: Running pre-flight validation") from agent.schema_validation import get_validation_errors, validate_network_model from agent.policy_engine import NetworkPolicy results = { 'schema_valid': False, 'policy_passed': False, 'ready_to_deploy': False, 'errors': [], 'warnings': [], 'info': [] } # Convert NetworkModel to dict for validation model_dict = model.to_dict() # 1. Schema Validation (Pydantic) logger.info("Running schema validation...") schema_errors = get_validation_errors(model_dict) if schema_errors: results['errors'].extend([f"Schema: {e}" for e in schema_errors]) logger.error(f"Schema validation failed with {len(schema_errors)} errors") else: results['schema_valid'] = True logger.info("✓ Schema validation passed") # 2. Policy Engine logger.info("Running policy checks...") policy = NetworkPolicy() violations = policy.check_network_model(model_dict) by_severity = policy.get_violations_by_severity() results['errors'].extend([str(v) for v in by_severity['ERROR']]) results['warnings'].extend([str(v) for v in by_severity['WARNING']]) results['info'].extend([str(v) for v in by_severity['INFO']]) if policy.has_errors(): logger.error(f"Policy validation failed with {len(by_severity['ERROR'])} errors") else: results['policy_passed'] = True logger.info(f"✓ Policy validation passed ({len(by_severity['WARNING'])} warnings, {len(by_severity['INFO'])} info)") # 3. Batfish Static Analysis logger.info("Running Batfish static analysis...") batfish_results = self._run_batfish_analysis(model) results['batfish_analysis'] = batfish_results results['batfish_passed'] = batfish_results.get('all_passed', False) # Add Batfish errors to overall results if not batfish_results.get('all_passed', False): if batfish_results.get('undefined_references'): results['errors'].append( f"Batfish: {len(batfish_results['undefined_references'])} undefined references" ) if batfish_results.get('routing_loops'): results['errors'].append( f"Batfish: {len(batfish_results['routing_loops'])} routing loops detected" ) if batfish_results.get('forwarding_errors'): results['errors'].append( f"Batfish: {len(batfish_results['forwarding_errors'])} forwarding errors" ) # Add recommendations if 'recommendations' in batfish_results: results['info'].extend(batfish_results['recommendations']) # Overall result (now includes Batfish) results['ready_to_deploy'] = ( results['schema_valid'] and results['policy_passed'] and results['batfish_passed'] ) if results['ready_to_deploy']: logger.info("✓ Pre-flight validation PASSED - ready to deploy") else: logger.warning(f"✗ Pre-flight validation FAILED - {len(results['errors'])} errors must be fixed") # Capture deployment failure for learning self._capture_validation_failure(model, results) return results def _run_batfish_analysis(self, model: NetworkModel) -> Dict[str, Any]: """ Run Batfish static analysis on network model Generates configs and analyzes them """ from agent.batfish_client import BatfishClient # Generate device configs from model configs = self._generate_configs_for_batfish(model) if not configs: logger.warning("No configs generated for Batfish analysis") return { 'all_passed': True, 'mock_mode': True, 'recommendations': ['No device configs to analyze'] } # Run Batfish analysis analysis = self.batfish.analyze_configs(configs, network_name=model.name) # Convert to dict and add recommendations results = analysis.to_dict() results['recommendations'] = self.batfish.generate_config_recommendations(analysis) results['mock_mode'] = self.batfish.mock_mode return results def _generate_configs_for_batfish(self, model: NetworkModel) -> Dict[str, str]: """ Generate device configurations from network model These are simple configs for Batfish validation Uses parallel execution when parallel_mode=True and >10 devices """ # 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 = {} # Generate basic configs for each device for device in model.devices: config_lines = [] # Hostname config_lines.append(f"hostname {device.name}") config_lines.append("!") # VLANs for vlan in model.vlans: config_lines.append(f"vlan {vlan['id']}") config_lines.append(f" name {vlan['name']}") config_lines.append("!") # Interfaces config_lines.append("interface Vlan1") config_lines.append(f" ip address {device.mgmt_ip} 255.255.255.0") config_lines.append(" no shutdown") config_lines.append("!") # Routing if model.routing: protocol = model.routing.get('protocol', 'static') if protocol == 'ospf': process_id = model.routing.get('process_id', 1) config_lines.append(f"router ospf {process_id}") for network in model.routing.get('networks', []): config_lines.append(f" network {network} area 0") config_lines.append("!") configs[device.name] = "\n".join(config_lines) return configs def _parallel_config_generation(self, model: NetworkModel) -> Dict[str, str]: """ Generate configs in parallel using Ray Scales to thousands of devices """ logger.info(f"Generating {len(model.devices)} configs in parallel using Ray") # Prepare device data for parallel processing device_data_list = [] for device in model.devices: device_data_list.append({ 'device_id': device.name, 'device': device, 'vlans': model.vlans, 'routing': model.routing }) # Define config generation function def generate_device_config(device_data: Dict[str, Any]) -> str: device = device_data['device'] vlans = device_data['vlans'] routing = device_data['routing'] config_lines = [] config_lines.append(f"hostname {device.name}") config_lines.append("!") for vlan in vlans: config_lines.append(f"vlan {vlan['id']}") config_lines.append(f" name {vlan['name']}") config_lines.append("!") config_lines.append("interface Vlan1") config_lines.append(f" ip address {device.mgmt_ip} 255.255.255.0") config_lines.append(" no shutdown") config_lines.append("!") if routing: protocol = routing.get('protocol', 'static') if protocol == 'ospf': process_id = routing.get('process_id', 1) config_lines.append(f"router ospf {process_id}") for network in routing.get('networks', []): config_lines.append(f" network {network} area 0") config_lines.append("!") return "\n".join(config_lines) # Execute in parallel results, progress = self.ray_executor.parallel_config_generation( devices=device_data_list, template_fn=generate_device_config, batch_size=100 ) logger.info(f"Config generation complete: {progress['completed']}/{progress['total_devices']} succeeded") # Extract successful configs configs = {} for result in results: if result.status.value == 'success': configs[result.device_id] = result.result else: logger.error(f"Failed to generate config for {result.device_id}: {result.error}") return configs def stage1_consultation(self, user_input: str) -> NetworkIntent: """ Stage 1: Capture user intent from natural language Uses LLM to extract structured requirements """ logger.info("Stage 1: Processing consultation input") from agent.consultation import NetworkConsultant consultant = NetworkConsultant() is_complete, output, intent_data = consultant.start_consultation(user_input) if is_complete and intent_data: # Consultation completed in one round intent = NetworkIntent( description=intent_data.get('description', user_input), business_requirements=intent_data.get('business_requirements', []), constraints=intent_data.get('constraints', []), timeline=intent_data.get('timeline'), budget=intent_data.get('budget') ) else: # Need more information - for now, use what we have # TODO: Support multi-turn consultation in UI logger.warning("Consultation incomplete - proceeding with available info") intent = NetworkIntent( description=user_input, business_requirements=["High availability", "Scalability"], constraints=["Budget conscious", "Easy to maintain"] ) return intent def _generate_clarifying_questions(self, intent: NetworkIntent) -> List[str]: """Deterministic clarifying questions for the UI when LLM chat is disabled.""" return [ "What is the target WAN bandwidth per site (e.g., 200 Mbps, 1 Gbps)?", "Do you need redundant internet links at HQ or any branch?", "Are guest and IoT networks required to be fully isolated from corporate traffic?", "Which vendors are approved for switches/routers/firewalls (Cisco/Arista/Fortinet/Ubiquiti)?", "Do you need WiFi voice roaming or only data for guests/corp?", ] def stage2_generate_sot(self, intent: NetworkIntent) -> NetworkModel: """ Stage 2: Generate Source of Truth from intent Creates the authoritative network data model """ logger.info("Stage 2: Generating source of truth") from agent.llm_client import LLMClient, LLMMessage import json llm = LLMClient() def _default_design() -> Dict[str, Any]: """Deterministic fallback design with concrete values for offline/demo runs.""" return { "vlans": [ {"id": 10, "name": "Management", "subnet": "10.10.10.0/24", "purpose": "Mgmt"}, {"id": 20, "name": "Users", "subnet": "10.20.0.0/22", "purpose": "Corp"}, {"id": 30, "name": "Guest", "subnet": "10.30.0.0/23", "purpose": "Guest WiFi"}, {"id": 40, "name": "IoT", "subnet": "10.40.0.0/23", "purpose": "Cameras/IoT"}, ], "subnets": [ {"network": "10.10.10.0/24", "gateway": "10.10.10.1", "vlan": 10, "purpose": "Mgmt"}, {"network": "10.20.0.0/22", "gateway": "10.20.0.1", "vlan": 20, "purpose": "Users"}, {"network": "10.30.0.0/23", "gateway": "10.30.0.1", "vlan": 30, "purpose": "Guest"}, {"network": "10.40.0.0/23", "gateway": "10.40.0.1", "vlan": 40, "purpose": "IoT"}, ], "devices": [ {"name": "hq-core-1", "role": "core", "model": "Cisco Catalyst 9300", "vendor": "cisco", "mgmt_ip": "10.10.10.11", "location": "HQ"}, {"name": "hq-core-2", "role": "core", "model": "Arista 7050", "vendor": "arista", "mgmt_ip": "10.10.10.12", "location": "HQ"}, {"name": "hq-fw", "role": "firewall", "model": "Fortinet FortiGate 60F", "vendor": "fortinet", "mgmt_ip": "10.10.10.21", "location": "HQ"}, {"name": "branch1-wan", "role": "edge", "model": "Cisco ISR 1100", "vendor": "cisco", "mgmt_ip": "10.10.10.31", "location": "Branch1"}, {"name": "branch2-wan", "role": "edge", "model": "Cisco ISR 1100", "vendor": "cisco", "mgmt_ip": "10.10.10.32", "location": "Branch2"}, {"name": "branch3-wan", "role": "edge", "model": "Cisco ISR 1100", "vendor": "cisco", "mgmt_ip": "10.10.10.33", "location": "Branch3"}, {"name": "hq-ap-1", "role": "wireless", "model": "Ubiquiti U6-Pro", "vendor": "ubiquiti", "mgmt_ip": "10.10.10.41", "location": "HQ"}, {"name": "hq-ap-2", "role": "wireless", "model": "Ubiquiti U6-Pro", "vendor": "ubiquiti", "mgmt_ip": "10.10.10.42", "location": "HQ"}, ], "services": ["DHCP", "DNS", "NTP", "Syslog", "RADIUS"], "routing": {"protocol": "ospf", "areas": ["0.0.0.0"], "process_id": 1, "networks": ["10.0.0.0/8"]}, } # Build prompt for network design design_prompt = f"""You are an expert network architect. Design a production-ready network based on these requirements: Description: {intent.description} Business Requirements: {', '.join(intent.business_requirements)} Constraints: {', '.join(intent.constraints)} Budget: {intent.budget or 'Not specified'} Timeline: {intent.timeline or 'Not specified'} Generate a complete network design with: 1. VLANs (ID, name, purpose, subnet) 2. Subnets (CIDR, gateway, purpose) 3. Devices (name, role, suggested model) 4. Services needed (DHCP, DNS, NTP, etc.) 5. Routing protocol recommendation Return ONLY a JSON object in this exact format: {{ "vlans": [ {{"id": 10, "name": "Management", "subnet": "10.0.10.0/24", "purpose": "Network management"}} ], "subnets": [ {{"network": "10.0.10.0/24", "gateway": "10.0.10.1", "vlan": 10, "purpose": "Management network"}} ], "devices": [ {{"name": "core-sw-01", "role": "core", "model": "Cisco Catalyst 9300", "mgmt_ip": "10.0.10.10"}} ], "services": ["DHCP", "DNS", "NTP"], "routing": {{"protocol": "OSPF", "areas": ["Area 0"]}} }} Be specific and practical. Use RFC1918 addressing. Consider scalability and security.""" try: # Get LLM response messages = [LLMMessage(role="user", content=design_prompt)] response = llm.chat(messages, temperature=0.3, max_tokens=3000) # Parse JSON from response json_start = response.find('{') json_end = response.rfind('}') + 1 if json_start >= 0 and json_end > json_start: design = json.loads(response[json_start:json_end]) else: raise ValueError("No JSON found in LLM response") # Build NetworkModel from design model = NetworkModel( name=f"network_{intent.description[:20].replace(' ', '_')}", version="1.0.0", intent=intent, devices=[], # populated below vlans=design.get('vlans', []), subnets=design.get('subnets', []), routing=design.get('routing', {}), services=design.get('services', ["DHCP", "DNS", "NTP"]) ) except Exception as e: logger.error(f"LLM design failed: {e}, using template") # Deterministic fallback template with real values design = _default_design() model = NetworkModel( name=f"network_{intent.description[:20].replace(' ', '_')}", version="1.0.0", intent=intent, devices=[], # populated below vlans=design.get('vlans', []), subnets=design.get('subnets', []), routing=design.get('routing', {}), services=design.get('services', ["DHCP", "DNS", "NTP"]) ) # Ensure we have meaningful design data even if LLM returned partials if not model.vlans or not model.subnets or not design.get("devices"): design = _default_design() model.vlans = design["vlans"] model.subnets = design["subnets"] model.routing = design["routing"] model.services = design["services"] # Populate devices from design and backfill mgmt IPs if missing devices: List[Device] = [] mgmt_seed = 11 for dev in design.get("devices", []): mgmt_ip = dev.get("mgmt_ip") or f"10.10.10.{mgmt_seed}" mgmt_seed += 1 # Normalize vendor/role for schema validation expectations vendor = (dev.get("vendor") or "other").lower() role = dev.get("role", "access").lower() if role == "access_point": role = "wireless" devices.append( Device( name=dev.get("name", f"device-{mgmt_seed}"), role=role, model=dev.get("model", "Generic Switch 48-port"), vendor=vendor, mgmt_ip=mgmt_ip, location=dev.get("location", "unspecified"), interfaces=dev.get("interfaces", []) ) ) # If no devices came through, fall back again to deterministic set if not devices: fallback = _default_design()["devices"] for dev in fallback: devices.append( Device( name=dev["name"], role=dev["role"], model=dev["model"], vendor=dev["vendor"], mgmt_ip=dev["mgmt_ip"], location=dev["location"], interfaces=[] ) ) model.devices = devices # Save to file (always, for backup) self.sot_file.write_text(model.to_yaml()) logger.info(f"Saved source of truth to {self.sot_file}") # Sync to NetBox if available if self.use_netbox and not self.netbox.mock_mode: try: logger.info("Syncing network model to NetBox...") summary = self.netbox.sync_network_model(design) logger.info(f"NetBox sync complete: {summary}") except Exception as e: logger.error(f"Failed to sync to NetBox: {e}") return model def stage3_generate_diagram(self, model: NetworkModel) -> Dict[str, str]: """ Stage 3: Generate network diagrams Returns multiple diagram formats """ logger.info("Stage 3: Generating network diagrams") from .topology_diagram import ( generate_ascii_diagram, generate_mermaid_diagram, generate_topology_summary ) # Convert model to topology format for diagram generation topology = { 'project': model.name, 'nodes': [{'name': d.name, 'node_type': d.role, 'status': 'planned'} for d in model.devices], 'links': [] } # Create simple synthetic links to make diagrams useful cores = [d for d in model.devices if d.role == "core"] firewalls = [d for d in model.devices if d.role == "firewall"] edges = [d for d in model.devices if d.role == "edge"] wireless = [d for d in model.devices if d.role == "wireless"] # Connect core devices together if len(cores) >= 2: topology['links'].append({'src': cores[0].name, 'dst': cores[1].name, 'status': 'planned'}) # Connect firewall to first core if cores and firewalls: topology['links'].append({'src': cores[0].name, 'dst': firewalls[0].name, 'status': 'planned'}) # Connect edges/branches to core or firewall for edge in edges: if firewalls: topology['links'].append({'src': firewalls[0].name, 'dst': edge.name, 'status': 'planned'}) elif cores: topology['links'].append({'src': cores[0].name, 'dst': edge.name, 'status': 'planned'}) # Connect wireless/APs to core for ap in wireless: if cores: topology['links'].append({'src': cores[0].name, 'dst': ap.name, 'status': 'planned'}) diagrams = { 'ascii': generate_ascii_diagram(topology), 'mermaid': generate_mermaid_diagram(topology), 'summary': generate_topology_summary(topology) } return diagrams def stage4_generate_bom(self, model: NetworkModel) -> BillOfMaterials: """ Stage 4: Generate Bill of Materials Creates shopping list for hardware/software """ logger.info("Stage 4: Generating bill of materials") from agent.hardware_pricing import ( estimate_device_cost, estimate_cable_cost, estimate_accessory_cost, PROCUREMENT_LINKS, ) devices = [] device_total = 0 procurement_links = [] # If we have devices in the model, price them if model.devices: for device in model.devices: cost = estimate_device_cost(device.model, device.vendor) devices.append({ 'quantity': 1, 'model': device.model, 'purpose': f"{device.role} - {device.name}", 'vendor': device.vendor, 'estimated_cost': cost, 'link': PROCUREMENT_LINKS.get(device.model) }) device_total += cost if device.model in PROCUREMENT_LINKS: procurement_links.append(f"{device.model}: {PROCUREMENT_LINKS[device.model]}") else: # Estimate based on VLANs/subnets if no devices specified num_vlans = len(model.vlans) if num_vlans > 0: # Assume need at least one core switch devices.append({ 'quantity': 1, 'model': 'Ubiquiti USW-Pro-24-PoE', 'purpose': 'Core switch', 'vendor': 'Ubiquiti', 'estimated_cost': 499, 'link': PROCUREMENT_LINKS.get("Ubiquiti USW-Pro-24-PoE") }) device_total += 499 if "Ubiquiti USW-Pro-24-PoE" in PROCUREMENT_LINKS: procurement_links.append(f"Ubiquiti USW-Pro-24-PoE: {PROCUREMENT_LINKS['Ubiquiti USW-Pro-24-PoE']}") # Add APs if we have guest/user networks if any('guest' in v.get('name', '').lower() or 'wifi' in v.get('name', '').lower() for v in model.vlans): ap_cost = estimate_device_cost('Ubiquiti U6-Pro') devices.append({ 'quantity': 2, 'model': 'Ubiquiti U6-Pro', 'purpose': 'Wireless Access Points', 'vendor': 'Ubiquiti', 'estimated_cost': ap_cost * 2, 'link': PROCUREMENT_LINKS.get("Ubiquiti U6-Pro") }) device_total += ap_cost * 2 if "Ubiquiti U6-Pro" in PROCUREMENT_LINKS: procurement_links.append(f"Ubiquiti U6-Pro: {PROCUREMENT_LINKS['Ubiquiti U6-Pro']}") # Cables cable_total = 0 cables = [ {'type': 'Cat6 Ethernet', 'length': '3ft', 'quantity': 10 + len(model.devices) * 2}, {'type': 'Fiber LC-LC', 'length': '10m', 'quantity': max(2, len(model.devices) // 3)} ] for cable in cables: cost = estimate_cable_cost(cable['type'], cable['quantity'], cable['length']) cable['estimated_cost'] = cost cable_total += cost # Accessories accessory_total = 0 accessories = [ {'name': '42U Server Rack', 'purpose': 'Equipment mounting'}, {'name': 'Console Cable Kit', 'purpose': 'Initial configuration'} ] for acc in accessories: cost = estimate_accessory_cost(acc['name']) acc['estimated_cost'] = cost accessory_total += cost total_cost = device_total + cable_total + accessory_total bom = BillOfMaterials( network_name=model.name, devices=devices, cables=cables, accessories=accessories, software_licenses=[], total_estimated_cost=total_cost, procurement_links=procurement_links ) # Save BOM self.bom_file.write_text(json.dumps(asdict(bom), indent=2)) logger.info(f"Saved BOM to {self.bom_file}") return bom def stage5_generate_setup_guide(self, model: NetworkModel, bom: BillOfMaterials) -> SetupGuide: """ Stage 5: Generate human deployment guide Includes physical setup + OOB network configuration """ logger.info("Stage 5: Generating setup guide") guide = SetupGuide( network_name=model.name, phases=[ { 'name': 'Physical Installation', 'duration': '4-6 hours', 'prerequisites': ['All equipment received', 'Rack installed', 'Power verified'], 'steps': [ 'Mount devices in rack following layout diagram', 'Connect power cables and verify PDU capacity', 'Install console cables for out-of-band access', 'Label all connections according to diagram' ] }, { 'name': 'Out-of-Band Network Setup', 'duration': '2-3 hours', 'prerequisites': ['Physical installation complete'], 'steps': [ 'Configure management switch with OOB VLAN', 'Connect console server to management network', 'Assign management IPs to all devices', 'Test SSH/console access to each device', 'Document all management IPs and credentials' ] }, { 'name': 'Handoff to Automation', 'duration': '30 minutes', 'prerequisites': ['OOB network operational', 'All devices reachable'], 'steps': [ 'Verify Overgrowth can reach all management IPs', 'Run connectivity test from automation server', 'Start autonomous agent deployment' ] } ], oob_network_config={ 'vlan': 999, 'subnet': '10.255.255.0/24', 'gateway': '10.255.255.1', 'dhcp_range': '10.255.255.100-10.255.255.200', 'dns': ['10.255.255.1'], 'ntp': ['10.255.255.1'] }, safety_checklist=[ 'Power off all equipment before installation', 'Verify proper grounding', 'Check environmental conditions (temp, humidity)', 'Have rollback plan ready', 'Document initial state' ], rollback_plan=[ 'Power down in reverse order of startup', 'Remove configurations and return to factory defaults', 'Restore from backup if configuration was attempted', 'Document what went wrong for post-mortem' ] ) # Save setup guide self.setup_guide_file.write_text(guide.to_markdown()) logger.info(f"Saved setup guide to {self.setup_guide_file}") return guide def stage6_autonomous_deploy(self, model: NetworkModel, credentials: Optional[Dict[str, str]] = None, dry_run: bool = False, parallel: bool = False) -> Dict[str, Any]: """ Stage 6: Autonomous configuration deployment to network devices Generates configs from templates and deploys to real network devices using Netmiko/NAPALM with automatic validation and rollback. Args: model: NetworkModel with device definitions credentials: Device credentials (username, password) dry_run: If True, validate but don't deploy parallel: Use Ray for parallel deployment Returns: Deployment results and summary """ logger.info(f"Stage 6: Starting autonomous deployment (dry_run={dry_run}, parallel={parallel})") from agent.deployment_engine import DeploymentEngine # Use default credentials if none provided if credentials is None: credentials = { 'username': 'admin', 'password': 'admin' } logger.warning("Using default credentials - override via credentials parameter") # Initialize deployment engine deployment_engine = DeploymentEngine( use_napalm=True, use_ray=parallel ) # Build network context for templates network_context = { 'vlans': model.vlans, 'routing': model.routing, 'domain_name': 'overgrowth.local', 'ntp_servers': ['0.pool.ntp.org', '1.pool.ntp.org'], 'dns_servers': ['8.8.8.8', '8.8.4.4'] } # Define validation checks default_pre_checks = [ 'command:show version', # Verify device accessible ] default_post_checks = [ 'command:show running-config', # Verify config applied ] # Deploy to all devices results = [] for device in model.devices: try: result = deployment_engine.generate_and_deploy( device=device, network_context=network_context, credentials=credentials, dry_run=dry_run, pre_checks=default_pre_checks, post_checks=default_post_checks ) results.append({ 'device_id': result.device_id, 'status': result.status.value, 'error': result.error, 'rolled_back': result.rolled_back, 'duration': result.duration_seconds, 'pre_checks_passed': all(result.pre_check_results.values()), 'post_checks_passed': all(result.post_check_results.values()) }) except Exception as e: logger.error(f"Deployment failed for {device.name}: {e}") results.append({ 'device_id': device.name, 'status': 'failed', 'error': str(e) }) # Get summary summary = deployment_engine.get_deployment_summary() # Cleanup deployment_engine.cleanup() return { 'status': 'completed', 'dry_run': dry_run, 'parallel': parallel, 'total_devices': len(model.devices), 'successful': summary['success_count'], 'failed': summary['failed_count'], 'rolled_back': summary['rolled_back_count'], 'success_rate': summary['success_rate'], 'results': results, 'summary': summary } def stage7_observability(self, model: NetworkModel) -> Dict[str, Any]: """ Stage 7: Set up observability stack SuzieQ for multi-vendor state collection and topology discovery """ logger.info("Stage 7: Configuring observability with SuzieQ") results = { 'status': 'partial', 'message': 'SuzieQ state collection configured', 'mock_mode': self.suzieq.mock_mode } # Collect initial network state devices = [{ 'name': d.name, 'ip': d.mgmt_ip, 'username': 'admin', # Would come from secrets/vault 'password': 'admin' } for d in model.devices] if devices: collection = self.suzieq.collect_network_state(devices) results['collection'] = collection logger.info(f"Collected state from {collection.get('devices_polled', 0)} devices") # Discover topology topology = self.suzieq.get_topology() results['topology'] = topology logger.info(f"Discovered {len(topology.get('nodes', []))} nodes in topology") # Get VLAN summary vlan_summary = self.suzieq.get_vlan_summary() results['vlans'] = vlan_summary return results def stage7b_drift_detection(self, model: NetworkModel) -> Dict[str, Any]: """ Stage 7b: Detect configuration drift Compare actual network state vs intended (SoT) """ logger.info("Stage 7b: Running drift detection") # Convert model to dict for comparison intended_state = model.to_dict() # Detect drift drift = self.suzieq.detect_drift(intended_state) results = { 'drift_detected': drift.has_drift, 'drift_score': drift.drift_score, 'devices_checked': drift.devices_checked, 'summary': { 'config_mismatches': len(drift.config_mismatches), 'missing_vlans': len(drift.missing_vlans), 'extra_vlans': len(drift.extra_vlans), 'ip_conflicts': len(drift.ip_conflicts), 'interfaces_down': len(drift.interface_down), 'routing_issues': len(drift.routing_issues) }, 'details': drift.to_dict(), 'mock_mode': self.suzieq.mock_mode } if drift.has_drift: logger.warning(f"Drift detected! Score: {drift.drift_score:.2f}") # Generate remediation plan remediation = self.suzieq.generate_remediation_plan(drift) results['remediation_plan'] = remediation auto_fix_count = sum(1 for r in remediation if r.get('auto_fix')) manual_count = sum(1 for r in remediation if not r.get('auto_fix')) logger.info(f"Remediation plan: {auto_fix_count} auto-fix, {manual_count} manual approval") else: logger.info("✓ No drift detected - network matches SoT") return results def stage8_validation(self, model: NetworkModel) -> Dict[str, Any]: """ Stage 8: Validate actual state matches intended state Continuous reconciliation with automatic remediation """ logger.info("Stage 8: Running validation and reconciliation") from datetime import datetime results = { 'status': 'completed', 'validation_passed': False, 'checks_performed': [] } # Run drift detection drift_results = self.stage7b_drift_detection(model) results['drift_detection'] = drift_results # Check if validation passed drift_score = drift_results.get('drift_score', 0.0) results['validation_passed'] = drift_score < 0.2 # Allow 20% drift tolerance # Generate compliance report compliance = { 'network_name': model.name, 'checked_at': datetime.now().isoformat(), 'drift_score': drift_score, 'status': 'COMPLIANT' if results['validation_passed'] else 'NON_COMPLIANT', 'findings': drift_results.get('summary', {}) } results['compliance_report'] = compliance # Apply auto-remediation if enabled if drift_results.get('remediation_plan'): logger.info("Applying automatic remediation for approved fixes...") remediation_results = self.suzieq.apply_remediation( drift_results['remediation_plan'], auto_approve=True # Only applies auto_fix=True items ) results['remediation'] = remediation_results logger.info(f"Remediation: {remediation_results['applied']} applied, " f"{remediation_results['skipped']} require approval") if results['validation_passed']: logger.info("✓ Validation PASSED - network state matches SoT") else: logger.warning(f"✗ Validation FAILED - drift score {drift_score:.2f} exceeds threshold") return results def run_full_pipeline(self, consultation_input: str) -> Dict[str, Any]: """ Execute the complete pipeline from consultation to production """ logger.info("Starting full Overgrowth pipeline") results = {} # Stage 1: Consultation intent = self.stage1_consultation(consultation_input) results['intent'] = asdict(intent) results['questions'] = self._generate_clarifying_questions(intent) # If the prompt is too short/vague, stop early and ask clarifying questions low_info = len(consultation_input.split()) < 8 or consultation_input.strip().lower() in { "i need a network", "i need a network!", "network", "build a network" } if low_info: results['needs_more_input'] = True return results # Stage 2: Source of Truth model = self.stage2_generate_sot(intent) results['model'] = model.to_dict() # Stage 0: Pre-flight Validation (runs AFTER SoT generation but BEFORE deployment) preflight = self.stage0_preflight(model) results['preflight'] = preflight # Only proceed with deployment if pre-flight passed if not preflight['ready_to_deploy']: logger.warning("Pre-flight validation failed - stopping before deployment") results['deployment_status'] = 'blocked' results['deployment_reason'] = f"{len(preflight['errors'])} validation errors" # Still generate diagrams and BOM for review diagrams = self.stage3_generate_diagram(model) results['diagrams'] = diagrams bom = self.stage4_generate_bom(model) results['bom'] = asdict(bom) results['shopping_list'] = bom.to_shopping_list() # Generate setup guide even when blocked so judges see it guide = self.stage5_generate_setup_guide(model, bom) results['setup_guide'] = guide.to_markdown() return results # Stage 3: Diagrams diagrams = self.stage3_generate_diagram(model) results['diagrams'] = diagrams # Stage 4: Bill of Materials bom = self.stage4_generate_bom(model) results['bom'] = asdict(bom) results['shopping_list'] = bom.to_shopping_list() # Stage 5: Setup Guide guide = self.stage5_generate_setup_guide(model, bom) results['setup_guide'] = guide.to_markdown() # Stages 6-8 results['deployment'] = self.stage6_autonomous_deploy( model=model, credentials=None, # Use defaults dry_run=True # Dry-run by default in full pipeline ) results['observability'] = self.stage7_observability(model) results['validation'] = self.stage8_validation(model) logger.info("Pipeline execution complete") return results def _capture_validation_failure(self, model: NetworkModel, validation_results: Dict[str, Any]): """ Capture validation failure as incident for learning Args: model: Network model that failed validation validation_results: Validation results with errors """ from agent.incident_learning import Incident from datetime import datetime # Extract error summary error_count = len(validation_results.get('errors', [])) error_types = [] if not validation_results.get('schema_valid'): error_types.append("schema validation") if not validation_results.get('policy_passed'): error_types.append("policy violation") if not validation_results.get('batfish_passed', True): error_types.append("batfish analysis") description = f"Pre-flight validation failed: {', '.join(error_types)} ({error_count} errors)" # Create incident incident_id = f"validation-{datetime.now().strftime('%Y%m%d-%H%M%S')}" incident = Incident( id=incident_id, timestamp=datetime.now().isoformat(), severity='medium', category='deployment_failure', description=description, affected_devices=[d.name for d in model.devices], network_model=model.to_dict(), validation_errors=[ {'error': e, 'type': 'validation'} for e in validation_results.get('errors', []) ] ) # Store in database try: self.incident_db.add_incident(incident) logger.info(f"Captured validation failure: {incident_id}") # Trigger learning async (don't block pipeline) # In production, this would be a background job # For now, just log that we would learn from it logger.info(f"Incident {incident_id} queued for root cause analysis") except Exception as e: logger.error(f"Failed to capture incident: {e}") def learn_from_recent_incidents(self, limit: int = 10) -> Dict[str, Any]: """ Analyze recent incidents and generate learnings Args: limit: Number of recent incidents to analyze Returns: Learning summary """ from agent.incident_learning import learn_from_incident # Get recent unresolved incidents incidents = self.incident_db.get_all_incidents(limit=limit) unresolved = [i for i in incidents if not i.resolution] learnings = [] for incident in unresolved[:5]: # Analyze top 5 try: learning = learn_from_incident(incident) learnings.append(learning) logger.info(f"Generated learnings for {incident.id}") except Exception as e: logger.error(f"Failed to learn from {incident.id}: {e}") return { 'total_incidents': len(incidents), 'unresolved': len(unresolved), 'analyzed': len(learnings), 'learnings': learnings } def enable_parallel_mode(self, ray_address: Optional[str] = None): """ Enable parallel execution mode for large-scale operations 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 self.ray_executor.initialize() logger.info(f"Parallel mode enabled - using Ray executor") resources = self.ray_executor.get_cluster_resources() logger.info(f"Available CPUs: {resources['available'].get('CPU', 0)}") def disable_parallel_mode(self): """Disable parallel execution mode""" self.parallel_mode = False if self.ray_executor: self.ray_executor.shutdown() logger.info("Parallel mode disabled") def parallel_deploy_fleet(self, model: NetworkModel, staggered: bool = True, stages: List[float] = [0.01, 0.1, 0.5, 1.0]) -> Dict[str, Any]: """ Deploy configs to entire device fleet in parallel Args: model: Network model with device configurations staggered: Use staggered rollout (canary deployment) stages: Rollout stages as percentages (default: 1%, 10%, 50%, 100%) Returns: Deployment results with progress tracking """ logger.info(f"Starting parallel deployment to {len(model.devices)} devices") if not self.parallel_mode: 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) if not configs: return { 'status': 'error', 'message': 'No configs generated for deployment' } # Mock GNS3 client for testing # In production, would use real GNS3/Netmiko/NAPALM class MockGNS3Client: def apply_config(self, device_id: str, config: str) -> Dict[str, Any]: import time time.sleep(0.1) # Simulate network delay return {'device_id': device_id, 'status': 'deployed'} gns3_client = MockGNS3Client() # Deploy with appropriate strategy if staggered: results, progress = self.ray_executor.staggered_rollout( deployments=configs, gns3_client=gns3_client, stages=stages, validation_fn=None # Could add validation between stages ) else: results, progress = self.ray_executor.parallel_deployment( deployments=configs, gns3_client=gns3_client, batch_size=50 ) # Compile results succeeded = [r for r in results if r.status.value == 'success'] failed = [r for r in results if r.status.value == 'failed'] return { 'status': 'completed' if len(failed) == 0 else 'partial', 'total_devices': len(model.devices), 'succeeded': len(succeeded), 'failed': len(failed), 'failed_devices': [r.device_id for r in failed], 'progress': progress, 'staggered_rollout': staggered, 'stages_used': stages if staggered else None }