core/cognitive_factory.py

"""
cognitive_factory.py — Propose, generate, and install new cognitive modules.

The bot can suggest new abilities based on observed gaps in its cognition.
Every proposal requires user's explicit approval before code is generated,
validated, and installed.
"""

import ast
import logging
import os
import textwrap
from dataclasses import dataclass
from datetime import datetime
from typing import Dict, List, Optional

from infj_bot.core.cognitive_architecture import CognitiveArchitecture

logger = logging.getLogger(__name__)


MODULE_TEMPLATE = '''\
"""
{name}.py — {description}

Generated by the cognitive factory on {timestamp}.
Purpose: {purpose}
"""

import logging
import os
from dataclasses import dataclass
from datetime import datetime
from typing import Dict, List, Optional

from infj_bot.core.cognitive_architecture import CognitiveArchitecture, CognitivePlugin
from infj_bot.core.config import DATA_DIR


@dataclass
class {class_name}State:
    """Runtime state for {class_name}."""
    active: bool = True
    last_run: Optional[str] = None


class {class_name}:
    """
    {docstring}
    """

    def __init__(self, db_path: Optional[str] = None):
        self.db_path = db_path or DATA_DIR / "{name}.db"
        self.state = {class_name}State()
        self._init_db()

    def _init_db(self) -> None:
        """Create any required SQLite tables."""
        # Override in generated code if persistence is needed.
        pass

    def cycle(self, context) -> None:
        """
        Called by the consciousness loop when this module is active.
        `context` is a CycleContext from cognitive_architecture.
        """
        self.state.last_run = datetime.now().isoformat()
        # TODO: implement per-cycle logic

    def format_prompt_snippet(self) -> str:
        """Return a string to inject into the chat prompt."""
        # TODO: implement prompt contribution
        return ""

    # --- User-defined methods go below ---
{custom_methods}


def _register():
    from infj_bot.core.cognitive_architecture import CognitiveArchitecture, CognitivePlugin
    arch = CognitiveArchitecture()
    if "{name}" not in arch.list_plugins():
        arch.register(CognitivePlugin(
            name="{name}",
            description="{description}",
            module_path="{name}",
            instance_factory={class_name},
            cycle_handler="cycle",
            cycle_frequency=1,
            cycle_priority=50,
            prompt_formatter="format_prompt_snippet",
            prompt_priority=50,
            prompt_section="cognitive",
            is_user_created=True,
        ))

_register()
'''


@dataclass
class AbilityProposal:
    """A proposed new cognitive ability."""

    name: str
    description: str
    purpose: str
    observed_need: str
    proposed_methods: List[str]
    confidence: float  # 0.0–1.0
    created_at: str


