Memory Architecture

OWL's memory system provides persistent context across sessions.

Three-Layer Design

┌─────────────────────────────────────────────────────────────┐
│                      Memory Architecture                     │
│                                                              │
│  ┌────────────────────────────────────────────────────────┐ │
│  │  Layer 1: Memory File (~/.owl/memory.md)               │ │
│  │  • Human-readable markdown                             │ │
│  │  • Always included in context                          │ │
│  │  • Preferences, notes, decisions                       │ │
│  └────────────────────────────────────────────────────────┘ │
│                            ↓                                 │
│  ┌────────────────────────────────────────────────────────┐ │
│  │  Layer 2: SQLite Database (~/.owl/memory/owl.db)       │ │
│  │  • Structured storage                                  │ │
│  │  • Conversations by session/project                    │ │
│  │  • Auto-extracted learnings                            │ │
│  └────────────────────────────────────────────────────────┘ │
│                            ↓                                 │
│  ┌────────────────────────────────────────────────────────┐ │
│  │  Layer 3: Session Summaries                            │ │
│  │  • Compressed old conversations                        │ │
│  │  • Prevents context overflow                           │ │
│  │  • Maintains continuity                                │ │
│  └────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘

Memory File

Implementation

# memory/memory_file.py
class MemoryFile:
    def __init__(self, path: Path):
        self.path = path

    def read(self) -> str:
        """Read entire memory file."""
        if self.path.exists():
            return self.path.read_text()
        return ""

    def remember(self, text: str, category: str = None) -> bool:
        """Add to memory with auto-categorization."""
        if category is None:
            category = self._categorize(text)

        content = self.read()
        # Find category section and append
        # ...
        self.path.write_text(new_content)
        return True

    def forget(self, text: str) -> bool:
        """Remove from memory by substring match."""
        content = self.read()
        # Find and remove matching line
        # ...

File Format

# Preferences
- I prefer tabs over spaces
- Use pytest for testing

# Notes
- API uses JWT authentication
- Database is PostgreSQL

# Decisions
- Chose FastAPI for performance

SQLite Store

Schema

-- Conversations
CREATE TABLE conversations (
    id INTEGER PRIMARY KEY,
    session_id TEXT NOT NULL,
    project_path TEXT,
    role TEXT NOT NULL,  -- user, assistant, tool
    content TEXT NOT NULL,
    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
    metadata JSON
);

CREATE INDEX idx_conv_session ON conversations(session_id);
CREATE INDEX idx_conv_project ON conversations(project_path);

-- Learnings
CREATE TABLE learnings (
    id INTEGER PRIMARY KEY,
    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
    category TEXT NOT NULL,  -- preference, project, style, etc.
    observation TEXT NOT NULL,
    learning TEXT NOT NULL,
    project_path TEXT,
    source TEXT DEFAULT 'auto',  -- user, auto
    confidence REAL DEFAULT 0.8,
    scope TEXT DEFAULT 'global'  -- global, project
);

-- Sessions
CREATE TABLE sessions (
    id TEXT PRIMARY KEY,
    started_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    last_active DATETIME,
    message_count INTEGER DEFAULT 0,
    summary TEXT
);

Implementation

# memory/store.py
class MemoryStore:
    def __init__(self, db_path: Path):
        self.conn = sqlite3.connect(str(db_path))
        self._init_schema()

    def add_message(
        self,
        session_id: str,
        role: str,
        content: str,
        project_path: str = None,
        metadata: dict = None
    ):
        """Store a conversation message."""
        self.conn.execute("""
            INSERT INTO conversations
            (session_id, project_path, role, content, metadata)
            VALUES (?, ?, ?, ?, ?)
        """, (session_id, project_path, role, content, json.dumps(metadata)))
        self.conn.commit()

    def get_conversation(
        self,
        session_id: str,
        limit: int = 20
    ) -> List[Message]:
        """Get recent messages for session."""
        cursor = self.conn.execute("""
            SELECT role, content, timestamp
            FROM conversations
            WHERE session_id = ?
            ORDER BY timestamp DESC
            LIMIT ?
        """, (session_id, limit))
        return [Message(*row) for row in reversed(cursor.fetchall())]

    def get_conversation_for_project(
        self,
        session_id: str,
        project_path: str,
        limit: int = 20
    ) -> List[Message]:
        """Get messages scoped to a project."""
        cursor = self.conn.execute("""
            SELECT role, content, timestamp
            FROM conversations
            WHERE session_id = ? AND project_path = ?
            ORDER BY timestamp DESC
            LIMIT ?
        """, (session_id, project_path, limit))
        return [Message(*row) for row in reversed(cursor.fetchall())]

