#!/usr/bin/env python3 """ GENESIS MASTER ORCHESTRATOR ============================ The top-level brain of the Genesis system. Responsibilities: 1. Reads all active projects from data/project_registry.json 2. Assigns agent teams to each project track 3. Monitors agent health and flags failed/stalled work 4. Routes tasks: Claude (reasoning) vs Gemini (execution) vs Playwright (browser) 5. Maintains global state in Redis (Elestio) 6. Reports status dashboard written to data/war_room_status.md 7. Triggers Evolution Engine on failures 8. Enforces ALL global rules (C: drive, SQLite, research-first, etc.) Architecture: GenesisCore (this) → TaskRouter → ExecutionLayer → Agents ↘ EvolutionEngine (failure monitoring) ↘ WarRoomDashboard (status reporting) Usage: from core.genesis_master_orchestrator import GenesisMasterOrchestrator orchestrator = GenesisMasterOrchestrator() status = orchestrator.get_system_status() orchestrator.run_health_check() """ import json import time import sys import os from datetime import datetime, timedelta from pathlib import Path from typing import Dict, List, Optional, Any, Tuple from dataclasses import dataclass, field, asdict from enum import Enum # Paths — all on E: drive GENESIS_ROOT = Path("E:/genesis-system") PROJECT_REGISTRY_PATH = GENESIS_ROOT / "data" / "project_registry.json" WAR_ROOM_STATUS_PATH = GENESIS_ROOT / "data" / "war_room_status.md" TASKS_JSON_PATH = GENESIS_ROOT / "loop" / "tasks.json" KG_ENTITIES_PATH = GENESIS_ROOT / "KNOWLEDGE_GRAPH" / "entities" FAILURE_LOG_PATH = KG_ENTITIES_PATH / "failure_evolution_log.jsonl" # Verify E: drive (RULE 6) assert str(GENESIS_ROOT).startswith("E:"), "CRITICAL: Must run on E: drive. C: drive is forbidden." class AgentRole(Enum): ORCHESTRATOR = "orchestrator" BUILDER = "builder" RESEARCHER = "researcher" VERIFIER = "verifier" MARKETING = "marketing" BROWSER = "browser" EVOLUTION = "evolution" class ProjectStatus(Enum): LIVE = "live" BUILDING = "building" BACKLOG = "backlog" BLOCKED = "blocked" PAUSED = "paused" RUNNING = "running" class ExecutorType(Enum): CLAUDE_OPUS = "claude-opus" # Complex reasoning, orchestration CLAUDE_SONNET = "claude-sonnet" # Standard implementation CLAUDE_HAIKU = "claude-haiku" # Simple ops, cost-sensitive GEMINI_SWARM = "gemini-swarm" # Bulk parallel execution KIMI_K2 = "kimi-k2" # 100-agent parallelism PLAYWRIGHT = "playwright" # Browser automation RLM_BLOODSTREAM = "rlm" # Memory operations @dataclass class AgentAssignment: """Tracks which agent is assigned to which project task.""" agent_id: str role: AgentRole project_id: str task_id: str executor: ExecutorType assigned_at: str = field(default_factory=lambda: datetime.now().isoformat()) last_heartbeat: Optional[str] = None status: str = "active" output_path: Optional[str] = None @dataclass class ProjectState: """Runtime state for a single Genesis project.""" id: str name: str status: ProjectStatus team: List[AgentAssignment] = field(default_factory=list) kpis: Dict[str, Any] = field(default_factory=dict) next_actions: List[str] = field(default_factory=list) last_updated: str = field(default_factory=lambda: datetime.now().isoformat()) health_score: float = 1.0 # 0.0 = dead, 1.0 = perfect blocked_reason: Optional[str] = None class GenesisMasterOrchestrator: """ The top-level orchestration brain for the entire Genesis system. This is the SINGLE SOURCE OF TRUTH for what all agents are doing, which projects are active, and what needs attention next. Singleton pattern — only one instance should exist per session. """ _instance = None def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._initialized = False return cls._instance def __init__(self): if self._initialized: return self._verify_e_drive() self.projects: Dict[str, ProjectState] = {} self.agent_assignments: List[AgentAssignment] = [] self.system_start_time = datetime.now().isoformat() self.health_checks_run = 0 self.redis_available = False # Load project registry self._load_project_registry() # Attempt Redis connection (Elestio) self._init_redis() self._initialized = True self._log_event("orchestrator_init", { "projects_loaded": len(self.projects), "redis": self.redis_available }) def _verify_e_drive(self): """RULE 6: Enforce E: drive. Die loudly if on C:.""" cwd = Path.cwd() if str(cwd).startswith("C:"): raise RuntimeError( "CRITICAL VIOLATION: Running on C: drive is FORBIDDEN.\n" "Switch to E:/genesis-system immediately.\n" "C: drive is critically low on space." ) def _init_redis(self): """Connect to Elestio Redis for global state. RULE 7: No SQLite.""" try: import redis # Elestio Redis config — loaded from environment redis_host = os.environ.get("REDIS_HOST", "redis-genesis.elestio.app") redis_port = int(os.environ.get("REDIS_PORT", "6379")) redis_password = os.environ.get("REDIS_PASSWORD", "") self.redis_client = redis.Redis( host=redis_host, port=redis_port, password=redis_password, decode_responses=True, socket_connect_timeout=3, socket_timeout=3 ) self.redis_client.ping() self.redis_available = True except Exception: # Redis not available in this session — operate in file-backed mode self.redis_client = None self.redis_available = False def _load_project_registry(self): """Load all projects from project_registry.json.""" if not PROJECT_REGISTRY_PATH.exists(): self._create_default_registry() with open(PROJECT_REGISTRY_PATH, "r") as f: registry = json.load(f) for proj in registry.get("projects", []): self.projects[proj["id"]] = ProjectState( id=proj["id"], name=proj.get("name", proj["id"]), status=ProjectStatus(proj.get("status", "backlog")), kpis=proj.get("kpis", {}), next_actions=proj.get("next_actions", []), blocked_reason=proj.get("blocked_reason") ) def _create_default_registry(self): """Create registry if missing.""" PROJECT_REGISTRY_PATH.parent.mkdir(parents=True, exist_ok=True) default = {"projects": []} with open(PROJECT_REGISTRY_PATH, "w") as f: json.dump(default, f, indent=2) def _log_event(self, event: str, data: Dict = None): """Append event to execution log.""" log_path = GENESIS_ROOT / "data" / "orchestrator_events.jsonl" log_path.parent.mkdir(parents=True, exist_ok=True) entry = { "timestamp": datetime.now().isoformat(), "event": event, "data": data or {} } with open(log_path, "a") as f: f.write(json.dumps(entry) + "\n") if self.redis_available: try: self.redis_client.lpush("genesis:events", json.dumps(entry)) self.redis_client.ltrim("genesis:events", 0, 999) # Keep last 1000 except Exception: pass def route_task(self, task: str, complexity: str = "medium") -> ExecutorType: """ Determine the optimal executor for a task. Delegates to TaskRouter for full logic. """ from core.task_router import TaskRouter router = TaskRouter() return router.classify_and_route(task, complexity) def assign_agent( self, project_id: str, task_id: str, role: AgentRole, executor: ExecutorType ) -> AgentAssignment: """Assign an agent to a project task and track it.""" assignment = AgentAssignment( agent_id=f"{role.value}-{project_id}-{int(time.time())}", role=role, project_id=project_id, task_id=task_id, executor=executor ) self.agent_assignments.append(assignment) if project_id in self.projects: self.projects[project_id].team.append(assignment) if self.redis_available: try: self.redis_client.hset( f"genesis:agents:{assignment.agent_id}", mapping={ "project": project_id, "task": task_id, "role": role.value, "executor": executor.value, "assigned_at": assignment.assigned_at, "status": "active" } ) except Exception: pass self._log_event("agent_assigned", { "agent_id": assignment.agent_id, "project": project_id, "task": task_id }) return assignment def run_health_check(self) -> Dict[str, Any]: """ Full system health check. Checks: project states, agent assignments, Redis, execution layer. Returns health report dict. """ self.health_checks_run += 1 report = { "timestamp": datetime.now().isoformat(), "check_number": self.health_checks_run, "projects": {}, "redis": self.redis_available, "total_active_agents": 0, "alerts": [] } # Check each project for proj_id, proj in self.projects.items(): active_agents = [a for a in proj.team if a.status == "active"] report["projects"][proj_id] = { "status": proj.status.value, "active_agents": len(active_agents), "next_actions": len(proj.next_actions), "health_score": proj.health_score } report["total_active_agents"] += len(active_agents) # Flag blocked projects if proj.status == ProjectStatus.BLOCKED: report["alerts"].append({ "severity": "HIGH", "project": proj_id, "message": f"BLOCKED: {proj.blocked_reason}" }) # Flag building projects with no agents if proj.status == ProjectStatus.BUILDING and len(active_agents) == 0: report["alerts"].append({ "severity": "MEDIUM", "project": proj_id, "message": "Building but no agents assigned" }) # Check tasks.json for overdue items pending_tasks = self._count_pending_tasks() if pending_tasks > 5: report["alerts"].append({ "severity": "MEDIUM", "message": f"{pending_tasks} pending stories in tasks.json — needs agent attention" }) self._log_event("health_check", report) return report def _count_pending_tasks(self) -> int: """Count pending stories in loop/tasks.json.""" try: with open(TASKS_JSON_PATH, "r") as f: tasks = json.load(f) return sum( 1 for s in tasks.get("stories", []) if not s.get("passes", False) and s.get("status") != "blocked" ) except Exception: return 0 def get_system_status(self) -> Dict[str, Any]: """ Get full system status snapshot. Used by WarRoomDashboard to generate status.md. """ return { "timestamp": datetime.now().isoformat(), "uptime_since": self.system_start_time, "projects": { pid: { "status": p.status.value, "health": p.health_score, "kpis": p.kpis, "next_actions": p.next_actions, "active_agents": len([a for a in p.team if a.status == "active"]), "blocked": p.blocked_reason } for pid, p in self.projects.items() }, "total_projects": len(self.projects), "live_projects": sum(1 for p in self.projects.values() if p.status == ProjectStatus.LIVE), "building_projects": sum(1 for p in self.projects.values() if p.status == ProjectStatus.BUILDING), "total_agents": len(self.agent_assignments), "redis_connected": self.redis_available, "health_checks_run": self.health_checks_run } def flag_failure(self, project_id: str, task_id: str, error: str, root_cause: str = ""): """ RULE 14: Flag a failure and trigger Evolution Engine. Every failure must be committed to memory. """ failure_entry = { "id": f"failure_{int(time.time())}", "date": datetime.now().strftime("%Y-%m-%d"), "project": project_id, "task": task_id, "what": error, "root_cause": root_cause or "Not yet determined", "guardrail_added": "pending", "rule_updated": "pending", "never_again": "pending" } # Write to failure log FAILURE_LOG_PATH.parent.mkdir(parents=True, exist_ok=True) with open(FAILURE_LOG_PATH, "a") as f: f.write(json.dumps(failure_entry) + "\n") # Update project health if project_id in self.projects: self.projects[project_id].health_score = max( 0.0, self.projects[project_id].health_score - 0.1 ) self._log_event("failure_flagged", failure_entry) # Trigger evolution engine try: from core.evolution_engine import EvolutionEngine engine = EvolutionEngine() engine.process_failure(failure_entry) except Exception: pass # Evolution engine is optional — don't cascade failures def reload_registry(self): """Hot-reload project registry without restart.""" self.projects.clear() self._load_project_registry() self._log_event("registry_reloaded", {"project_count": len(self.projects)}) # Singleton accessor _master_orchestrator = None def get_orchestrator() -> GenesisMasterOrchestrator: """Get the singleton master orchestrator.""" global _master_orchestrator if _master_orchestrator is None: _master_orchestrator = GenesisMasterOrchestrator() return _master_orchestrator if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Genesis Master Orchestrator") parser.add_argument("command", choices=["status", "health", "projects"]) args = parser.parse_args() orch = get_orchestrator() if args.command == "status": print(json.dumps(orch.get_system_status(), indent=2)) elif args.command == "health": report = orch.run_health_check() print(json.dumps(report, indent=2)) elif args.command == "projects": for pid, proj in orch.projects.items(): print(f" {pid}: {proj.status.value} | health={proj.health_score:.1f}")