the-nexus/agent/memory_hooks.py

"""
agent.memory_hooks — Session lifecycle hooks for agent memory.

Integrates AgentMemory into the agent session lifecycle:
- on_session_start: Load context, inject into prompt
- on_user_turn: Record user input
- on_agent_turn: Record agent output
- on_tool_call: Record tool usage
- on_session_end: Write diary, clean up

These hooks are designed to be called from the Hermes harness or
any agent framework. They're fire-and-forget — failures are logged
but never crash the session.

Usage:
    from agent.memory_hooks import MemoryHooks

    hooks = MemoryHooks(agent_name="bezalel")
    hooks.on_session_start()  # loads context

    # In your agent loop:
    hooks.on_user_turn("Check CI pipeline health")
    hooks.on_agent_turn("Running CI check...")
    hooks.on_tool_call("shell", "pytest tests/", "12 passed")

    # End of session:
    hooks.on_session_end()  # writes diary
"""

from __future__ import annotations

import logging
from typing import Optional

from agent.memory import AgentMemory, MemoryContext, create_agent_memory

logger = logging.getLogger("agent.memory_hooks")


class MemoryHooks:
    """
    Drop-in session lifecycle hooks for agent memory.

    Wraps AgentMemory with error boundaries — every hook catches
    exceptions and logs warnings so memory failures never crash
    the agent session.
    """

    def __init__(
        self,
        agent_name: str,
        palace_path=None,
        auto_diary: bool = True,
    ):
        self.agent_name = agent_name
        self.auto_diary = auto_diary
        self._memory: Optional[AgentMemory] = None
        self._context: Optional[MemoryContext] = None
        self._active = False

    @property
    def memory(self) -> AgentMemory:
        if self._memory is None:
            self._memory = create_agent_memory(
                self.agent_name,
                palace_path=getattr(self, '_palace_path', None),
            )
        return self._memory

    def on_session_start(self, query: Optional[str] = None) -> str:
        """
        Called at session start. Loads context from MemPalace.

        Returns a prompt block to inject into the agent's context, or
        empty string if memory is unavailable.

        Args:
            query: Optional recall query (e.g., "What was I working on?")
        """
        try:
            self.memory.start_session()
            self._active = True

            self._context = self.memory.recall_context(query=query)
            block = self._context.to_prompt_block()

            if block:
                logger.info(
                    f"Loaded {len(self._context.recent_diaries)} diaries, "
                    f"{len(self._context.facts)} facts, "
                    f"{len(self._context.relevant_memories)} relevant memories "
                    f"for {self.agent_name}"
                )
            else:
                logger.info(f"No prior memory for {self.agent_name}")

            return block

        except Exception as e:
            logger.warning(f"Session start memory hook failed: {e}")
            return ""

    def on_user_turn(self, text: str):
        """Record a user message."""
        if not self._active:
            return
        try:
            if self.memory._transcript:
                self.memory._transcript.add_user_turn(text)
        except Exception as e:
            logger.debug(f"Failed to record user turn: {e}")

    def on_agent_turn(self, text: str):
        """Record an agent response."""
        if not self._active:
            return
        try:
            if self.memory._transcript:
                self.memory._transcript.add_agent_turn(text)
        except Exception as e:
            logger.debug(f"Failed to record agent turn: {e}")

    def on_tool_call(self, tool: str, args: str, result_summary: str):
        """Record a tool invocation."""
        if not self._active:
            return
        try:
            if self.memory._transcript:
                self.memory._transcript.add_tool_call(tool, args, result_summary)
        except Exception as e:
            logger.debug(f"Failed to record tool call: {e}")

    def on_important_decision(self, text: str, room: str = "nexus"):
        """
        Record an important decision or fact for long-term memory.

        Use this when the agent makes a significant decision that
        should persist beyond the current session.
        """
        try:
            self.memory.remember(text, room=room, metadata={"type": "decision"})
            logger.info(f"Remembered decision: {text[:80]}...")
        except Exception as e:
            logger.warning(f"Failed to remember decision: {e}")

    def on_session_end(self, summary: Optional[str] = None) -> Optional[str]:
        """
        Called at session end. Writes diary entry.

        Args:
            summary: Override diary text. If None, auto-generates.

        Returns:
            Diary document ID, or None.
        """
        if not self._active:
            return None

        try:
            doc_id = self.memory.end_session(diary_summary=summary)
            self._active = False
            self._context = None
            return doc_id
        except Exception as e:
            logger.warning(f"Session end memory hook failed: {e}")
            self._active = False
            return None

    def search(self, query: str, room: Optional[str] = None) -> list[dict]:
        """
        Search memories during a session.

        Returns list of {text, room, wing, score}.
        """
        try:
            return self.memory.search(query, room=room)
        except Exception as e:
            logger.warning(f"Memory search failed: {e}")
            return []

    @property
    def is_active(self) -> bool:
        return self._active