self_improvement.py

"""
Recursive Self-Improvement Engine — goal formulation, procedural generation,
simulation, evaluation, and adaptive refinement.
"""

import logging
import time
from typing import Any, Dict, List, Optional

from nima_unified.training.goal_formulator import GoalFormulator

logger = logging.getLogger("nima_unified.training.self_improvement")


class RecursiveSelfImprovementEngine:
    """
    Master recursive self-improvement engine integrating goal formulation,
    procedural generation, simulation, evaluation, and adaptive refinement.
    """

    def __init__(self):
        self.goal_formulator = GoalFormulator()
        self.procedural_generator = None   # Set by backend
        self.simulation_loop = None        # Set by backend
        self.current_improvement_cycle = 0
        self.improvement_history: List[Dict[str, Any]] = []
        logger.info("RecursiveSelfImprovementEngine initialized")

    async def execute_improvement_cycle(
        self,
        capabilities: Dict[str, float],
        performance_feedback: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        """Execute a complete self-improvement cycle."""
        self.current_improvement_cycle += 1
        cycle_start = time.time()
        logger.info(f"Starting improvement cycle #{self.current_improvement_cycle}")

        gap_analysis = self.goal_formulator.analyze_capabilities(capabilities)
        goals = self.goal_formulator.formulate_goals(gap_analysis, performance_feedback)

        cycle_result = {
            "cycle_id": self.current_improvement_cycle,
            "started_at": cycle_start,
            "gap_analysis": gap_analysis,
            "goals_formulated": len(goals),
            "goals": goals,
            "status": "in_progress",
        }

        self.improvement_history.append(cycle_result)
        logger.info(f"Cycle #{self.current_improvement_cycle}: {len(goals)} goals formulated")
        return cycle_result

    def get_improvement_history(self, limit: int = 20) -> List[Dict[str, Any]]:
        return [c.copy() for c in self.improvement_history[-limit:]]

    def get_current_cycle_status(self) -> Dict[str, Any]:
        if self.improvement_history:
            return self.improvement_history[-1].copy()
        return {"status": "no_cycles_run", "cycle_id": 0}