hermes-agent/agent/nexus_deployment.py

#!/usr/bin/env python3
"""
Nexus Deployment System

Real-time deployment system for Nexus Three.js modules.
Provides hot-reload, validation, rollback, and versioning capabilities.

Features:
- Hot-reload Three.js modules without page refresh
- Syntax validation and Three.js API compliance checking
- Rollback on error
- Versioning for nexus modules
- Module registry and dependency tracking

Usage:
    from agent.nexus_deployment import NexusDeployer
    
    deployer = NexusDeployer()
    
    # Deploy with hot-reload
    result = deployer.deploy_module(room_code, module_name="zen_garden")
    
    # Rollback if needed
    deployer.rollback_module("zen_garden")
    
    # Get module status
    status = deployer.get_module_status("zen_garden")
"""

import json
import logging
import re
import os
import hashlib
from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum

# Import validation from existing nexus_architect (avoid circular imports)
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

def _import_validation():
    """Lazy import to avoid circular dependencies."""
    try:
        from tools.nexus_architect import validate_three_js_code, sanitize_three_js_code
        return validate_three_js_code, sanitize_three_js_code
    except ImportError:
        # Fallback: define local validation functions
        def validate_three_js_code(code, strict_mode=False):
            """Fallback validation."""
            errors = []
            if "eval(" in code:
                errors.append("Security violation: eval detected")
            if "Function(" in code:
                errors.append("Security violation: Function constructor detected")
            return type('ValidationResult', (), {
                'is_valid': len(errors) == 0,
                'errors': errors,
                'warnings': []
            })()
        
        def sanitize_three_js_code(code):
            """Fallback sanitization."""
            return code
        
        return validate_three_js_code, sanitize_three_js_code

logger = logging.getLogger(__name__)


# =============================================================================
# Deployment States
# =============================================================================

class DeploymentStatus(Enum):
    """Status of a module deployment."""
    PENDING = "pending"
    VALIDATING = "validating"
    DEPLOYING = "deploying"
    ACTIVE = "active"
    FAILED = "failed"
    ROLLING_BACK = "rolling_back"
    ROLLED_BACK = "rolled_back"


# =============================================================================
# Data Models
# =============================================================================

@dataclass
class ModuleVersion:
    """Version information for a Nexus module."""
    version_id: str
    module_name: str
    code_hash: str
    timestamp: str
    changes: str = ""
    author: str = "nexus_architect"
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "version_id": self.version_id,
            "module_name": self.module_name,
            "code_hash": self.code_hash,
            "timestamp": self.timestamp,
            "changes": self.changes,
            "author": self.author,
        }


@dataclass
class DeployedModule:
    """A deployed Nexus module."""
    name: str
    code: str
    status: DeploymentStatus
    version: str
    deployed_at: str
    last_updated: str
    validation_result: Dict[str, Any] = field(default_factory=dict)
    error_log: List[str] = field(default_factory=list)
    dependencies: Set[str] = field(default_factory=set)
    hot_reload_supported: bool = True
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "name": self.name,
            "status": self.status.value,
            "version": self.version,
            "deployed_at": self.deployed_at,
            "last_updated": self.last_updated,
            "validation": self.validation_result,
            "dependencies": list(self.dependencies),
            "hot_reload_supported": self.hot_reload_supported,
            "code_preview": self.code[:200] + "..." if len(self.code) > 200 else self.code,
        }


# =============================================================================
# Nexus Deployer
# =============================================================================

