#!/usr/bin/env python3 """ GENESIS ORCHESTRATOR ===================== Master orchestrator that ties all Genesis components into a unified autonomous system. Responsibilities: - Initialize all subsystems in correct order - Coordinate between agents, memory, learning, and verification - Manage the autonomous execution loop - Handle system health and recovery - Provide unified interface for all operations Usage: orchestrator = GenesisOrchestrator() orchestrator.start() orchestrator.submit_task(task) """ import asyncio import json import os import signal import sys import threading import time from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from typing import Dict, List, Any, Optional, Callable # Genesis module imports try: from multi_agent_coordinator import MultiAgentCoordinator, Task, TaskComplexity, AgentType COORDINATOR_AVAILABLE = True except ImportError: COORDINATOR_AVAILABLE = False try: from learning_loop import LearningLoop LEARNING_AVAILABLE = True except ImportError: LEARNING_AVAILABLE = False try: from verified_executor import VerifiedExecutor VERIFIER_AVAILABLE = True except ImportError: VERIFIER_AVAILABLE = False try: from event_bus import EventBus, Event, get_event_bus EVENT_BUS_AVAILABLE = True except ImportError: EVENT_BUS_AVAILABLE = False try: from performance_profiler import PerformanceProfiler PROFILER_AVAILABLE = True except ImportError: PROFILER_AVAILABLE = False try: from rollback_system import RollbackSystem ROLLBACK_AVAILABLE = True except ImportError: ROLLBACK_AVAILABLE = False try: from genesis_memory_cortex import GenesisCortex MEMORY_AVAILABLE = True except ImportError: MEMORY_AVAILABLE = False @dataclass class OrchestratorState: """Current state of the orchestrator.""" status: str = "initialized" started_at: Optional[str] = None tasks_completed: int = 0 tasks_failed: int = 0 total_cost: float = 0.0 current_task: Optional[str] = None agents_active: int = 0 last_checkpoint: Optional[str] = None @dataclass class SubsystemStatus: """Status of a subsystem.""" name: str available: bool healthy: bool last_check: str details: Dict = field(default_factory=dict) class GenesisOrchestrator: """ Master orchestrator for the Genesis autonomous system. Coordinates all subsystems: - Multi-Agent Coordinator (task execution) - Learning Loop (continuous improvement) - Verified Executor (adversarial verification) - Event Bus (inter-module communication) - Performance Profiler (metrics and optimization) - Rollback System (recovery) - Memory Cortex (knowledge persistence) """ def __init__( self, genesis_root: Path = None, enable_learning: bool = True, enable_verification: bool = True, enable_profiling: bool = True ): self.genesis_root = genesis_root or Path(__file__).parent.parent self.state = OrchestratorState() self._running = False self._shutdown_event = threading.Event() self._task_queue: List[Task] = [] self._lock = threading.RLock() # Configuration self.enable_learning = enable_learning self.enable_verification = enable_verification self.enable_profiling = enable_profiling # Initialize subsystems self._init_subsystems() # Set up signal handlers signal.signal(signal.SIGINT, self._signal_handler) signal.signal(signal.SIGTERM, self._signal_handler) def _init_subsystems(self): """Initialize all Genesis subsystems.""" self.subsystems: Dict[str, SubsystemStatus] = {} # Event Bus (initialize first for inter-module communication) self.event_bus = None if EVENT_BUS_AVAILABLE: try: self.event_bus = get_event_bus() self.subsystems["event_bus"] = SubsystemStatus( name="Event Bus", available=True, healthy=True, last_check=datetime.now().isoformat() ) except Exception as e: self.subsystems["event_bus"] = SubsystemStatus( name="Event Bus", available=False, healthy=False, last_check=datetime.now().isoformat(), details={"error": str(e)} ) # Multi-Agent Coordinator self.coordinator = None if COORDINATOR_AVAILABLE: try: self.coordinator = MultiAgentCoordinator(enable_learning=self.enable_learning) self.subsystems["coordinator"] = SubsystemStatus( name="Multi-Agent Coordinator", available=True, healthy=True, last_check=datetime.now().isoformat() ) except Exception as e: self.subsystems["coordinator"] = SubsystemStatus( name="Multi-Agent Coordinator", available=False, healthy=False, last_check=datetime.now().isoformat(), details={"error": str(e)} ) # Learning Loop self.learning = None if LEARNING_AVAILABLE and self.enable_learning: try: self.learning = LearningLoop() self.subsystems["learning"] = SubsystemStatus( name="Learning Loop", available=True, healthy=True, last_check=datetime.now().isoformat() ) except Exception as e: self.subsystems["learning"] = SubsystemStatus( name="Learning Loop", available=False, healthy=False, last_check=datetime.now().isoformat(), details={"error": str(e)} ) # Verified Executor self.verifier = None if VERIFIER_AVAILABLE and self.enable_verification: try: self.verifier = VerifiedExecutor() self.subsystems["verifier"] = SubsystemStatus( name="Verified Executor", available=True, healthy=True, last_check=datetime.now().isoformat() ) except Exception as e: self.subsystems["verifier"] = SubsystemStatus( name="Verified Executor", available=False, healthy=False, last_check=datetime.now().isoformat(), details={"error": str(e)} ) # Performance Profiler self.profiler = None if PROFILER_AVAILABLE and self.enable_profiling: try: self.profiler = PerformanceProfiler() self.subsystems["profiler"] = SubsystemStatus( name="Performance Profiler", available=True, healthy=True, last_check=datetime.now().isoformat() ) except Exception as e: self.subsystems["profiler"] = SubsystemStatus( name="Performance Profiler", available=False, healthy=False, last_check=datetime.now().isoformat(), details={"error": str(e)} ) # Rollback System self.rollback = None if ROLLBACK_AVAILABLE: try: self.rollback = RollbackSystem(self.genesis_root) self.subsystems["rollback"] = SubsystemStatus( name="Rollback System", available=True, healthy=True, last_check=datetime.now().isoformat() ) except Exception as e: self.subsystems["rollback"] = SubsystemStatus( name="Rollback System", available=False, healthy=False, last_check=datetime.now().isoformat(), details={"error": str(e)} ) # Memory Cortex self.memory = None if MEMORY_AVAILABLE: try: self.memory = GenesisCortex() self.subsystems["memory"] = SubsystemStatus( name="Memory Cortex", available=True, healthy=True, last_check=datetime.now().isoformat() ) except Exception as e: self.subsystems["memory"] = SubsystemStatus( name="Memory Cortex", available=False, healthy=False, last_check=datetime.now().isoformat(), details={"error": str(e)} ) def _signal_handler(self, signum, frame): """Handle shutdown signals.""" print("\n[Orchestrator] Shutdown signal received...") self.stop() def start(self): """Start the orchestrator.""" if self._running: return self._running = True self.state.status = "running" self.state.started_at = datetime.now().isoformat() # Create initial checkpoint if self.rollback: self.rollback.create_checkpoint("Orchestrator start") self.state.last_checkpoint = datetime.now().isoformat() # Publish start event if self.event_bus: self.event_bus.publish( "orchestrator.started", {"timestamp": self.state.started_at} ) print(f"[Orchestrator] Started at {self.state.started_at}") print(f"[Orchestrator] Subsystems: {len([s for s in self.subsystems.values() if s.healthy])}/{len(self.subsystems)} healthy") def stop(self): """Stop the orchestrator gracefully.""" if not self._running: return self._running = False self._shutdown_event.set() self.state.status = "stopped" # Create shutdown checkpoint if self.rollback: self.rollback.create_checkpoint("Orchestrator shutdown") # Publish stop event if self.event_bus: self.event_bus.publish( "orchestrator.stopped", { "tasks_completed": self.state.tasks_completed, "tasks_failed": self.state.tasks_failed, "total_cost": self.state.total_cost } ) print(f"[Orchestrator] Stopped. Tasks: {self.state.tasks_completed} completed, {self.state.tasks_failed} failed") def submit_task( self, task_id: str, title: str, description: str, task_type: str = "general", complexity: str = "moderate", verify: bool = None ) -> Dict: """ Submit a task for execution. Args: task_id: Unique task identifier title: Task title description: Task description task_type: Type of task complexity: Task complexity verify: Whether to run verification (default: use orchestrator setting) Returns: Execution result dictionary """ if not self._running: return {"error": "Orchestrator not running", "task_id": task_id} # Create task if COORDINATOR_AVAILABLE: task = Task( id=task_id, title=title, description=description, task_type=task_type, complexity=TaskComplexity[complexity.upper()] if hasattr(TaskComplexity, complexity.upper()) else TaskComplexity.MODERATE ) else: return {"error": "Coordinator not available", "task_id": task_id} self.state.current_task = task_id # Publish task start event if self.event_bus: self.event_bus.publish("task.started", {"task_id": task_id, "title": title}) # Profile execution start_time = time.time() try: # Execute with verification if enabled should_verify = verify if verify is not None else self.enable_verification if should_verify and self.verifier: result = self.verifier.execute_and_verify(task) success = result.verification_passed output = result.output cost = result.total_cost elif self.coordinator: exec_result = self.coordinator.execute_task(task) success = exec_result.success output = exec_result.output cost = exec_result.cost else: return {"error": "No execution engine available", "task_id": task_id} duration = time.time() - start_time # Record to profiler if self.profiler: self.profiler.record( operation=f"task:{task_type}", duration=duration, cost=cost, success=success ) # Update state if success: self.state.tasks_completed += 1 else: self.state.tasks_failed += 1 self.state.total_cost += cost self.state.current_task = None # Publish completion event if self.event_bus: self.event_bus.publish( "task.completed" if success else "task.failed", {"task_id": task_id, "success": success, "duration": duration} ) return { "task_id": task_id, "success": success, "output": output, "duration": duration, "cost": cost } except Exception as e: self.state.tasks_failed += 1 self.state.current_task = None if self.event_bus: self.event_bus.publish("task.error", {"task_id": task_id, "error": str(e)}) return {"task_id": task_id, "error": str(e)} def execute_batch( self, tasks: List[Dict], parallel: bool = True ) -> List[Dict]: """Execute a batch of tasks.""" results = [] if parallel and self.coordinator: # Convert to Task objects task_objects = [] for t in tasks: if COORDINATOR_AVAILABLE: task_objects.append(Task( id=t.get("id", f"batch-{len(task_objects)}"), title=t.get("title", ""), description=t.get("description", ""), task_type=t.get("task_type", "general"), complexity=TaskComplexity.MODERATE )) # Execute in parallel exec_results = self.coordinator.execute_parallel(task_objects) for task_id, result in exec_results.items(): results.append({ "task_id": task_id, "success": result.success, "output": result.output, "cost": result.cost }) else: # Execute sequentially for t in tasks: result = self.submit_task( task_id=t.get("id", f"seq-{len(results)}"), title=t.get("title", ""), description=t.get("description", ""), task_type=t.get("task_type", "general"), complexity=t.get("complexity", "moderate") ) results.append(result) return results def get_status(self) -> Dict: """Get orchestrator status.""" return { "state": { "status": self.state.status, "started_at": self.state.started_at, "tasks_completed": self.state.tasks_completed, "tasks_failed": self.state.tasks_failed, "total_cost": self.state.total_cost, "current_task": self.state.current_task }, "subsystems": { name: { "available": s.available, "healthy": s.healthy, "last_check": s.last_check } for name, s in self.subsystems.items() }, "summary": { "healthy_subsystems": len([s for s in self.subsystems.values() if s.healthy]), "total_subsystems": len(self.subsystems), "success_rate": self.state.tasks_completed / max(1, self.state.tasks_completed + self.state.tasks_failed) } } def health_check(self) -> Dict: """Run health check on all subsystems.""" results = {} for name, status in self.subsystems.items(): status.last_check = datetime.now().isoformat() # In a real implementation, we'd ping each subsystem results[name] = { "healthy": status.healthy, "available": status.available } return results def create_checkpoint(self, description: str = None) -> Optional[str]: """Create a system checkpoint.""" if not self.rollback: return None desc = description or f"Checkpoint at {datetime.now().isoformat()}" checkpoint = self.rollback.create_checkpoint(desc) self.state.last_checkpoint = datetime.now().isoformat() if self.event_bus: self.event_bus.publish("checkpoint.created", {"checkpoint_id": checkpoint.checkpoint_id}) return checkpoint.checkpoint_id def main(): """CLI for Genesis Orchestrator.""" import argparse parser = argparse.ArgumentParser(description="Genesis Orchestrator") parser.add_argument("command", choices=["start", "status", "health", "demo"]) parser.add_argument("--no-learning", action="store_true", help="Disable learning") parser.add_argument("--no-verify", action="store_true", help="Disable verification") args = parser.parse_args() orchestrator = GenesisOrchestrator( enable_learning=not args.no_learning, enable_verification=not args.no_verify ) if args.command == "start": print("Starting Genesis Orchestrator...") orchestrator.start() print("Press Ctrl+C to stop") try: while orchestrator._running: time.sleep(1) except KeyboardInterrupt: orchestrator.stop() elif args.command == "status": status = orchestrator.get_status() print("Genesis Orchestrator Status:") print(json.dumps(status, indent=2)) elif args.command == "health": health = orchestrator.health_check() print("Health Check Results:") for name, result in health.items(): status = "✓" if result["healthy"] else "✗" print(f" {status} {name}") elif args.command == "demo": print("Running demo...") orchestrator.start() # Submit a demo task result = orchestrator.submit_task( task_id="demo-001", title="Demo Task", description="This is a demonstration task", task_type="demo", complexity="simple", verify=False # Skip verification for demo ) print(f"\nDemo result: {json.dumps(result, indent=2)}") print(f"\nFinal status: {json.dumps(orchestrator.get_status(), indent=2)}") orchestrator.stop() if __name__ == "__main__": main()