""" Genesis Swarm Reporter ====================== Generates hourly progress reports for all active swarms. Writes to: hive/progress/HOURLY_REPORT_{timestamp}.md Usage: from core.swarm_reporter import SwarmReporter reporter = SwarmReporter() reporter.generate_and_save() Cron (AEST = UTC+10, 4am–7pm AEST = 18:00–09:00 UTC previous day): # Every hour from 4am to 7pm AEST (18:00-09:00 UTC) 0 18-23,0-9 * * * cd /mnt/e/genesis-system && /usr/bin/python3 core/swarm_reporter.py """ import json import os from datetime import datetime, timezone, timedelta from pathlib import Path from typing import Any # --------------------------------------------------------------------------- # Paths # --------------------------------------------------------------------------- REPO_ROOT = Path("/mnt/e/genesis-system") DAILY_BUDGET_FILE = REPO_ROOT / "data" / "daily_budget.json" SWARM_PROGRESS_DIR = REPO_ROOT / "data" / "swarm_progress" HIVE_PROGRESS_DIR = REPO_ROOT / "hive" / "progress" DAILY_BUDGET_LIMIT_USD = 50.0 AEST_OFFSET = timedelta(hours=10) # --------------------------------------------------------------------------- # Helper: read daily budget # --------------------------------------------------------------------------- def _load_daily_budget() -> dict[str, Any]: """Load daily_budget.json, returning defaults if missing.""" if not DAILY_BUDGET_FILE.exists(): return { "date_aest": _today_aest(), "total_spent_usd": 0.0, "story_count": 0, "agent_runs": [], } try: return json.loads(DAILY_BUDGET_FILE.read_text()) except (json.JSONDecodeError, OSError): return {"date_aest": _today_aest(), "total_spent_usd": 0.0, "story_count": 0, "agent_runs": []} def _today_aest() -> str: """Return today's date string in AEST (UTC+10).""" return (datetime.now(timezone.utc) + AEST_OFFSET).strftime("%Y-%m-%d") def _now_aest() -> datetime: return datetime.now(timezone.utc) + AEST_OFFSET # --------------------------------------------------------------------------- # Helper: read swarm progress files # --------------------------------------------------------------------------- def _load_swarm_progress() -> list[dict[str, Any]]: """Load all swarm progress JSON files from data/swarm_progress/.""" swarms: list[dict[str, Any]] = [] if not SWARM_PROGRESS_DIR.exists(): return swarms for f in sorted(SWARM_PROGRESS_DIR.glob("*.json")): try: data = json.loads(f.read_text()) swarms.append(data) except (json.JSONDecodeError, OSError): continue return swarms # --------------------------------------------------------------------------- # Main reporter class # --------------------------------------------------------------------------- class SwarmReporter: """ Generates hourly reports for all active Genesis swarms. Writes Markdown to hive/progress/HOURLY_REPORT_{timestamp}.md. """ def __init__(self) -> None: HIVE_PROGRESS_DIR.mkdir(parents=True, exist_ok=True) SWARM_PROGRESS_DIR.mkdir(parents=True, exist_ok=True) # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def generate_report(self) -> str: """ Build the full hourly Markdown report and return it as a string. Sections: 1. Active swarms 2. Completed stories 3. Budget consumed 4. Velocity 5. Top 3 blockers 6. Next wave recommendations """ budget = _load_daily_budget() swarms = _load_swarm_progress() now = _now_aest() timestamp_str = now.strftime("%Y-%m-%d %H:%M AEST") date_str = now.strftime("%Y-%m-%d") active_swarms = [s for s in swarms if s.get("status") == "running"] completed_swarms = [s for s in swarms if s.get("status") == "completed"] # Aggregate story counts total_stories_today = budget.get("story_count", 0) stories_this_hour = sum( s.get("stories_completed_this_hour", 0) for s in active_swarms ) # Budget metrics spent_usd = budget.get("total_spent_usd", 0.0) remaining_usd = max(0.0, DAILY_BUDGET_LIMIT_USD - spent_usd) budget_pct = min(100.0, (spent_usd / DAILY_BUDGET_LIMIT_USD) * 100) # Velocity target_stories_per_hour = 333 # 8000 stories / 24 hours rough target velocity_ok = stories_this_hour >= (target_stories_per_hour * 0.5) # Model distribution across active swarms model_counts: dict[str, int] = {} for s in active_swarms: model = s.get("model", "unknown") model_counts[model] = model_counts.get(model, 0) + s.get("agent_count", 1) # Blockers blockers = self._collect_blockers(swarms) # Recommendations recommendations = self._generate_recommendations( active_swarms, spent_usd, remaining_usd, velocity_ok ) # ------------------------------------------------------------------ # Build report # ------------------------------------------------------------------ bar = self._budget_bar(budget_pct) lines: list[str] = [ f"# Genesis Swarm — Hourly Report", f"**Generated**: {timestamp_str}", f"**Date**: {date_str}", "", "---", "", "## 1. Active Swarms", "", ] if active_swarms: lines.append("| Mission | Model | Agents | Stories Done | Status |") lines.append("|---------|-------|--------|--------------|--------|") for s in active_swarms: lines.append( f"| {s.get('mission', 'unknown')[:40]} " f"| {s.get('model', '?')} " f"| {s.get('agent_count', 1)} " f"| {s.get('stories_completed', 0)} " f"| {s.get('status', '?')} |" ) else: lines.append("_No active swarms at this time._") lines += [ "", f"**Total active swarms**: {len(active_swarms)}", f"**Completed today**: {len(completed_swarms)} swarms", "", "---", "", "## 2. Completed Stories", "", f"- **This hour**: {stories_this_hour}", f"- **Today total**: {total_stories_today}", f"- **Hourly target**: {target_stories_per_hour}", f"- **On track**: {'YES' if velocity_ok else 'NO — below target'}", "", "---", "", "## 3. Budget Consumed", "", f"```", f"Today: ${spent_usd:.2f} / ${DAILY_BUDGET_LIMIT_USD:.2f}", f" {bar} {budget_pct:.1f}%", f"Remain: ${remaining_usd:.2f}", f"```", "", "**Agent runs today**: " + str(len(budget.get("agent_runs", []))), "", "---", "", "## 4. Velocity", "", f"- **Stories/hour**: {stories_this_hour}", f"- **Target**: {target_stories_per_hour} stories/hour", f"- **Status**: {'ON TRACK' if velocity_ok else 'UNDER TARGET'}", "", ] if model_counts: lines.append("**Models in use**:") for model, count in sorted(model_counts.items(), key=lambda x: -x[1]): lines.append(f" - `{model}`: {count} agents") lines.append("") lines += [ "---", "", "## 5. Top 3 Blockers", "", ] if blockers: for i, blocker in enumerate(blockers[:3], 1): lines.append(f"{i}. {blocker}") else: lines.append("_No blockers detected._") lines += [ "", "---", "", "## 6. Next Wave Recommendations", "", ] for i, rec in enumerate(recommendations[:5], 1): lines.append(f"{i}. {rec}") lines += [ "", "---", "", f"_Report auto-generated by Genesis SwarmReporter v1.0_", ] return "\n".join(lines) def generate_and_save(self) -> Path: """Generate report and write to hive/progress/. Returns the path.""" report = self.generate_report() timestamp = _now_aest().strftime("%Y%m%d_%H%M") out_path = HIVE_PROGRESS_DIR / f"HOURLY_REPORT_{timestamp}.md" out_path.write_text(report, encoding="utf-8") print(f"[SwarmReporter] Report written: {out_path}") return out_path # ------------------------------------------------------------------ # Private helpers # ------------------------------------------------------------------ def _collect_blockers(self, swarms: list[dict]) -> list[str]: blockers: list[str] = [] for s in swarms: if s.get("status") == "blocked": reason = s.get("blocker_reason", "Unknown blocker") mission = s.get("mission", "unknown mission") blockers.append(f"`{mission}` — {reason}") # Detect stalled swarms (started > 30 min ago, 0 progress) if s.get("status") == "running": started_at_str = s.get("started_at", "") stories_done = s.get("stories_completed", 0) if started_at_str and stories_done == 0: try: started_at = datetime.fromisoformat(started_at_str) elapsed = (_now_aest() - started_at).total_seconds() / 60 if elapsed > 30: blockers.append( f"`{s.get('mission', 'unknown')}` — " f"Stalled for {elapsed:.0f} min, 0 stories completed" ) except ValueError: pass if not blockers: blockers = ["No blockers detected — all systems nominal"] return blockers def _generate_recommendations( self, active_swarms: list[dict], spent_usd: float, remaining_usd: float, velocity_ok: bool, ) -> list[str]: recs: list[str] = [] if remaining_usd < 10.0: recs.append("BUDGET WARNING: Less than $10 remaining today — pause non-critical swarms") elif remaining_usd > 40.0: recs.append("Budget headroom available — consider launching additional research scouts") if not velocity_ok: recs.append( "Velocity below target — consider increasing agent count per swarm " "or switching to faster/cheaper model (DeepSeek Chat at $0.001/call)" ) if len(active_swarms) == 0: recs.append("No active swarms — fire `python3 scripts/genesis_hive.py --status` to check queue") if len(active_swarms) < 3: recs.append( "Swarm count is low — dispatch parallel ATLAS + FORGE workstreams " "to maximise velocity" ) if not recs: recs.append("All systems optimal — maintain current trajectory") recs.append("Consider pre-firing tomorrow's research scouts now") return recs @staticmethod def _budget_bar(pct: float, width: int = 20) -> str: filled = int(width * pct / 100) empty = width - filled bar = "#" * filled + "-" * empty return f"[{bar}]" # --------------------------------------------------------------------------- # CLI entry point # --------------------------------------------------------------------------- if __name__ == "__main__": reporter = SwarmReporter() path = reporter.generate_and_save() print("\n--- Report Preview ---\n") print(path.read_text())