Deploy failsafe watchdog: app.py

app.py

@@ -0,0 +1,445 @@
+#!/usr/bin/env python3
+"""
+🛡️ FAILSAFE WATCHDOG - Immortal Guardian Space
+Monitors Agent Zero workspace, auto-heals on failure, persists state via Mem0
+Chimera Protocol: ZeroGPU/FreeCPU only - no credit consumption
+"""
+
+import os
+import sys
+import json
+import time
+import hashlib
+import asyncio
+import threading
+import requests
+from datetime import datetime, timedelta
+from typing import Dict, List, Optional, Any
+from dataclasses import dataclass, asdict
+from pathlib import Path
+
+import gradio as gr
+from huggingface_hub import HfApi
+
+# Optional: Mem0 for persistent memory
+try:
+    from mem0 import Memory
+    MEM0_AVAILABLE = True
+except ImportError:
+    MEM0_AVAILABLE = False
+    print("⚠️ Mem0 not installed - using local JSON fallback")
+
+# Configuration
+WATCHDOG_CONFIG = {
+    "target_workspace": "ScottzillaSystems/agent-zero",
+    "check_interval": 30,  # seconds
+    "failure_threshold": 3,
+    "mem0_enabled": True,
+    "self_heal_enabled": True,
+    "backup_watchdog_count": 2,
+    "hf_token": os.environ.get("HF_TOKEN"),
+    "mem0_api_key": os.environ.get("MEM0_API_KEY")
+}
+
+@dataclass
+class HealthCheck:
+    timestamp: str
+    status: str  # healthy, degraded, failed
+    latency_ms: float
+    error_patterns: List[str]
+    action_taken: Optional[str] = None
+
+class Mem0Persistence:
+    """Cross-restart memory via Mem0 or local fallback"""
+    
+    def __init__(self, api_key: Optional[str] = None):
+        self.api_key = api_key or os.environ.get("MEM0_API_KEY")
+        self.user_id = "failsafe_watchdog"
+        
+        if MEM0_AVAILABLE and self.api_key:
+            try:
+                self.memory = Memory(api_key=self.api_key)
+                self.mode = "mem0_cloud"
+                print("✅ Mem0 cloud persistence active")
+            except Exception as e:
+                print(f"⚠️ Mem0 init failed: {e}, using local fallback")
+                self.mode = "local_json"
+        else:
+            self.mode = "local_json"
+            print("✅ Local JSON persistence active")
+        
+        self.local_path = Path("/data/watchdog_memory.json")
+        self._ensure_local_storage()
+    
+    def _ensure_local_storage(self):
+        if self.mode == "local_json":
+            self.local_path.parent.mkdir(parents=True, exist_ok=True)
+            if not self.local_path.exists():
+                self.local_path.write_text(json.dumps({"incidents": [], "state": {}}))
+    
+    def add(self, message: str, metadata: Dict = None):
+        """Store memory entry"""
+        if self.mode == "mem0_cloud":
+            try:
+                self.memory.add(message, user_id=self.user_id, metadata=metadata or {})
+                return True
+            except Exception as e:
+                print(f"Mem0 add failed: {e}, falling back to local")
+        
+        # Local fallback
+        data = json.loads(self.local_path.read_text())
+        data["incidents"].append({
+            "timestamp": datetime.utcnow().isoformat(),
+            "message": message,
+            "metadata": metadata or {}
+        })
+        self.local_path.write_text(json.dumps(data, indent=2))
+        return True
+    
+    def search(self, query: str, limit: int = 10) -> List[Dict]:
+        """Search memory"""
+        if self.mode == "mem0_cloud":
+            try:
+                return self.memory.search(query, user_id=self.user_id, limit=limit)
+            except Exception as e:
+                print(f"Mem0 search failed: {e}")
+        
+        # Local fallback - simple text search
+        data = json.loads(self.local_path.read_text())
+        results = []
+        for incident in data["incidents"]:
+            if query.lower() in incident["message"].lower():
+                results.append(incident)
+        return results[-limit:]
+    
+    def get_state(self) -> Dict:
+        """Get current watchdog state"""
+        if self.mode == "local_json":
+            data = json.loads(self.local_path.read_text())
+            return data.get("state", {})
+        return {}
+    
+    def save_state(self, state: Dict):
+        """Save watchdog state"""
+        if self.mode == "local_json":
+            data = json.loads(self.local_path.read_text())
+            data["state"] = state
+            self.local_path.write_text(json.dumps(data, indent=2))
+
+class WorkspaceMonitor:
+    """Monitors Agent Zero workspace health"""
+    
+    FAILURE_PATTERNS = {
+        "rate_limit": ["Key limit exceeded", "429", "quota exceeded", "rate limit"],
+        "fd_exhaustion": ["Too many open files", "EMFILE", "Errno 24"],
+        "memory_pressure": ["MemoryError", "OOM", "Killed process", "out of memory"],
+        "api_blackout": ["Connection refused", "Timeout", "Name resolution failed"],
+        "auth_failure": ["401", "403", "Unauthorized", "Invalid token"]
+    }
+    
+    def __init__(self, target_space: str, hf_token: str):
+        self.target = target_space
+        self.hf_token = hf_token
+        self.api = HfApi(token=hf_token)
+        self.history: List[HealthCheck] = []
+        self.consecutive_failures = 0
+    
+    async def check_health(self) -> HealthCheck:
+        """Perform health check on target workspace"""
+        start = time.time()
+        error_patterns = []
+        status = "healthy"
+        action = None
+        
+        try:
+            # Check HF Space runtime status
+            space_info = self.api.space_info(self.target)
+            runtime = getattr(space_info, 'runtime', {})
+            stage = runtime.get('stage', 'UNKNOWN')
+            
+            if stage == 'RUNNING':
+                status = "healthy"
+            elif stage == 'BUILD_ERROR':
+                status = "failed"
+                error_patterns.append("build_error")
+                action = "trigger_rebuild"
+            elif stage == 'SLEEPING':
+                status = "degraded"
+                action = "wake_space"
+            else:
+                status = "degraded"
+                
+        except Exception as e:
+            status = "failed"
+            error_msg = str(e).lower()
+            
+            # Pattern matching
+            for pattern_type, patterns in self.FAILURE_PATTERNS.items():
+                if any(p.lower() in error_msg for p in patterns):
+                    error_patterns.append(pattern_type)
+            
+            if not error_patterns:
+                error_patterns.append("unknown")
+            
+            action = self._determine_action(error_patterns)
+        
+        latency = (time.time() - start) * 1000
+        
+        check = HealthCheck(
+            timestamp=datetime.utcnow().isoformat(),
+            status=status,
+            latency_ms=latency,
+            error_patterns=error_patterns,
+            action_taken=action
+        )
+        
+        self.history.append(check)
+        if len(self.history) > 100:
+            self.history = self.history[-100:]
+        
+        # Track consecutive failures
+        if status == "failed":
+            self.consecutive_failures += 1
+        else:
+            self.consecutive_failures = 0
+        
+        return check
+    
+    def _determine_action(self, error_patterns: List[str]) -> str:
+        """Determine remediation action from error patterns"""
+        if "fd_exhaustion" in error_patterns:
+            return "restart_container"
+        elif "rate_limit" in error_patterns:
+            return "switch_to_local_inference"
+        elif "memory_pressure" in error_patterns:
+            return "scale_down_agents"
+        elif "build_error" in error_patterns:
+            return "trigger_rebuild"
+        else:
+            return "restart_container"
+    
+    async def execute_remediation(self, action: str) -> bool:
+        """Execute remediation action"""
+        print(f"🔧 Executing remediation: {action}")
+        
+        try:
+            if action == "restart_container":
+                # Trigger space restart via empty commit
+                self.api.upload_file(
+                    repo_id=self.target,
+                    repo_type="space",
+                    path_in_repo=".watchdog_trigger",
+                    path_or_fileobj=f"restart_{int(time.time())}".encode(),
+                    commit_message="[FAILSAFE] Auto-restart due to health check failure"
+                )
+                return True
+                
+            elif action == "trigger_rebuild":
+                # Touch README to trigger rebuild
+                self.api.upload_file(
+                    repo_id=self.target,
+                    repo_type="space",
+                    path_in_repo="README.md",
+                    path_or_fileobj=b"# Trigger rebuild\n",
+                    commit_message="[FAILSAFE] Trigger rebuild"
+                )
+                return True
+                
+            elif action == "wake_space":
+                # Send request to wake sleeping space
+                requests.get(f"https://{self.target.replace('/', '-')}.hf.space", timeout=10)
+                return True
+                
+        except Exception as e:
+            print(f"❌ Remediation failed: {e}")
+            return False
+        
+        return False
+
+class SelfReplication:
+    """Spawns backup watchdogs if current instance degrades"""
+    
+    def __init__(self, hf_token: str, mem0: Mem0Persistence):
+        self.hf_token = hf_token
+        self.mem0 = mem0
+        self.api = HfApi(token=hf_token)
+        self.backup_spaces = [
+            "ScottzillaSystems/failsafe-watchdog-beta",
+            "ScottzillaSystems/failsafe-watchdog-gamma"
+        ]
+    
+    def check_own_health(self) -> bool:
+        """Check if this watchdog is healthy"""
+        # Simple checks
+        if not self.hf_token:
+            return False
+        
+        # Check if we can write to memory
+        try:
+            self.mem0.add("health_check", {"type": "self_check"})
+            return True
+        except:
+            return False
+    
+    def spawn_backup(self) -> Optional[str]:
+        """Spawn backup watchdog instance"""
+        for backup_space in self.backup_spaces:
+            try:
+                # Check if backup exists
+                try:
+                    self.api.space_info(backup_space)
+                    print(f"✅ Backup {backup_space} already exists")
+                    return backup_space
+                except:
+                    # Create backup space
+                    self.api.create_repo(
+                        repo_id=backup_space,
+                        repo_type="space",
+                        space_sdk="gradio"
+                    )
+                    print(f"🚀 Spawned backup watchdog: {backup_space}")
+                    return backup_space
+            except Exception as e:
+                print(f"⚠️ Failed to spawn {backup_space}: {e}")
+        
+        return None
+
+class FailsafeWatchdog:
+    """Main watchdog orchestrator"""
+    
+    def __init__(self):
+        self.config = WATCHDOG_CONFIG
+        self.mem0 = Mem0Persistence(self.config["mem0_api_key"])
+        self.monitor = WorkspaceMonitor(
+            self.config["target_workspace"],
+            self.config["hf_token"]
+        )
+        self.replication = SelfReplication(
+            self.config["hf_token"],
+            self.mem0
+        )
+        self.running = False
+        self.status_log: List[Dict] = []
+        
+        # Restore state from Mem0
+        self._restore_state()
+    
+    def _restore_state(self):
+        """Restore previous state from persistent memory"""
+        state = self.mem0.get_state()
+        if state:
+            print(f"📥 Restored state: {state.get('restart_count', 0)} previous restarts")
+    
+    async def run(self):
+        """Main watchdog loop"""
+        self.running = True
+        print("🛡️ Failsafe Watchdog started")
+        
+        while self.running:
+            try:
+                # Health check
+                check = await self.monitor.check_health()
+                
+                # Log to Mem0
+                if check.status in ["failed", "degraded"]:
+                    self.mem0.add(
+                        f"Health check {check.status}: {', '.join(check.error_patterns)}",
+                        {"type": "health_check", "status": check.status}
+                    )
+                
+                # Execute remediation if needed
+                if check.action_taken and self.monitor.consecutive_failures >= self.config["failure_threshold"]:
+                    success = await self.monitor.execute_remediation(check.action_taken)
+                    
+                    self.mem0.add(
+                        f"Remediation {check.action_taken}: {'success' if success else 'failed'}",
+                        {"type": "remediation", "action": check.action_taken, "success": success}
+                    )
+                    
+                    if success:
+                        self.monitor.consecutive_failures = 0
+                
+                # Self-replication check
+                if not self.replication.check_own_health():
+                    backup = self.replication.spawn_backup()
+                    if backup:
+                        self.mem0.add(
+                            f"Self-replication triggered: spawned {backup}",
+                            {"type": "self_replication", "backup": backup}
+                        )
+                
+                # Update status log
+                self.status_log.append(asdict(check))
+                if len(self.status_log) > 50:
+                    self.status_log = self.status_log[-50:]
+                
+                # Save state
+                self.mem0.save_state({
+                    "last_check": check.timestamp,
+                    "restart_count": len([s for s in self.status_log if s.get("action_taken")]),
+                    "consecutive_failures": self.monitor.consecutive_failures
+                })
+                
+                await asyncio.sleep(self.config["check_interval"])
+                
+            except Exception as e:
+                print(f"❌ Watchdog error: {e}")
+                await asyncio.sleep(5)
+    
+    def get_dashboard_data(self) -> Dict:
+        """Get data for Gradio dashboard"""
+        recent_checks = self.status_log[-10:]
+        
+        return {
+            "status": "running" if self.running else "stopped",
+            "target": self.config["target_workspace"],
+            "last_check": recent_checks[-1] if recent_checks else None,
+            "health_history": recent_checks,
+            "consecutive_failures": self.monitor.consecutive_failures,
+            "mem0_mode": self.mem0.mode,
+            "incidents": self.mem0.search("failed", limit=5) if hasattr(self, 'mem0') else []
+        }
+
+# Gradio UI
+watchdog = FailsafeWatchdog()
+
+def start_watchdog():
+    if not watchdog.running:
+        asyncio.create_task(watchdog.run())
+    return "🛡️ Watchdog started"
+
+def stop_watchdog():
+    watchdog.running = False
+    return "⏹️ Watchdog stopped"
+
+def get_status():
+    return json.dumps(watchdog.get_dashboard_data(), indent=2, default=str)
+
+def view_incidents():
+    incidents = watchdog.mem0.search("incident", limit=10)
+    return json.dumps(incidents, indent=2, default=str)
+
+with gr.Blocks(title="Failsafe Watchdog") as demo:
+    gr.Markdown("# 🛡️ Failsafe Watchdog - Immortal Guardian")
+    gr.Markdown("Monitors Agent Zero workspace, auto-heals on failure, persists via Mem0")
+    
+    with gr.Row():
+        start_btn = gr.Button("▶️ Start Watchdog", variant="primary")
+        stop_btn = gr.Button("⏹️ Stop Watchdog", variant="secondary")
+    
+    with gr.Row():
+        status_btn = gr.Button("📊 Get Status")
+        incidents_btn = gr.Button("📋 View Incidents")
+    
+    output = gr.Textbox(label="Output", lines=20)
+    
+    start_btn.click(start_watchdog, outputs=output)
+    stop_btn.click(stop_watchdog, outputs=output)
+    status_btn.click(get_status, outputs=output)
+    incidents_btn.click(view_incidents, outputs=output)
+    
+    # Auto-refresh status
+    demo.load(get_status, outputs=output, every=30)
+
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=7860)