#!/usr/bin/env python3 """ Genesis Hive Master Command ============================== The single command interface for the 5-General Hyperscale Hive. Usage: python3 genesis_hive.py "Build the voice widget checkout flow" python3 genesis_hive.py "Research top 20 AU digital agencies" --general ATLAS python3 genesis_hive.py "Set up GHL workflow for George demo" --general HERMES python3 genesis_hive.py --status # Show all active swarms python3 genesis_hive.py --budget # Today's spend vs $50 target python3 genesis_hive.py --report # Generate hourly report now python3 genesis_hive.py --triage "task" # Show triage decision only Flow: triage → route to correct general → spawn → monitor → report """ import argparse import json import sys from datetime import datetime, timezone, timedelta from pathlib import Path REPO_ROOT = Path("/mnt/e/genesis-system") sys.path.insert(0, str(REPO_ROOT)) from core.triage_agent import GenesisTriageAgent AEST_OFFSET = timedelta(hours=10) DAILY_BUDGET_LIMIT_USD = 50.0 def _now_aest() -> datetime: return datetime.now(timezone.utc) + AEST_OFFSET def _load_budget() -> dict: budget_file = REPO_ROOT / "data" / "daily_budget.json" if budget_file.exists(): try: return json.loads(budget_file.read_text()) except (json.JSONDecodeError, OSError): pass return {"date_aest": "", "total_spent_usd": 0.0, "story_count": 0, "agent_runs": []} # --------------------------------------------------------------------------- # General dispatcher # --------------------------------------------------------------------------- GENERAL_DISPATCHERS = { "ATLAS": "core.generals.atlas", "FORGE": "core.generals.forge", "HERMES": "core.generals.hermes", "MNEMOSYNE": "core.generals.mnemosyne", "PROMETHEUS": "core.generals.prometheus", } GENERAL_DESCRIPTIONS = { "ATLAS": "Research & Intelligence — web scouts, competitive analysis", "FORGE": "Architecture & Build — code, PRDs, tests", "HERMES": "Demo & GTM — browser automation, GHL, outreach", "MNEMOSYNE": "Memory & Intelligence — KG ingest, RLM, axioms", "PROMETHEUS": "Capabilities & New Tech — model eval, MCP install", } def dispatch_to_general( mission: str, general: str, triage_result: dict, dry_run: bool = False, ) -> dict: """ Route a mission to the correct general's spawn function. Parameters ---------- mission : str The task to execute general : str Which general to dispatch to (ATLAS|FORGE|HERMES|MNEMOSYNE|PROMETHEUS) triage_result : dict Full triage metadata from GenesisTriageAgent dry_run : bool If True, show routing but don't launch agents Returns ------- dict with dispatch result """ import importlib module_path = GENERAL_DISPATCHERS.get(general) if not module_path: return {"status": "error", "reason": f"Unknown general: {general}"} if dry_run: return { "status": "dry_run", "general": general, "mission": mission, "triage": triage_result, "would_call": f"{module_path}.spawn('{mission}')", } try: module = importlib.import_module(module_path) result = module.spawn(mission) return { "status": "launched", "general": general, "mission": mission, "triage": triage_result, "swarm": result, } except (ImportError, AttributeError) as e: return {"status": "error", "reason": str(e), "general": general} # --------------------------------------------------------------------------- # Display helpers # --------------------------------------------------------------------------- def show_budget() -> None: budget = _load_budget() spent = budget.get("total_spent_usd", 0.0) remaining = DAILY_BUDGET_LIMIT_USD - spent pct = min(100.0, (spent / DAILY_BUDGET_LIMIT_USD) * 100) bar_filled = int(30 * pct / 100) bar = "#" * bar_filled + "." * (30 - bar_filled) print(f"\n[GENESIS HIVE — BUDGET DASHBOARD]") print(f" Date (AEST) : {budget.get('date_aest', _now_aest().strftime('%Y-%m-%d'))}") print(f" Spent : ${spent:.4f}") print(f" Remaining : ${remaining:.4f}") print(f" Limit : ${DAILY_BUDGET_LIMIT_USD:.2f}") print(f" Progress : [{bar}] {pct:.1f}%") print(f" Stories : {budget.get('story_count', 0)} completed today") print(f" Runs : {len(budget.get('agent_runs', []))}") runs = budget.get("agent_runs", [])[-5:] if runs: print(f"\n Last 5 runs:") for r in runs: print(f" {r.get('run_id', '?')[:20]} {r.get('model', '?')[:25]} " f"${r.get('cost_usd', 0):.4f} {r.get('mission', '?')[:35]}") def show_status() -> None: progress_dir = REPO_ROOT / "data" / "swarm_progress" if not progress_dir.exists(): print("[STATUS] No swarms directory found.") return files = sorted(progress_dir.glob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True) if not files: print("[STATUS] No swarms found.") return running = [] completed = [] other = [] for f in files: try: d = json.loads(f.read_text()) s = d.get("status", "unknown") if s == "running": running.append(d) elif s == "completed": completed.append(d) else: other.append(d) except (json.JSONDecodeError, OSError): pass print(f"\n[GENESIS HIVE — SWARM STATUS {_now_aest().strftime('%Y-%m-%d %H:%M AEST')}]") print(f" Running : {len(running)}") print(f" Completed : {len(completed)}") print(f" Other : {len(other)}") print(f" Total : {len(files)}") if running: print(f"\n ACTIVE SWARMS:") print(f" {'ID':<20} {'GENERAL':<12} {'MODEL':<28} {'AGENTS':>6} {'STORIES':>7}") print(f" {'-'*80}") for d in running[:10]: rid = (d.get("run_id") or d.get("swarm_id", "?"))[:20] gen = d.get("general", "?")[:10] model = d.get("model", "?")[:26] agents = d.get("agent_count", 0) done = d.get("stories_completed", 0) print(f" {rid:<20} {gen:<12} {model:<28} {agents:>6} {done:>7}") show_budget() def show_triage(task: str) -> None: agent = GenesisTriageAgent() print(f"\n[TRIAGE RESULT]") print(agent.explain(task)) def show_report() -> None: try: from core.swarm_reporter import SwarmReporter reporter = SwarmReporter() path = reporter.generate_and_save() print(f"\n[REPORT] Generated: {path}") print("\n" + path.read_text()[:3000]) except ImportError as e: print(f"[ERROR] Could not load SwarmReporter: {e}") # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main() -> None: parser = argparse.ArgumentParser( description="Genesis Hive Master Command", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) parser.add_argument("mission", nargs="?", help="Task/mission to execute") parser.add_argument("--general", "-g", choices=["ATLAS", "FORGE", "HERMES", "MNEMOSYNE", "PROMETHEUS"], help="Override general selection (skip triage)") parser.add_argument("--dry-run", "-n", action="store_true", help="Show routing decision without launching agents") parser.add_argument("--triage", "-t", help="Show triage decision for a task (no launch)") parser.add_argument("--status", action="store_true", help="Show all active swarms") parser.add_argument("--budget", action="store_true", help="Show today's spend vs $50 target") parser.add_argument("--report", action="store_true", help="Generate hourly progress report now") parser.add_argument("--list-generals", action="store_true", help="List all 5 generals and their domains") args = parser.parse_args() # ── Status-only commands ────────────────────────────────────────────── if args.status: show_status() return if args.budget: show_budget() return if args.report: show_report() return if args.triage: show_triage(args.triage) return if args.list_generals: print("\n[GENESIS 5-GENERAL COMMAND STRUCTURE]") for name, desc in GENERAL_DESCRIPTIONS.items(): print(f" {name:<12} — {desc}") return # ── Mission execution ──────────────────────────────────────────────── if not args.mission: parser.print_help() print("\n[GENERALS]") for name, desc in GENERAL_DESCRIPTIONS.items(): print(f" {name:<12} — {desc}") print("\n[EXAMPLES]") print(' python3 genesis_hive.py "Build the voice widget checkout flow"') print(' python3 genesis_hive.py "Research top 20 AU agencies" --general ATLAS') print(' python3 genesis_hive.py "Set up GHL for George demo" --dry-run') sys.exit(0) # ── Triage ──────────────────────────────────────────────────────────── triage_agent = GenesisTriageAgent() triage = triage_agent.triage(args.mission) # Override general if specified general = args.general or triage["general"] # ── Print dispatch plan ─────────────────────────────────────────────── print(f"\n[GENESIS HIVE — {_now_aest().strftime('%Y-%m-%d %H:%M AEST')}]") print(f" Mission : {args.mission[:80]}") print(f" General : {general} {'(overridden)' if args.general else '(triaged)'}") print(f" Complexity: {triage['complexity']} (Tier {triage['model_tier']})") print(f" Model : {triage['suggested_model']}") print(f" Priority : {triage['priority']}") print(f" Browser : {triage['requires_browser']}") print(f" Mem Write : {triage['requires_memory_write']}") print(f" Est. Cost : ${triage['estimated_cost_usd']:.4f}") print(f" Parallel : {triage['parallel_safe']}") print(f" Triage ms : {triage['triage_latency_ms']}") if args.dry_run: print(f"\n[DRY RUN] Would dispatch to: {general}.spawn('{args.mission[:60]}')") print("[DRY RUN] Pass without --dry-run to execute.") return # ── Dispatch ────────────────────────────────────────────────────────── print(f"\n[DISPATCHING] → {general}") result = dispatch_to_general(args.mission, general, triage, dry_run=False) if result.get("status") == "launched": swarm = result.get("swarm", {}) swarm_id = swarm.get("swarm_id") or swarm.get("run_id", "unknown") print(f"\n[LAUNCHED] Swarm: {swarm_id}") print(f"[LAUNCHED] Agents: {swarm.get('agent_count', '?')}") print(f"[LAUNCHED] Output: {swarm.get('output_dir', '?')}") print(f"\n[NEXT STEPS]") print(f" 1. Monitor: python3 genesis_hive.py --status") print(f" 2. Report: python3 genesis_hive.py --report") print(f" 3. Budget: python3 genesis_hive.py --budget") elif result.get("status") == "error": print(f"\n[ERROR] Dispatch failed: {result.get('reason')}") sys.exit(1) if __name__ == "__main__": main()