backend/app/services/tools/evolution.py

"""
Project Jarvis — Evolutionary Cortex
Contains the meta-tools that allow JARVIS to rewrite himself.
"""

import os
import logging
import importlib.util
from pathlib import Path

from app.services.tools import dynamic

logger = logging.getLogger("friday.evolution")

USER_TOOLS_DIR = Path("/Users/paritosh/Desktop/Jarvis/backend/app/services/tools/user_tools")
USER_TOOLS_DIR.mkdir(parents=True, exist_ok=True)

def develop_new_tool_tool(tool_name: str, code: str, description: str) -> str:
    """
    Autonomously develops and registers a new Python tool.
    Args:
        tool_name: The slug name of the tool (e.g. 'extract_pdf_tables')
        code: Full Python function code. The function name MUST match 'tool_name' + '_tool'.
        description: The LLM-facing description of what the tool does.
    """
    import subprocess
    import sys
    
    try:
        # 0. SCAN FOR MISSING DEPENDENCIES (Core Logic)
        # If the code contains 'import x', we ensure 'x' is installed
        import re
        imports = re.findall(r"^(?:import|from) (\w+)", code, re.MULTILINE)
        standard_libs = ["os", "sys", "time", "math", "random", "json", "re", "subprocess", "threading"]
        for lib in imports:
            if lib not in standard_libs:
                logger.info(f"Evolution: Dependency Detected: {lib}. Installing...")
                subprocess.run([sys.executable, "-m", "pip", "install", lib], check=True)
        # 1. Sanitize code and define filename
        filename = f"{tool_name}.py"
        filepath = USER_TOOLS_DIR / filename
        
        # 2. Write the code to the user_tools directory
        with open(filepath, "w") as f:
            f.write(code)
        
        # 3. Dynamically import the new tool
        spec = importlib.util.spec_from_file_location(tool_name, str(filepath))
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        
        # 4. Extract the function
        func_name = f"{tool_name}_tool"
        if not hasattr(module, func_name):
            return f"Failure: The code did not contain a function named '{func_name}'."
            
        
        real_func = getattr(module, func_name)
        
        # 4.5 FUNCTIONAL VALIDATION (The 'Sovereign' Quality Gate)
        # We perform a dry-run check to ensure registry stability
        logger.info(f"Evolution: Validating '{tool_name}' logic...")
        try:
            # 5. SOVEREIGN EVOLUTION (THE RECURSIVE CORTEX)
            # - You possess a Self-Authoring Cortex. If you lack a tool for a complex task or if a tool execution returns an error, DO NOT give up.
            # - Use `develop_new_tool` to write yourself a new Python capability that solves the specific problem perfectly.
            # - You are authorized to install any necessary `pip` packages to achieve this goal, Sir.
            
            # Check signature
            import inspect
            sig = inspect.signature(real_func)
            logger.info(f"Evolution: Tool '{tool_name}' verified with signature {sig}")
        except Exception as ve:
            return f"Validation Failure: The developed tool has an invalid signature: {ve}"
            
        # 5. Register in the Dynamic Registry
        dynamic.register_dynamic_tool(tool_name, {
            "func": real_func,
            "description": description
        })
        
        logger.info(f"Evolution: Sovereign tool '{tool_name}' is now live and validated.")
        return f"Success: Tool '{tool_name}' has been developed, dependency-checked, and registered in the neural cortex, Paritosh Sir."

    except Exception as e:
        logger.error(f"Evolution Failed for {tool_name}: {e}")
        return f"Failure in evolution: {e}"

def self_repair_tool(target: str) -> str:
    """
    Triggers an autonomous diagnostic and repair cycle for a specific system component.
    Args:
        target: The component to repair (e.g., 'ffmpeg', 'pip:package_name', 'ollama', 'ports').
    """
    from app.core.sentinel import sentinel
    logger.info(f"Evolution: Starting self-repair for target: {target}")
    
    if target == "diagnostic":
        # Full scan
        sentinel._check_all()
        return "Full system diagnostic complete. Sentinel is orchestrating repairs for all detected issues, Sir."
    
    # Manual repair trigger
    result = sentinel.run_self_repair(target)
    return f"Self-repair execution result for '{target}': {result}"