class NexusDeployer:
    """
    Deployment system for Nexus Three.js modules.
    
    Provides:
    - Hot-reload deployment
    - Validation before deployment
    - Automatic rollback on failure
    - Version tracking
    - Module registry
    """
    
    def __init__(self, modules_dir: Optional[str] = None):
        """
        Initialize the Nexus Deployer.
        
        Args:
            modules_dir: Directory to store deployed modules (optional)
        """
        self.modules: Dict[str, DeployedModule] = {}
        self.version_history: Dict[str, List[ModuleVersion]] = {}
        self.modules_dir = modules_dir or os.path.expanduser("~/.nexus/modules")
        
        # Ensure modules directory exists
        os.makedirs(self.modules_dir, exist_ok=True)
        
        # Hot-reload configuration
        self.hot_reload_enabled = True
        self.auto_rollback = True
        self.strict_validation = True
        
        logger.info(f"NexusDeployer initialized. Modules dir: {self.modules_dir}")
    
    def deploy_module(
        self,
        module_code: str,
        module_name: str,
        version: Optional[str] = None,
        dependencies: Optional[List[str]] = None,
        hot_reload: bool = True,
        validate: bool = True
    ) -> Dict[str, Any]:
        """
        Deploy a Nexus module with hot-reload support.
        
        Args:
            module_code: The Three.js module code
            module_name: Unique module identifier
            version: Optional version string (auto-generated if not provided)
            dependencies: List of dependent module names
            hot_reload: Enable hot-reload for this module
            validate: Run validation before deployment
        
        Returns:
            Dict with deployment results
        """
        timestamp = datetime.now().isoformat()
        version = version or self._generate_version(module_name, module_code)
        
        result = {
            "success": True,
            "module_name": module_name,
            "version": version,
            "timestamp": timestamp,
            "hot_reload": hot_reload,
            "validation": {},
            "deployment": {},
        }
        
        # Check for existing module (hot-reload scenario)
        existing_module = self.modules.get(module_name)
        if existing_module and not hot_reload:
            return {
                "success": False,
                "error": f"Module '{module_name}' already exists. Use hot_reload=True to update."
            }
        
        # Validation phase
        if validate:
            validation = self._validate_module(module_code)
            result["validation"] = validation
            
            if not validation["is_valid"]:
                result["success"] = False
                result["error"] = "Validation failed"
                result["message"] = "Module deployment aborted due to validation errors"
                
                if self.auto_rollback:
                    result["rollback_triggered"] = False  # Nothing to rollback yet
                
                return result
        
        # Create deployment backup for rollback
        if existing_module:
            self._create_backup(existing_module)
        
        # Deployment phase
        try:
            deployed = DeployedModule(
                name=module_name,
                code=module_code,
                status=DeploymentStatus.DEPLOYING,
                version=version,
                deployed_at=timestamp if not existing_module else existing_module.deployed_at,
                last_updated=timestamp,
                validation_result=result.get("validation", {}),
                dependencies=set(dependencies or []),
                hot_reload_supported=hot_reload,
            )
            
            # Save to file system
            self._save_module_file(deployed)
            
            # Update registry
            deployed.status = DeploymentStatus.ACTIVE
            self.modules[module_name] = deployed
            
            # Record version
            self._record_version(module_name, version, module_code)
            
            result["deployment"] = {
                "status": "active",
                "hot_reload_ready": hot_reload,
                "file_path": self._get_module_path(module_name),
            }
            result["message"] = f"Module '{module_name}' v{version} deployed successfully"
            
            if existing_module:
                result["message"] += " (hot-reload update)"
            
            logger.info(f"Deployed module: {module_name} v{version}")
            
        except Exception as e:
            result["success"] = False
            result["error"] = str(e)
            result["deployment"] = {"status": "failed"}
            
            # Attempt rollback if deployment failed
            if self.auto_rollback and existing_module:
                rollback_result = self.rollback_module(module_name)
                result["rollback_result"] = rollback_result
            
            logger.error(f"Deployment failed for {module_name}: {e}")
        
        return result
    
    def hot_reload_module(self, module_name: str, new_code: str) -> Dict[str, Any]:
        """
        Hot-reload an active module with new code.
        
        Args:
            module_name: Name of the module to reload
            new_code: New module code
        
        Returns:
            Dict with reload results
        """
        if module_name not in self.modules:
            return {
                "success": False,
                "error": f"Module '{module_name}' not found. Deploy it first."
            }
        
        module = self.modules[module_name]
        if not module.hot_reload_supported:
            return {
                "success": False,
                "error": f"Module '{module_name}' does not support hot-reload"
            }
        
        # Use deploy_module with hot_reload=True
        return self.deploy_module(
            module_code=new_code,
            module_name=module_name,
            hot_reload=True,
            validate=True
        )
    
    def rollback_module(self, module_name: str, to_version: Optional[str] = None) -> Dict[str, Any]:
        """
        Rollback a module to a previous version.
        
        Args:
            module_name: Module to rollback
            to_version: Specific version to rollback to (latest backup if not specified)
        
        Returns:
            Dict with rollback results
        """
        if module_name not in self.modules:
            return {
                "success": False,
                "error": f"Module '{module_name}' not found"
            }
        
        module = self.modules[module_name]
        module.status = DeploymentStatus.ROLLING_BACK
        
        try:
            if to_version:
                # Restore specific version
                version_data = self._get_version(module_name, to_version)
                if not version_data:
                    return {
                        "success": False,
                        "error": f"Version '{to_version}' not found for module '{module_name}'"
                    }
                # Would restore from version data
            else:
                # Restore from backup
                backup_code = self._get_backup(module_name)
                if backup_code:
                    module.code = backup_code
                    module.last_updated = datetime.now().isoformat()
                else:
                    return {
                        "success": False,
                        "error": f"No backup available for '{module_name}'"
                    }
            
            module.status = DeploymentStatus.ROLLED_BACK
            self._save_module_file(module)
            
            logger.info(f"Rolled back module: {module_name}")
            
            return {
                "success": True,
                "module_name": module_name,
                "message": f"Module '{module_name}' rolled back successfully",
                "status": module.status.value,
            }
            
        except Exception as e:
            module.status = DeploymentStatus.FAILED
            logger.error(f"Rollback failed for {module_name}: {e}")
            return {
                "success": False,
                "error": str(e)
            }
    
    def validate_module(self, module_code: str) -> Dict[str, Any]:
        """
        Validate Three.js module code without deploying.
        
        Args:
            module_code: Code to validate
        
        Returns:
            Dict with validation results
        """
        return self._validate_module(module_code)
    
    def get_module_status(self, module_name: str) -> Optional[Dict[str, Any]]:
        """
        Get status of a deployed module.
        
        Args:
            module_name: Module name
        
        Returns:
            Module status dict or None if not found
        """
        if module_name in self.modules:
            return self.modules[module_name].to_dict()
        return None
    
    def get_all_modules(self) -> Dict[str, Any]:
        """
        Get status of all deployed modules.
        
        Returns:
            Dict with all module statuses
        """
        return {
            "modules": {
                name: module.to_dict()
                for name, module in self.modules.items()
            },
            "total_count": len(self.modules),
            "active_count": sum(1 for m in self.modules.values() if m.status == DeploymentStatus.ACTIVE),
        }
    
    def get_version_history(self, module_name: str) -> List[Dict[str, Any]]:
        """
        Get version history for a module.
        
        Args:
            module_name: Module name
        
        Returns:
            List of version dicts
        """
        history = self.version_history.get(module_name, [])
        return [v.to_dict() for v in history]
    
    def remove_module(self, module_name: str) -> Dict[str, Any]:
        """
        Remove a deployed module.
        
        Args:
            module_name: Module to remove
        
        Returns:
            Dict with removal results
        """
        if module_name not in self.modules:
            return {
                "success": False,
                "error": f"Module '{module_name}' not found"
            }
        
        try:
            # Remove file
            module_path = self._get_module_path(module_name)
            if os.path.exists(module_path):
                os.remove(module_path)
            
            # Remove from registry
            del self.modules[module_name]
            
            logger.info(f"Removed module: {module_name}")
            
            return {
                "success": True,
                "message": f"Module '{module_name}' removed successfully"
            }
            
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }
    
    def _validate_module(self, code: str) -> Dict[str, Any]:
        """Internal validation method."""
        # Use existing validation from nexus_architect (lazy import)
        validate_fn, _ = _import_validation()
        validation_result = validate_fn(code, strict_mode=self.strict_validation)
        
        # Check Three.js API compliance
        three_api_issues = self._check_three_js_api_compliance(code)
        
        return {
            "is_valid": validation_result.is_valid and len(three_api_issues) == 0,
            "syntax_valid": validation_result.is_valid,
            "api_compliant": len(three_api_issues) == 0,
            "errors": validation_result.errors + three_api_issues,
            "warnings": validation_result.warnings,
            "safety_score": max(0, 100 - len(validation_result.errors) * 20 - len(validation_result.warnings) * 5),
        }
    
    def _check_three_js_api_compliance(self, code: str) -> List[str]:
        """Check for Three.js API compliance issues."""
        issues = []
        
        # Check for required patterns
        if "THREE.Group" not in code and "new THREE" not in code:
            issues.append("No Three.js objects created")
        
        # Check for deprecated APIs
        deprecated_patterns = [
            (r"THREE\.Face3", "THREE.Face3 is deprecated, use BufferGeometry"),
            (r"THREE\.Geometry\(", "THREE.Geometry is deprecated, use BufferGeometry"),
        ]
        
        for pattern, message in deprecated_patterns:
            if re.search(pattern, code):
                issues.append(f"Deprecated API: {message}")
        
        return issues
    
    def _generate_version(self, module_name: str, code: str) -> str:
        """Generate version string from code hash."""
        code_hash = hashlib.md5(code.encode()).hexdigest()[:8]
        timestamp = datetime.now().strftime("%Y%m%d%H%M")
        return f"{timestamp}-{code_hash}"
    
    def _create_backup(self, module: DeployedModule) -> None:
        """Create backup of existing module."""
        backup_path = os.path.join(
            self.modules_dir,
            f"{module.name}.{module.version}.backup.js"
        )
        with open(backup_path, 'w') as f:
            f.write(module.code)
    
    def _get_backup(self, module_name: str) -> Optional[str]:
        """Get backup code for module."""
        if module_name not in self.modules:
            return None
        
        module = self.modules[module_name]
        backup_path = os.path.join(
            self.modules_dir,
            f"{module.name}.{module.version}.backup.js"
        )
        
        if os.path.exists(backup_path):
            with open(backup_path, 'r') as f:
                return f.read()
        return None
    
    def _save_module_file(self, module: DeployedModule) -> None:
        """Save module to file system."""
        module_path = self._get_module_path(module.name)
        with open(module_path, 'w') as f:
            f.write(f"// Nexus Module: {module.name}\n")
            f.write(f"// Version: {module.version}\n")
            f.write(f"// Status: {module.status.value}\n")
            f.write(f"// Updated: {module.last_updated}\n")
            f.write(f"// Hot-Reload: {module.hot_reload_supported}\n")
            f.write("\n")
            f.write(module.code)
    
    def _get_module_path(self, module_name: str) -> str:
        """Get file path for module."""
        return os.path.join(self.modules_dir, f"{module_name}.nexus.js")
    
    def _record_version(self, module_name: str, version: str, code: str) -> None:
        """Record version in history."""
        if module_name not in self.version_history:
            self.version_history[module_name] = []
        
        version_info = ModuleVersion(
            version_id=version,
            module_name=module_name,
            code_hash=hashlib.md5(code.encode()).hexdigest()[:16],
            timestamp=datetime.now().isoformat(),
        )
        
        self.version_history[module_name].insert(0, version_info)
        
        # Keep only last 10 versions
        self.version_history[module_name] = self.version_history[module_name][:10]
    
    def _get_version(self, module_name: str, version: str) -> Optional[ModuleVersion]:
        """Get specific version info."""
        history = self.version_history.get(module_name, [])
        for v in history:
            if v.version_id == version:
                return v
        return None