Summarizer

Purpose

Prevent context overflow by compressing old conversations.

Configuration

SUMMARIZE_THRESHOLD = 20  # When to trigger
SUMMARIZE_BATCH = 10      # How many to summarize
KEEP_RECENT = 10          # Always keep fresh

Implementation

# memory/summarizer.py
class Summarizer:
    def __init__(self, llm: LLMClient):
        self.llm = llm

    def summarize_async(self, session_id: str):
        """Trigger async summarization if needed."""
        thread = threading.Thread(
            target=self._summarize,
            args=(session_id,)
        )
        thread.start()

    def _summarize(self, session_id: str):
        messages = self.memory.get_conversation(session_id)

        if len(messages) < SUMMARIZE_THRESHOLD:
            return

        # Get oldest batch to summarize
        to_summarize = messages[:SUMMARIZE_BATCH]

        # Generate summary using LLM
        prompt = f"Summarize this conversation in 2-4 sentences:\n{format_messages(to_summarize)}"
        summary = self.llm.simple_query(prompt)

        # Store summary
        self.memory.update_session_summary(session_id, summary)

        # Delete summarized messages
        for msg in to_summarize:
            self.memory.delete_message(msg.id)

Summary Format

Session Summary:
- Discussed Python testing approaches, settled on pytest
- Reviewed authentication flow, identified security issue
- Implemented user registration with email verification

Learnings

Extraction

Learnings are extracted during reflection:

# soul/reflector.py
class SoulReflector:
    def reflect_async(self, session_id: str, project_path: str = None):
        thread = threading.Thread(
            target=self._reflect,
            args=(session_id, project_path)
        )
        thread.start()

    def _reflect(self, session_id: str, project_path: str):
        # Get recent conversation
        messages = self.memory.get_conversation(session_id, limit=6)

        # Ask LLM to extract learnings
        prompt = """
        Extract learnings from this conversation.
        Format as JSON array:
        [{"category": "preference|project|style|technical",
          "observation": "what happened",
          "learning": "what to remember",
          "scope": "global|project"}]
        """

        result = self.llm.simple_query(prompt + format_messages(messages))
        learnings = json.loads(result)

        # Store learnings
        for learning in learnings:
            self.memory.add_learning(
                category=learning["category"],
                observation=learning["observation"],
                learning=learning["learning"],
                project_path=project_path if learning["scope"] == "project" else None,
                scope=learning["scope"]
            )

Categories

Category	Scope	Example
preference	global	"User prefers functional style"
project	project	"Uses PostgreSQL database"
style	global	"Prefers descriptive names"
technical	project	"API uses JWT authentication"
feedback	global	"Likes concise responses"

Context Building

How Memory is Used

# llm/context.py
class ContextBuilder:
    def build(self, user_message: str, history: List[Message]) -> List[Message]:
        sections = []

        # Memory file (always included)
        memory_content = self.memory_file.read()
        if memory_content:
            sections.append(f"## MY MEMORY\n{memory_content}")

        # Session summary (if exists)
        summary = self.memory.get_session_summary(self.session_id)
        if summary:
            sections.append(f"## PREVIOUS CONTEXT\n{summary}")

        # Build system prompt
        system_prompt = "\n\n".join(sections)

        return [
            Message(role="system", content=system_prompt),
            *history,
            Message(role="user", content=user_message)
        ]

Data Lifecycle

User Message
    │
    ├──→ Stored in SQLite (conversations)
    │
    ├──→ Every 3 exchanges → Reflection
    │         │
    │         └──→ Learnings extracted → SQLite (learnings)
    │                   │
    │                   └──→ Every 10 learnings → Evolution eligible
    │
    └──→ Every 20 messages → Summarization
              │
              └──→ Old messages compressed
                        │
                        └──→ Summary stored in sessions table