#!/usr/bin/env python3 """ Genesis Health API ================== Simple HTTP API for health monitoring, callable from n8n webhooks. Usage: python3 health_api.py [port] # Default port: 8765 Endpoints: GET /health - Full health status GET /heartbeat - Heartbeat status only GET /circuits - Circuit breaker status only GET /ping - Fast health check (no initialization) GET /mcp-status - MCP sync status POST /pulse - Trigger heartbeat pulse POST /mcp-sync - Flush MCP sync queue """ import json import sys import threading import time from datetime import datetime from http.server import HTTPServer, BaseHTTPRequestHandler from urllib.parse import urlparse # Add current directory to path sys.path.insert(0, '/mnt/e/genesis-system/core') # Response cache TTL in seconds (5 minutes to avoid slow refreshes during normal operation) CACHE_TTL = 300 class ComponentCache: """ Singleton cache for heavy Genesis components. Initializes once at startup, reuses for all requests. """ _instance = None _lock = threading.Lock() def __new__(cls): if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._initialized = False return cls._instance def initialize(self): """Initialize all components once.""" if self._initialized: return print("[HealthAPI] Initializing component cache...") # Response cache for expensive operations self._heartbeat_status_cache = None self._heartbeat_status_time = 0 self._mcp_status_cache = None self._mcp_status_time = 0 # Heartbeat (heavy - Redis, vectors, etc.) self._heartbeat = None self._heartbeat_error = None try: from genesis_heartbeat import GenesisHeartbeat self._heartbeat = GenesisHeartbeat() print("[OK] GenesisHeartbeat initialized") except Exception as e: self._heartbeat_error = str(e) print(f"[WARN] GenesisHeartbeat failed: {e}") # MCP Sync Manager (lighter) self._mcp_manager = None self._mcp_error = None try: from reliable_mcp_sync import MCPSyncManager self._mcp_manager = MCPSyncManager() print("[OK] MCPSyncManager initialized") except Exception as e: self._mcp_error = str(e) print(f"[WARN] MCPSyncManager failed: {e}") self._initialized = True print("[HealthAPI] Component cache ready") @property def heartbeat(self): return self._heartbeat @property def heartbeat_error(self): return self._heartbeat_error @property def mcp_manager(self): return self._mcp_manager @property def mcp_error(self): return self._mcp_error def get_heartbeat_status_cached(self) -> dict: """Get heartbeat status with caching (stale-while-revalidate pattern).""" now = time.time() cache_age = now - self._heartbeat_status_time if self._heartbeat_status_cache else float('inf') # Return cached data if available (even if stale) if self._heartbeat_status_cache and cache_age < CACHE_TTL: return self._heartbeat_status_cache # If cache is stale but exists, return it and refresh in background if self._heartbeat_status_cache: # Start background refresh if not already running if not getattr(self, '_heartbeat_refresh_running', False): self._heartbeat_refresh_running = True thread = threading.Thread(target=self._refresh_heartbeat_cache, daemon=True) thread.start() return self._heartbeat_status_cache # No cache - must fetch synchronously (first request) return self._fetch_heartbeat_status() def _fetch_heartbeat_status(self) -> dict: """Fetch fresh heartbeat status.""" if self._heartbeat is None: return {'error': self._heartbeat_error, 'running': False} try: status = self._heartbeat.status() self._heartbeat_status_cache = status self._heartbeat_status_time = time.time() return status except Exception as e: return {'error': str(e), 'running': False} def _refresh_heartbeat_cache(self): """Background refresh of heartbeat cache.""" try: self._fetch_heartbeat_status() finally: self._heartbeat_refresh_running = False def get_mcp_status_cached(self) -> dict: """Get MCP status with caching (stale-while-revalidate pattern).""" now = time.time() cache_age = now - self._mcp_status_time if self._mcp_status_cache else float('inf') # Return cached data if available and fresh if self._mcp_status_cache and cache_age < CACHE_TTL: return self._mcp_status_cache # If cache is stale but exists, return it and refresh in background if self._mcp_status_cache: if not getattr(self, '_mcp_refresh_running', False): self._mcp_refresh_running = True thread = threading.Thread(target=self._refresh_mcp_cache, daemon=True) thread.start() return self._mcp_status_cache # No cache - must fetch synchronously return self._fetch_mcp_status() def _fetch_mcp_status(self) -> dict: """Fetch fresh MCP status.""" if self._mcp_manager is None: return {'error': self._mcp_error} try: status = self._mcp_manager.get_sync_status() self._mcp_status_cache = status self._mcp_status_time = time.time() return status except Exception as e: return {'error': str(e)} def _refresh_mcp_cache(self): """Background refresh of MCP cache.""" try: self._fetch_mcp_status() finally: self._mcp_refresh_running = False # Global component cache _cache = None def get_cache() -> ComponentCache: """Get the initialized component cache.""" global _cache if _cache is None: _cache = ComponentCache() _cache.initialize() return _cache class HealthHandler(BaseHTTPRequestHandler): """HTTP request handler for Genesis health endpoints.""" def _send_json(self, data: dict, status: int = 200): """Send JSON response.""" self.send_response(status) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(json.dumps(data, indent=2).encode()) def _send_error(self, message: str, status: int = 500): """Send error response.""" self._send_json({'error': message}, status) def do_GET(self): """Handle GET requests.""" path = urlparse(self.path).path if path == '/ping': # Ultra-fast endpoint - no component access self._send_json({'status': 'ok', 'service': 'genesis-health-api'}) elif path == '/health': self._handle_health() elif path == '/heartbeat': self._handle_heartbeat() elif path == '/circuits': self._handle_circuits() elif path == '/mcp-status': self._handle_mcp_status() else: self._send_error('Not found', 404) def do_POST(self): """Handle POST requests.""" path = urlparse(self.path).path if path == '/pulse': self._handle_pulse() elif path == '/mcp-sync': self._handle_mcp_sync() else: self._send_error('Not found', 404) def do_OPTIONS(self): """Handle CORS preflight.""" self.send_response(200) self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type') self.end_headers() def _handle_health(self): """Full health check using cached components.""" try: cache = get_cache() # Use cached status to avoid slow repeated calls heartbeat = cache.get_heartbeat_status_cached() circuits = self._get_circuit_status() mcp = cache.get_mcp_status_cached() healthy = ( not heartbeat.get('error') and circuits.get('healthy', True) and mcp.get('stats', {}).get('total_failed', 0) <= 3 ) self._send_json({ 'healthy': healthy, 'timestamp': datetime.now().isoformat(), 'cache_ttl_seconds': CACHE_TTL, 'heartbeat': heartbeat, 'circuit_breakers': circuits, 'mcp_sync': mcp }) except Exception as e: self._send_error(str(e)) def _handle_heartbeat(self): """Heartbeat status only using cached component.""" try: cache = get_cache() status = cache.get_heartbeat_status_cached() self._send_json(status) except Exception as e: self._send_error(str(e)) def _handle_circuits(self): """Circuit breaker status only.""" try: status = self._get_circuit_status() self._send_json(status) except Exception as e: self._send_error(str(e)) def _handle_mcp_status(self): """MCP sync status using cached component.""" try: cache = get_cache() status = cache.get_mcp_status_cached() self._send_json(status) except Exception as e: self._send_error(str(e)) def _handle_pulse(self): """Trigger heartbeat pulse using cached component.""" try: cache = get_cache() if cache.heartbeat is None: self._send_error(f'Heartbeat not available: {cache.heartbeat_error}') return result = cache.heartbeat.pulse() self._send_json({'success': True, 'result': result}) except Exception as e: self._send_error(str(e)) def _handle_mcp_sync(self): """Flush MCP sync queue using cached component.""" try: cache = get_cache() if cache.mcp_manager is None: self._send_error(f'MCP manager not available: {cache.mcp_error}') return results = cache.mcp_manager.flush_queue() status = cache.mcp_manager.get_sync_status() self._send_json({ 'success': True, 'flushed': len(results) if results else 0, 'status': status }) except Exception as e: self._send_error(str(e)) def _get_circuit_status(self) -> dict: """Get circuit breaker status.""" try: from circuit_breaker import get_all_status return get_all_status() except Exception as e: return {'error': str(e), 'healthy': True} def log_message(self, format, *args): """Log HTTP requests.""" print(f"[HealthAPI] {args[0]}") def run_server(port: int = 8765): """Run the health API server.""" # Pre-initialize components before starting server print(f"Genesis Health API initializing on port {port}...") cache = get_cache() # Pre-populate response cache (this is the slow part) print("[HealthAPI] Pre-populating response cache...") cache.get_heartbeat_status_cached() cache.get_mcp_status_cached() print("[HealthAPI] Cache ready") server = HTTPServer(('0.0.0.0', port), HealthHandler) print(f"\nGenesis Health API running on http://0.0.0.0:{port}") print("Endpoints:") print(" GET /ping - Fast health check (instant)") print(" GET /health - Full health status") print(" GET /heartbeat - Heartbeat status") print(" GET /circuits - Circuit breaker status") print(" GET /mcp-status - MCP sync status") print(" POST /pulse - Trigger heartbeat pulse") print(" POST /mcp-sync - Flush MCP sync queue") print() try: server.serve_forever() except KeyboardInterrupt: print("\nShutting down...") server.shutdown() if __name__ == '__main__': port = int(sys.argv[1]) if len(sys.argv) > 1 else 8765 run_server(port)