timmy-config/scripts/local-first/dependency_monitor.py

#!/usr/bin/env python3
"""
Monitor dependencies for local-first AI fallbacks.
Issue #483: [AUDIT][RISK] Maintain local-first fallbacks for all cloud AI
"""
import json
import time
import argparse
import requests
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from datetime import datetime

@dataclass
class DependencyStatus:
    """Status of a dependency."""
    name: str
    available: bool
    response_time: float
    last_checked: str
    error_message: Optional[str] = None

class DependencyMonitor:
    """Monitor local and external AI dependencies."""
    
    def __init__(self):
        self.local_dependencies = {
            "ollama": {
                "url": "http://localhost:11434",
                "check": self._check_ollama
            },
            "hermes4": {
                "url": "http://localhost:11434",
                "check": lambda: self._check_model("hermes4")
            },
            "llama3-8b": {
                "url": "http://localhost:11434",
                "check": lambda: self._check_model("llama3-8b")
            }
        }
        
        self.external_dependencies = {
            "perplexity": {
                "check": self._check_perplexity
            }
        }
    
    def _check_ollama(self) -> DependencyStatus:
        """Check if Ollama is running."""
        start_time = time.time()
        try:
            response = requests.get("http://localhost:11434/api/tags", timeout=5)
            response_time = time.time() - start_time
            
            if response.status_code == 200:
                return DependencyStatus(
                    name="ollama",
                    available=True,
                    response_time=response_time,
                    last_checked=datetime.now().isoformat()
                )
            else:
                return DependencyStatus(
                    name="ollama",
                    available=False,
                    response_time=response_time,
                    last_checked=datetime.now().isoformat(),
                    error_message=f"HTTP {response.status_code}"
                )
        except Exception as e:
            response_time = time.time() - start_time
            return DependencyStatus(
                name="ollama",
                available=False,
                response_time=response_time,
                last_checked=datetime.now().isoformat(),
                error_message=str(e)
            )
    
    def _check_model(self, model_name: str) -> DependencyStatus:
        """Check if a specific model is available."""
        start_time = time.time()
        try:
            response = requests.get("http://localhost:11434/api/tags", timeout=5)
            response_time = time.time() - start_time
            
            if response.status_code == 200:
                models = [m["name"] for m in response.json().get("models", [])]
                available = model_name in models
                
                return DependencyStatus(
                    name=model_name,
                    available=available,
                    response_time=response_time,
                    last_checked=datetime.now().isoformat(),
                    error_message=None if available else "Model not found"
                )
            else:
                return DependencyStatus(
                    name=model_name,
                    available=False,
                    response_time=response_time,
                    last_checked=datetime.now().isoformat(),
                    error_message=f"HTTP {response.status_code}"
                )
        except Exception as e:
            response_time = time.time() - start_time
            return DependencyStatus(
                name=model_name,
                available=False,
                response_time=response_time,
                last_checked=datetime.now().isoformat(),
                error_message=str(e)
            )
    
    def _check_perplexity(self) -> DependencyStatus:
        """Check if Perplexity is available."""
        # Simplified check - in reality would test actual API
        start_time = time.time()
        try:
            # Simulate API check
            time.sleep(0.1)
            response_time = time.time() - start_time
            
            # For now, assume available if we have an API key
            import os
            available = bool(os.environ.get("PERPLEXITY_API_KEY"))
            
            return DependencyStatus(
                name="perplexity",
                available=available,
                response_time=response_time,
                last_checked=datetime.now().isoformat(),
                error_message=None if available else "API key not set"
            )
        except Exception as e:
            response_time = time.time() - start_time
            return DependencyStatus(
                name="perplexity",
                available=False,
                response_time=response_time,
                last_checked=datetime.now().isoformat(),
                error_message=str(e)
            )
    
    def check_all(self) -> Dict[str, List[DependencyStatus]]:
        """Check all dependencies."""
        results = {
            "local": [],
            "external": []
        }
        
        # Check local dependencies
        for name, config in self.local_dependencies.items():
            status = config["check"]()
            results["local"].append(status)
        
        # Check external dependencies
        for name, config in self.external_dependencies.items():
            status = config["check"]()
            results["external"].append(status)
        
        return results
    
    def generate_report(self, results: Dict[str, List[DependencyStatus]]) -> Dict[str, Any]:
        """Generate a monitoring report."""
        report = {
            "timestamp": datetime.now().isoformat(),
            "summary": {
                "total_dependencies": 0,
                "available": 0,
                "unavailable": 0,
                "local_available": 0,
                "external_available": 0
            },
            "details": {
                "local": [],
                "external": []
            },
            "recommendations": []
        }
        
        # Process local dependencies
        for status in results["local"]:
            report["summary"]["total_dependencies"] += 1
            if status.available:
                report["summary"]["available"] += 1
                report["summary"]["local_available"] += 1
            else:
                report["summary"]["unavailable"] += 1
            
            report["details"]["local"].append({
                "name": status.name,
                "available": status.available,
                "response_time": status.response_time,
                "last_checked": status.last_checked,
                "error_message": status.error_message
            })
        
        # Process external dependencies
        for status in results["external"]:
            report["summary"]["total_dependencies"] += 1
            if status.available:
                report["summary"]["available"] += 1
                report["summary"]["external_available"] += 1
            else:
                report["summary"]["unavailable"] += 1
            
            report["details"]["external"].append({
                "name": status.name,
                "available": status.available,
                "response_time": status.response_time,
                "last_checked": status.last_checked,
                "error_message": status.error_message
            })
        
        # Generate recommendations
        if report["summary"]["local_available"] == 0:
            report["recommendations"].append({
                "priority": "high",
                "message": "No local models available - check Ollama installation",
                "action": "Run: ollama serve"
            })
        
        if report["summary"]["external_available"] == 0:
            report["recommendations"].append({
                "priority": "medium",
                "message": "No external services available - check API keys",
                "action": "Set PERPLEXITY_API_KEY environment variable"
            })
        
        if report["summary"]["local_available"] > 0 and report["summary"]["external_available"] == 0:
            report["recommendations"].append({
                "priority": "low",
                "message": "Running in local-only mode",
                "action": "Consider setting up external services for better quality"
            })
        
        return report

