""" 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 import os import time import base64 import subprocess import socket from distutils.util import strtobool from .netbox_client import NetBoxClient from agent.network_ops import create_gns3_project, build_network_from_description, get_lab_topology logger = logging.getLogger(__name__) SITE_COUNT_MIN = 1 SITE_COUNT_MAX = 10 @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 self.gns3_server = os.getenv("GNS3_SERVER") self.gns3_web_url = os.getenv("GNS3_WEB_URL") self.gns3_project_name_default = os.getenv("GNS3_PROJECT_NAME") self.gns3_project_id_default = os.getenv("GNS3_PROJECT_ID") self.enable_gns3_build = strtobool(os.getenv("OG_ENABLE_GNS3_BUILD", "1")) == 1 # Enable deployment by default; set OG_DEPLOY_ENABLED=0 to disable in restricted envs. self.deploy_enabled = strtobool(os.getenv("OG_DEPLOY_ENABLED", "1")) == 1 self.ssh_on_seed_fail = strtobool(os.getenv("OG_SSH_ON_SEED_FAIL", "0")) == 1 self.enable_seed = strtobool(os.getenv("OG_ENABLE_SEED", "1")) == 1 self.wg_status = "not_configured" self._ensure_wireguard() def _slugify(self, text: str, length: int = 24) -> str: import re slug = re.sub(r"[^a-zA-Z0-9]+", "-", text.lower()).strip("-") if not slug: slug = "network" return slug[:length] def _parse_site_count(self, text: str) -> int: """ Extract requested site/branch count from the description. Defaults to 3 if not specified. """ import re words = { "one": 1, "two": 2, "three": 3, "four": 4, "five": 5, "six": 6, "seven": 7, "eight": 8, "nine": 9, "ten": 10, } text_lower = text.lower() for word, val in words.items(): if f"{word}-site" in text_lower or f"{word} site" in text_lower: return max(SITE_COUNT_MIN, min(SITE_COUNT_MAX, val)) m = re.search(r"(\d+)[-\s]?site", text_lower) if m: try: return max(SITE_COUNT_MIN, min(SITE_COUNT_MAX, int(m.group(1)))) except Exception: pass m = re.search(r"(\d+)\s*sites?", text_lower) if m: try: return max(SITE_COUNT_MIN, min(SITE_COUNT_MAX, int(m.group(1)))) except Exception: pass m = re.search(r"(\d+)\s*(locations?|branches?)", text_lower) if m: try: return max(SITE_COUNT_MIN, min(SITE_COUNT_MAX, int(m.group(1)))) except Exception: pass m = re.search(r"(\d+)\s*(stores?|shops?)", text_lower) if m: try: return max(SITE_COUNT_MIN, min(SITE_COUNT_MAX, int(m.group(1)))) except Exception: pass return 3 def _site_names(self, site_cnt: int) -> List[str]: # Match the naming used by the GNS3 builder (store clouds) return [f"☕-Store-{idx}" for idx in range(1, site_cnt + 1)] def _count_branch_sites( self, nodes: List[Dict[str, Any]], expected_site_count: Optional[int] = None ) -> Dict[str, Any]: """ Count branch/store sites using strict naming from the retail builder. Only counts clouds named ☕-Store-N and switches named SW-Store-N. """ import re cloud_pat = re.compile(r"^☕-Store-(\d+)$") sw_pat = re.compile(r"^SW-Store-(\d+)$") cloud_sites: List[str] = [] switch_sites: List[str] = [] site_indices: set[int] = set() for node in nodes or []: name = str(node.get("name", "")).strip() node_type = (node.get("node_type") or "").lower() m_cloud = cloud_pat.match(name) m_sw = sw_pat.match(name) if m_cloud and node_type == "cloud": idx = int(m_cloud.group(1)) site_indices.add(idx) cloud_sites.append(name) elif m_sw: idx = int(m_sw.group(1)) site_indices.add(idx) switch_sites.append(name) site_count = len(site_indices) missing: List[str] = [] if expected_site_count: expected_names = [f"☕-Store-{i}" for i in range(1, expected_site_count + 1)] missing = [name for name in expected_names if name not in cloud_sites] return { "count": site_count, "indices": sorted(site_indices), "cloud_sites": sorted(cloud_sites), "switch_sites": sorted(switch_sites), "missing": missing, } def _ensure_wireguard(self): """ If WG_CONFIG_B64 is provided, bring up wg0 using that config. """ self.wg_status = "not_configured" wg_cfg_b64 = os.getenv("WG_CONFIG_B64") if not wg_cfg_b64: return # If wg0 already exists, skip try: already_up = subprocess.run( ["ip", "link", "show", "wg0"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=2, ) if already_up.returncode == 0: logger.info("WireGuard wg0 already up; skipping bring-up") self.wg_status = "up" return except Exception: pass try: cfg_bytes = base64.b64decode(wg_cfg_b64) cfg_path = Path("/tmp/wg0.conf") cfg_path.write_bytes(cfg_bytes) # Lock down perms to avoid wg-quick warnings try: cfg_path.chmod(0o600) except Exception: pass logger.info("Bringing up WireGuard interface wg0 from WG_CONFIG_B64") result = subprocess.run( ["wg-quick", "up", str(cfg_path)], capture_output=True, text=True, timeout=15, ) # Persist debug info for post-run inspection try: dbg_path = self.workspace_dir / "wireguard_debug.txt" dbg_path.write_text( f"cmd: wg-quick up {cfg_path}\n" f"returncode: {result.returncode}\n" f"stdout:\n{result.stdout}\n\nstderr:\n{result.stderr}\n" ) except Exception: pass if result.returncode != 0: stderr = (result.stderr or "").strip() logger.error(f"WireGuard bring-up failed: {stderr}") if "Operation not permitted" in stderr: self.wg_status = "error: wg-quick needs NET_ADMIN on this Space" else: self.wg_status = f"error: {stderr or 'wg-quick failed'}" else: logger.info("WireGuard wg0 up") self.wg_status = "up" except Exception as e: logger.error(f"WireGuard setup error: {e}") self.wg_status = f"error: {e}" def _extract_builder_summary(self, build_resp: Dict[str, Any]) -> Dict[str, Any]: """ Extract builder summary (site counts, nodes) from the MCP response. Handles both parsed dicts and JSON-encoded 'text' fields. """ candidates: List[Dict[str, Any]] = [] if isinstance(build_resp, dict): candidates.append(build_resp) txt = build_resp.get("text") if isinstance(txt, str): try: candidates.append(json.loads(txt)) except Exception: pass raw = build_resp.get("raw", {}) if isinstance(raw, dict): content = raw.get("content") or [] if content and isinstance(content, list) and isinstance(content[0], dict): txt2 = content[0].get("text") if isinstance(txt2, str): try: candidates.append(json.loads(txt2)) except Exception: pass for cand in candidates: if isinstance(cand, dict) and ( "site_count_built" in cand or "site_count_requested" in cand or "nodes" in cand ): return cand return {} def _prepare_lab_info(self, intent: NetworkIntent, create_project: bool = False) -> Dict[str, Any]: """ Return lab link info and optionally create a project via MCP to get the project_id. """ if not self.gns3_server: return {} project_name = self._generate_project_name(intent) requested_project_id = None lab_info: Dict[str, Any] = { "api": self.gns3_server, "project_name": project_name, "project_id": requested_project_id, } if create_project: try: resp = create_gns3_project(project_name=project_name, project_id=requested_project_id) data = resp if isinstance(resp, dict) else {} if not data and isinstance(resp, dict): txt = resp.get("text") or "" if txt: try: data = json.loads(txt) except Exception: data = {} project = (data or {}).get("project", {}) if project: lab_info["project_id"] = project.get("id") or requested_project_id lab_info["project_name"] = project.get("name") or project_name lab_info["create_success"] = bool((data or {}).get("success")) if data.get("error"): lab_info["error"] = data["error"] except Exception as e: lab_info["error"] = str(e) if self.gns3_web_url: lab_info["web_url"] = self.gns3_web_url elif lab_info.get("project_id"): lab_info["web_url"] = f"{self.gns3_server.rstrip('/')}/static/web-ui/server/1/project/{lab_info['project_id']}" else: lab_info["web_url"] = f"{self.gns3_server.rstrip('/')}/static/webUi" return lab_info def _generate_project_name(self, intent: NetworkIntent) -> str: """ Generate a readable, unique project name (LLM-assisted with safe fallback). """ fallback = f"og-{self._slugify(intent.description)}-{int(time.time())}" try: from agent.llm_client import LLMClient, LLMMessage llm = LLMClient() prompt = ( "Create a short, hyphenated project name for a network lab build. " "It should be 3-6 words, lowercase, letters/numbers only, hyphen separated. " "Base it on this request:\n" f"\"{intent.description}\"" ) resp = llm.chat([LLMMessage(role="user", content=prompt)], temperature=0.2, max_tokens=30) # Extract first line, slugify, append timestamp for uniqueness candidate = resp.strip().split("\n")[0] slug = self._slugify(candidate, length=40) if not slug: return fallback return f"{slug}-{int(time.time())}" except Exception as e: logger.warning(f"LLM project name generation failed, using fallback: {e}") return fallback def _ssh_preflight(self, model: NetworkModel, port: int = 22, timeout: int = 3) -> List[Dict[str, Any]]: """Attempt TCP connection to each device mgmt_ip to surface reachability before deployment.""" results = [] for device in model.devices: ip = getattr(device, "mgmt_ip", None) status = {"device": device.name, "ip": ip, "port": port} if not ip: status["status"] = "missing_ip" status["error"] = "No mgmt_ip set" results.append(status) continue try: with socket.create_connection((ip, port), timeout=timeout): status["status"] = "reachable" except Exception as e: status["status"] = "unreachable" status["error"] = str(e) results.append(status) return results def _build_model_from_dict(self, data: Dict[str, Any], description: str, constraints: Optional[List[str]] = None) -> NetworkModel: """ Normalize raw data into a NetworkModel with safe defaults. Used by brownfield import paths (NetBox, GNS3, existing YAML). """ constraints = constraints or [] intent_data = data.get('intent') or {} intent = NetworkIntent( description=intent_data.get('description', description), business_requirements=intent_data.get('business_requirements', ["Maintain current network safely"]), constraints=intent_data.get('constraints', []) + constraints, timeline=intent_data.get('timeline'), budget=intent_data.get('budget') ) allowed_roles = {"core", "distribution", "access", "edge", "firewall", "router", "wireless"} allowed_vendors = {"cisco", "juniper", "arista", "hp", "dell", "ubiquiti", "mikrotik", "other"} role_synonyms = { "ap": "wireless", "access_point": "wireless", "access-point": "wireless", "wifi": "wireless", "server": "access", } devices: List[Device] = [] mgmt_seed = 11 for dev in data.get("devices", []): mgmt_ip = dev.get("mgmt_ip") or f"10.255.0.{mgmt_seed}" mgmt_seed += 1 vendor = (dev.get("vendor") or "other").lower() role = (dev.get("role") or "access").lower().replace(" ", "_") role = role_synonyms.get(role, role) if role not in allowed_roles: role = "access" if vendor not in allowed_vendors: vendor = "other" devices.append( Device( name=dev.get("name", f"device-{mgmt_seed}"), role=role, model=dev.get("model", "Unknown"), vendor=vendor, mgmt_ip=mgmt_ip, location=dev.get("location", "unspecified"), interfaces=dev.get("interfaces", []) ) ) if not devices: devices.append( Device( name="imported-edge-01", role="edge", model="Imported", vendor="other", mgmt_ip="10.255.0.10", location="unspecified", interfaces=[] ) ) vlans = data.get("vlans") or [{ "id": 999, "name": "brownfield_mgmt", "subnet": "10.255.0.0/24", "purpose": "Imported management network" }] subnets = data.get("subnets") or [{ "network": "10.255.0.0/24", "gateway": "10.255.0.1", "vlan": vlans[0]["id"], "purpose": "Brownfield management overlay" }] routing = data.get("routing") or {"protocol": "static", "networks": [s.get("network") for s in subnets if s.get("network")]} services = data.get("services") or ["DHCP", "DNS", "NTP"] return NetworkModel( name=data.get("name") or "brownfield_network", version=data.get("version") or "1.0.0", intent=intent, devices=devices, vlans=vlans, subnets=subnets, routing=routing, services=services ) def _model_from_topology(self, topology: Dict[str, Any]) -> Dict[str, Any]: """Convert a GNS3 topology snapshot into a basic NetworkModel dict.""" nodes = topology.get("nodes", []) devices = [] mgmt_seed = 11 for node in nodes: node_type = (node.get("node_type") or "").lower() name = node.get("name", f"node-{mgmt_seed}") role = "access" if "fw" in name.lower() or "firewall" in name.lower(): role = "firewall" elif node_type in {"router", "qemu"}: role = "router" elif node_type in {"switch", "sw"}: role = "distribution" elif "ap" in name.lower(): role = "wireless" devices.append({ "name": name, "role": role, "model": node.get("properties", {}).get("symbol") or "GNS3 node", "vendor": "other", "mgmt_ip": f"10.254.0.{mgmt_seed}", "location": topology.get("project", "gns3"), "interfaces": [] }) mgmt_seed += 1 vlan = { "id": 999, "name": "brownfield_mgmt", "subnet": "10.254.0.0/24", "purpose": "Imported from GNS3" } subnet = { "network": "10.254.0.0/24", "gateway": "10.254.0.1", "vlan": vlan["id"], "purpose": "Management overlay" } return { "name": topology.get("project", "gns3-import"), "version": "1.0.0", "devices": devices, "vlans": [vlan], "subnets": [subnet], "routing": {"protocol": "static", "networks": [subnet["network"]]}, "services": ["DHCP", "DNS", "NTP"], "links": topology.get("links", []) } def run_brownfield_pipeline( self, consultation_input: str, source: str = "netbox", project_name: Optional[str] = None, sot_path: Optional[str] = None, simulation_only: bool = True, read_only_validation: bool = True ) -> Dict[str, Any]: """ Brownfield/read-only workflow: import an existing network and validate without pushing changes. Supports NetBox, GNS3, or an existing SoT file. """ logger.info(f"Starting brownfield pipeline from source={source}") source = (source or "netbox").lower() import_summary = {"source": source, "project": project_name} topology = None raw_model: Dict[str, Any] = {} if source == "netbox": if self.use_netbox and not self.netbox.mock_mode: raw_model = self.netbox.export_network_model() import_summary["note"] = "Imported from NetBox/Nautobot" else: logger.warning("NetBox unavailable; falling back to local SoT file") import_summary["note"] = "NetBox unavailable - using local SoT" source = "file" if source == "gns3": try: from agent.network_ops import get_lab_topology topo_name = project_name or os.getenv("GNS3_PROJECT_NAME", "overgrowth") topology = get_lab_topology(topo_name) raw_model = self._model_from_topology(topology) import_summary["note"] = f"Discovered topology from GNS3 project '{topo_name}'" except Exception as e: logger.error(f"Failed to import from GNS3: {e}") import_summary["note"] = f"GNS3 import failed: {e}" if source == "file" or not raw_model: path = Path(sot_path or self.sot_file) try: content = path.read_text() if path.suffix.lower() in [".yaml", ".yml"]: raw_model = yaml.safe_load(content) or {} else: raw_model = json.loads(content) import_summary["note"] = f"Loaded SoT from {path}" except Exception as e: logger.error(f"Could not load SoT from {path}: {e}") import_summary["note"] = f"File load failed ({path}): {e}" raw_model = raw_model or {} model = self._build_model_from_dict( raw_model, description=consultation_input or "Brownfield validation", constraints=["brownfield-import"] ) # Persist SoT snapshot even in read-only mode self.sot_file.write_text(model.to_yaml()) logger.info(f"Saved imported source of truth to {self.sot_file}") results: Dict[str, Any] = { "mode": { "source": source, "simulation_only": simulation_only, "read_only_validation": read_only_validation, "import_summary": import_summary.get("note") }, "intent": asdict(model.intent), "questions": [], "import_summary": import_summary, "wg_status": self.wg_status } lab_info = self._prepare_lab_info(model.intent, create_project=False) if lab_info: results["lab"] = lab_info results["model"] = model.to_dict() results["preflight"] = self.stage0_preflight(model) diagrams = self.stage3_generate_diagram(model) # Log mermaid for debugging try: mermaid_dbg = diagrams.get("mermaid") if mermaid_dbg: logger.info(f"Mermaid diagram:\n{mermaid_dbg}") try: dbg_file = self.workspace_dir / "mermaid_debug.mmd" dbg_file.write_text(mermaid_dbg) except Exception: pass except Exception: pass if topology: diagrams["ascii"] = topology.get("ascii_diagram", diagrams.get("ascii")) diagrams["mermaid"] = topology.get("mermaid_diagram", diagrams.get("mermaid")) diagrams["summary"] = topology.get("summary", diagrams.get("summary")) results["diagrams"] = {"ascii": "Diagrams disabled; use GNS3 UI."} bom = self.stage4_generate_bom(model) results["bom"] = asdict(bom) results["shopping_list"] = bom.to_shopping_list() guide = self.stage5_generate_setup_guide(model, bom) results["setup_guide"] = guide.to_markdown() # Deployment is skipped unless explicitly requested and validation passed if (not simulation_only) and results["preflight"].get("ready_to_deploy") and (not read_only_validation) and self.deploy_enabled: results["deployment"] = self.stage6_autonomous_deploy( model=model, credentials=None, dry_run=False, lab_info=lab_info ) if lab_info: results["deployment"]["lab"] = lab_info else: results["deployment_status"] = "skipped" if not self.deploy_enabled: results["deployment_reason"] = "deployment_disabled" elif simulation_only: results["deployment_reason"] = "simulation_only" elif read_only_validation: results["deployment_reason"] = "read_only_validation" results["observability"] = self.stage7_observability(model) results["validation"] = self.stage8_validation( model, apply_remediation=not read_only_validation ) return results 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() site_count = self._parse_site_count(intent.description) def _default_design(site_cnt: int) -> Dict[str, Any]: """Deterministic fallback design with concrete values for offline/demo runs.""" sites = [] 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"}, ] for idx in range(1, site_cnt + 1): site = f"Site{idx}" sites.append(site) devices.append({ "name": f"{site.lower()}-wan", "role": "edge", "model": "Cisco ISR 1100", "vendor": "cisco", "mgmt_ip": f"10.10.{10+idx}.31", "location": site }) devices.extend([ {"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"}, ]) 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": "Corp"}, {"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"}, ] for idx, site in enumerate(sites, start=1): subnets.append({ "network": f"10.{60+idx}.0.0/24", "gateway": f"10.{60+idx}.0.1", "vlan": 200 + idx, "purpose": f"{site} LAN" }) 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": subnets, "devices": devices, "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(site_count) 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(site_count) 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 allowed_roles = {"core", "distribution", "access", "edge", "firewall", "router", "wireless"} allowed_vendors = {"cisco", "juniper", "arista", "hp", "dell", "ubiquiti", "mikrotik", "other"} role_synonyms = { "ap": "wireless", "access_point": "wireless", "access-point": "wireless", "wifi": "wireless", "server": "access", } 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().replace(" ", "_") role = role_synonyms.get(role, role) if role not in allowed_roles: role = "access" if vendor not in allowed_vendors: vendor = "other" 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", []) ) ) # Ensure we have at least the requested number of branches/sites branch_devices = [d for d in devices if d.role == "edge" or "branch" in d.location.lower() or "site" in d.location.lower()] missing_branches = max(0, site_count - len(branch_devices)) for idx in range(1, missing_branches + 1): site_idx = len(branch_devices) + idx site_name = f"Site{site_idx}" devices.append( Device( name=f"{site_name.lower()}-wan", role="edge", model="Cisco ISR 1100", vendor="cisco", mgmt_ip=f"10.10.{10+site_idx}.50", location=site_name, interfaces=[] ) ) # Add per-site subnets if missing if model.subnets is not None: existing_purposes = {s.get("purpose", "").lower() for s in model.subnets if isinstance(s, dict)} for idx in range(1, site_count + 1): purpose = f"site{idx} lan" if purpose not in existing_purposes: model.subnets.append({ "network": f"10.{70+idx}.0.0/24", "gateway": f"10.{70+idx}.0.1", "vlan": 300 + idx, "purpose": purpose }) # If no devices came through, fall back again to deterministic set if not devices: fallback = _default_design(site_count)["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: Diagram generation disabled for hackathon. Use the GNS3 project UI for topology visualization. """ logger.info("Stage 3: Diagram generation disabled (use GNS3 UI)") return {"ascii": "Diagrams disabled; use GNS3 project UI."} 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, lab_info: Optional[Dict[str, Any]] = None) -> 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})") deploy_mode = os.getenv("OG_DEPLOY_MODE", "hybrid").lower() project_for_seed = None if lab_info and lab_info.get("project_name"): project_for_seed = { "name": lab_info.get("project_name"), "id": lab_info.get("project_id") } # Deployment credentials from environment env_username = os.getenv("DEPLOY_USERNAME") env_password = os.getenv("DEPLOY_PASSWORD") env_port = int(os.getenv("DEPLOY_PORT", "22")) if credentials is None: credentials = { 'username': env_username or 'admin', 'password': env_password or 'admin' } ssh_preflight = self._ssh_preflight(model, port=env_port) 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 = [] seed_results = [] for device in model.devices: try: # Seed via GNS3 if available and mode allows seeded = False if self.enable_seed and deploy_mode in ("hybrid", "gns3") and project_for_seed: from agent.network_ops import seed_gns3_node_config # Generate config for seeding via GNS3 cfg_text = deployment_engine.template_engine.generate_device_config( device=device, network_context=network_context ) seed_resp = seed_gns3_node_config( project_name=project_for_seed["name"], device_name=device.name, config=cfg_text ) seed_results.append({ "device": device.name, "seed_success": bool(seed_resp.get("success")), "seed_error": seed_resp.get("error"), "project_id": seed_resp.get("project_id"), }) seeded = bool(seed_resp.get("success")) # Skip SSH only if deploy_mode is gns3-only if deploy_mode == "gns3": results.append({ 'device_id': device.name, 'status': 'seeded', 'error': None, 'rolled_back': False, 'duration': 0, 'pre_checks_passed': bool(seed_resp.get("success")) if self.enable_seed else False, 'post_checks_passed': bool(seed_resp.get("success")) if self.enable_seed else False }) continue 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, 'seed_results': seed_results, 'deploy_mode': deploy_mode, 'ssh_preflight': ssh_preflight } 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, apply_remediation: bool = True) -> 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': [], 'read_only': not apply_remediation } # 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 apply_remediation and 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") elif drift_results.get('remediation_plan'): # Document skipped remediation when running in read-only mode results['remediation'] = { 'applied': 0, 'skipped': len(drift_results['remediation_plan']), 'reason': 'read-only validation mode' } 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) lab_info = self._prepare_lab_info(intent, create_project=True) results['wg_status'] = self.wg_status if lab_info: results['lab'] = lab_info site_count = self._parse_site_count(intent.description) site_names = self._site_names(site_count) results["requested_site_count"] = site_count if lab_info is not None: lab_info["requested_site_count"] = site_count lab_info["site_names"] = site_names # 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() # Build GNS3 lab if enabled and server configured def _nodes_from_resp(build_resp: Dict[str, Any]) -> List[Dict[str, Any]]: if not isinstance(build_resp, dict): return [] if "nodes" in build_resp and isinstance(build_resp["nodes"], list): return build_resp["nodes"] topo = build_resp.get("topology") if isinstance(topo, dict) and isinstance(topo.get("nodes"), list): return topo["nodes"] txt = build_resp.get("text") if isinstance(build_resp, dict) else None if txt: try: data = json.loads(txt) if isinstance(data, dict): if "nodes" in data and isinstance(data["nodes"], list): return data["nodes"] topo2 = data.get("topology", {}) if isinstance(topo2, dict) and isinstance(topo2.get("nodes"), list): return topo2["nodes"] except Exception: return [] return [] if self.gns3_server and self.enable_gns3_build and lab_info: builder_site_count: Optional[int] = None try: base_desc = ( intent.description + f"\nBuild EXACTLY {site_count} branch/site nodes named {', '.join(site_names)} plus HQ." + f" Each site must include an edge router and POS + WiFi + Cameras hosts." + " Do not omit any site. Power on all nodes." + " Use simple images; fewer image types is fine." ) build_resp = build_network_from_description( description=base_desc, project_name=lab_info.get("project_name") or "overgrowth", auto_configure=True, site_count=site_count, ) lab_info["build_result"] = build_resp nodes = _nodes_from_resp(build_resp) builder_summary = self._extract_builder_summary(build_resp) builder_site_count = builder_summary.get("site_count_built") # Refresh project identifiers from builder summary if available so UI links are correct. if builder_summary.get("project_id"): lab_info["project_id"] = builder_summary.get("project_id") if builder_summary.get("project_name"): lab_info["project_name"] = builder_summary.get("project_name") lab_info["builder_site_count"] = builder_site_count results["builder_site_count"] = builder_site_count lab_info["builder_summary"] = builder_summary site_summary_resp = self._count_branch_sites(nodes, expected_site_count=site_count) lab_info["initial_site_count"] = site_summary_resp.get("count") lab_info["initial_missing_sites"] = site_summary_resp.get("missing", []) except Exception as e: lab_info["build_error"] = str(e) try: topology = get_lab_topology(lab_info.get("project_name") or "overgrowth") site_summary = self._count_branch_sites(topology.get("nodes", []), expected_site_count=site_count) topology_count = site_summary["count"] missing_sites = site_summary.get("missing") or [] lab_info["topology_site_count"] = topology_count results["topology_site_count"] = topology_count lab_info["missing_sites"] = missing_sites lab_info["site_nodes"] = site_summary["cloud_sites"] or site_summary["switch_sites"] lab_info["topology"] = { "nodes": topology.get("nodes", []), "links": topology.get("links", []), } actual_count = builder_site_count if builder_site_count is not None else topology_count lab_info["actual_site_count"] = actual_count results["actual_site_count"] = actual_count mismatch = None if ( builder_site_count is not None and builder_site_count != site_count ) or (topology_count is not None and topology_count != site_count): mismatch = { "requested": site_count, "builder": builder_site_count, "topology": topology_count, } elif builder_site_count is not None and topology_count is not None and builder_site_count != topology_count: mismatch = { "requested": site_count, "builder": builder_site_count, "topology": topology_count, } if mismatch: lab_info["site_count_mismatch"] = mismatch results["site_count_mismatch"] = mismatch except Exception as e: lab_info["topology_error"] = str(e) # Update web_url now that project_id/project_name may have been refreshed. try: if lab_info.get("api") and lab_info.get("project_id"): lab_info["web_url"] = f"{lab_info['api'].rstrip('/')}/static/web-ui/server/1/project/{lab_info['project_id']}" except Exception: pass results["lab"] = lab_info # 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 if self.deploy_enabled: results['deployment'] = self.stage6_autonomous_deploy( model=model, credentials=None, # Use defaults dry_run=True, # Dry-run by default in full pipeline lab_info=lab_info ) lab_info = results.get('lab') if lab_info: results['deployment']['lab'] = lab_info else: results['deployment_status'] = 'skipped' results['deployment_reason'] = 'deployment_disabled' 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 }