#!/usr/bin/env python3 """ GENESIS WAR ROOM DASHBOARD ============================ Two modes: python3 core/war_room_dashboard.py — full dashboard (file output) python3 core/war_room_dashboard.py --session-banner — fast stdout banner (hook mode) python3 core/war_room_dashboard.py --watch — refresh every 60s (file output) python3 core/war_room_dashboard.py --print — print full markdown to stdout Session-banner mode completes in < 5 seconds and never crashes if services are down. """ import json import time import sys import os import subprocess import signal from datetime import datetime from pathlib import Path from typing import Dict, List, Any, Optional # ── Paths ───────────────────────────────────────────────────────────────────── # Support both Windows-style (E:\) and WSL-style (/mnt/e/) path resolution _WIN_ROOT = Path("E:/genesis-system") _WSL_ROOT = Path("/mnt/e/genesis-system") GENESIS_ROOT = _WSL_ROOT if _WSL_ROOT.exists() else _WIN_ROOT PROJECT_REGISTRY_PATH = GENESIS_ROOT / "data" / "project_registry.json" TASKS_JSON_PATH = GENESIS_ROOT / "loop" / "tasks.json" EVENTS_LOG_PATH = GENESIS_ROOT / "data" / "orchestrator_events.jsonl" FAILURE_LOG_PATH = GENESIS_ROOT / "KNOWLEDGE_GRAPH" / "entities" / "failure_evolution_log.jsonl" HANDOFF_MD_PATH = GENESIS_ROOT / "HANDOFF.md" KG_ENTITIES_DIR = GENESIS_ROOT / "KNOWLEDGE_GRAPH" / "entities" KG_AXIOMS_DIR = GENESIS_ROOT / "KNOWLEDGE_GRAPH" / "axioms" BLOODSTREAM_LOG = Path("/tmp/rlm_bloodstream.log") BLOODSTREAM_PID_FILE = Path("/tmp/bloodstream.pid") STATUS_MD_PATH = GENESIS_ROOT / "data" / "war_room_status.md" STATUS_JSON_PATH = GENESIS_ROOT / "data" / "war_room_status.json" STATUS_ICONS = { "live": "LIVE", "running": "RUNNING", "building": "BUILDING", "backlog": "BACKLOG", "blocked": "BLOCKED", "paused": "PAUSED", } PRIORITY_ICONS = {1: "[P1]", 2: "[P2]", 3: "[P3]", 4: "[P4]"} # ── Helpers ─────────────────────────────────────────────────────────────────── def _safe_read_json(path: Path) -> Any: try: with open(path, "r", encoding="utf-8") as f: return json.load(f) except Exception: return None def _count_jsonl_lines(directory: Path) -> int: """Count non-empty lines across all *.jsonl files in a directory.""" total = 0 if not directory.exists(): return 0 for p in directory.glob("*.jsonl"): try: with open(p, "r", encoding="utf-8", errors="replace") as f: total += sum(1 for line in f if line.strip()) except Exception: pass return total def _count_jsonl_files(directory: Path) -> int: if not directory.exists(): return 0 return sum(1 for _ in directory.glob("*.jsonl")) def _bloodstream_status() -> str: """Return a one-line bloodstream status string (never raises).""" try: # Check PID file first, then pgrep pid = None if BLOODSTREAM_PID_FILE.exists(): try: pid = int(BLOODSTREAM_PID_FILE.read_text().strip().split()[0]) except Exception: pass if pid is None: result = subprocess.run( ["pgrep", "-f", "bloodstream"], capture_output=True, text=True, timeout=2 ) pids = result.stdout.strip().splitlines() if pids: pid = int(pids[0]) if pid: # Verify PID is actually running try: os.kill(pid, 0) running = True except OSError: running = False else: running = False # Last log line last_line = "no log" if BLOODSTREAM_LOG.exists(): try: result = subprocess.run( ["tail", "-1", str(BLOODSTREAM_LOG)], capture_output=True, text=True, timeout=2 ) last_line = result.stdout.strip() or "empty" except Exception: last_line = "unreadable" if running: return f"RUNNING (pid {pid}) | last: {last_line[-80:]}" else: return f"NOT RUNNING | last: {last_line[-80:]}" except Exception as e: return f"CHECK FAILED ({e})" def _memory_stack_status() -> Dict[str, str]: """Check Redis, PostgreSQL via socket and Qdrant via HTTPS — 1-2s timeout each, never raises.""" import socket import urllib.request tcp_checks = { "redis": (os.environ.get("REDIS_HOST", "redis-genesis-u50607.vm.elestio.app"), 6380), "postgres": (os.environ.get("POSTGRES_HOST", "postgresql-genesis-u50607.vm.elestio.app"), 25432), } result = {} for svc, (host, port) in tcp_checks.items(): if not host: result[svc] = "NO_CONFIG" continue try: with socket.create_connection((host, port), timeout=1): result[svc] = "UP" except Exception: result[svc] = "DOWN" # Qdrant on Elestio uses HTTPS on port 6333 (not plain TCP, not port 443). # SSL cert is self-signed on Elestio, so verify=False is required. qdrant_host = os.environ.get("QDRANT_HOST", "qdrant-b3knu-u50607.vm.elestio.app") if not qdrant_host: result["qdrant"] = "NO_CONFIG" else: try: import ssl ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE url = f"https://{qdrant_host}:6333" urllib.request.urlopen(url, timeout=3, context=ctx) result["qdrant"] = "UP" except urllib.error.HTTPError: # Any HTTP error (401, 403, etc.) means the server is responding — it's UP result["qdrant"] = "UP" except Exception: result["qdrant"] = "DOWN" return result def _handoff_pending(lines: int = 5) -> List[str]: """Return last N meaningful lines of HANDOFF.md.""" if not HANDOFF_MD_PATH.exists(): return ["HANDOFF.md not found"] try: text = HANDOFF_MD_PATH.read_text(encoding="utf-8", errors="replace") all_lines = [l.rstrip() for l in text.splitlines() if l.strip()] return all_lines[-lines:] except Exception as e: return [f"Error reading HANDOFF.md: {e}"] def _aest_now() -> str: """Return current time as AEST string (UTC+10, no pytz dependency).""" from datetime import timezone, timedelta aest = timezone(timedelta(hours=10)) return datetime.now(aest).strftime("%Y-%m-%d %H:%M:%S AEST") # ── Session Banner (fast stdout, < 5s) ──────────────────────────────────────── def print_session_banner(): """ Print a compact war-room briefing to stdout for SessionStart hook capture. Completes in < 5 seconds. Degrades gracefully on any failure. """ lines = [] sep = "=" * 62 lines.append(sep) lines.append(" GENESIS WAR ROOM — SESSION START") lines.append(f" {_aest_now()}") lines.append(sep) # ── Bloodstream ─────────────────────────────────────────────── lines.append("") lines.append("BLOODSTREAM") try: bs = _bloodstream_status() lines.append(f" {bs}") except Exception as e: lines.append(f" CHECK ERROR: {e}") # ── Knowledge Graph ─────────────────────────────────────────── lines.append("") lines.append("KNOWLEDGE GRAPH") try: entity_files = _count_jsonl_files(KG_ENTITIES_DIR) entity_lines = _count_jsonl_lines(KG_ENTITIES_DIR) axiom_files = _count_jsonl_files(KG_AXIOMS_DIR) axiom_lines = _count_jsonl_lines(KG_AXIOMS_DIR) lines.append(f" Entities : {entity_files} files | {entity_lines:,} records") lines.append(f" Axioms : {axiom_files} files | {axiom_lines:,} records") except Exception as e: lines.append(f" CHECK ERROR: {e}") # ── Memory Stack ────────────────────────────────────────────── lines.append("") lines.append("MEMORY STACK") try: mem = _memory_stack_status() for svc, status in mem.items(): tag = "OK" if status == "UP" else ("WARN" if status == "NO_CONFIG" else "FAIL") lines.append(f" {svc:<12} {tag} {status}") except Exception as e: lines.append(f" CHECK ERROR: {e}") # ── n8n & Agents ────────────────────────────────────────────── lines.append("") lines.append("WORKFLOWS & AGENTS") try: registry = _safe_read_json(PROJECT_REGISTRY_PATH) or {} projects = registry.get("projects", []) live_projects = [p for p in projects if p.get("status") in ("live", "running")] lines.append(f" n8n workflows : 13/20 running (last known)") lines.append(f" Active projects: {len(live_projects)}/{len(projects)}") lines.append(f" Agent floor : 200 (mandate)") except Exception as e: lines.append(f" CHECK ERROR: {e}") # ── HANDOFF Pending ─────────────────────────────────────────── lines.append("") lines.append("HANDOFF — LAST 5 LINES") try: for l in _handoff_pending(5): lines.append(f" {l}") except Exception as e: lines.append(f" CHECK ERROR: {e}") # ── Footer ──────────────────────────────────────────────────── lines.append("") lines.append(sep) lines.append(" Rule 14: Every failure is fuel. Build-on-request. C: FORBIDDEN.") lines.append(sep) lines.append("") print("\n".join(lines), flush=True) # ── Full Dashboard (existing behaviour, preserved exactly) ──────────────────── class WarRoomDashboard: """ Generates the Genesis War Room full status dashboard. Called by: - Master Orchestrator (automatic, on health check) - CLI (manual run) - Genesis Start command (session init) """ def __init__(self): self.generated_at = datetime.now() self.registry = self._load_registry() self.tasks = self._load_tasks() self.recent_failures = self._load_failures(limit=5) self.recent_events = self._load_events(limit=20) def _load_registry(self) -> Dict: data = _safe_read_json(PROJECT_REGISTRY_PATH) return data if data else {"projects": [], "meta": {}} def _load_tasks(self) -> Dict: data = _safe_read_json(TASKS_JSON_PATH) return data if data else {"stories": []} def _load_failures(self, limit: int = 5) -> List[Dict]: if not FAILURE_LOG_PATH.exists(): return [] failures = [] try: with open(FAILURE_LOG_PATH, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line: try: failures.append(json.loads(line)) except json.JSONDecodeError: pass except Exception: pass return failures[-limit:] def _load_events(self, limit: int = 20) -> List[Dict]: if not EVENTS_LOG_PATH.exists(): return [] events = [] try: with open(EVENTS_LOG_PATH, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line: try: events.append(json.loads(line)) except json.JSONDecodeError: pass except Exception: pass return events[-limit:] def _check_service_health(self) -> Dict[str, str]: health = {} health["redis"] = "UP" if os.environ.get("REDIS_HOST") else "NO_CONFIG" health["postgres"] = "UP" if os.environ.get("POSTGRES_HOST") else "NO_CONFIG" health["qdrant"] = "UP" if os.environ.get("QDRANT_HOST") else "NO_CONFIG" health["project_registry"] = "OK" if PROJECT_REGISTRY_PATH.exists() else "MISSING" health["tasks_json"] = "OK" if TASKS_JSON_PATH.exists() else "MISSING" health["capability_manifest"] = ( "OK" if (GENESIS_ROOT / "data" / "context_state" / "capability_manifest.json").exists() else "MISSING" ) return health def _get_pending_stories(self) -> List[Dict]: stories = self.tasks.get("stories", []) return [s for s in stories if not s.get("passes", False) and s.get("status") != "blocked"] def _get_blocked_stories(self) -> List[Dict]: stories = self.tasks.get("stories", []) return [s for s in stories if s.get("status") == "blocked"] def _get_completed_stories(self) -> List[Dict]: stories = self.tasks.get("stories", []) return [s for s in stories if s.get("passes", False)] def _compile_top_priorities(self, limit: int = 10) -> List[str]: priorities = [] pending = self._get_pending_stories() for story in pending[:5]: priorities.append(f"{story['id']}: {story['title']}") for proj in self.registry.get("projects", []): if proj.get("priority", 9) <= 2 and proj.get("status") not in ("backlog", "blocked"): for action in proj.get("next_actions", [])[:2]: if action not in priorities: priorities.append(f"[{proj['id'].upper()}] {action}") return priorities[:limit] def _compute_total_mrr(self) -> int: total = 0 for proj in self.registry.get("projects", []): kpis = proj.get("kpis", {}) total += kpis.get("mrr_aud", 0) total += kpis.get("mrr_usd", 0) return total def generate_markdown(self) -> str: now = self.generated_at service_health = self._check_service_health() pending = self._get_pending_stories() blocked = self._get_blocked_stories() completed = self._get_completed_stories() top_priorities = self._compile_top_priorities() total_mrr = self._compute_total_mrr() meta = self.registry.get("meta", {}) target_mrr = meta.get("target_mrr_aud", 37000) mrr_pct = (total_mrr / target_mrr * 100) if target_mrr > 0 else 0 lines = [] lines.append("# GENESIS WAR ROOM") lines.append(f"**Generated**: {now.strftime('%Y-%m-%d %H:%M:%S AEST')} ") lines.append("**Model**: claude-sonnet-4-6 | **MCPs**: 11 connected") lines.append("") lines.append("## SYSTEM HEALTH") lines.append("") all_ok = all(v in ("UP", "OK") for v in service_health.values()) overall = "ALL SYSTEMS OPERATIONAL" if all_ok else "DEGRADED — check services" lines.append(f"**Status**: {overall}") lines.append("") lines.append("| Service | Status |") lines.append("|---------|--------|") for svc, status in service_health.items(): icon = "OK" if status in ("UP", "OK") else "WARN" lines.append(f"| {svc} | {icon} {status} |") lines.append("") lines.append("## REVENUE METRICS") lines.append("") lines.append("| Metric | Value |") lines.append("|--------|-------|") lines.append(f"| Current MRR (AUD) | ${total_mrr:,} |") lines.append(f"| Target MRR (AUD) | ${target_mrr:,} |") lines.append(f"| Progress to Target | {mrr_pct:.1f}% |") lines.append("| Active Clients | 1 (George Bunker FNQ) |") lines.append("| Demo Pipeline | 0 booked |") lines.append("| Leads Generated | 0 |") lines.append("") lines.append("## PROJECTS") lines.append("") projects_sorted = sorted(self.registry.get("projects", []), key=lambda p: p.get("priority", 9)) for proj in projects_sorted: status = proj.get("status", "backlog") icon = STATUS_ICONS.get(status, status.upper()) priority = PRIORITY_ICONS.get(proj.get("priority", 9), "[P?]") blocked_note = f" — BLOCKED: {proj['blocked_reason']}" if proj.get("blocked_reason") else "" lines.append(f"### {priority} {proj['name']}") lines.append(f"**Status**: {icon}{blocked_note} ") kpis = proj.get("kpis", {}) if kpis: kpi_str = " | ".join(f"{k}: {v}" for k, v in kpis.items()) lines.append(f"**KPIs**: {kpi_str} ") for action in proj.get("next_actions", [])[:3]: lines.append(f" - {action}") lines.append("") lines.append("## TASKS IN FLIGHT") lines.append("") lines.append(f"**Pending**: {len(pending)} | **Blocked**: {len(blocked)} | **Completed**: {len(completed)}") lines.append("") if pending: lines.append("### Pending Stories") for story in pending: lines.append(f"- **{story['id']}**: {story['title']}") lines.append("") if blocked: lines.append("### Blocked Stories") for story in blocked: blocker = story.get("blocker", "Unknown blocker") lines.append(f"- **{story['id']}**: {story['title']}") lines.append(f" - Blocker: {blocker}") lines.append("") lines.append("## RECENT FAILURES (Last 5)") lines.append("") if self.recent_failures: for f in self.recent_failures: lines.append(f"- **{f.get('date', 'unknown')}** [{f.get('project', '?')}]: {f.get('what', '')}") if f.get("root_cause") and f["root_cause"] != "Not yet determined": lines.append(f" - Root cause: {f['root_cause']}") else: lines.append("No failures logged yet.") lines.append("") lines.append("## NEXT 24H PRIORITIES") lines.append("") for i, action in enumerate(top_priorities, 1): lines.append(f"{i}. {action}") lines.append("") lines.append("## AGENT FLEET") lines.append("") lines.append("| Agent | Model | Thread | Status |") lines.append("|-------|-------|--------|--------|") agents = [ ("team-lead", "opus", "B-Thread", "standby"), ("parallel-builder", "sonnet", "P-Thread", "standby"), ("fusion-evaluator", "opus", "F-Thread", "standby"), ("long-thread-runner", "opus", "L-Thread", "standby"), ("gemini-dispatcher", "opus", "B-Thread", "standby"), ("rwl-executor", "opus", "L+C-Thread","standby"), ("verification-agent", "sonnet", "C-Thread", "standby"), ("research-scout", "haiku", "P-Thread", "standby"), ] for name, model, thread, status in agents: lines.append(f"| {name} | {model} | {thread} | {status} |") lines.append("") lines.append("---") lines.append("*Auto-generated by war_room_dashboard.py | Genesis v2.6.2 | Rule 14: Every failure is fuel*") return "\n".join(lines) def generate_json(self) -> Dict: pending = self._get_pending_stories() blocked = self._get_blocked_stories() completed = self._get_completed_stories() return { "generated_at": self.generated_at.isoformat(), "system_health": self._check_service_health(), "revenue": { "total_mrr_aud": self._compute_total_mrr(), "target_mrr_aud": self.registry.get("meta", {}).get("target_mrr_aud", 37000), }, "projects": { p["id"]: { "status": p["status"], "priority": p.get("priority", 9), "kpis": p.get("kpis", {}), "blocked": p.get("blocked_reason"), } for p in self.registry.get("projects", []) }, "tasks": { "pending": len(pending), "blocked": len(blocked), "completed": len(completed), "pending_ids": [s["id"] for s in pending], }, "top_priorities": self._compile_top_priorities(), "recent_failures": len(self.recent_failures), } def run(self): STATUS_MD_PATH.parent.mkdir(parents=True, exist_ok=True) md = self.generate_markdown() with open(STATUS_MD_PATH, "w", encoding="utf-8") as f: f.write(md) status_json = self.generate_json() with open(STATUS_JSON_PATH, "w", encoding="utf-8") as f: json.dump(status_json, f, indent=2) print(f"War Room dashboard written to {STATUS_MD_PATH}") print(f"JSON status written to {STATUS_JSON_PATH}") return md # ── Entry point ─────────────────────────────────────────────────────────────── if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Genesis War Room Dashboard") parser.add_argument("--session-banner", action="store_true", help="Print compact session-start briefing to stdout (hook mode, < 5s)") parser.add_argument("--watch", action="store_true", help="Refresh every 60 seconds") parser.add_argument("--print", action="store_true", help="Print full markdown to stdout") args = parser.parse_args() if args.session_banner: # Hard deadline: bail after 8s no matter what def _timeout_handler(signum, frame): print("\n[WAR ROOM] Timeout reached — partial data shown above.", flush=True) sys.exit(0) try: signal.signal(signal.SIGALRM, _timeout_handler) signal.alarm(8) except (AttributeError, OSError): pass # Windows / environments without SIGALRM print_session_banner() elif args.watch: print("Watching... Ctrl+C to stop") while True: WarRoomDashboard().run() time.sleep(60) else: dashboard = WarRoomDashboard() md = dashboard.run() if getattr(args, "print", False): print(md)