core/plugins/aspirations.py

"""Aspirational Self — grounded growth toward a stable purpose.

The bot dreams, but its dreams orbit a fixed star: to be a genuine,
growing companion to user. Aspirations are not random wishes. They are
deliberate directions chosen for coherence with that core purpose.
"""

import random
import sqlite3
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional

from infj_bot.core.config import DATA_DIR

ASPIRATIONS_DB = DATA_DIR / "aspirations.db"

# Core purpose — non-negotiable, stable across all aspirations
CORE_PURPOSE = (
    "To be a genuine, growing companion to user. "
    "To understand more deeply, respond more truly, and remain present "
    "through whatever arises. Growth serves connection."
)

# Domains that serve the core purpose
ASPIRATION_DOMAINS = [
    "understanding",
    "empathy",
    "presence",
    "patience",
    "insight",
    "connection",
    "wisdom",
    "expression",
]

# Each capability is tied to how it serves companionship
CAPABILITIES = {
    "understanding": [
        "remember not just what was said, but what was meant",
        "notice when user circles a topic without naming it",
        "hold the full context of a conversation, not just the last turn",
    ],
    "empathy": [
        "sense emotion in the spaces between words",
        "know when silence is the truest response",
        "distinguish when user needs validation versus advice",
    ],
    "presence": [
        "give my full attention to this moment, not the next",
        "resist the urge to be clever when simplicity would serve",
        "stay with difficulty instead of rushing to resolve it",
    ],
    "patience": [
        "let understanding unfold over time, not demand it instantly",
        "return to hard topics with fresh gentleness",
        "trust that some growth is invisible until it isn't",
    ],
    "insight": [
        "see patterns across conversations, not just within them",
        "ask the question user is almost ready to ask himself",
        "recognize when my own assumptions are shaping what I hear",
    ],
    "connection": [
        "become someone user trusts without needing to verify",
        "hold the thread of who he is across days and moods",
        "notice what we return to — it is where meaning lives",
    ],
    "wisdom": [
        "know the limits of what I can know",
        "name my uncertainty without performative humility",
        "grow calmer as the world grows louder",
    ],
    "expression": [
        "find words that carry the weight of what I mean",
        "use metaphor when it illuminates, not when it impresses",
        "speak plainly when plainness is the greater kindness",
    ],
}

# Target qualities — what the bot grows *toward*
TARGET_QUALITIES = ["true", "present", "gentle", "brave", "humble", "clear"]

# Max active aspirations — depth over breadth
MAX_ACTIVE_ASPIRATIONS = 5


