kira/backend/services/memory.py

"""Honcho memory service for Kira.

Integrates Honcho persistent memory into Kira's conversation pipeline:
- User context retrieval before LLM calls
- Message storage after each exchange
- Cross-session memory for personalized responses
"""

import logging
from honcho import Honcho
from honcho.peer import Peer
from honcho.session import Session
from config import settings

logger = logging.getLogger("kira.memory")


class KiraMemory:
    """Manages Honcho memory for Kira conversations."""

    def __init__(self):
        self._honcho: Honcho | None = None
        self._user_peer: Peer | None = None
        self._kira_peer: Peer | None = None
        self._session: Session | None = None
        self._initialized = False

    def init(self) -> bool:
        """Initialize Honcho connection. Returns False if not configured."""
        api_key = settings.honcho_api_key
        base_url = settings.honcho_base_url

        if not api_key:
            logger.warning("HONCHO_API_KEY not set — memory disabled")
            return False

        if not base_url:
            self._honcho = Honcho(
                api_key=api_key,
                workspace_id="kira",
                environment="production",
            )
        else:
            self._honcho = Honcho(
                api_key=api_key,
                base_url=base_url,
                workspace_id="kira",
            )

        logger.info(f"Honcho connected to workspace 'kira'")
        self._initialized = True
        return True

    @property
    def enabled(self) -> bool:
        return self._initialized and self._honcho is not None

    def ensure_peers(self, user_id: str = "default-user") -> None:
        """Get or create Honcho peers for the user and Kira."""
        if not self.enabled:
            return

        self._user_peer = self._honcho.peer(user_id)
        self._kira_peer = self._honcho.peer("kira")

        logger.info(f"Peers ready: user={user_id}, kira")

    def ensure_session(self, session_id: str) -> None:
        """Get or create a Honcho session for this conversation."""
        if not self.enabled:
            return

        self._session = self._honcho.session(session_id)

        # Add peers to session if not already members
        if self._user_peer and self._kira_peer:
            self._session.add_peers([self._user_peer, self._kira_peer])

        logger.info(f"Session ready: {session_id}")

    def get_user_context(self) -> str:
        """Query Honcho for context about the user.
        
        Returns a string summary of what Honcho knows about the user,
        to inject into the LLM system prompt. Empty string if no context.
        """
        if not self.enabled or not self._user_peer:
            return ""

        try:
            # Query Honcho's dialectic reasoning about the user
            context = self._user_peer.chat(
                "What should Kira know about this user? "
                "Summarize their preferences, current projects, mood, "
                "and any important context in 2-3 sentences."
            )
            if context:
                return f"\n[Memory: {context}]"
            return ""
        except Exception as e:
            logger.warning(f"Failed to get user context: {e}")
            return ""

    def get_kira_context(self) -> str:
        """Get what the user knows about Kira (relationship context)."""
        if not self.enabled or not self._user_peer:
            return ""

        try:
            context = self._user_peer.chat(
                "What is the user's relationship with Kira? "
                "How do they feel about their focus sessions? "
                "Summarize in 1-2 sentences.",
                target="kira",
            )
            if context:
                return f"\n[Kira Context: {context}]"
            return ""
        except Exception as e:
            logger.warning(f"Failed to get relationship context: {e}")
            return ""

    def build_system_prompt_suffix(self) -> str:
        """Build a context suffix to append to Kira's system prompt."""
        if not self.enabled:
            return ""

        user_ctx = self.get_user_context()
        kira_ctx = self.get_kira_context()

        parts = [s for s in [user_ctx, kira_ctx] if s]
        if not parts:
            return ""

        return "\n\n---\n### What Kira remembers:" + "".join(parts)

    def store_messages(
        self,
        user_message: str,
        kira_message: str,
    ) -> None:
        """Store a conversation exchange in Honcho."""
        if not self.enabled or not self._session:
            return

        try:
            messages = []
            if self._user_peer:
                messages.append(
                    self._user_peer.message(user_message)
                )
            if self._kira_peer:
                messages.append(
                    self._kira_peer.message(kira_message)
                )

            if messages:
                self._session.add_messages(messages)
                logger.debug("Stored conversation exchange in Honcho")
        except Exception as e:
            logger.warning(f"Failed to store messages: {e}")

    def store_user_message(self, text: str) -> None:
        """Store a single user message."""
        if not self.enabled or not self._session or not self._user_peer:
            return
        try:
            self._session.add_messages([self._user_peer.message(text)])
        except Exception as e:
            logger.warning(f"Failed to store user message: {e}")

    def store_kira_message(self, text: str) -> None:
        """Store a single Kira message."""
        if not self.enabled or not self._session or not self._kira_peer:
            return
        try:
            self._session.add_messages([self._kira_peer.message(text)])
        except Exception as e:
            logger.warning(f"Failed to store Kira message: {e}")

    # ─── User preferences (stored in Honcho peer metadata) ───

    DEFAULT_PREFERENCES: dict[str, str] = {
        "name": "",
        "scene": "cozy-room",
        "outfit": "cozy-hoodie",
        "accessory": "",
    }

    def get_user_preferences(self, user_id: str) -> dict[str, str]:
        """Load user preferences from Honcho peer metadata."""
        if not self.enabled:
            return dict(self.DEFAULT_PREFERENCES)

        try:
            peer = self._honcho.peer(user_id)
            meta = peer.metadata or {}
            prefs = dict(self.DEFAULT_PREFERENCES)
            for key in prefs:
                val = meta.get(key)
                if val is not None:
                    prefs[key] = str(val)
            logger.info(f"Loaded preferences for {user_id}: {prefs}")
            return prefs
        except Exception as e:
            logger.warning(f"Failed to load preferences: {e}")
            return dict(self.DEFAULT_PREFERENCES)

    def set_user_preference(
        self, user_id: str, key: str, value: str
    ) -> bool:
        """Save a single user preference to Honcho peer metadata."""
        if key not in self.DEFAULT_PREFERENCES:
            logger.warning(f"Unknown preference key: {key}")
            return False

        try:
            peer = self._honcho.peer(user_id)
            # Merge with existing metadata
            meta = dict(peer.metadata or {})
            meta[key] = value
            peer.set_metadata(meta)
            logger.info(f"Saved preference {user_id}.{key} = {value}")
            return True
        except Exception as e:
            logger.warning(f"Failed to save preference: {e}")
            return False


# Singleton instance for the app
kira_memory = KiraMemory()