# Generated by Cloudflare Build Swarm — Continuous_Dev_Daemon # Date: 2026-02-26T14:24:09.828162 # Model: gemini-3-flash-preview # Tokens: 950 import os import sys import time import json import signal import argparse import subprocess import requests from datetime import datetime, timedelta from pathlib import Path from typing import Dict, Any, List, Optional # Import pattern as requested sys.path.insert(0, str(Path(__file__).parent.parent)) from core.gemini_executor import GeminiExecutor class ContinuousDevDaemon: """ A continuous development daemon for the SubAIVA Durable Objects project. Monitors health, runs tests, analyzes results with Gemini, and manages KG entities. """ def __init__(self, args: argparse.Namespace): self.interval = args.interval * 60 # Convert minutes to seconds self.url = args.url self.auto_fix = args.auto_fix self.verbose = args.verbose self.project_root = Path(__file__).parent.parent / "subaiva-do" self.reports_dir = self.project_root / "reports" self.pid_file = Path("/tmp/subaiva_dev_daemon.pid") self.running = True self.gemini = GeminiExecutor() # Ensure directories exist self.reports_dir.mkdir(parents=True, exist_ok=True) # Setup signal handling signal.signal(signal.SIGINT, self._handle_exit) signal.signal(signal.SIGTERM, self._handle_exit) def _handle_exit(self, signum, frame): print(f"\n[Daemon] Received signal {signum}. Shutting down gracefully...") self.running = False if self.pid_file.exists(): self.pid_file.unlink() sys.exit(0) def _write_pid(self): self.pid_file.write_text(str(os.getpid())) def log(self, message: str): if self.verbose: timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"[{timestamp}] {message}") def check_health(self) -> bool: """Checks if the SubAIVA service is deployed and responding.""" try: response = requests.get(f"{self.url}/health", timeout=10) return response.status_code == 200 except Exception as e: self.log(f"Health check failed: {e}") return False def run_tests(self, deployed: bool) -> Dict[str, Any]: """Runs smoke tests or local dev tests via npm/wrangler.""" self.log(f"Running {'deployed' if deployed else 'local'} tests...") cmd = ["npm", "test"] if not deployed else ["npm", "run", "test:deployed"] try: result = subprocess.run( cmd, cwd=self.project_root, capture_output=True, text=True, timeout=300 ) return { "success": result.returncode == 0, "stdout": result.stdout, "stderr": result.stderr, "exit_code": result.returncode } except Exception as e: return {"success": False, "error": str(e)} def analyze_results(self, test_results: Dict[str, Any], health_status: bool) -> Dict[str, Any]: """Uses Gemini to analyze logs and suggest improvements or fixes.""" prompt = f""" Analyze the following SubAIVA Durable Objects development cycle data: Health Status: {"Healthy" if health_status else "Unhealthy"} Test Success: {test_results.get('success')} Stdout: {test_results.get('stdout', '')[:2000]} Stderr: {test_results.get('stderr', '')[:2000]} Tasks: 1. Identify any anomalies or bugs. 2. Suggest specific code improvements. 3. If there are simple errors (missing imports, type mismatches), provide a JSON 'fix' object. 4. Generate a Knowledge Graph (KG) entity summary for any failures. Respond in JSON format with keys: 'analysis', 'suggestions', 'kg_entities', 'auto_fix_patch'. """ try: response = self.gemini.execute(prompt) # Assuming GeminiExecutor returns a string that needs parsing or a dict if isinstance(response, str): return json.loads(response) return response except Exception as e: self.log(f"Gemini analysis failed: {e}") return {"analysis": "Failed to analyze", "error": str(e)} def write_to_kg(self, entities: List[Dict[str, Any]]): """Simulates writing bug reports/observations to the Knowledge Graph.""" for entity in entities: self.log(f"Writing KG Entity: {entity.get('name', 'Unnamed Observation')}") # Implementation for KG insertion goes here def apply_auto_fix(self, patch: Dict[str, Any]): """Optionally applies simple fixes suggested by Gemini.""" if not self.auto_fix or not patch: return self.log("Attempting auto-fix...") # Logic to apply patches to local files safely # E.g., patch contains {"file": "src/index.ts", "content": "..."} def save_cycle_report(self, data: Dict[str, Any]): """Writes cycle results to a timestamped JSON file.""" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") report_path = self.reports_dir / f"cycle_{timestamp}.json" with open(report_path, "w") as f: json.dump(data, f, indent=2) return report_path def generate_daily_summary(self): """Aggregates reports from the last 24 hours into a markdown file.""" self.log("Generating daily summary...") summary_path = self.project_root / "DAILY_SUMMARY.md" yesterday = datetime.now() - timedelta(days=1) relevant_reports = [] for report_file in self.reports_dir.glob("cycle_*.json"): file_time = datetime.fromtimestamp(report_file.stat().st_mtime) if file_time > yesterday: with open(report_file, "r") as f: relevant_reports.append(json.load(f)) with open(summary_path, "w") as f: f.write(f"# SubAIVA Continuous Dev Summary - {datetime.now().date()}\n\n") f.write(f"Total Cycles: {len(relevant_reports)}\n") # Logic to summarize findings... for report in relevant_reports: status = "✅" if report.get("test_results", {}).get("success") else "❌" f.write(f"- {report.get('timestamp')}: {status} {report.get('analysis', {}).get('summary', 'No summary')}\n") def run(self): """Main daemon loop.""" self._write_pid() self.log(f"Daemon started. Interval: {self.interval/60} minutes.") last_summary_time = datetime.now() while self.running: try: cycle_data = {"timestamp": datetime.now().isoformat()} # 1. Health Check is_deployed = self.check_health() cycle_data["deployed"] = is_deployed # 2. Run Tests test_results = self.run_tests(is_deployed) cycle_data["test_results"] = test_results # 3. Gemini Analysis analysis = self.analyze_results(test_results, is_deployed) cycle_data["analysis"] = analysis # 4. KG Integration if "kg_entities" in analysis: self.write_to_kg(analysis["kg_entities"]) # 5. Auto-Fix if self.auto_fix and "auto_fix_patch" in analysis: self.apply_auto_fix(analysis["auto_fix_patch"]) # 6. Save Report self.save_cycle_report(cycle_data) # 7. Daily Summary Check if datetime.now() - last_summary_time > timedelta(days=1): self.generate_daily_summary() last_summary_time = datetime.now() except Exception as e: self.log(f"Error in cycle: {e}") self.log(f"Cycle complete. Sleeping for {self.interval/60} minutes...") time.sleep(self.interval) if __name__ == "__main__": parser = argparse.ArgumentParser(description="SubAIVA Continuous Development Daemon") parser.add_argument("--interval", type=int, default=30, help="Interval in minutes between cycles") parser.add_argument("--url", type=str, default="http://localhost:8787", help="Base URL of the SubAIVA service") parser.add_argument("--auto-fix", action="store_true", help="Allow Gemini to attempt auto-fixing simple issues") parser.add_argument("--verbose", action="store_true", help="Enable verbose logging") args = parser.parse_args() daemon = ContinuousDevDaemon(args) daemon.run()