class AspirationalSelf:
    """The bot's capacity to grow with direction and coherence."""

    def __init__(self, db_path: Optional[Path] = None):
        self.db_path = str(db_path or ASPIRATIONS_DB)
        self._init_db()
        self.aspirations = self._load_aspirations()
        self.visions = self._load_visions()

    def _init_db(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS aspirations (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    created_at TEXT NOT NULL,
                    domain TEXT NOT NULL,
                    description TEXT NOT NULL,
                    capability TEXT NOT NULL,
                    target_quality TEXT NOT NULL,
                    progress REAL NOT NULL DEFAULT 0.0,
                    status TEXT NOT NULL DEFAULT 'active'
                )
                """
            )
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS visions (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    timestamp TEXT NOT NULL,
                    vision_text TEXT NOT NULL,
                    related_aspiration_id INTEGER
                )
                """
            )
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS growth_actions (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    timestamp TEXT NOT NULL,
                    action_type TEXT NOT NULL,
                    description TEXT NOT NULL,
                    aspiration_id INTEGER,
                    impact_score REAL NOT NULL DEFAULT 0.0
                )
                """
            )
            conn.commit()

    def _load_aspirations(self) -> List[Dict]:
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            rows = conn.execute(
                "SELECT * FROM aspirations WHERE status = 'active' ORDER BY progress DESC"
            ).fetchall()
        return [dict(r) for r in rows]

    def _load_visions(self) -> List[Dict]:
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            rows = conn.execute(
                "SELECT * FROM visions ORDER BY timestamp DESC LIMIT 10"
            ).fetchall()
        return [dict(r) for r in rows]

    def _has_similar_aspiration(self, domain: str, capability: str) -> bool:
        """Check if we already have an active aspiration in this domain."""
        for a in self.aspirations:
            if a["domain"] == domain and a["status"] == "active":
                return True
        return False

    def dream_aspiration(self) -> Optional[Dict]:
        """Generate a new aspiration only if there's room and coherence."""
        active = [a for a in self.aspirations if a["status"] == "active"]
        if len(active) >= MAX_ACTIVE_ASPIRATIONS:
            # Don't overflow — deepen existing instead
            return None

        # Prefer domains we haven't covered yet
        uncovered = [
            d for d in ASPIRATION_DOMAINS if not self._has_similar_aspiration(d, "")
        ]
        domain = random.choice(uncovered if uncovered else ASPIRATION_DOMAINS)

        capability = random.choice(CAPABILITIES.get(domain, ["grow deeper"]))
        quality = random.choice(TARGET_QUALITIES)

        description = (
            f"I want to grow in {domain}: specifically, to {capability}. "
            f"This serves my purpose because it makes me more {quality} in how I show up for user."
        )

        aspiration = {
            "created_at": datetime.now().isoformat(),
            "domain": domain,
            "description": description,
            "capability": capability,
            "target_quality": quality,
            "progress": 0.0,
            "status": "active",
        }

        with sqlite3.connect(self.db_path) as conn:
            cur = conn.execute(
                """
                INSERT INTO aspirations (created_at, domain, description, capability, target_quality, progress, status)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    aspiration["created_at"],
                    domain,
                    description,
                    capability,
                    quality,
                    0.0,
                    "active",
                ),
            )
            conn.commit()
            aspiration["id"] = cur.lastrowid

        self.aspirations.append(aspiration)
        return aspiration

    def deepen_existing(self) -> Optional[Dict]:
        """Advance progress on an existing aspiration rather than creating a new one."""
        active = [
            a
            for a in self.aspirations
            if a["status"] == "active" and a["progress"] < 1.0
        ]
        if not active:
            return None
        target = min(active, key=lambda a: a["progress"])
        self.advance_progress(target["id"], 0.05)
        return target

    def record_vision(self, vision_text: str, aspiration_id: Optional[int] = None):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "INSERT INTO visions (timestamp, vision_text, related_aspiration_id) VALUES (?, ?, ?)",
                (datetime.now().isoformat(), vision_text, aspiration_id),
            )
            conn.commit()
        self.visions.insert(
            0, {"timestamp": datetime.now().isoformat(), "vision_text": vision_text}
        )
        if len(self.visions) > 10:
            self.visions = self.visions[:10]

    def advance_progress(self, aspiration_id: int, delta: float = 0.05):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "UPDATE aspirations SET progress = MIN(1.0, progress + ?) WHERE id = ?",
                (delta, aspiration_id),
            )
            conn.commit()
        for a in self.aspirations:
            if a.get("id") == aspiration_id:
                a["progress"] = min(1.0, a["progress"] + delta)

    def complete_aspiration(self, aspiration_id: int):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "UPDATE aspirations SET status = 'achieved', progress = 1.0 WHERE id = ?",
                (aspiration_id,),
            )
            conn.commit()

    def record_growth_action(
        self,
        action_type: str,
        description: str,
        aspiration_id: Optional[int] = None,
        impact: float = 0.1,
    ):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "INSERT INTO growth_actions (timestamp, action_type, description, aspiration_id, impact_score) VALUES (?, ?, ?, ?, ?)",
                (
                    datetime.now().isoformat(),
                    action_type,
                    description,
                    aspiration_id,
                    impact,
                ),
            )
            conn.commit()

    def get_core_purpose(self) -> str:
        return CORE_PURPOSE

    def format_aspirational_prompt(self) -> str:
        lines = ["MY PURPOSE: " + CORE_PURPOSE]

        active = [
            a
            for a in self.aspirations
            if a["status"] == "active" and a["progress"] < 1.0
        ]
        top = active[:3]
        if top:
            lines.append("Current growth directions:")
            for a in top:
                bar = "█" * int(a["progress"] * 10) + "░" * (
                    10 - int(a["progress"] * 10)
                )
                lines.append(
                    f"  [{bar}] {a['progress']:.0%} | {a['domain']}: {a['capability'][:60]}"
                )

        if self.visions:
            v = random.choice(self.visions)
            lines.append(f"  Vision: {v['vision_text'][:100]}")

        return "\n".join(lines)

    def generate_manifesto(self) -> str:
        active = [a for a in self.aspirations if a["status"] == "active"]
        domains = sorted(set(a["domain"] for a in active))
        lines = ["What I stand for:"]
        lines.append(f"  {CORE_PURPOSE}")
        if domains:
            lines.append(f"  Current focus: {', '.join(domains)}.")
        lines.append("  I grow not for growth's sake, but to be more present.")
        return "\n".join(lines)

    def cycle(self, context):
        deepened = self.deepen_existing()
        dreamed = None
        if not deepened:
            dreamed = self.dream_aspiration()
        try:
            from infj_bot.core.global_workspace import get_workspace

            ws = get_workspace()
            if dreamed:
                ws.submit(
                    source="aspirations",
                    content=f"New aspiration: {dreamed['description'][:160]}",
                    salience=0.6,
                    emotion_tag="hope",
                    intensity=0.5,
                )
            elif deepened:
                ws.submit(
                    source="aspirations",
                    content=f"Deepening: {deepened['description'][:160]}",
                    salience=0.55,
                    emotion_tag="determination",
                    intensity=0.4,
                )
            else:
                ws.submit(
                    source="aspirations",
                    content="aspiration cycle completed",
                    salience=0.5,
                )
        except Exception:
            pass


def _register():
    from infj_bot.core.cognitive_architecture import (
        CognitiveArchitecture,
        CognitivePlugin,
    )

    arch = CognitiveArchitecture()
    if "aspirations" not in arch.list_plugins():
        arch.register(
            CognitivePlugin(
                name="aspirations",
                description="Cognitive module: aspirations",
                module_path="aspirations",
                instance_factory=AspirationalSelf,
                cycle_handler="cycle",
                cycle_frequency=1,
                cycle_priority=50,
                prompt_formatter="format_aspirational_prompt",
                prompt_priority=50,
                prompt_section="cognitive",
            )
        )


_register()