# =============================================================================
# Convenience Functions
# =============================================================================

_deployer_instance: Optional[NexusDeployer] = None


def get_deployer() -> NexusDeployer:
    """Get or create the NexusDeployer singleton."""
    global _deployer_instance
    if _deployer_instance is None:
        _deployer_instance = NexusDeployer()
    return _deployer_instance


def deploy_nexus_module(
    module_code: str,
    module_name: str,
    test: bool = True,
    hot_reload: bool = True
) -> Dict[str, Any]:
    """
    Deploy a Nexus module with validation.
    
    Args:
        module_code: Three.js module code
        module_name: Unique module identifier
        test: Run validation tests before deployment
        hot_reload: Enable hot-reload support
    
    Returns:
        Dict with deployment results
    """
    deployer = get_deployer()
    return deployer.deploy_module(
        module_code=module_code,
        module_name=module_name,
        hot_reload=hot_reload,
        validate=test
    )


def hot_reload_module(module_name: str, new_code: str) -> Dict[str, Any]:
    """
    Hot-reload an existing module.
    
    Args:
        module_name: Module to reload
        new_code: New module code
    
    Returns:
        Dict with reload results
    """
    deployer = get_deployer()
    return deployer.hot_reload_module(module_name, new_code)


def validate_nexus_code(code: str) -> Dict[str, Any]:
    """
    Validate Three.js code without deploying.
    
    Args:
        code: Three.js code to validate
    
    Returns:
        Dict with validation results
    """
    deployer = get_deployer()
    return deployer.validate_module(code)


