#!/usr/bin/env python3 """ CONTINUOUS DEVELOPMENT WATCHDOG ================================ Monitors AIVA orchestrator for 8-hour continuous development sessions. Ensures no gaps > 5 minutes, auto-restarts if needed, timestamps everything. Usage: python continuous_watchdog.py start --hours 8 python continuous_watchdog.py status python continuous_watchdog.py report """ import os import sys import json import time import logging import subprocess import argparse from datetime import datetime, timedelta from pathlib import Path from typing import Optional, Dict, Any # ============================================================ # CONFIGURATION # ============================================================ GENESIS_ROOT = Path("/mnt/e/genesis-system") LOG_DIR = GENESIS_ROOT / "logs" DATA_DIR = GENESIS_ROOT / "data" HOURLY_DIR = LOG_DIR / "hourly" LOG_FILE = LOG_DIR / "continuous_dev.log" STATUS_FILE = DATA_DIR / "continuous_session.json" # Ensure directories exist LOG_DIR.mkdir(exist_ok=True) DATA_DIR.mkdir(exist_ok=True) HOURLY_DIR.mkdir(exist_ok=True) # Configuration MAX_GAP_MINUTES = 5 CHECK_INTERVAL_SECONDS = 60 AIVA_SCRIPT = GENESIS_ROOT / "core" / "aiva_orchestrator.py" # ============================================================ # LOGGING SETUP # ============================================================ logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] WATCHDOG: %(message)s', handlers=[ logging.StreamHandler(sys.stdout), logging.FileHandler(LOG_FILE) ] ) logger = logging.getLogger("continuous_watchdog") # ============================================================ # SESSION STATE # ============================================================ class SessionState: """Manages continuous session state with persistence.""" def __init__(self): self.session_id: str = "" self.start_time: datetime = None self.end_time: datetime = None self.duration_hours: int = 8 self.tasks_completed: int = 0 self.last_activity: datetime = None self.gaps_detected: int = 0 self.restarts: int = 0 self.hourly_completions: Dict[int, int] = {} self.status: str = "INITIALIZING" def load(self): """Load state from file.""" if STATUS_FILE.exists(): try: data = json.loads(STATUS_FILE.read_text()) self.session_id = data.get("session_id", "") self.start_time = datetime.fromisoformat(data["start_time"]) if data.get("start_time") else None self.end_time = datetime.fromisoformat(data["end_time"]) if data.get("end_time") else None self.duration_hours = data.get("duration_hours", 8) self.tasks_completed = data.get("tasks_completed", 0) self.last_activity = datetime.fromisoformat(data["last_activity"]) if data.get("last_activity") else None self.gaps_detected = data.get("gaps_detected", 0) self.restarts = data.get("restarts", 0) self.hourly_completions = data.get("hourly_completions", {}) self.status = data.get("status", "UNKNOWN") logger.info(f"Loaded session state: {self.session_id}") except Exception as e: logger.error(f"Failed to load state: {e}") def save(self): """Save state to file.""" data = { "session_id": self.session_id, "start_time": self.start_time.isoformat() if self.start_time else None, "end_time": self.end_time.isoformat() if self.end_time else None, "duration_hours": self.duration_hours, "tasks_completed": self.tasks_completed, "last_activity": self.last_activity.isoformat() if self.last_activity else None, "gaps_detected": self.gaps_detected, "restarts": self.restarts, "hourly_completions": self.hourly_completions, "status": self.status, "updated_at": datetime.now().isoformat() } STATUS_FILE.write_text(json.dumps(data, indent=2)) def start_session(self, hours: int = 8): """Initialize a new session.""" self.session_id = f"CONTINUOUS-{datetime.now().strftime('%Y-%m-%d-%H%M')}" self.start_time = datetime.now() self.end_time = self.start_time + timedelta(hours=hours) self.duration_hours = hours self.tasks_completed = 0 self.last_activity = datetime.now() self.gaps_detected = 0 self.restarts = 0 self.hourly_completions = {} self.status = "RUNNING" self.save() logger.info(f"Started session {self.session_id} for {hours} hours") logger.info(f"End time: {self.end_time.strftime('%Y-%m-%d %H:%M:%S')}") def record_task_completion(self, task_id: str, description: str): """Record a completed task with timestamp.""" now = datetime.now() self.tasks_completed += 1 self.last_activity = now # Track hourly completions hour = (now - self.start_time).seconds // 3600 if self.start_time else 0 hour_key = str(hour) self.hourly_completions[hour_key] = self.hourly_completions.get(hour_key, 0) + 1 # Log with timestamp log_entry = f"[{now.isoformat()}] TASK_COMPLETE {task_id} {description[:50]}" logger.info(log_entry) self.save() def check_for_gap(self) -> bool: """Check if there's been a gap > MAX_GAP_MINUTES.""" if not self.last_activity: return False gap = datetime.now() - self.last_activity gap_minutes = gap.total_seconds() / 60 if gap_minutes > MAX_GAP_MINUTES: self.gaps_detected += 1 logger.warning(f"GAP DETECTED: {gap_minutes:.1f} minutes since last activity") self.save() return True return False def record_restart(self): """Record an AIVA restart.""" self.restarts += 1 logger.info(f"AIVA restart #{self.restarts}") self.save() def is_session_active(self) -> bool: """Check if session is still within time bounds.""" if not self.end_time: return False return datetime.now() < self.end_time and self.status == "RUNNING" def time_remaining(self) -> timedelta: """Get time remaining in session.""" if not self.end_time: return timedelta(0) remaining = self.end_time - datetime.now() return max(remaining, timedelta(0)) # ============================================================ # AIVA MONITORING # ============================================================ def check_aiva_running() -> tuple[bool, Optional[int]]: """Check if AIVA orchestrator is running.""" try: result = subprocess.run( ["pgrep", "-f", "aiva_orchestrator"], capture_output=True, text=True ) if result.returncode == 0: pid = int(result.stdout.strip().split('\n')[0]) return True, pid return False, None except Exception as e: logger.error(f"Error checking AIVA: {e}") return False, None def start_aiva() -> bool: """Start AIVA orchestrator.""" try: logger.info("Starting AIVA orchestrator...") subprocess.Popen( [sys.executable, str(AIVA_SCRIPT), "start"], cwd=str(GENESIS_ROOT / "core"), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL ) time.sleep(5) # Wait for startup running, pid = check_aiva_running() if running: logger.info(f"AIVA started successfully (PID: {pid})") return True else: logger.error("AIVA failed to start") return False except Exception as e: logger.error(f"Error starting AIVA: {e}") return False def get_redis_queue_status() -> Dict[str, int]: """Get Redis queue depths.""" try: import redis r = redis.Redis(host='localhost', port=6379, decode_responses=True) return { "pending": r.llen("genesis:task_queue"), "active": r.llen("genesis:active_tasks"), "completed": r.llen("genesis:completed_tasks"), "failed": r.llen("genesis:failed_tasks") } except Exception as e: logger.warning(f"Redis unavailable: {e}") return {"pending": -1, "active": -1, "completed": -1, "failed": -1} def get_last_completion_time() -> Optional[datetime]: """Get timestamp of last completed task from Redis.""" try: import redis r = redis.Redis(host='localhost', port=6379, decode_responses=True) # Get most recent completed task last = r.lindex("genesis:completed_tasks", -1) if last: task = json.loads(last) if "completed_at" in task: return datetime.fromisoformat(task["completed_at"]) return None except Exception: return None # ============================================================ # HOURLY REPORT # ============================================================ def generate_hourly_report(state: SessionState, hour: int): """Generate hourly status report.""" report = { "session_id": state.session_id, "hour": hour, "timestamp": datetime.now().isoformat(), "tasks_this_hour": state.hourly_completions.get(str(hour), 0), "tasks_total": state.tasks_completed, "gaps_detected": state.gaps_detected, "restarts": state.restarts, "queue_status": get_redis_queue_status(), "time_remaining": str(state.time_remaining()) } report_file = HOURLY_DIR / f"hour_{hour:02d}_{datetime.now().strftime('%H%M')}.json" report_file.write_text(json.dumps(report, indent=2)) logger.info(f"=== HOURLY REPORT (Hour {hour}) ===") logger.info(f"Tasks this hour: {report['tasks_this_hour']}") logger.info(f"Tasks total: {report['tasks_total']}") logger.info(f"Queue pending: {report['queue_status']['pending']}") logger.info(f"Time remaining: {report['time_remaining']}") return report # ============================================================ # MAIN WATCHDOG LOOP # ============================================================ def run_watchdog(hours: int = 8): """Main watchdog loop for continuous development.""" state = SessionState() state.start_session(hours) logger.info("=" * 60) logger.info("CONTINUOUS DEVELOPMENT WATCHDOG STARTED") logger.info(f"Session: {state.session_id}") logger.info(f"Duration: {hours} hours") logger.info(f"End time: {state.end_time}") logger.info("=" * 60) last_hour_reported = -1 last_completed_count = 0 try: while state.is_session_active(): # Check current hour for hourly reports current_hour = int((datetime.now() - state.start_time).total_seconds() // 3600) if current_hour > last_hour_reported: generate_hourly_report(state, current_hour) last_hour_reported = current_hour # Check if AIVA is running running, pid = check_aiva_running() if not running: logger.warning("AIVA not running - attempting restart") if start_aiva(): state.record_restart() else: logger.error("Failed to restart AIVA - waiting before retry") time.sleep(30) continue # Check queue status queue = get_redis_queue_status() current_completed = queue.get("completed", 0) # Track new completions if current_completed > last_completed_count: new_tasks = current_completed - last_completed_count for _ in range(new_tasks): state.record_task_completion("task", "Completed via Gemini Flash") last_completed_count = current_completed # Check for gaps if state.check_for_gap(): logger.warning("Activity gap detected - checking AIVA health") if not running: start_aiva() state.record_restart() # Status output every check remaining = state.time_remaining() logger.debug( f"Status: AIVA={pid}, Queue={queue['pending']}, " f"Completed={state.tasks_completed}, Remaining={remaining}" ) # Sleep until next check time.sleep(CHECK_INTERVAL_SECONDS) # Session complete state.status = "COMPLETED" state.save() logger.info("=" * 60) logger.info("SESSION COMPLETE") logger.info(f"Total tasks: {state.tasks_completed}") logger.info(f"Total gaps: {state.gaps_detected}") logger.info(f"Total restarts: {state.restarts}") logger.info("=" * 60) # Final report generate_hourly_report(state, hours) except KeyboardInterrupt: logger.info("Watchdog interrupted by user") state.status = "INTERRUPTED" state.save() except Exception as e: logger.error(f"Watchdog error: {e}") state.status = "ERROR" state.save() raise def show_status(): """Show current session status.""" state = SessionState() state.load() print("\n" + "=" * 60) print("CONTINUOUS DEVELOPMENT SESSION STATUS") print("=" * 60) print(f"Session ID: {state.session_id}") print(f"Status: {state.status}") print(f"Start Time: {state.start_time}") print(f"End Time: {state.end_time}") print(f"Time Remaining: {state.time_remaining()}") print(f"Tasks Completed: {state.tasks_completed}") print(f"Gaps Detected: {state.gaps_detected}") print(f"Restarts: {state.restarts}") print("-" * 60) # AIVA status running, pid = check_aiva_running() print(f"AIVA Running: {'Yes (PID: ' + str(pid) + ')' if running else 'No'}") # Queue status queue = get_redis_queue_status() print(f"Queue Pending: {queue['pending']}") print(f"Queue Active: {queue['active']}") print(f"Queue Completed: {queue['completed']}") print(f"Queue Failed: {queue['failed']}") print("=" * 60 + "\n") def show_report(): """Show session report.""" state = SessionState() state.load() print("\n" + "=" * 60) print("CONTINUOUS DEVELOPMENT SESSION REPORT") print("=" * 60) print(f"Session ID: {state.session_id}") print(f"Duration: {state.duration_hours} hours") print(f"Status: {state.status}") print("-" * 60) print("HOURLY BREAKDOWN:") for hour, count in sorted(state.hourly_completions.items(), key=lambda x: int(x[0])): print(f" Hour {hour}: {count} tasks") print("-" * 60) print(f"TOTAL TASKS: {state.tasks_completed}") print(f"TOTAL GAPS: {state.gaps_detected}") print(f"TOTAL RESTARTS: {state.restarts}") print("=" * 60 + "\n") # ============================================================ # CLI # ============================================================ def main(): parser = argparse.ArgumentParser(description="Continuous Development Watchdog") subparsers = parser.add_subparsers(dest="command", help="Commands") # Start command start_parser = subparsers.add_parser("start", help="Start watchdog session") start_parser.add_argument("--hours", type=int, default=8, help="Session duration in hours") # Status command subparsers.add_parser("status", help="Show session status") # Report command subparsers.add_parser("report", help="Show session report") args = parser.parse_args() if args.command == "start": run_watchdog(args.hours) elif args.command == "status": show_status() elif args.command == "report": show_report() else: parser.print_help() if __name__ == "__main__": main()