feat: Add Stage 0 pre-flight validation (Todo #2)

- Created schema_validation.py with Pydantic models for VLANs, subnets, devices, routing
- Validates VLAN IDs (1-4094), subnet CIDR notation, gateway within network, no duplicates
- Created policy_engine.py enforcing industry best practices:
* RFC1918 private addressing
* No VLAN 1 in production
* Naming conventions (hostnames, VLANs)
* Security (guest isolation, management VLAN)
* Design (redundancy, DHCP ranges, routing protocols)
- Integrated stage0_preflight() into pipeline - runs AFTER SoT generation, BEFORE deployment
- Pipeline blocks deployment if validation errors exist
- Updated UI to show pre-flight results (errors, warnings, info)
- Added pydantic>=2.0.0 to requirements
- Comprehensive test suites (test_validation.py, test_preflight.py) - all passing

Pre-flight prevents bad configs from ever touching devices. Schema catches typos/ranges,
policy enforces security/design standards. Batfish integration pending for static analysis.

Phase 2 of research recommendations - 2 of 6 todos complete.

agent/pipeline_engine.py

@@ -180,6 +180,73 @@ class OvergrowthPipeline:
                 logger.warning("NetBox not available - falling back to YAML files")
                 self.use_netbox = 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 (TODO)
+        # TODO: Implement Batfish integration
+        # - Snapshot network configs
+        # - Validate reachability
+        # - Check ACL behavior
+        # - Find routing loops
+        results['batfish_status'] = 'not_implemented'
+        
+        # Overall result
+        results['ready_to_deploy'] = results['schema_valid'] and results['policy_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")
+        
+        return results
+    
     def stage1_consultation(self, user_input: str) -> NetworkIntent:
         """
         Stage 1: Capture user intent from natural language
@@ -571,6 +638,25 @@ Be specific and practical. Use RFC1918 addressing. Consider scalability and secu
         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)
+            
+            return results
+        
         # Stage 3: Diagrams
         diagrams = self.stage3_generate_diagram(model)
         results['diagrams'] = diagrams

agent/policy_engine.py

@@ -0,0 +1,353 @@
+"""
+Network Policy Engine
+Enforces security, naming, and design best practices
+"""
+
+from typing import List, Dict, Any, Tuple
+from ipaddress import ip_network, ip_address
+import re
+
+
+class PolicyViolation:
+    """Represents a policy violation"""
+    def __init__(self, severity: str, category: str, message: str, location: str = ""):
+        self.severity = severity  # ERROR, WARNING, INFO
+        self.category = category  # security, naming, design, addressing
+        self.message = message
+        self.location = location
+    
+    def __str__(self):
+        loc = f" [{self.location}]" if self.location else ""
+        return f"{self.severity} ({self.category}){loc}: {self.message}"
+
+
+class NetworkPolicy:
+    """
+    Enforces network design policies
+    Checks against industry best practices and organizational standards
+    """
+    
+    def __init__(self, config: Dict[str, Any] = None):
+        """
+        Initialize policy engine with configuration
+        
+        Default policies:
+        - RFC1918 private addressing required
+        - VLAN 1 not allowed for production
+        - Management VLAN required
+        - Secure naming conventions
+        - Gateway is first usable IP in subnet
+        """
+        self.config = config or {}
+        self.violations: List[PolicyViolation] = []
+    
+    def check_network_model(self, model: Dict[str, Any]) -> List[PolicyViolation]:
+        """
+        Run all policy checks on network model
+        Returns list of violations
+        """
+        self.violations = []
+        
+        # Naming policies
+        self._check_naming_conventions(model)
+        
+        # Addressing policies
+        self._check_addressing_policies(model)
+        
+        # VLAN policies
+        self._check_vlan_policies(model)
+        
+        # Security policies
+        self._check_security_policies(model)
+        
+        # Design best practices
+        self._check_design_practices(model)
+        
+        return self.violations
+    
+    def _check_naming_conventions(self, model: Dict[str, Any]):
+        """Check naming follows standards"""
+        # Device naming
+        for device in model.get('devices', []):
+            name = device.get('name', '')
+            role = device.get('role', '')
+            
+            # Should include role in name
+            if role and role not in name.lower():
+                self.violations.append(PolicyViolation(
+                    severity="WARNING",
+                    category="naming",
+                    message=f"Device name should indicate role '{role}'",
+                    location=f"device:{name}"
+                ))
+            
+            # Should use hyphens not underscores
+            if '_' in name:
+                self.violations.append(PolicyViolation(
+                    severity="INFO",
+                    category="naming",
+                    message="Consider using hyphens instead of underscores in device names",
+                    location=f"device:{name}"
+                ))
+            
+            # Check for sequential numbering
+            if not re.search(r'\d{2}$', name):
+                self.violations.append(PolicyViolation(
+                    severity="INFO",
+                    category="naming",
+                    message="Device name should end with 2-digit number for scalability",
+                    location=f"device:{name}"
+                ))
+        
+        # VLAN naming
+        for vlan in model.get('vlans', []):
+            name = vlan.get('name', '')
+            
+            # No spaces in VLAN names
+            if ' ' in name:
+                self.violations.append(PolicyViolation(
+                    severity="WARNING",
+                    category="naming",
+                    message="VLAN names should not contain spaces",
+                    location=f"vlan:{vlan.get('id')}"
+                ))
+            
+            # Should be descriptive
+            if len(name) < 3:
+                self.violations.append(PolicyViolation(
+                    severity="INFO",
+                    category="naming",
+                    message="VLAN name should be descriptive (3+ characters)",
+                    location=f"vlan:{vlan.get('id')}"
+                ))
+    
+    def _check_addressing_policies(self, model: Dict[str, Any]):
+        """Check IP addressing follows best practices"""
+        
+        # Check for RFC1918 private addressing
+        require_private = self.config.get('require_rfc1918', True)
+        
+        for subnet in model.get('subnets', []):
+            network = subnet.get('network', '')
+            gateway = subnet.get('gateway', '')
+            
+            try:
+                net = ip_network(network, strict=False)
+                
+                # Check if using private addressing
+                if require_private and not net.is_private:
+                    self.violations.append(PolicyViolation(
+                        severity="ERROR",
+                        category="addressing",
+                        message=f"Non-private address space detected: {network} (use RFC1918: 10/8, 172.16/12, 192.168/16)",
+                        location=f"subnet:{network}"
+                    ))
+                
+                # Check gateway is first usable IP
+                if gateway:
+                    gw = ip_address(gateway)
+                    first_usable = list(net.hosts())[0] if net.num_addresses > 2 else None
+                    
+                    if first_usable and gw != first_usable:
+                        self.violations.append(PolicyViolation(
+                            severity="INFO",
+                            category="addressing",
+                            message=f"Gateway {gateway} is not first usable IP ({first_usable})",
+                            location=f"subnet:{network}"
+                        ))
+                
+                # Warn on wasteful subnets (e.g., /24 for 2 devices)
+                if net.num_addresses > 256:
+                    self.violations.append(PolicyViolation(
+                        severity="INFO",
+                        category="addressing",
+                        message=f"Large subnet ({net.num_addresses} IPs) - consider smaller subnets for better security segmentation",
+                        location=f"subnet:{network}"
+                    ))
+                
+            except Exception as e:
+                self.violations.append(PolicyViolation(
+                    severity="ERROR",
+                    category="addressing",
+                    message=f"Invalid subnet: {e}",
+                    location=f"subnet:{network}"
+                ))
+        
+        # Check for overlapping subnets
+        subnets_list = [s.get('network') for s in model.get('subnets', [])]
+        for i, subnet1 in enumerate(subnets_list):
+            for subnet2 in subnets_list[i+1:]:
+                try:
+                    net1 = ip_network(subnet1, strict=False)
+                    net2 = ip_network(subnet2, strict=False)
+                    if net1.overlaps(net2):
+                        self.violations.append(PolicyViolation(
+                            severity="ERROR",
+                            category="addressing",
+                            message=f"Overlapping subnets: {subnet1} and {subnet2}",
+                            location="subnets"
+                        ))
+                except Exception:
+                    pass
+    
+    def _check_vlan_policies(self, model: Dict[str, Any]):
+        """Check VLAN configuration policies"""
+        
+        vlan_ids = [v.get('id') for v in model.get('vlans', [])]
+        vlan_names = {v.get('id'): v.get('name', '') for v in model.get('vlans', [])}
+        
+        # VLAN 1 should not be used
+        if 1 in vlan_ids:
+            self.violations.append(PolicyViolation(
+                severity="WARNING",
+                category="security",
+                message="VLAN 1 detected - default VLAN should not be used for production",
+                location="vlan:1"
+            ))
+        
+        # Check for management VLAN
+        mgmt_vlans = [v for v in model.get('vlans', []) 
+                     if 'mgmt' in v.get('name', '').lower() or 'management' in v.get('name', '').lower()]
+        
+        if not mgmt_vlans and len(model.get('devices', [])) > 0:
+            self.violations.append(PolicyViolation(
+                severity="WARNING",
+                category="design",
+                message="No dedicated management VLAN found - consider creating one for device management",
+                location="vlans"
+            ))
+        
+        # Check VLAN ID ranges (common practice: 10-99 infrastructure, 100+ user)
+        for vlan in model.get('vlans', []):
+            vid = vlan.get('id')
+            name = vlan.get('name', '')
+            
+            if vid and vid >= 1006 and vid <= 1024:
+                self.violations.append(PolicyViolation(
+                    severity="WARNING",
+                    category="design",
+                    message=f"VLAN {vid} is in extended range reserved range - may not be supported on all devices",
+                    location=f"vlan:{vid}"
+                ))
+    
+    def _check_security_policies(self, model: Dict[str, Any]):
+        """Check security best practices"""
+        
+        # Check for guest network isolation
+        guest_vlans = [v for v in model.get('vlans', []) 
+                      if 'guest' in v.get('name', '').lower() or 'visitor' in v.get('name', '').lower()]
+        
+        if guest_vlans:
+            # Guest networks should be isolated (different subnet range)
+            guest_subnets = [s for s in model.get('subnets', []) 
+                           if any(s.get('vlan') == gv.get('id') for gv in guest_vlans)]
+            
+            if not guest_subnets:
+                self.violations.append(PolicyViolation(
+                    severity="WARNING",
+                    category="security",
+                    message="Guest VLAN exists but no dedicated subnet configured",
+                    location="guest network"
+                ))
+        
+        # Check for DHCP snooping (implied by DHCP configuration)
+        for subnet in model.get('subnets', []):
+            if subnet.get('dhcp_enabled'):
+                # Should have DHCP range defined
+                if not subnet.get('dhcp_range_start') or not subnet.get('dhcp_range_end'):
+                    self.violations.append(PolicyViolation(
+                        severity="WARNING",
+                        category="design",
+                        message="DHCP enabled but no pool range defined",
+                        location=f"subnet:{subnet.get('network')}"
+                    ))
+        
+        # Warn if no redundancy in design
+        core_devices = [d for d in model.get('devices', []) if d.get('role') == 'core']
+        if len(core_devices) < 2 and len(model.get('devices', [])) > 3:
+            self.violations.append(PolicyViolation(
+                severity="WARNING",
+                category="design",
+                message="No redundant core devices - consider adding redundancy for HA",
+                location="devices"
+            ))
+    
+    def _check_design_practices(self, model: Dict[str, Any]):
+        """Check general design best practices"""
+        
+        # Every subnet should have a VLAN
+        vlans_with_subnets = {s.get('vlan') for s in model.get('subnets', []) if s.get('vlan')}
+        all_vlans = {v.get('id') for v in model.get('vlans', [])}
+        
+        vlans_without_subnets = all_vlans - vlans_with_subnets
+        if vlans_without_subnets:
+            self.violations.append(PolicyViolation(
+                severity="INFO",
+                category="design",
+                message=f"VLANs without subnets: {sorted(vlans_without_subnets)}",
+                location="vlans/subnets"
+            ))
+        
+        # Check routing protocol makes sense for network size
+        routing = model.get('routing')
+        device_count = len(model.get('devices', []))
+        
+        if routing:
+            protocol = routing.get('protocol')
+            
+            if protocol == 'ospf' and device_count < 3:
+                self.violations.append(PolicyViolation(
+                    severity="INFO",
+                    category="design",
+                    message="OSPF may be overkill for small network (<3 devices) - consider static routing",
+                    location="routing"
+                ))
+            
+            if protocol == 'static' and device_count > 10:
+                self.violations.append(PolicyViolation(
+                    severity="WARNING",
+                    category="design",
+                    message="Static routing challenging for large networks - consider dynamic protocol",
+                    location="routing"
+                ))
+        
+        # Check for services configuration
+        services = model.get('services', [])
+        recommended_services = {'DHCP', 'DNS', 'NTP'}
+        missing_services = recommended_services - set(services)
+        
+        if missing_services:
+            self.violations.append(PolicyViolation(
+                severity="INFO",
+                category="design",
+                message=f"Consider adding services: {', '.join(missing_services)}",
+                location="services"
+            ))
+    
+    def get_violations_by_severity(self) -> Dict[str, List[PolicyViolation]]:
+        """Group violations by severity"""
+        result = {'ERROR': [], 'WARNING': [], 'INFO': []}
+        for v in self.violations:
+            result[v.severity].append(v)
+        return result
+    
+    def has_errors(self) -> bool:
+        """Check if there are any ERROR-level violations"""
+        return any(v.severity == 'ERROR' for v in self.violations)
+    
+    def format_violations(self) -> str:
+        """Format violations as readable text"""
+        if not self.violations:
+            return "✓ No policy violations detected"
+        
+        lines = []
+        by_severity = self.get_violations_by_severity()
+        
+        for severity in ['ERROR', 'WARNING', 'INFO']:
+            viols = by_severity[severity]
+            if viols:
+                lines.append(f"\n{severity}S ({len(viols)}):")
+                for v in viols:
+                    lines.append(f"  • {v}")
+        
+        return "\n".join(lines)

agent/schema_validation.py

@@ -0,0 +1,319 @@
+"""
+Schema Validation for Network Models
+Uses Pydantic for type checking and data validation
+"""
+
+from typing import List, Optional, Dict, Any, Literal
+from pydantic import BaseModel, Field, validator, IPvAnyAddress, IPvAnyNetwork
+from ipaddress import ip_network, ip_address
+import re
+
+
+class VLANModel(BaseModel):
+    """VLAN configuration schema"""
+    id: int = Field(..., ge=1, le=4094, description="VLAN ID (1-4094)")
+    name: str = Field(..., min_length=1, max_length=32, description="VLAN name")
+    purpose: Optional[str] = Field(None, description="VLAN purpose/description")
+    subnet: Optional[str] = Field(None, description="Associated subnet in CIDR notation")
+    
+    @validator('name')
+    def validate_vlan_name(cls, v):
+        """Ensure VLAN name follows naming conventions"""
+        # No spaces, special characters
+        if not re.match(r'^[a-zA-Z0-9_-]+$', v):
+            raise ValueError("VLAN name must contain only letters, numbers, underscore, or hyphen")
+        return v
+    
+    @validator('subnet')
+    def validate_subnet_format(cls, v):
+        """Validate CIDR notation"""
+        if v:
+            try:
+                ip_network(v, strict=False)
+            except ValueError as e:
+                raise ValueError(f"Invalid subnet format: {e}")
+        return v
+    
+    class Config:
+        schema_extra = {
+            "example": {
+                "id": 10,
+                "name": "Management",
+                "purpose": "Network management and monitoring",
+                "subnet": "10.0.10.0/24"
+            }
+        }
+
+
+class SubnetModel(BaseModel):
+    """IP Subnet schema"""
+    network: str = Field(..., description="Network address in CIDR notation")
+    gateway: str = Field(..., description="Default gateway IP address")
+    vlan: Optional[int] = Field(None, ge=1, le=4094, description="Associated VLAN ID")
+    purpose: Optional[str] = Field(None, description="Subnet purpose")
+    dhcp_enabled: bool = Field(False, description="DHCP server enabled")
+    dhcp_range_start: Optional[str] = Field(None, description="DHCP pool start IP")
+    dhcp_range_end: Optional[str] = Field(None, description="DHCP pool end IP")
+    
+    @validator('network')
+    def validate_network(cls, v):
+        """Ensure valid CIDR notation"""
+        try:
+            net = ip_network(v, strict=False)
+            # Ensure it's a private network (RFC1918) unless explicitly public
+            # if not net.is_private:
+            #     raise ValueError("Use RFC1918 private addressing (10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16)")
+            return str(net)
+        except ValueError as e:
+            raise ValueError(f"Invalid network: {e}")
+    
+    @validator('gateway')
+    def validate_gateway(cls, v, values):
+        """Ensure gateway is within the network"""
+        try:
+            gw = ip_address(v)
+            if 'network' in values:
+                net = ip_network(values['network'], strict=False)
+                if gw not in net:
+                    raise ValueError(f"Gateway {v} not in network {values['network']}")
+        except ValueError as e:
+            raise ValueError(f"Invalid gateway: {e}")
+        return v
+    
+    @validator('dhcp_range_end')
+    def validate_dhcp_range(cls, v, values):
+        """Ensure DHCP range is valid"""
+        if v and 'dhcp_range_start' in values and values['dhcp_range_start']:
+            start = ip_address(values['dhcp_range_start'])
+            end = ip_address(v)
+            if end <= start:
+                raise ValueError("DHCP range end must be greater than start")
+            
+            # Ensure both are in the network
+            if 'network' in values:
+                net = ip_network(values['network'], strict=False)
+                if start not in net or end not in net:
+                    raise ValueError("DHCP range must be within subnet")
+        return v
+    
+    class Config:
+        schema_extra = {
+            "example": {
+                "network": "10.0.10.0/24",
+                "gateway": "10.0.10.1",
+                "vlan": 10,
+                "purpose": "Management network",
+                "dhcp_enabled": True,
+                "dhcp_range_start": "10.0.10.100",
+                "dhcp_range_end": "10.0.10.200"
+            }
+        }
+
+
+class InterfaceModel(BaseModel):
+    """Network interface schema"""
+    name: str = Field(..., description="Interface name (e.g., GigabitEthernet1/0/1)")
+    type: Literal["ethernet", "management", "loopback", "vlan", "port-channel"] = Field(..., description="Interface type")
+    speed: Optional[str] = Field(None, description="Interface speed (e.g., 1000, 10G)")
+    mode: Optional[Literal["access", "trunk"]] = Field(None, description="Switchport mode")
+    vlan: Optional[int] = Field(None, description="Access VLAN or native VLAN for trunk")
+    allowed_vlans: Optional[List[int]] = Field(None, description="Allowed VLANs for trunk ports")
+    ip_address: Optional[str] = Field(None, description="IP address if L3 interface")
+    description: Optional[str] = Field(None, max_length=240, description="Interface description")
+    enabled: bool = Field(True, description="Interface administratively up")
+    
+    @validator('allowed_vlans')
+    def validate_allowed_vlans(cls, v):
+        """Ensure VLAN IDs are valid"""
+        if v:
+            for vlan_id in v:
+                if vlan_id < 1 or vlan_id > 4094:
+                    raise ValueError(f"Invalid VLAN ID {vlan_id} (must be 1-4094)")
+        return v
+    
+    class Config:
+        schema_extra = {
+            "example": {
+                "name": "GigabitEthernet1/0/1",
+                "type": "ethernet",
+                "speed": "1000",
+                "mode": "trunk",
+                "vlan": 1,
+                "allowed_vlans": [10, 20, 30],
+                "description": "Uplink to core switch",
+                "enabled": True
+            }
+        }
+
+
+class DeviceModel(BaseModel):
+    """Network device schema"""
+    name: str = Field(..., min_length=1, max_length=64, description="Device hostname")
+    role: Literal["core", "distribution", "access", "edge", "firewall", "router", "wireless"] = Field(..., description="Device role in network")
+    model: str = Field(..., description="Hardware model")
+    vendor: Literal["cisco", "juniper", "arista", "hp", "dell", "ubiquiti", "mikrotik", "other"] = Field(..., description="Vendor")
+    mgmt_ip: str = Field(..., description="Management IP address")
+    location: str = Field(..., description="Physical location")
+    interfaces: List[InterfaceModel] = Field(default_factory=list, description="Network interfaces")
+    os_version: Optional[str] = Field(None, description="OS version")
+    serial_number: Optional[str] = Field(None, description="Device serial number")
+    
+    @validator('name')
+    def validate_hostname(cls, v):
+        """Ensure hostname follows RFC1123"""
+        if not re.match(r'^[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?$', v):
+            raise ValueError("Invalid hostname format (RFC1123)")
+        return v
+    
+    @validator('mgmt_ip')
+    def validate_mgmt_ip(cls, v):
+        """Ensure valid IP address"""
+        try:
+            ip_address(v)
+        except ValueError as e:
+            raise ValueError(f"Invalid management IP: {e}")
+        return v
+    
+    class Config:
+        schema_extra = {
+            "example": {
+                "name": "core-sw-01",
+                "role": "core",
+                "model": "Catalyst 9300",
+                "vendor": "cisco",
+                "mgmt_ip": "10.0.10.10",
+                "location": "Main Office - IDF1",
+                "os_version": "17.9.4",
+                "interfaces": []
+            }
+        }
+
+
+class RoutingModel(BaseModel):
+    """Routing protocol configuration"""
+    protocol: Literal["static", "ospf", "bgp", "eigrp", "rip", "is-is"] = Field(..., description="Routing protocol")
+    autonomous_system: Optional[int] = Field(None, ge=1, le=4294967295, description="AS number for BGP/EIGRP")
+    process_id: Optional[int] = Field(None, ge=1, le=65535, description="Process ID for OSPF/EIGRP")
+    router_id: Optional[str] = Field(None, description="Router ID")
+    areas: Optional[List[str]] = Field(None, description="OSPF areas or IS-IS levels")
+    networks: Optional[List[str]] = Field(None, description="Networks to advertise")
+    neighbors: Optional[List[str]] = Field(None, description="BGP neighbors or static routes")
+    
+    @validator('router_id')
+    def validate_router_id(cls, v):
+        """Ensure router ID is valid IP format"""
+        if v:
+            try:
+                ip_address(v)
+            except ValueError as e:
+                raise ValueError(f"Invalid router ID format: {e}")
+        return v
+    
+    class Config:
+        schema_extra = {
+            "example": {
+                "protocol": "ospf",
+                "process_id": 1,
+                "router_id": "10.0.0.1",
+                "areas": ["0"],
+                "networks": ["10.0.0.0/8"]
+            }
+        }
+
+
+class NetworkModelSchema(BaseModel):
+    """Complete network data model with validation"""
+    name: str = Field(..., min_length=1, max_length=64, description="Network name")
+    version: str = Field(..., pattern=r'^\d+\.\d+\.\d+$', description="Schema version (semantic versioning)")
+    description: Optional[str] = Field(None, description="Network description")
+    business_requirements: Optional[List[str]] = Field(default_factory=list, description="Business requirements")
+    constraints: Optional[List[str]] = Field(default_factory=list, description="Design constraints")
+    intent: Optional[Dict[str, Any]] = Field(None, description="Original network intent")
+    
+    devices: List[DeviceModel] = Field(default_factory=list, description="Network devices")
+    vlans: List[VLANModel] = Field(default_factory=list, description="VLANs")
+    subnets: List[SubnetModel] = Field(default_factory=list, description="IP subnets")
+    routing: Optional[Dict[str, Any]] = Field(None, description="Routing configuration")
+    services: List[str] = Field(default_factory=lambda: ["DHCP", "DNS", "NTP"], description="Network services")
+    
+    @validator('vlans')
+    def check_unique_vlan_ids(cls, v):
+        """Ensure no duplicate VLAN IDs"""
+        vlan_ids = [vlan.id for vlan in v]
+        if len(vlan_ids) != len(set(vlan_ids)):
+            duplicates = [vid for vid in vlan_ids if vlan_ids.count(vid) > 1]
+            raise ValueError(f"Duplicate VLAN IDs found: {duplicates}")
+        return v
+    
+    @validator('devices')
+    def check_unique_device_names(cls, v):
+        """Ensure no duplicate device names"""
+        device_names = [d.name for d in v]
+        if len(device_names) != len(set(device_names)):
+            duplicates = [name for name in device_names if device_names.count(name) > 1]
+            raise ValueError(f"Duplicate device names found: {duplicates}")
+        return v
+    
+    @validator('devices')
+    def check_unique_mgmt_ips(cls, v):
+        """Ensure no duplicate management IPs"""
+        mgmt_ips = [d.mgmt_ip for d in v]
+        if len(mgmt_ips) != len(set(mgmt_ips)):
+            duplicates = [ip for ip in mgmt_ips if mgmt_ips.count(ip) > 1]
+            raise ValueError(f"Duplicate management IPs found: {duplicates}")
+        return v
+    
+    @validator('subnets')
+    def check_subnet_vlan_references(cls, v, values):
+        """Ensure referenced VLANs exist"""
+        if 'vlans' in values:
+            valid_vlan_ids = {vlan.id for vlan in values['vlans']}
+            for subnet in v:
+                if subnet.vlan and subnet.vlan not in valid_vlan_ids:
+                    raise ValueError(f"Subnet {subnet.network} references non-existent VLAN {subnet.vlan}")
+        return v
+    
+    class Config:
+        schema_extra = {
+            "example": {
+                "name": "corporate-network",
+                "version": "1.0.0",
+                "description": "Corporate office network",
+                "business_requirements": ["High availability", "Secure guest access"],
+                "constraints": ["Budget under $10K", "Cisco preferred"],
+                "devices": [],
+                "vlans": [],
+                "subnets": [],
+                "routing": None,
+                "services": ["DHCP", "DNS", "NTP"]
+            }
+        }
+
+
+def validate_network_model(data: Dict[str, Any]) -> NetworkModelSchema:
+    """
+    Validate network model data against schema
+    Raises ValidationError if invalid
+    """
+    return NetworkModelSchema(**data)
+
+
+def get_validation_errors(data: Dict[str, Any]) -> List[str]:
+    """
+    Get human-readable validation errors
+    Returns empty list if valid
+    """
+    try:
+        validate_network_model(data)
+        return []
+    except Exception as e:
+        # Parse pydantic validation errors
+        errors = []
+        if hasattr(e, 'errors'):
+            for error in e.errors():
+                loc = ' -> '.join(str(l) for l in error['loc'])
+                msg = error['msg']
+                errors.append(f"{loc}: {msg}")
+        else:
+            errors.append(str(e))
+        return errors

app.py

@@ -265,17 +265,53 @@ def build_ui():
                 # Run the pipeline
                 results = pipeline.run_full_pipeline(user_input)
                 
+                # Check pre-flight validation
+                preflight = results.get('preflight', {})
+                ready_to_deploy = preflight.get('ready_to_deploy', False)
+                
                 # Format status
                 status = "## 🚀 Pipeline Execution Complete!\n\n"
-                status += "### ✅ Completed Stages:\n"
+                
+                # Pre-flight validation section
+                if ready_to_deploy:
+                    status += "### ✅ Pre-flight Validation PASSED\n"
+                    status += f"- Schema validation: ✓ Passed\n"
+                    status += f"- Policy checks: ✓ Passed\n"
+                    if preflight.get('warnings'):
+                        status += f"- Warnings: {len(preflight['warnings'])}\n"
+                    if preflight.get('info'):
+                        status += f"- Info: {len(preflight['info'])}\n"
+                    status += "\n"
+                else:
+                    status += "### ❌ Pre-flight Validation FAILED\n"
+                    status += "**Deployment blocked until errors are fixed:**\n\n"
+                    for error in preflight.get('errors', []):
+                        status += f"- ❌ {error}\n"
+                    status += "\n"
+                    if preflight.get('warnings'):
+                        status += "**Warnings:**\n"
+                        for warning in preflight.get('warnings', []):
+                            status += f"- ⚠️  {warning}\n"
+                        status += "\n"
+                
+                # Completed stages
+                status += "### Completed Stages:\n"
+                status += "0. " + ("✅" if ready_to_deploy else "❌") + " Pre-flight Validation\n"
                 status += "1. ✅ Consultation - Intent captured\n"
                 status += "2. ✅ Source of Truth - Network model generated\n"
                 status += "3. ✅ Diagrams - Visualizations created\n"
                 status += "4. ✅ Bill of Materials - Shopping list ready\n"
                 status += "5. ✅ Setup Guide - Deployment instructions generated\n"
-                status += "6. ⏳ Autonomous Deploy - Ready for execution\n"
-                status += "7. ⏳ Observability - Ready for setup\n"
-                status += "8. ⏳ Validation - Ready for verification\n\n"
+                
+                if ready_to_deploy:
+                    status += "6. ⏳ Autonomous Deploy - Ready for execution\n"
+                    status += "7. ⏳ Observability - Ready for setup\n"
+                    status += "8. ⏳ Validation - Ready for verification\n\n"
+                else:
+                    status += "6. 🚫 Autonomous Deploy - BLOCKED (fix validation errors)\n"
+                    status += "7. 🚫 Observability - BLOCKED\n"
+                    status += "8. 🚫 Validation - BLOCKED\n\n"
+                
                 status += "### 📁 Files Created:\n"
                 status += "- `infra/network_model.yaml` - Source of Truth\n"
                 status += "- `infra/bill_of_materials.json` - BOM data\n"
@@ -294,6 +330,8 @@ def build_ui():
                 
             except Exception as e:
                 error = f"## ❌ Pipeline Error\n\n{str(e)}"
+                import traceback
+                error += f"\n\n```\n{traceback.format_exc()}\n```"
                 return error, "", "", "", ""
 
         # Event Handlers

requirements.txt

@@ -8,3 +8,4 @@ python-dotenv
 anthropic>=0.40.0
 pyyaml
 pynetbox>=7.0.0
+pydantic>=2.0.0

test_preflight.py

@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+"""
+Test Stage 0: Pre-flight validation integration
+"""
+
+import logging
+from pathlib import Path
+from agent.pipeline_engine import OvergrowthPipeline, NetworkIntent
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+def test_preflight_pass():
+    """Test pre-flight with valid network"""
+    print("\n=== Test 1: Pre-flight Validation - PASS ===")
+    
+    pipeline = OvergrowthPipeline()
+    
+    # Create a good network intent
+    intent = NetworkIntent(
+        description="Corporate office network with guest WiFi",
+        business_requirements=["Secure guest access", "High availability", "QoS for VoIP"],
+        constraints=["Budget under $15K", "Cisco preferred"],
+        budget="$15000",
+        timeline="2 weeks"
+    )
+    
+    # Generate source of truth
+    model = pipeline.stage2_generate_sot(intent)
+    
+    # Run pre-flight validation
+    results = pipeline.stage0_preflight(model)
+    
+    print(f"\nValidation Results:")
+    print(f"  Schema Valid: {results['schema_valid']}")
+    print(f"  Policy Passed: {results['policy_passed']}")
+    print(f"  Ready to Deploy: {results['ready_to_deploy']}")
+    print(f"  Errors: {len(results['errors'])}")
+    print(f"  Warnings: {len(results['warnings'])}")
+    print(f"  Info: {len(results['info'])}")
+    
+    if results['errors']:
+        print("\nErrors:")
+        for err in results['errors']:
+            print(f"  - {err}")
+    
+    if results['warnings']:
+        print("\nWarnings:")
+        for warn in results['warnings'][:5]:  # Show first 5
+            print(f"  - {warn}")
+    
+    if results['ready_to_deploy']:
+        print("\n✓ Pre-flight validation PASSED - ready to deploy")
+    else:
+        print("\n✗ Pre-flight validation FAILED")
+
+
+def test_full_pipeline_with_validation():
+    """Test complete pipeline including pre-flight"""
+    print("\n=== Test 2: Full Pipeline with Validation ===")
+    
+    pipeline = OvergrowthPipeline()
+    
+    consultation = "I need a network for a small retail store with guest WiFi, employee WiFi, POS systems, and security cameras. Budget is $5000."
+    
+    results = pipeline.run_full_pipeline(consultation)
+    
+    print(f"\nPipeline Results:")
+    print(f"  Intent captured: {'intent' in results}")
+    print(f"  Model generated: {'model' in results}")
+    print(f"  Pre-flight complete: {'preflight' in results}")
+    
+    if 'preflight' in results:
+        pf = results['preflight']
+        print(f"\nPre-flight:")
+        print(f"  Ready to deploy: {pf['ready_to_deploy']}")
+        print(f"  Errors: {len(pf['errors'])}")
+        print(f"  Warnings: {len(pf['warnings'])}")
+        
+        if 'deployment_status' in results:
+            print(f"\nDeployment Status: {results['deployment_status']}")
+            if 'deployment_reason' in results:
+                print(f"  Reason: {results['deployment_reason']}")
+    
+    print(f"\nOutputs:")
+    print(f"  Diagrams: {'diagrams' in results}")
+    print(f"  BOM: {'bom' in results}")
+    print(f"  Setup Guide: {'setup_guide' in results}")
+    
+    if 'bom' in results:
+        bom = results['bom']
+        print(f"\nBill of Materials:")
+        print(f"  Total Cost: ${bom.get('total_estimated_cost', 0):,.2f}")
+        print(f"  Devices: {len(bom.get('devices', []))}")
+    
+    print("\n✓ Full pipeline test complete")
+
+
+if __name__ == "__main__":
+    print("\n" + "="*60)
+    print("Stage 0: Pre-flight Validation Test")
+    print("="*60)
+    
+    try:
+        test_preflight_pass()
+        test_full_pipeline_with_validation()
+        
+        print("\n" + "="*60)
+        print("✓ All pre-flight tests passed!")
+        print("="*60 + "\n")
+        
+    except Exception as e:
+        print(f"\n✗ Test failed: {e}")
+        import traceback
+        traceback.print_exc()
+        exit(1)

test_validation.py

@@ -0,0 +1,344 @@
+#!/usr/bin/env python3
+"""
+Test schema validation and policy engine
+"""
+
+import logging
+from agent.schema_validation import (
+    validate_network_model, 
+    get_validation_errors,
+    VLANModel,
+    SubnetModel,
+    DeviceModel
+)
+from agent.policy_engine import NetworkPolicy
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+def test_valid_network():
+    """Test validation with a valid network model"""
+    print("\n=== Test 1: Valid Network Model ===")
+    
+    model = {
+        "name": "test-network",
+        "version": "1.0.0",
+        "description": "Test network",
+        "business_requirements": ["High availability"],
+        "constraints": ["Budget friendly"],
+        "vlans": [
+            {"id": 10, "name": "Management", "purpose": "Network mgmt", "subnet": "10.0.10.0/24"},
+            {"id": 20, "name": "Users", "purpose": "Employee workstations", "subnet": "10.0.20.0/24"}
+        ],
+        "subnets": [
+            {
+                "network": "10.0.10.0/24",
+                "gateway": "10.0.10.1",
+                "vlan": 10,
+                "purpose": "Management",
+                "dhcp_enabled": True,
+                "dhcp_range_start": "10.0.10.100",
+                "dhcp_range_end": "10.0.10.200"
+            },
+            {
+                "network": "10.0.20.0/24",
+                "gateway": "10.0.20.1",
+                "vlan": 20,
+                "purpose": "Users"
+            }
+        ],
+        "devices": [
+            {
+                "name": "core-sw-01",
+                "role": "core",
+                "model": "Catalyst 9300",
+                "vendor": "cisco",
+                "mgmt_ip": "10.0.10.10",
+                "location": "Main IDF",
+                "interfaces": []
+            }
+        ],
+        "services": ["DHCP", "DNS", "NTP"]
+    }
+    
+    errors = get_validation_errors(model)
+    if errors:
+        print(f"✗ Validation failed:")
+        for err in errors:
+            print(f"  - {err}")
+    else:
+        validated = validate_network_model(model)
+        print(f"✓ Network model validated successfully")
+        print(f"  - Name: {validated.name}")
+        print(f"  - VLANs: {len(validated.vlans)}")
+        print(f"  - Subnets: {len(validated.subnets)}")
+        print(f"  - Devices: {len(validated.devices)}")
+
+
+def test_invalid_vlan():
+    """Test VLAN validation"""
+    print("\n=== Test 2: Invalid VLAN ID ===")
+    
+    model = {
+        "name": "test",
+        "version": "1.0.0",
+        "description": "Test",
+        "vlans": [
+            {"id": 5000, "name": "Invalid"}  # VLAN ID out of range
+        ],
+        "devices": [],
+        "subnets": [],
+        "services": []
+    }
+    
+    errors = get_validation_errors(model)
+    if errors:
+        print(f"✓ Correctly caught validation error:")
+        for err in errors:
+            print(f"  - {err}")
+    else:
+        print("✗ Should have failed validation")
+
+
+def test_invalid_subnet():
+    """Test subnet validation"""
+    print("\n=== Test 3: Invalid Subnet ===")
+    
+    model = {
+        "name": "test",
+        "version": "1.0.0",
+        "description": "Test",
+        "vlans": [{"id": 10, "name": "Test"}],
+        "subnets": [
+            {
+                "network": "10.0.10.0/24",
+                "gateway": "192.168.1.1",  # Gateway not in subnet
+                "vlan": 10
+            }
+        ],
+        "devices": [],
+        "services": []
+    }
+    
+    errors = get_validation_errors(model)
+    if errors:
+        print(f"✓ Correctly caught validation error:")
+        for err in errors:
+            print(f"  - {err}")
+    else:
+        print("✗ Should have failed validation")
+
+
+def test_duplicate_vlan_ids():
+    """Test duplicate VLAN detection"""
+    print("\n=== Test 4: Duplicate VLAN IDs ===")
+    
+    model = {
+        "name": "test",
+        "version": "1.0.0",
+        "description": "Test",
+        "vlans": [
+            {"id": 10, "name": "VLAN10"},
+            {"id": 10, "name": "AlsoVLAN10"}  # Duplicate!
+        ],
+        "devices": [],
+        "subnets": [],
+        "services": []
+    }
+    
+    errors = get_validation_errors(model)
+    if errors:
+        print(f"✓ Correctly caught duplicate VLAN IDs:")
+        for err in errors:
+            print(f"  - {err}")
+    else:
+        print("✗ Should have failed validation")
+
+
+def test_policy_engine():
+    """Test policy engine checks"""
+    print("\n=== Test 5: Policy Engine ===")
+    
+    model = {
+        "name": "test-network",
+        "version": "1.0.0",
+        "description": "Test network",
+        "vlans": [
+            {"id": 1, "name": "Default"},  # VLAN 1 - should warn
+            {"id": 10, "name": "Mgmt"},
+            {"id": 20, "name": "Guest WiFi"}  # Space in name - should warn
+        ],
+        "subnets": [
+            {"network": "10.0.10.0/24", "gateway": "10.0.10.1", "vlan": 10},
+            {"network": "10.0.20.0/24", "gateway": "10.0.20.1", "vlan": 20}
+        ],
+        "devices": [
+            {
+                "name": "switch1",  # No role in name, no 2-digit suffix
+                "role": "core",
+                "model": "Test",
+                "vendor": "cisco",
+                "mgmt_ip": "10.0.10.10",
+                "location": "Office"
+            }
+        ],
+        "services": ["DHCP"]  # Missing DNS, NTP
+    }
+    
+    policy = NetworkPolicy()
+    violations = policy.check_network_model(model)
+    
+    print(f"Found {len(violations)} policy violations:")
+    print(policy.format_violations())
+    
+    by_severity = policy.get_violations_by_severity()
+    print(f"\n✓ Policy check complete:")
+    print(f"  - Errors: {len(by_severity['ERROR'])}")
+    print(f"  - Warnings: {len(by_severity['WARNING'])}")
+    print(f"  - Info: {len(by_severity['INFO'])}")
+
+
+def test_overlapping_subnets():
+    """Test overlapping subnet detection"""
+    print("\n=== Test 6: Overlapping Subnets ===")
+    
+    model = {
+        "name": "test",
+        "version": "1.0.0",
+        "description": "Test",
+        "vlans": [
+            {"id": 10, "name": "Network1"},
+            {"id": 20, "name": "Network2"}
+        ],
+        "subnets": [
+            {"network": "10.0.0.0/16", "gateway": "10.0.0.1", "vlan": 10},
+            {"network": "10.0.10.0/24", "gateway": "10.0.10.1", "vlan": 20}  # Overlaps!
+        ],
+        "devices": [],
+        "services": []
+    }
+    
+    policy = NetworkPolicy()
+    violations = policy.check_network_model(model)
+    
+    overlap_errors = [v for v in violations if 'overlapping' in v.message.lower()]
+    
+    if overlap_errors:
+        print(f"✓ Correctly detected overlapping subnets:")
+        for err in overlap_errors:
+            print(f"  - {err}")
+    else:
+        print("✗ Should have detected overlapping subnets")
+
+
+def test_complete_validation():
+    """Test complete validation flow"""
+    print("\n=== Test 7: Complete Validation Flow ===")
+    
+    model = {
+        "name": "production-network",
+        "version": "1.0.0",
+        "description": "Production corporate network",
+        "business_requirements": ["HA", "Secure guest access", "QoS for VoIP"],
+        "constraints": ["Cisco only", "Budget $25K"],
+        "vlans": [
+            {"id": 10, "name": "Management", "purpose": "Network management"},
+            {"id": 20, "name": "Users", "purpose": "Employee workstations"},
+            {"id": 30, "name": "Guest", "purpose": "Guest WiFi"},
+            {"id": 40, "name": "Voice", "purpose": "VoIP phones"}
+        ],
+        "subnets": [
+            {"network": "10.0.10.0/24", "gateway": "10.0.10.1", "vlan": 10, "purpose": "Management"},
+            {"network": "10.0.20.0/22", "gateway": "10.0.20.1", "vlan": 20, "purpose": "Users"},
+            {"network": "10.0.30.0/24", "gateway": "10.0.30.1", "vlan": 30, "purpose": "Guest"},
+            {"network": "10.0.40.0/24", "gateway": "10.0.40.1", "vlan": 40, "purpose": "Voice"}
+        ],
+        "devices": [
+            {
+                "name": "core-sw-01",
+                "role": "core",
+                "model": "Catalyst 9300",
+                "vendor": "cisco",
+                "mgmt_ip": "10.0.10.10",
+                "location": "Main IDF"
+            },
+            {
+                "name": "core-sw-02",
+                "role": "core",
+                "model": "Catalyst 9300",
+                "vendor": "cisco",
+                "mgmt_ip": "10.0.10.11",
+                "location": "Main IDF"
+            },
+            {
+                "name": "access-sw-01",
+                "role": "access",
+                "model": "Catalyst 2960X",
+                "vendor": "cisco",
+                "mgmt_ip": "10.0.10.20",
+                "location": "Floor 1"
+            }
+        ],
+        "routing": {
+            "protocol": "ospf",
+            "process_id": 1,
+            "router_id": "10.0.0.1",
+            "areas": ["0"]
+        },
+        "services": ["DHCP", "DNS", "NTP", "Syslog"]
+    }
+    
+    # Schema validation
+    errors = get_validation_errors(model)
+    if errors:
+        print(f"✗ Schema validation failed:")
+        for err in errors:
+            print(f"  - {err}")
+        return
+    
+    print("✓ Schema validation passed")
+    validated = validate_network_model(model)
+    print(f"  - {len(validated.vlans)} VLANs")
+    print(f"  - {len(validated.subnets)} subnets")
+    print(f"  - {len(validated.devices)} devices")
+    
+    # Policy validation
+    policy = NetworkPolicy()
+    violations = policy.check_network_model(model)
+    
+    print(f"\n✓ Policy validation complete:")
+    by_severity = policy.get_violations_by_severity()
+    print(f"  - Errors: {len(by_severity['ERROR'])}")
+    print(f"  - Warnings: {len(by_severity['WARNING'])}")
+    print(f"  - Info: {len(by_severity['INFO'])}")
+    
+    if policy.has_errors():
+        print("\n✗ Cannot proceed - fix errors first")
+    else:
+        print("\n✓ Ready for deployment")
+
+
+if __name__ == "__main__":
+    print("\n" + "="*60)
+    print("Schema Validation & Policy Engine Test Suite")
+    print("="*60)
+    
+    try:
+        test_valid_network()
+        test_invalid_vlan()
+        test_invalid_subnet()
+        test_duplicate_vlan_ids()
+        test_policy_engine()
+        test_overlapping_subnets()
+        test_complete_validation()
+        
+        print("\n" + "="*60)
+        print("✓ All validation tests completed!")
+        print("="*60 + "\n")
+        
+    except Exception as e:
+        print(f"\n✗ Test failed: {e}")
+        import traceback
+        traceback.print_exc()
+        exit(1)