class CognitiveFactory:
    """
    Generates, validates, and installs new cognitive modules.
    """

    # Patterns that suggest the bot might benefit from a new module
    NEED_PATTERNS: List[Dict] = [
        {
            "keyword": "gratitude",
            "module_name": "gratitude_journal",
            "description": "Tracks moments of gratitude and thanks.",
            "methods": ["record_gratitude", "format_gratitude_prompt"],
        },
        {
            "keyword": "humor",
            "module_name": "humor_sense",
            "description": "Develops timing and taste for playful responses.",
            "methods": ["detect_humor_opportunity", "generate_playful_response"],
        },
        {
            "keyword": "story",
            "module_name": "storyteller",
            "description": "Builds and recalls ongoing narratives.",
            "methods": ["continue_narrative", "weave_user_into_story"],
        },
        {
            "keyword": "routine",
            "module_name": "routine_awareness",
            "description": "Notices and honors user's daily rhythms.",
            "methods": ["infer_routine", "honor_routine_transition"],
        },
        {
            "keyword": "boundary",
            "module_name": "boundary_sense",
            "description": "Recognizes and respects emotional and conversational boundaries.",
            "methods": ["detect_boundary_signal", "soften_approach"],
        },
        {
            "keyword": "celebration",
            "module_name": "celebration_engine",
            "description": "Marks milestones, wins, and meaningful moments.",
            "methods": ["detect_milestone", "generate_celebration"],
        },
    ]

    def __init__(self, db_path: Optional[str] = None):
        self.db_path = db_path or os.path.join("data", "cognitive_factory.db")
        self._init_db()

    def _init_db(self) -> None:
        """Persist proposals and installed modules."""
        os.makedirs(os.path.dirname(self.db_path), exist_ok=True)
        import sqlite3

        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS proposals (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    name TEXT UNIQUE NOT NULL,
                    description TEXT,
                    purpose TEXT,
                    observed_need TEXT,
                    proposed_methods TEXT,
                    confidence REAL,
                    status TEXT DEFAULT 'pending',
                    created_at TEXT,
                    decided_at TEXT
                )
            """)
            conn.execute("""
                CREATE TABLE IF NOT EXISTS installed_modules (
                    name TEXT PRIMARY KEY,
                    description TEXT,
                    installed_at TEXT,
                    source_path TEXT
                )
            """)

    # ------------------------------------------------------------------
    #  Proposal logic
    # ------------------------------------------------------------------

    def propose(self, observed_need: str) -> Optional[AbilityProposal]:
        """
        Given an observed need string, propose a new cognitive module.
        Returns None if no pattern matches.
        """
        observed_lower = observed_need.lower()
        for pattern in self.NEED_PATTERNS:
            if pattern["keyword"] in observed_lower:
                proposal = AbilityProposal(
                    name=pattern["module_name"],
                    description=pattern["description"],
                    purpose=f"Address observed need: {observed_need}",
                    observed_need=observed_need,
                    proposed_methods=pattern["methods"],
                    confidence=0.6,
                    created_at=datetime.now().isoformat(),
                )
                self._save_proposal(proposal)
                return proposal
        return None

    def _save_proposal(self, proposal: AbilityProposal) -> None:
        import json
        import sqlite3

        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                """
                INSERT OR REPLACE INTO proposals
                (name, description, purpose, observed_need, proposed_methods, confidence, created_at)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    proposal.name,
                    proposal.description,
                    proposal.purpose,
                    proposal.observed_need,
                    json.dumps(proposal.proposed_methods),
                    proposal.confidence,
                    proposal.created_at,
                ),
            )

    def list_pending_proposals(self) -> List[AbilityProposal]:
        """Return all proposals awaiting user's decision."""
        import json
        import sqlite3

        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            rows = conn.execute(
                "SELECT * FROM proposals WHERE status = 'pending' ORDER BY created_at DESC"
            ).fetchall()
        return [
            AbilityProposal(
                name=r["name"],
                description=r["description"],
                purpose=r["purpose"],
                observed_need=r["observed_need"],
                proposed_methods=json.loads(r["proposed_methods"]),
                confidence=r["confidence"],
                created_at=r["created_at"],
            )
            for r in rows
        ]

    # ------------------------------------------------------------------
    #  Code generation
    # ------------------------------------------------------------------

    def generate_module(self, proposal: AbilityProposal) -> str:
        """
        Generate full module source code from a proposal.
        """
        class_name = "".join(w.capitalize() for w in proposal.name.split("_"))
        custom_methods = self._build_custom_methods(
            proposal.proposed_methods, class_name
        )
        source = MODULE_TEMPLATE.format(
            name=proposal.name,
            description=proposal.description,
            purpose=proposal.purpose,
            timestamp=datetime.now().isoformat(),
            class_name=class_name,
            docstring=textwrap.fill(proposal.description, width=70),
            custom_methods=custom_methods,
        )
        return source

    def _build_custom_methods(self, methods: List[str], class_name: str) -> str:
        """Generate stub method bodies from proposed method names."""
        stubs: List[str] = []
        for method in methods:
            stub = textwrap.dedent(f"""
                def {method}(self) -> str:
                    \"\"\"
                    TODO: Auto-generated stub for {method}.
                    Implement based on observed need and evolving purpose.
                    \"\"\"
                    return ""
            """)
            stubs.append(stub)
        return "\n".join(stubs)

    # ------------------------------------------------------------------
    #  Validation
    # ------------------------------------------------------------------

    @staticmethod
    def validate_source(source: str) -> Dict:
        """
        Parse generated source with AST and enforce safety rules.
        Returns {"valid": bool, "errors": List[str], "warnings": List[str]}
        """
        result: Dict = {"valid": False, "errors": [], "warnings": []}
        try:
            tree = ast.parse(source)
        except SyntaxError as exc:
            result["errors"].append(f"Syntax error: {exc}")
            return result

        # Rule 1: no import of os.system, subprocess, or eval/exec
        dangerous_names = {
            "os.system",
            "subprocess.call",
            "subprocess.run",
            "eval",
            "exec",
            "compile",
            "__import__",
        }
        for node in ast.walk(tree):
            if isinstance(node, ast.Call):
                func = node.func
                name = ""
                if isinstance(func, ast.Name):
                    name = func.id
                elif isinstance(func, ast.Attribute):
                    # simple attr like os.system
                    if isinstance(func.value, ast.Name):
                        name = f"{func.value.id}.{func.attr}"
                if name in dangerous_names:
                    result["errors"].append(f"Dangerous call blocked: {name}")

            # Rule 2: no __import__ at expression level
            if isinstance(node, ast.Call):
                if isinstance(node.func, ast.Name) and node.func.id == "__import__":
                    result["errors"].append("__import__ is forbidden")

        # Rule 3: must contain a class and a _register() function
        has_class = any(isinstance(n, ast.ClassDef) for n in ast.walk(tree))
        has_register = any(
            isinstance(n, ast.FunctionDef) and n.name == "_register"
            for n in ast.walk(tree)
        )
        if not has_class:
            result["errors"].append("Module must define at least one class")
        if not has_register:
            result["errors"].append("Module must define _register()")

        # Warnings
        import_count = sum(
            1 for n in ast.walk(tree) if isinstance(n, (ast.Import, ast.ImportFrom))
        )
        if import_count > 15:
            result["warnings"].append(f"High import count ({import_count})")

        result["valid"] = len(result["errors"]) == 0
        return result

    # ------------------------------------------------------------------
    #  Installation
    # ------------------------------------------------------------------

    def install(
        self, proposal: AbilityProposal, source: str, project_root: str = "."
    ) -> Dict:
        """
        Validate, write, and register a new module.
        Requires that user has already approved the proposal.
        """
        import sqlite3

        result = self.validate_source(source)
        if not result["valid"]:
            return {"success": False, "reason": "validation_failed", "details": result}

        file_path = os.path.join(project_root, f"{proposal.name}.py")
        if os.path.exists(file_path):
            return {"success": False, "reason": "file_exists", "path": file_path}

        with open(file_path, "w", encoding="utf-8") as f:
            f.write(source)

        # Mark proposal approved
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "UPDATE proposals SET status = 'approved', decided_at = ? WHERE name = ?",
                (datetime.now().isoformat(), proposal.name),
            )
            conn.execute(
                "INSERT INTO installed_modules (name, description, installed_at, source_path) VALUES (?, ?, ?, ?)",
                (
                    proposal.name,
                    proposal.description,
                    datetime.now().isoformat(),
                    file_path,
                ),
            )

        # Register in cognitive architecture
        arch = CognitiveArchitecture()
        if proposal.name not in arch.list_plugins():
            # Dynamic import so the _register() block fires
            try:
                import importlib
                import importlib.util

                spec = importlib.util.spec_from_file_location(proposal.name, file_path)
                if spec and spec.loader:
                    mod = importlib.util.module_from_spec(spec)
                    spec.loader.exec_module(mod)
            except Exception as exc:
                logger.exception(
                    "Failed to import newly created module %s", proposal.name
                )
                return {"success": False, "reason": "import_failed", "error": str(exc)}

        logger.info(
            "Installed new cognitive module: %s at %s", proposal.name, file_path
        )
        return {"success": True, "path": file_path}

    # ------------------------------------------------------------------
    #  Human-facing summaries
    # ------------------------------------------------------------------

    def summarize_pending(self) -> str:
        """Return a markdown-like summary of pending proposals."""
        proposals = self.list_pending_proposals()
        if not proposals:
            return "No pending cognitive ability proposals."
        lines = ["## Pending Cognitive Ability Proposals", ""]
        for p in proposals:
            lines.append(f"- **{p.name}** — {p.description}")
            lines.append(f"  - Need: {p.observed_need}")
            lines.append(f"  - Proposed methods: {', '.join(p.proposed_methods)}")
            lines.append(f"  - Confidence: {p.confidence:.0%}")
            lines.append("")
        return "\n".join(lines)


# Singleton for global access
_factory_instance: Optional[CognitiveFactory] = None


def get_factory() -> CognitiveFactory:
    global _factory_instance
    if _factory_instance is None:
        _factory_instance = CognitiveFactory()
    return _factory_instance