def get_deployment_status() -> Dict[str, Any]:
    """Get status of all deployed modules."""
    deployer = get_deployer()
    return deployer.get_all_modules()


# =============================================================================
# Tool Schemas
# =============================================================================

NEXUS_DEPLOYMENT_SCHEMAS = {
    "deploy_nexus_module": {
        "name": "deploy_nexus_module",
        "description": "Deploy a Nexus Three.js module with validation and hot-reload support",
        "parameters": {
            "type": "object",
            "properties": {
                "module_code": {"type": "string"},
                "module_name": {"type": "string"},
                "test": {"type": "boolean", "default": True},
                "hot_reload": {"type": "boolean", "default": True},
            },
            "required": ["module_code", "module_name"]
        }
    },
    "hot_reload_module": {
        "name": "hot_reload_module",
        "description": "Hot-reload an existing Nexus module with new code",
        "parameters": {
            "type": "object",
            "properties": {
                "module_name": {"type": "string"},
                "new_code": {"type": "string"},
            },
            "required": ["module_name", "new_code"]
        }
    },
    "validate_nexus_code": {
        "name": "validate_nexus_code",
        "description": "Validate Three.js code for Nexus deployment without deploying",
        "parameters": {
            "type": "object",
            "properties": {
                "code": {"type": "string"}
            },
            "required": ["code"]
        }
    },
    "get_deployment_status": {
        "name": "get_deployment_status",
        "description": "Get status of all deployed Nexus modules",
        "parameters": {"type": "object", "properties": {}}
    },
}


if __name__ == "__main__":
    # Demo
    print("Nexus Deployment System - Demo")
    print("=" * 50)
    
    deployer = NexusDeployer()
    
    # Sample module code
    sample_code = """
(function() {
    function createDemoRoom() {
        const room = new THREE.Group();
        room.name = 'demo_room';
        
        const light = new THREE.AmbientLight(0x404040, 0.5);
        room.add(light);
        
        return room;
    }
    
    window.NexusRooms = window.NexusRooms || {};
    window.NexusRooms.demo_room = createDemoRoom;
    
    return { createDemoRoom };
})();
"""
    
    # Deploy
    result = deployer.deploy_module(sample_code, "demo_room")
    print(f"\nDeployment result: {result['message']}")
    print(f"Validation: {result['validation'].get('is_valid', False)}")
    print(f"Safety score: {result['validation'].get('safety_score', 0)}/100")
    
    # Get status
    status = deployer.get_all_modules()
    print(f"\nTotal modules: {status['total_count']}")
    print(f"Active: {status['active_count']}")