#!/usr/bin/env python3 """ Mission Control - Story 5 ========================= Main orchestrator for the Claude Code Capability Discovery Swarm. Coordinates: - Agent lifecycle management - Parallel agent execution - Result aggregation - Scoreboard updates - 4am report generation """ import asyncio import json from pathlib import Path from datetime import datetime from typing import Dict, List, Any, Optional, Type from dataclasses import dataclass, field import logging from .registry import CapabilityRegistry, Capability from .scoreboard import Scoreboard from .scheduler import SwarmScheduler, ScheduleConfig from .agents.base_agent import BaseAgent, AgentResult, AgentConfig, AgentTeam @dataclass class SwarmConfig: """Configuration for the swarm.""" max_parallel_agents: int = 5 scout_run_interval_seconds: int = 300 # 5 minutes audit_run_interval_seconds: int = 600 # 10 minutes data_dir: Optional[str] = None enable_notifications: bool = True slack_webhook: Optional[str] = None class MissionControl: """ Central orchestrator for the capability discovery swarm. Usage: # Initialize mc = MissionControl() # Register agents mc.register_agent(DocsScraperAgent()) mc.register_agent(SourceAnalyzerAgent()) mc.register_agent(GapFinderAgent()) # Run manually results = await mc.run_cycle() # Or run with scheduler (8pm-4am AEST) await mc.run_scheduled() # Get status status = mc.get_status() """ def __init__(self, config: Optional[SwarmConfig] = None): self.config = config or SwarmConfig() self.logger = logging.getLogger("swarm.mission_control") # Data directory if self.config.data_dir: self.data_dir = Path(self.config.data_dir) else: self.data_dir = Path(__file__).parent / "data" self.data_dir.mkdir(parents=True, exist_ok=True) # Core components self.registry = CapabilityRegistry(self.data_dir / "capability_registry.db") self.scoreboard = Scoreboard(self.data_dir / "scoreboard.db") self.scheduler = SwarmScheduler() # Registered agents self.scout_agents: Dict[str, BaseAgent] = {} self.audit_agents: Dict[str, BaseAgent] = {} # State self.is_running = False self.current_cycle = 0 self.last_results: Dict[str, AgentResult] = {} def register_agent(self, agent: BaseAgent): """Register an agent with mission control.""" agent_id = agent.config.agent_id team = agent.config.team # Inject shared registry and scoreboard agent._registry = self.registry agent._scoreboard = self.scoreboard if team == AgentTeam.SCOUT: self.scout_agents[agent_id] = agent self.logger.info(f"Registered SCOUT agent: {agent_id}") else: self.audit_agents[agent_id] = agent self.logger.info(f"Registered AUDIT agent: {agent_id}") # Register with scoreboard self.scoreboard.register_agent(agent_id, team.value, agent.config.name) def unregister_agent(self, agent_id: str): """Unregister an agent.""" if agent_id in self.scout_agents: del self.scout_agents[agent_id] elif agent_id in self.audit_agents: del self.audit_agents[agent_id] async def run_cycle(self) -> Dict[str, Any]: """ Run a single cycle of the swarm. 1. Run SCOUT agents in parallel 2. Run AUDIT agents in parallel 3. Aggregate results 4. Update scoreboard Returns summary of the cycle. """ self.current_cycle += 1 cycle_start = datetime.now() self.logger.info(f"Starting cycle {self.current_cycle}") results = { "cycle": self.current_cycle, "start_time": cycle_start.isoformat(), "scout_results": {}, "audit_results": {}, "summary": {} } # Phase 1: Run SCOUT agents self.logger.info(f"Running {len(self.scout_agents)} SCOUT agents...") scout_results = await self._run_agents_parallel( list(self.scout_agents.values()), max_parallel=self.config.max_parallel_agents ) results["scout_results"] = { agent_id: result.to_dict() if result else {"error": "No result"} for agent_id, result in scout_results.items() } # Phase 2: Run AUDIT agents self.logger.info(f"Running {len(self.audit_agents)} AUDIT agents...") audit_results = await self._run_agents_parallel( list(self.audit_agents.values()), max_parallel=self.config.max_parallel_agents ) results["audit_results"] = { agent_id: result.to_dict() if result else {"error": "No result"} for agent_id, result in audit_results.items() } # Store results self.last_results = {**scout_results, **audit_results} # Calculate summary total_discovered = sum( len(r.capabilities_found) for r in scout_results.values() if r ) total_implemented = sum( len(r.capabilities_implemented) for r in scout_results.values() if r ) total_gaps = sum( len(r.gaps_found) for r in audit_results.values() if r ) total_bugs = sum( len(r.bugs_found) for r in audit_results.values() if r ) cycle_end = datetime.now() results["summary"] = { "capabilities_discovered": total_discovered, "capabilities_implemented": total_implemented, "gaps_found": total_gaps, "bugs_found": total_bugs, "runtime_seconds": (cycle_end - cycle_start).total_seconds() } results["end_time"] = cycle_end.isoformat() # Save cycle results self._save_cycle_results(results) self.logger.info(f"Cycle {self.current_cycle} complete: " f"{total_discovered} discovered, {total_implemented} implemented, " f"{total_gaps} gaps, {total_bugs} bugs") return results async def _run_agents_parallel(self, agents: List[BaseAgent], max_parallel: int = 5) -> Dict[str, AgentResult]: """Run agents in parallel with concurrency limit.""" semaphore = asyncio.Semaphore(max_parallel) results = {} async def run_with_semaphore(agent: BaseAgent) -> tuple: async with semaphore: result = await agent.execute() return agent.config.agent_id, result tasks = [run_with_semaphore(agent) for agent in agents] completed = await asyncio.gather(*tasks, return_exceptions=True) for item in completed: if isinstance(item, Exception): self.logger.error(f"Agent execution error: {item}") else: agent_id, result = item results[agent_id] = result return results def _save_cycle_results(self, results: Dict): """Save cycle results to file.""" results_dir = self.data_dir / "cycle_results" results_dir.mkdir(exist_ok=True) filename = f"cycle_{self.current_cycle}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" with open(results_dir / filename, 'w') as f: json.dump(results, f, indent=2, default=str) async def run_scheduled(self): """ Run the swarm on schedule (8pm-4am AEST). This blocks and runs continuously until stopped. """ self.is_running = True async def on_window_start(): self.logger.info("=== SWARM WINDOW OPENED ===") self.current_cycle = 0 async def on_window_end(): self.logger.info("=== SWARM WINDOW CLOSED ===") # Generate 4am report report = self.generate_4am_report() self.logger.info(report) if self.config.enable_notifications: await self._send_notification(report) async def swarm_iteration(): if not self.scheduler.is_near_window_end(): await self.run_cycle() # Wait before next cycle await asyncio.sleep(self.config.scout_run_interval_seconds) try: await self.scheduler.run( swarm_iteration, on_window_start=on_window_start, on_window_end=on_window_end ) finally: self.is_running = False def stop(self): """Stop the scheduled run.""" self.scheduler.stop() self.is_running = False def generate_4am_report(self) -> str: """Generate the 4am summary report.""" return self.scoreboard.generate_nightly_report() async def _send_notification(self, message: str): """Send notification (e.g., to Slack).""" if not self.config.slack_webhook: return try: import aiohttp async with aiohttp.ClientSession() as session: await session.post( self.config.slack_webhook, json={"text": f"```{message}```"} ) except Exception as e: self.logger.error(f"Failed to send notification: {e}") def get_status(self) -> Dict[str, Any]: """Get current swarm status.""" registry_stats = self.registry.get_stats() team_scores = self.scoreboard.get_team_scores() scheduler_status = self.scheduler.get_status() return { "is_running": self.is_running, "current_cycle": self.current_cycle, "scheduler": scheduler_status, "agents": { "scout": list(self.scout_agents.keys()), "audit": list(self.audit_agents.keys()), "total": len(self.scout_agents) + len(self.audit_agents) }, "registry": registry_stats, "scores": { "scout": team_scores.get('scout', {}).total_points if 'scout' in team_scores else 0, "audit": team_scores.get('audit', {}).total_points if 'audit' in team_scores else 0 }, "timestamp": datetime.now().isoformat() } def get_leaderboard(self, limit: int = 10) -> List[Dict]: """Get current leaderboard.""" return self.scoreboard.get_agent_leaderboard(limit=limit) # === Seed Data === def seed_known_capabilities(self): """Seed the registry with known Claude Code capabilities.""" known_caps = [ # CLI Features ("cli_dangerously_skip", "Dangerously Skip Permissions", "cli_features", "Flag to skip permission prompts: --dangerously-skip-permissions"), ("cli_model_select", "Model Selection", "cli_features", "Select model via --model flag (haiku, sonnet, opus)"), ("cli_resume", "Resume Session", "cli_features", "Resume previous session with --resume flag"), ("cli_continue", "Continue Conversation", "cli_features", "Continue last conversation with --continue"), # Tool System ("tool_read", "Read Tool", "tool_system", "Read files with line numbers, supports images and PDFs"), ("tool_write", "Write Tool", "tool_system", "Write new files to filesystem"), ("tool_edit", "Edit Tool", "tool_system", "Make precise string replacements in files"), ("tool_bash", "Bash Tool", "tool_system", "Execute shell commands with timeout"), ("tool_glob", "Glob Tool", "tool_system", "Fast file pattern matching"), ("tool_grep", "Grep Tool", "tool_system", "Ripgrep-based content search"), ("tool_webfetch", "WebFetch Tool", "tool_system", "Fetch and process web content"), ("tool_websearch", "WebSearch Tool", "tool_system", "Search the web for information"), ("tool_task", "Task Tool", "tool_system", "Launch subagents for complex tasks"), ("tool_lsp", "LSP Tool", "tool_system", "Language Server Protocol integration"), # MCP Integration ("mcp_lazy_loading", "MCP Lazy Loading", "mcp_integration", "Servers load on-demand when tools are used"), ("mcp_server_config", "MCP Server Config", "mcp_integration", "Configure MCP servers in settings"), # Hooks System ("hooks_pre_tool", "Pre-Tool Hooks", "hooks_system", "Execute code before tool calls"), ("hooks_post_tool", "Post-Tool Hooks", "hooks_system", "Execute code after tool calls"), ("hooks_notification", "Notification Hooks", "hooks_system", "Receive notifications on events"), # Skills System ("skills_custom", "Custom Skills", "skills_system", "Define custom skills in .claude/skills/"), ("skills_builtin", "Built-in Skills", "skills_system", "Use /commit, /pr and other built-in skills"), # Advanced Features ("parallel_tools", "Parallel Tool Execution", "advanced_features", "Execute multiple independent tools in parallel"), ("plan_mode", "Plan Mode", "advanced_features", "Enter plan mode for complex implementation planning"), ("todo_tracking", "Todo Tracking", "advanced_features", "Track tasks with TodoWrite tool"), # Memory & Context ("context_summarization", "Context Summarization", "memory_context", "Automatic conversation summarization"), ("claudemd", "CLAUDE.md", "memory_context", "Project instructions in CLAUDE.md file"), ] seeded = 0 for cap_id, name, category, description in known_caps: cap = Capability( id=cap_id, name=name, category=category, description=description, discovered_by="seed_data", discovered_at=datetime.now().isoformat(), status="discovered" ) if self.registry.add_capability(cap): seeded += 1 self.logger.info(f"Seeded {seeded} known capabilities") return seeded # CLI interface async def main(): import argparse parser = argparse.ArgumentParser(description="Capability Discovery Swarm") parser.add_argument("command", choices=["init", "status", "run", "report", "seed"]) parser.add_argument("--cycles", type=int, default=1, help="Number of cycles to run") args = parser.parse_args() logging.basicConfig(level=logging.INFO) mc = MissionControl() if args.command == "init": print("Initializing swarm...") print(f"Data directory: {mc.data_dir}") print("Registry and scoreboard initialized") elif args.command == "status": status = mc.get_status() print(json.dumps(status, indent=2, default=str)) elif args.command == "run": print(f"Running {args.cycles} cycle(s)...") for i in range(args.cycles): results = await mc.run_cycle() print(f"Cycle {i+1} complete: {results['summary']}") elif args.command == "report": print(mc.generate_4am_report()) elif args.command == "seed": count = mc.seed_known_capabilities() print(f"Seeded {count} capabilities") if __name__ == "__main__": asyncio.run(main())