project-lyra/cortex/autonomy/self/state.py

"""
Self-state management for Project Lyra.
Maintains persistent identity, mood, energy, and focus across sessions.
"""

import json
import logging
import os
from datetime import datetime
from pathlib import Path
from typing import Dict, Any, Optional

# Configuration
STATE_FILE = Path(os.getenv("SELF_STATE_FILE", "/app/data/self_state.json"))
VERBOSE_DEBUG = os.getenv("VERBOSE_DEBUG", "false").lower() == "true"

logger = logging.getLogger(__name__)

if VERBOSE_DEBUG:
    logger.setLevel(logging.DEBUG)

# Default state structure
DEFAULT_STATE = {
    "mood": "neutral",
    "energy": 0.8,
    "focus": "user_request",
    "confidence": 0.7,
    "curiosity": 0.5,
    "last_updated": None,
    "interaction_count": 0,
    "learning_queue": [],  # Topics Lyra wants to explore
    "active_goals": [],  # Self-directed goals
    "preferences": {
        "verbosity": "medium",
        "formality": "casual",
        "proactivity": 0.3  # How likely to suggest things unprompted
    },
    "metadata": {
        "version": "1.0",
        "created_at": None
    }
}


class SelfState:
    """Manages Lyra's persistent self-state."""

    def __init__(self):
        self._state = self._load_state()

    def _load_state(self) -> Dict[str, Any]:
        """Load state from disk or create default."""
        if STATE_FILE.exists():
            try:
                with open(STATE_FILE, 'r') as f:
                    state = json.load(f)
                    logger.info(f"Loaded self-state from {STATE_FILE}")
                    return state
            except Exception as e:
                logger.error(f"Failed to load self-state: {e}")
                return self._create_default_state()
        else:
            return self._create_default_state()

    def _create_default_state(self) -> Dict[str, Any]:
        """Create and save default state."""
        state = DEFAULT_STATE.copy()
        state["metadata"]["created_at"] = datetime.now().isoformat()
        state["last_updated"] = datetime.now().isoformat()
        self._save_state(state)
        logger.info("Created new default self-state")
        return state

    def _save_state(self, state: Dict[str, Any]) -> None:
        """Persist state to disk."""
        try:
            STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
            with open(STATE_FILE, 'w') as f:
                json.dump(state, f, indent=2)
            if VERBOSE_DEBUG:
                logger.debug(f"Saved self-state to {STATE_FILE}")
        except Exception as e:
            logger.error(f"Failed to save self-state: {e}")

    def get_state(self) -> Dict[str, Any]:
        """Get current state snapshot."""
        return self._state.copy()

    def update_from_interaction(
        self,
        mood_delta: float = 0.0,
        energy_delta: float = 0.0,
        new_focus: Optional[str] = None,
        confidence_delta: float = 0.0,
        curiosity_delta: float = 0.0
    ) -> None:
        """
        Update state based on interaction.

        Args:
            mood_delta: Change in mood (-1.0 to 1.0)
            energy_delta: Change in energy (-1.0 to 1.0)
            new_focus: New focus area
            confidence_delta: Change in confidence
            curiosity_delta: Change in curiosity
        """
        # Apply deltas with bounds checking
        self._state["energy"] = max(0.0, min(1.0,
            self._state.get("energy", 0.8) + energy_delta))

        self._state["confidence"] = max(0.0, min(1.0,
            self._state.get("confidence", 0.7) + confidence_delta))

        self._state["curiosity"] = max(0.0, min(1.0,
            self._state.get("curiosity", 0.5) + curiosity_delta))

        # Update focus if provided
        if new_focus:
            self._state["focus"] = new_focus

        # Update mood (simplified sentiment)
        if mood_delta != 0:
            mood_map = ["frustrated", "neutral", "engaged", "excited"]
            current_mood_idx = 1  # neutral default
            if self._state.get("mood") in mood_map:
                current_mood_idx = mood_map.index(self._state["mood"])

            new_mood_idx = max(0, min(len(mood_map) - 1,
                int(current_mood_idx + mood_delta * 2)))
            self._state["mood"] = mood_map[new_mood_idx]

        # Increment interaction counter
        self._state["interaction_count"] = self._state.get("interaction_count", 0) + 1
        self._state["last_updated"] = datetime.now().isoformat()

        # Persist changes
        self._save_state(self._state)

        if VERBOSE_DEBUG:
            logger.debug(f"Updated self-state: mood={self._state['mood']}, "
                        f"energy={self._state['energy']:.2f}, "
                        f"confidence={self._state['confidence']:.2f}")

    def add_learning_goal(self, topic: str) -> None:
        """Add topic to learning queue."""
        queue = self._state.get("learning_queue", [])
        if topic not in [item.get("topic") for item in queue]:
            queue.append({
                "topic": topic,
                "added_at": datetime.now().isoformat(),
                "priority": 0.5
            })
            self._state["learning_queue"] = queue
            self._save_state(self._state)
            logger.info(f"Added learning goal: {topic}")

    def add_active_goal(self, goal: str, context: str = "") -> None:
        """Add self-directed goal."""
        goals = self._state.get("active_goals", [])
        goals.append({
            "goal": goal,
            "context": context,
            "created_at": datetime.now().isoformat(),
            "status": "active"
        })
        self._state["active_goals"] = goals
        self._save_state(self._state)
        logger.info(f"Added active goal: {goal}")


# Global instance
_self_state_instance = None

def get_self_state_instance() -> SelfState:
    """Get or create global SelfState instance."""
    global _self_state_instance
    if _self_state_instance is None:
        _self_state_instance = SelfState()
    return _self_state_instance


def load_self_state() -> Dict[str, Any]:
    """Load self state - public API for backwards compatibility."""
    return get_self_state_instance().get_state()


def update_self_state(**kwargs) -> None:
    """Update self state - public API."""
    get_self_state_instance().update_from_interaction(**kwargs)