def main():
    parser = argparse.ArgumentParser(description="Monitor AI dependencies")
    parser.add_argument("--output", help="Output file for report")
    parser.add_argument("--format", default="json", choices=["json", "text"],
                       help="Output format")
    
    args = parser.parse_args()
    
    # Create monitor
    monitor = DependencyMonitor()
    
    # Check dependencies
    print("Checking dependencies...")
    results = monitor.check_all()
    
    # Generate report
    report = monitor.generate_report(results)
    
    # Output report
    if args.format == "json":
        output = json.dumps(report, indent=2)
    else:
        # Text format
        output = f"""Dependency Monitor Report
Generated: {report['timestamp']}

Summary:
  Total dependencies: {report['summary']['total_dependencies']}
  Available: {report['summary']['available']}
  Unavailable: {report['summary']['unavailable']}
  Local available: {report['summary']['local_available']}
  External available: {report['summary']['external_available']}

Local Dependencies:
"""
        for dep in report['details']['local']:
            status = "✓" if dep['available'] else "✗"
            output += f"  {status} {dep['name']}: {dep['response_time']:.3f}s"
            if dep['error_message']:
                output += f" - {dep['error_message']}"
            output += "\n"
        
        output += "\nExternal Dependencies:\n"
        for dep in report['details']['external']:
            status = "✓" if dep['available'] else "✗"
            output += f"  {status} {dep['name']}: {dep['response_time']:.3f}s"
            if dep['error_message']:
                output += f" - {dep['error_message']}"
            output += "\n"
        
        if report['recommendations']:
            output += "\nRecommendations:\n"
            for rec in report['recommendations']:
                output += f"  [{rec['priority'].upper()}] {rec['message']}\n"
                output += f"    Action: {rec['action']}\n"
    
    # Print or save
    if args.output:
        with open(args.output, 'w') as f:
            f.write(output)
        print(f"Report saved to {args.output}")
    else:
        print(output)
    
    return 0

if __name__ == "__main__":
    import sys
    sys.exit(main())