""" AIVA Strategic Loop - PM-032 1-hour strategic review loop. Evaluates goals, adjusts priorities, and logs strategic decisions. """ import os import json import logging import asyncio from datetime import datetime, timedelta from typing import Dict, List, Optional, Any from dataclasses import dataclass, asdict, field from pathlib import Path # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) @dataclass class StrategicDecision: """A strategic decision made during review.""" decision_id: str decision_type: str # priority_change, resource_allocation, goal_adjustment description: str rationale: str impact: str # high, medium, low timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat()) @dataclass class StrategicReviewEntry: """Entry from a strategic review cycle.""" review_id: str timestamp: str objectives_evaluated: int goals_adjusted: int priorities_changed: int decisions: List[StrategicDecision] performance_summary: Dict duration_ms: float class StrategicLoop: """ 1-hour strategic review loop for AIVA. Reviews objectives, evaluates progress, and adjusts priorities. Usage: loop = StrategicLoop(objectives_loader, rank_tracker) await loop.start() """ def __init__( self, objectives_loader=None, rank_tracker=None, monitoring_dashboard=None, interval_seconds: int = 3600, # 1 hour log_dir: str = "logs" ): """ Initialize the strategic loop. Args: objectives_loader: StrategicObjectivesLoader instance rank_tracker: RankTracker instance monitoring_dashboard: MonitoringDashboard instance interval_seconds: Review interval (default 1 hour) log_dir: Directory for logs """ self.objectives_loader = objectives_loader self.rank_tracker = rank_tracker self.monitoring_dashboard = monitoring_dashboard self.interval_seconds = interval_seconds self.log_dir = Path(log_dir) self.log_dir.mkdir(parents=True, exist_ok=True) self._running = False self._task: Optional[asyncio.Task] = None self.review_history: List[StrategicReviewEntry] = [] self.pending_decisions: List[StrategicDecision] = [] logger.info(f"StrategicLoop initialized: interval={interval_seconds}s") async def start(self) -> None: """Start the strategic loop.""" if self._running: logger.warning("Strategic loop already running") return self._running = True self._task = asyncio.create_task(self._loop()) logger.info("Strategic loop started") async def stop(self) -> None: """Stop the strategic loop.""" self._running = False if self._task: self._task.cancel() try: await self._task except asyncio.CancelledError: pass logger.info("Strategic loop stopped") async def _loop(self) -> None: """Main loop execution.""" while self._running: try: await self.review_strategy() await asyncio.sleep(self.interval_seconds) except asyncio.CancelledError: break except Exception as e: logger.error(f"Strategic loop error: {e}") await asyncio.sleep(300) # Wait 5 min before retry async def review_strategy(self) -> StrategicReviewEntry: """ Execute a strategic review cycle. Returns: StrategicReviewEntry with results """ import time start_time = time.time() review_id = f"strat_{int(time.time() * 1000)}" logger.info(f"Starting strategic review {review_id}") decisions = [] objectives_evaluated = 0 goals_adjusted = 0 priorities_changed = 0 performance_summary = {} try: # 1. Evaluate current objectives objectives_evaluated = await self._evaluate_objectives() # 2. Analyze performance trends performance_summary = await self._analyze_performance() # 3. Identify necessary adjustments adjustments = await self._identify_adjustments(performance_summary) # 4. Make and record decisions for adj in adjustments: decision = await self._make_decision(adj) if decision: decisions.append(decision) if decision.decision_type == "goal_adjustment": goals_adjusted += 1 elif decision.decision_type == "priority_change": priorities_changed += 1 # 5. Apply priority changes await self._apply_priority_changes(decisions) except Exception as e: logger.error(f"Strategic review failed: {e}") duration = (time.time() - start_time) * 1000 entry = StrategicReviewEntry( review_id=review_id, timestamp=datetime.utcnow().isoformat(), objectives_evaluated=objectives_evaluated, goals_adjusted=goals_adjusted, priorities_changed=priorities_changed, decisions=decisions, performance_summary=performance_summary, duration_ms=duration ) self.review_history.append(entry) self._log_review(entry) logger.info( f"Strategic review {review_id} complete: " f"{objectives_evaluated} objectives, {len(decisions)} decisions " f"in {duration:.0f}ms" ) return entry async def _evaluate_objectives(self) -> int: """Evaluate current objectives.""" if not self.objectives_loader: return 0 try: objectives = self.objectives_loader.load_objectives() return len(objectives) except Exception as e: logger.error(f"Objective evaluation failed: {e}") return 0 async def _analyze_performance(self) -> Dict: """Analyze performance trends.""" summary = { "period": "1_hour", "metrics": {} } if self.rank_tracker: summary["metrics"]["current_rank"] = self.rank_tracker.current_rank.value summary["metrics"]["tasks_completed"] = self.rank_tracker.metrics.tasks_completed summary["metrics"]["success_rate"] = self.rank_tracker.metrics.success_rate if self.monitoring_dashboard: try: status = self.monitoring_dashboard.health_status() summary["health_status"] = status.get("overall_status", "unknown") summary["metrics"]["queue_depth"] = status.get("metrics", {}).get("queue_depth", {}).get("value", 0) except Exception: pass return summary async def _identify_adjustments(self, performance: Dict) -> List[Dict]: """Identify needed adjustments based on performance.""" adjustments = [] metrics = performance.get("metrics", {}) # Check success rate success_rate = metrics.get("success_rate", 1.0) if success_rate < 0.8: adjustments.append({ "type": "priority_change", "reason": f"Low success rate ({success_rate:.1%})", "action": "Prioritize simpler tasks" }) # Check queue depth queue_depth = metrics.get("queue_depth", 0) if queue_depth > 100: adjustments.append({ "type": "resource_allocation", "reason": f"High queue depth ({queue_depth})", "action": "Consider parallel execution" }) # Check health status if performance.get("health_status") == "warning": adjustments.append({ "type": "goal_adjustment", "reason": "System health warning", "action": "Reduce load, focus on stability" }) return adjustments async def _make_decision(self, adjustment: Dict) -> Optional[StrategicDecision]: """Make a strategic decision based on adjustment need.""" decision = StrategicDecision( decision_id=f"dec_{int(datetime.utcnow().timestamp() * 1000)}", decision_type=adjustment["type"], description=adjustment["action"], rationale=adjustment["reason"], impact="medium" ) logger.info(f"Strategic decision: {decision.decision_type} - {decision.description}") return decision async def _apply_priority_changes(self, decisions: List[StrategicDecision]) -> None: """Apply any priority changes from decisions.""" for decision in decisions: if decision.decision_type == "priority_change": # In production, this would update task priorities logger.debug(f"Applied priority change: {decision.description}") def _log_review(self, entry: StrategicReviewEntry) -> None: """Log review to file.""" log_file = self.log_dir / "strategic_loop.jsonl" try: data = asdict(entry) data["decisions"] = [asdict(d) for d in entry.decisions] with open(log_file, "a") as f: f.write(json.dumps(data) + "\n") except Exception as e: logger.error(f"Failed to log review: {e}") def get_recent_reviews(self, limit: int = 10) -> List[Dict]: """Get recent review entries.""" entries = [] for r in self.review_history[-limit:]: data = asdict(r) data["decisions"] = [asdict(d) for d in r.decisions] entries.append(data) return entries def get_recent_decisions(self, limit: int = 20) -> List[Dict]: """Get recent strategic decisions.""" decisions = [] for review in reversed(self.review_history): for decision in review.decisions: decisions.append(asdict(decision)) if len(decisions) >= limit: return decisions return decisions def get_stats(self) -> Dict: """Get strategic loop statistics.""" if not self.review_history: return { "total_reviews": 0, "total_decisions": 0, "avg_duration_ms": 0 } total_decisions = sum(len(r.decisions) for r in self.review_history) return { "total_reviews": len(self.review_history), "total_decisions": total_decisions, "total_objectives_evaluated": sum(r.objectives_evaluated for r in self.review_history), "avg_duration_ms": sum(r.duration_ms for r in self.review_history) / len(self.review_history), "is_running": self._running } # Singleton instance _strategic_loop: Optional[StrategicLoop] = None def get_strategic_loop(**kwargs) -> StrategicLoop: """Get or create singleton StrategicLoop.""" global _strategic_loop if _strategic_loop is None: _strategic_loop = StrategicLoop(**kwargs) return _strategic_loop if __name__ == "__main__": import asyncio async def main(): loop = StrategicLoop() # Run one review cycle entry = await loop.review_strategy() print(f"\nStrategic Review Complete:") print(f" ID: {entry.review_id}") print(f" Objectives Evaluated: {entry.objectives_evaluated}") print(f" Decisions Made: {len(entry.decisions)}") print(f" Duration: {entry.duration_ms:.0f}ms") if entry.decisions: print(f"\nDecisions:") for d in entry.decisions: print(f" [{d.decision_type}] {d.description}") print(f"\nStats: {loop.get_stats()}") asyncio.run(main())