feat(core-ai): Implement multi-tiered conversation memory

Introduces a comprehensive, multi-tiered memory system to provide conversation history and context for the AI agent. This lays the foundation for more stateful and intelligent interactions.

Key components of this implementation:

- **Multi-Tiered Memory Architecture:**
  - **Tier 1 (Working Memory):** A fast, in-memory buffer (`ConversationBufferMemory`) that holds the most recent turns of a conversation for immediate access.
  - **Tier 3 (Long-Term Memory):** A persistent, semantic search-based memory store using Qdrant (`QdrantConversationMemory`). It stores all conversation turns as vector embeddings, enabling long-term recall and similarity search.

- **Qdrant Integration:**
  - The `qdrant-client` is added to manage collections and perform vector search operations.
  - Each user is assigned a dedicated Qdrant collection for multi-tenancy.

- **Ollama Embedding Client:**
  - A new `OllamaEmbeddingClient` generates text embeddings via the Ollama API, replacing the need for local sentence-transformer models. This significantly reduces the service's dependency footprint.

- **Configuration and Stack Updates:**
  - The `config.py` and `core-ai.yml` stack file are updated with new settings for enabling memory, configuring Qdrant, and specifying the embedding model.

- **Utility and Schema Additions:**
  - New Pydantic schemas (`memory/schemas.py`) define the data structures for conversation turns and memory management.
  - Utility functions (`utils.py`) are added for user ID sanitization and collection naming.

This feature enhances the agent's capabilities by allowing it to maintain context across multiple turns and sessions, leading to more coherent and relevant responses.

services/core-ai/main.py

@@ -20,6 +20,7 @@ from src.agents import (
     PYDANTIC_AI_AVAILABLE
 )
 from src.tools import get_all_tools
+from src.utils import extract_user_id_from_request
 
 async def chat_completions(request):
     """
@@ -42,11 +43,15 @@ async def chat_completions(request):
         conversation_id = data.get("conversation_id")
         enable_tools = data.get("enable_tools", True)  # Tools enabled by default
 
+        # Extract user ID from request (supports user_id, user_email, or falls back to default)
+        user_id = extract_user_id_from_request(data)
+
         if not messages:
             raise web.HTTPBadRequest(reason="'messages' field is required")
 
         # Get the agent instance (default: PydanticAI agent with tools)
-        agent = get_pydantic_agent(discover_tools=enable_tools)
+        # Note: Agent is cached per user_id, so each user gets their own agent instance with their memory
+        agent = get_pydantic_agent(discover_tools=enable_tools, user_id=user_id)
 
         # For non-streaming requests, collect the full response
         if not stream:
@@ -215,11 +220,14 @@ async def chat_pydantic(request):
         conversation_id = data.get("conversation_id")
         enable_tools = data.get("enable_tools", True)
 
+        # Extract user ID from request
+        user_id = extract_user_id_from_request(data)
+
         if not messages:
             raise web.HTTPBadRequest(reason="'messages' field is required")
 
         # Get PydanticAI agent with or without tools
-        agent = get_pydantic_agent(discover_tools=enable_tools)
+        agent = get_pydantic_agent(discover_tools=enable_tools, user_id=user_id)
 
         # Non-streaming response
         if not stream:

services/core-ai/requirements.txt

@@ -1,5 +1,5 @@
-# PydanticAI and dependencies
-pydantic-ai  # Full library with Ollama support
+# PydanticAI and dependencies (slim to reduce bloat)
+pydantic-ai-slim  # Minimal library - Ollama uses OpenAI-compatible API
 pydantic>=2.10.3  # Let pydantic-ai determine the compatible version
 pydantic-settings==2.6.1
 
@@ -12,6 +12,9 @@ aiohttp-cors==0.7.0
 python-dotenv>=1.1.0
 httpx==0.28.1
 
+# Memory system
+qdrant-client>=1.12.0  # Vector database client
+
 # Testing
 pytest==8.3.4
 pytest-asyncio==0.24.0

services/core-ai/src/agents/pydantic_agent.py

@@ -23,6 +23,8 @@ except ImportError:
 
 from src.config import get_settings
 from src.prompts import get_prompt
+from src.memory import get_memory_manager_for_user, MessageRole
+from src.utils import sanitize_email_to_user_id
 
 logger = logging.getLogger(__name__)
 
@@ -37,7 +39,7 @@ class PydanticAgent:
         response = await agent.chat_completion(messages=[{"role": "user", "content": "Hello"}])
     """
 
-    def __init__(self, tools: List = None, discover_tools: bool = False):
+    def __init__(self, tools: List = None, discover_tools: bool = False, user_id: Optional[str] = None, enable_memory: Optional[bool] = None):
         if not PYDANTIC_AI_AVAILABLE:
             raise ImportError("PydanticAI not available. Install with: pip install pydantic-ai")
 
@@ -45,6 +47,26 @@ class PydanticAgent:
 
         self.settings = get_settings()
 
+        # Memory configuration
+        self.enable_memory = enable_memory if enable_memory is not None else self.settings.memory_enabled
+        self.user_id = user_id or self.settings.default_user_id
+
+        # Initialize memory manager if enabled
+        if self.enable_memory:
+            try:
+                self.memory_manager = get_memory_manager_for_user(
+                    user_id=self.user_id,
+                    buffer_max_turns=self.settings.memory_tier1_size
+                )
+                logger.info(f"PydanticAgent: Memory enabled for user '{self.user_id}'")
+            except Exception as e:
+                logger.warning(f"PydanticAgent: Failed to initialize memory: {e}. Continuing without memory.")
+                self.enable_memory = False
+                self.memory_manager = None
+        else:
+            self.memory_manager = None
+            logger.info("PydanticAgent: Memory disabled")
+
         # Tools can be provided explicitly or discovered
         if tools is not None:
             # Explicit tools provided
@@ -132,6 +154,18 @@ class PydanticAgent:
             user_query = user_messages[-1]["content"]
             logger.info(f"📤 PydanticAgent: User query: {user_query[:100]}...")
 
+            # Store user message in memory
+            if self.enable_memory and conversation_id:
+                try:
+                    await self.memory_manager.add_turn(
+                        conversation_id=conversation_id,
+                        role=MessageRole.USER,
+                        content=user_query
+                    )
+                    logger.debug(f"Stored user message in memory for conversation {conversation_id}")
+                except Exception as e:
+                    logger.warning(f"Failed to store user message in memory: {e}")
+
             # Run the agent
             if stream:
                 # Streaming response - collect chunks to avoid async context issues
@@ -159,6 +193,19 @@ class PydanticAgent:
                 # Final chunk with finish reason
                 yield {"type": "content", "content": "", "finish_reason": "stop"}
                 logger.info(f"📥 PydanticAgent: Streaming complete")
+
+                # Store assistant response in memory (streaming)
+                if self.enable_memory and conversation_id:
+                    try:
+                        await self.memory_manager.add_turn(
+                            conversation_id=conversation_id,
+                            role=MessageRole.ASSISTANT,
+                            content=previous_text
+                        )
+                        logger.debug(f"Stored assistant response in memory for conversation {conversation_id}")
+                    except Exception as e:
+                        logger.warning(f"Failed to store assistant response in memory: {e}")
+
             else:
                 # Non-streaming response
                 result = await self.agent.run(user_query)
@@ -166,6 +213,18 @@ class PydanticAgent:
                 logger.info(f"📥 PydanticAgent: Response: {str(response_text)[:100]}...")
                 yield {"type": "content", "content": str(response_text), "finish_reason": "stop"}
 
+                # Store assistant response in memory (non-streaming)
+                if self.enable_memory and conversation_id:
+                    try:
+                        await self.memory_manager.add_turn(
+                            conversation_id=conversation_id,
+                            role=MessageRole.ASSISTANT,
+                            content=str(response_text)
+                        )
+                        logger.debug(f"Stored assistant response in memory for conversation {conversation_id}")
+                    except Exception as e:
+                        logger.warning(f"Failed to store assistant response in memory: {e}")
+
         except Exception as e:
             logger.error(f"PydanticAgent: Error during chat: {e}", exc_info=True)
             yield {
@@ -202,7 +261,7 @@ class PydanticAgent:
 
 
 @lru_cache()
-def get_pydantic_agent(tools: tuple = None, discover_tools: bool = False) -> PydanticAgent:
+def get_pydantic_agent(tools: tuple = None, discover_tools: bool = False, user_id: str = None, enable_memory: bool = None) -> PydanticAgent:
     """
     Get cached PydanticAI agent instance.
 
@@ -212,9 +271,11 @@ def get_pydantic_agent(tools: tuple = None, discover_tools: bool = False) -> Pyd
     Args:
         tools: Tuple of tool functions (None to use discovery)
         discover_tools: Whether to discover tools from registry
+        user_id: Optional user ID for memory (defaults to config default_user_id)
+        enable_memory: Optional memory enable flag (defaults to config memory_enabled)
 
     Returns:
         Cached PydanticAgent instance
     """
     tools_list = list(tools) if tools is not None else None
-    return PydanticAgent(tools=tools_list, discover_tools=discover_tools)
+    return PydanticAgent(tools=tools_list, discover_tools=discover_tools, user_id=user_id, enable_memory=enable_memory)

services/core-ai/src/config.py

@@ -38,6 +38,20 @@ class Settings(BaseSettings):
     simple_enabled: bool = True   # Enable simple endpoint
     adk_enabled: bool = True      # Enable ADK endpoint
 
+    # Memory System Configuration
+    memory_enabled: bool = True
+    memory_tier1_size: int = 10  # Max turns in RAM buffer
+
+    # Qdrant Configuration (for conversation memory)
+    qdrant_url: str = "http://qdrant:6333"
+    qdrant_collection_prefix: str = "core_ai_user"  # Prefix for user collections
+
+    # Embedding Configuration
+    embedding_model: str = "nomic-embed-text"  # Ollama embedding model
+    embedding_dimension: int = 768  # nomic-embed-text dimension
+
+    # Default User (until external auth is integrated)
+    default_user_id: str = "llmdefault_at_schweitz_net"
 
     class Config:
         env_file = ".env"

services/core-ai/src/memory/__init__.py

@@ -0,0 +1,56 @@
+"""
+Multi-tenant memory system for conversation persistence
+
+Architecture:
+- Tier 1: ConversationBufferMemory (in-memory, fast, last 10 turns) - per user
+- Tier 2/3: QdrantConversationMemory (persistent + semantic search) - separate collection per user
+- Manager: MemoryManager (orchestrates all tiers) - per user instance
+
+Multi-tenancy:
+- Each user gets their own Qdrant collection: core_ai_user_{user_id}
+- Complete data isolation between users
+- Easy GDPR compliance (delete entire user collection)
+"""
+from .tier1_buffer import ConversationBufferMemory, get_buffer_memory
+from .qdrant_memory import QdrantConversationMemory, get_qdrant_memory_for_user
+from .manager import MemoryManager, get_memory_manager_for_user, clear_memory_manager_cache
+from .schemas import (
+    ConversationTurn,
+    ConversationBuffer,
+    ConversationMetadata,
+    ConversationSummary,
+    MemoryQuery,
+    MemoryResult,
+    MessageRole,
+    TokenUsage,
+    ConversationListResponse,
+    ConversationDetailResponse,
+    ConversationSearchRequest,
+    ConversationSearchResponse,
+)
+
+__all__ = [
+    # Manager (primary interface)
+    "MemoryManager",
+    "get_memory_manager_for_user",
+    "clear_memory_manager_cache",
+    # Tier 1
+    "ConversationBufferMemory",
+    "get_buffer_memory",
+    # Tier 2/3
+    "QdrantConversationMemory",
+    "get_qdrant_memory_for_user",
+    # Schemas
+    "ConversationTurn",
+    "ConversationBuffer",
+    "ConversationMetadata",
+    "ConversationSummary",
+    "MemoryQuery",
+    "MemoryResult",
+    "MessageRole",
+    "TokenUsage",
+    "ConversationListResponse",
+    "ConversationDetailResponse",
+    "ConversationSearchRequest",
+    "ConversationSearchResponse",
+]

services/core-ai/src/memory/base.py

@@ -0,0 +1,169 @@
+"""
+Base classes for memory system
+"""
+from abc import ABC, abstractmethod
+from typing import List, Optional
+from .schemas import ConversationTurn, ConversationBuffer, MemoryQuery, MemoryResult
+
+
+class BaseMemory(ABC):
+    """Base class for all memory tiers"""
+
+    @abstractmethod
+    async def add_turn(self, conversation_id: str, turn: ConversationTurn) -> None:
+        """
+        Add a new turn to memory
+
+        Args:
+            conversation_id: Unique conversation identifier
+            turn: The conversation turn to store
+        """
+        pass
+
+    @abstractmethod
+    async def get_turns(
+        self,
+        conversation_id: str,
+        limit: Optional[int] = None,
+        offset: int = 0
+    ) -> List[ConversationTurn]:
+        """
+        Retrieve turns from memory
+
+        Args:
+            conversation_id: Unique conversation identifier
+            limit: Maximum number of turns to retrieve
+            offset: Number of turns to skip
+
+        Returns:
+            List of conversation turns
+        """
+        pass
+
+    @abstractmethod
+    async def clear_conversation(self, conversation_id: str) -> None:
+        """
+        Clear all turns for a conversation
+
+        Args:
+            conversation_id: Unique conversation identifier
+        """
+        pass
+
+    @abstractmethod
+    async def conversation_exists(self, conversation_id: str) -> bool:
+        """
+        Check if a conversation exists in this memory tier
+
+        Args:
+            conversation_id: Unique conversation identifier
+
+        Returns:
+            True if conversation exists
+        """
+        pass
+
+
+class Tier1Memory(BaseMemory):
+    """Base class for Tier 1 (working memory)"""
+
+    @abstractmethod
+    async def get_buffer(self, conversation_id: str) -> Optional[ConversationBuffer]:
+        """
+        Get the full conversation buffer
+
+        Args:
+            conversation_id: Unique conversation identifier
+
+        Returns:
+            ConversationBuffer or None if not found
+        """
+        pass
+
+    @abstractmethod
+    async def prune(self, conversation_id: str, keep_last: int = 5) -> None:
+        """
+        Prune old turns, keeping only the most recent ones
+
+        Args:
+            conversation_id: Unique conversation identifier
+            keep_last: Number of recent turns to keep
+        """
+        pass
+
+
+class Tier2Memory(BaseMemory):
+    """Base class for Tier 2 (short-term memory with summaries)"""
+
+    @abstractmethod
+    async def add_summary(
+        self,
+        conversation_id: str,
+        summary_text: str,
+        turn_range_start: int,
+        turn_range_end: int
+    ) -> None:
+        """
+        Add a conversation summary
+
+        Args:
+            conversation_id: Unique conversation identifier
+            summary_text: The summarized text
+            turn_range_start: First turn number in summary
+            turn_range_end: Last turn number in summary
+        """
+        pass
+
+    @abstractmethod
+    async def get_summaries(self, conversation_id: str) -> List[dict]:
+        """
+        Get all summaries for a conversation
+
+        Args:
+            conversation_id: Unique conversation identifier
+
+        Returns:
+            List of summary dictionaries
+        """
+        pass
+
+
+class Tier3Memory(BaseMemory):
+    """Base class for Tier 3 (long-term vector memory)"""
+
+    @abstractmethod
+    async def add_turn_with_embedding(
+        self,
+        conversation_id: str,
+        turn: ConversationTurn,
+        embedding: List[float]
+    ) -> None:
+        """
+        Add a turn with its vector embedding
+
+        Args:
+            conversation_id: Unique conversation identifier
+            turn: The conversation turn
+            embedding: Vector embedding of the turn content
+        """
+        pass
+
+    @abstractmethod
+    async def similarity_search(
+        self,
+        query_embedding: List[float],
+        conversation_id: Optional[str] = None,
+        limit: int = 5
+    ) -> List[dict]:
+        """
+        Perform semantic similarity search
+
+        Args:
+            query_embedding: Vector embedding of the search query
+            conversation_id: Optional filter to specific conversation
+            limit: Maximum number of results
+
+        Returns:
+            List of matching turns with scores
+        """
+        pass

services/core-ai/src/memory/manager.py

@@ -0,0 +1,337 @@
+"""
+Multi-tenant Memory Manager: Orchestrates all memory tiers with per-user isolation
+
+Coordinates:
+- Tier 1: ConversationBufferMemory (RAM, fast, last N turns) - per user
+- Tier 2/3: QdrantConversationMemory (persistent + semantic) - separate collection per user
+
+Provides unified interface for memory operations with automatic
+tier management and per-user data isolation.
+"""
+import logging
+import asyncio
+from typing import List, Optional, Dict, Any
+from datetime import datetime
+
+from .tier1_buffer import ConversationBufferMemory
+from .qdrant_memory import QdrantConversationMemory, get_qdrant_memory_for_user
+from .schemas import ConversationTurn, MessageRole, TokenUsage
+from src.config import get_settings
+
+logger = logging.getLogger(__name__)
+settings = get_settings()
+
+
+class MemoryManager:
+    """
+    Multi-tenant unified memory manager orchestrating all tiers
+
+    Features:
+    - Per-user data isolation (separate Qdrant collections)
+    - Per-user in-memory buffers
+    - Automatic consolidation from buffer to Qdrant
+    - Semantic search within user's conversations
+    - Memory lifecycle management
+
+    Responsibilities:
+    - Add turns to appropriate tiers
+    - Retrieve conversation history (buffer + persistent)
+    - Consolidate buffer to persistent storage
+    - Semantic search across user's conversations
+    - Memory lifecycle management
+    """
+
+    def __init__(
+        self,
+        user_id: str,
+        buffer_max_turns: int = 10,
+        auto_consolidate: bool = True
+    ):
+        """
+        Initialize memory manager for a specific user
+
+        Args:
+            user_id: Sanitized user ID (email format: username_at_domain_com)
+            buffer_max_turns: Max turns to keep in RAM buffer
+            auto_consolidate: Automatically consolidate when buffer threshold reached
+        """
+        self.user_id = user_id
+        self.auto_consolidate = auto_consolidate
+
+        # Create user-specific buffer (in-memory)
+        self.buffer_memory = ConversationBufferMemory(max_turns=buffer_max_turns)
+
+        # Create user-specific Qdrant memory (separate collection)
+        self.qdrant_memory = get_qdrant_memory_for_user(user_id)
+
+        logger.info(
+            f"MemoryManager initialized for user '{user_id}' "
+            f"(auto_consolidate={auto_consolidate}, buffer_max={buffer_max_turns})"
+        )
+
+    async def add_turn(
+        self,
+        conversation_id: str,
+        role: MessageRole,
+        content: str,
+        tokens: Optional[TokenUsage] = None,
+        metadata: Optional[Dict[str, Any]] = None
+    ) -> ConversationTurn:
+        """
+        Add a conversation turn to memory
+
+        Automatically:
+        1. Adds to Tier 1 (buffer)
+        2. Adds to Tier 2/3 (Qdrant) immediately
+        3. Auto-prunes buffer if max turns reached
+
+        Args:
+            conversation_id: Unique conversation identifier
+            role: Message role (user, assistant, system)
+            content: Message content
+            tokens: Optional token usage
+            metadata: Optional metadata
+
+        Returns:
+            The created conversation turn
+        """
+        # Get current buffer to determine turn number
+        buffer = await self.buffer_memory.get_buffer(conversation_id)
+        turn_number = (buffer.metadata.turn_count + 1) if buffer else 1
+
+        # Create turn with user_id
+        turn = ConversationTurn(
+            role=role,
+            content=content,
+            timestamp=datetime.utcnow(),
+            turn_number=turn_number,
+            user_id=self.user_id,
+            tokens=tokens,
+            metadata=metadata or {}
+        )
+
+        # Add to Tier 1 (buffer) - fast RAM storage
+        await self.buffer_memory.add_turn(conversation_id, turn)
+        logger.debug(
+            f"Turn {turn_number} added to buffer for user '{self.user_id}' "
+            f"conversation {conversation_id}"
+        )
+
+        # Add to Tier 2/3 (Qdrant) immediately - persistent storage with embeddings
+        try:
+            await self.qdrant_memory.add_turn(conversation_id, turn)
+            logger.debug(
+                f"Turn {turn_number} added to Qdrant for user '{self.user_id}' "
+                f"conversation {conversation_id}"
+            )
+        except Exception as e:
+            logger.error(
+                f"Error adding turn to Qdrant for user '{self.user_id}': {e}"
+            )
+            # Don't fail the whole operation if Qdrant fails
+            # Buffer still has the turn
+
+        return turn
+
+    async def get_recent_turns(
+        self,
+        conversation_id: str,
+        limit: int = 10
+    ) -> List[ConversationTurn]:
+        """
+        Get recent conversation turns (from buffer)
+
+        Args:
+            conversation_id: Unique conversation identifier
+            limit: Maximum number of turns to retrieve
+
+        Returns:
+            List of recent conversation turns
+        """
+        return await self.buffer_memory.get_recent_turns(conversation_id, limit)
+
+    async def get_full_history(
+        self,
+        conversation_id: str,
+        include_buffer: bool = True
+    ) -> List[ConversationTurn]:
+        """
+        Get complete conversation history
+
+        Retrieves from Qdrant (Tier 2) - buffer is just a cache
+
+        Args:
+            conversation_id: Unique conversation identifier
+            include_buffer: Ignored (kept for API compatibility)
+
+        Returns:
+            Complete conversation history, sorted chronologically
+        """
+        # Get from Qdrant (source of truth)
+        turns = await self.qdrant_memory.get_turns(conversation_id)
+
+        # Sort chronologically (should already be sorted, but ensure it)
+        turns.sort(key=lambda t: t.turn_number)
+
+        return turns
+
+    async def search_conversations(
+        self,
+        query: str,
+        conversation_id: Optional[str] = None,
+        limit: int = 5
+    ) -> List[Dict[str, Any]]:
+        """
+        Semantic search across user's conversations (Tier 3 mode)
+
+        Searches only within this user's collection.
+
+        Args:
+            query: Search query
+            conversation_id: Optional filter to specific conversation
+            limit: Maximum number of results
+
+        Returns:
+            List of matching turns with scores
+        """
+        return await self.qdrant_memory.similarity_search(
+            query=query,
+            conversation_id=conversation_id,
+            limit=limit
+        )
+
+    async def clear_conversation(
+        self,
+        conversation_id: str,
+        clear_buffer: bool = True,
+        clear_qdrant: bool = True
+    ) -> None:
+        """
+        Clear conversation from memory
+
+        Args:
+            conversation_id: Unique conversation identifier
+            clear_buffer: Clear from Tier 1 buffer
+            clear_qdrant: Clear from Tier 2/3 Qdrant
+        """
+        if clear_buffer:
+            await self.buffer_memory.clear_conversation(conversation_id)
+            logger.info(
+                f"Cleared buffer for user '{self.user_id}' conversation {conversation_id}"
+            )
+
+        if clear_qdrant:
+            await self.qdrant_memory.clear_conversation(conversation_id)
+            logger.info(
+                f"Cleared Qdrant for user '{self.user_id}' conversation {conversation_id}"
+            )
+
+    async def clear_all_user_data(self) -> None:
+        """
+        Clear ALL data for this user (GDPR compliance)
+
+        Deletes:
+        - All buffer data for this user
+        - Entire Qdrant collection for this user
+        """
+        # Clear all buffers (in-memory)
+        conversation_ids = await self.buffer_memory.get_all_conversation_ids()
+        for conv_id in conversation_ids:
+            await self.buffer_memory.clear_conversation(conv_id)
+
+        # Delete entire Qdrant collection
+        await self.qdrant_memory.clear_all_data()
+
+        logger.info(f"Cleared ALL data for user '{self.user_id}'")
+
+    async def get_conversation_stats(
+        self,
+        conversation_id: str
+    ) -> Dict[str, Any]:
+        """
+        Get conversation statistics across all tiers
+
+        Args:
+            conversation_id: Unique conversation identifier
+
+        Returns:
+            Dictionary with stats from buffer and Qdrant
+        """
+        # Get buffer stats
+        buffer = await self.buffer_memory.get_buffer(conversation_id)
+        buffer_stats = {
+            "buffer_turns": buffer.metadata.turn_count if buffer else 0,
+            "buffer_tokens": buffer.metadata.total_tokens if buffer else 0
+        }
+
+        # Get Qdrant stats
+        qdrant_stats = await self.qdrant_memory.get_conversation_stats(conversation_id)
+
+        # Combine
+        return {
+            "user_id": self.user_id,
+            "conversation_id": conversation_id,
+            **buffer_stats,
+            "qdrant_turns": qdrant_stats["total_turns"],
+            "qdrant_tokens": qdrant_stats["total_tokens"],
+            "exists_in_buffer": buffer is not None,
+            "exists_in_qdrant": qdrant_stats["exists"]
+        }
+
+    async def list_conversations(self) -> List[str]:
+        """
+        List all conversation IDs for this user
+
+        Returns:
+            List of conversation IDs
+        """
+        return await self.qdrant_memory.list_conversations()
+
+
+# Per-user memory manager cache
+_memory_managers: Dict[str, MemoryManager] = {}
+
+
+def get_memory_manager_for_user(
+    user_id: str,
+    buffer_max_turns: int = 10,
+    auto_consolidate: bool = True
+) -> MemoryManager:
+    """
+    Get or create memory manager instance for a specific user
+
+    Args:
+        user_id: Sanitized user ID
+        buffer_max_turns: Max turns to keep in RAM buffer
+        auto_consolidate: Automatically consolidate when buffer threshold reached
+
+    Returns:
+        MemoryManager instance for the user
+    """
+    if user_id not in _memory_managers:
+        _memory_managers[user_id] = MemoryManager(
+            user_id=user_id,
+            buffer_max_turns=buffer_max_turns,
+            auto_consolidate=auto_consolidate
+        )
+        logger.info(f"Created new MemoryManager for user '{user_id}'")
+
+    return _memory_managers[user_id]
+
+
+def clear_memory_manager_cache(user_id: Optional[str] = None) -> None:
+    """
+    Clear memory manager cache
+
+    Args:
+        user_id: Optional user ID to clear (None = clear all)
+    """
+    global _memory_managers
+
+    if user_id:
+        if user_id in _memory_managers:
+            del _memory_managers[user_id]
+            logger.info(f"Cleared MemoryManager cache for user '{user_id}'")
+    else:
+        _memory_managers.clear()
+        logger.info("Cleared all MemoryManager caches")

services/core-ai/src/memory/qdrant_memory.py

@@ -0,0 +1,465 @@
+"""
+Unified Tier 2/3: Qdrant-based conversation memory with collection-per-user
+
+Multi-tenant architecture:
+- Each user gets their own Qdrant collection: core_ai_user_{user_id}
+- Collections created on-demand
+- Complete data isolation between users
+- Easy GDPR compliance (delete entire collection)
+
+Dual-mode operation:
+- Tier 2: Historical retrieval (filter by conversation_id, time-based)
+- Tier 3: Semantic search (vector similarity across user's conversations)
+"""
+import logging
+import uuid
+import re
+from typing import List, Optional, Dict, Any
+from datetime import datetime
+from qdrant_client import QdrantClient
+from qdrant_client.models import (
+    Distance,
+    VectorParams,
+    PointStruct,
+    Filter,
+    FieldCondition,
+    MatchValue,
+)
+
+from .base import BaseMemory
+from .schemas import ConversationTurn, MessageRole
+from src.config import get_settings
+from src.models.embeddings_ollama import get_embedding_client
+
+logger = logging.getLogger(__name__)
+settings = get_settings()
+
+
+class QdrantConversationMemory(BaseMemory):
+    """
+    Multi-tenant conversation memory using Qdrant with collection-per-user.
+
+    Each user gets a dedicated collection for complete data isolation.
+    Stores all conversation turns with vectors for semantic search.
+    """
+
+    def __init__(
+        self,
+        user_id: str,
+        collection_prefix: Optional[str] = None,
+        qdrant_url: Optional[str] = None
+    ):
+        """
+        Initialize Qdrant memory for a specific user
+
+        Args:
+            user_id: Sanitized user ID (email format: username_at_domain_com)
+            collection_prefix: Collection name prefix (default: core_ai_user)
+            qdrant_url: Qdrant connection URL (default from settings)
+        """
+        self.user_id = user_id
+        self.collection_prefix = collection_prefix or "core_ai_user"
+        self.collection_name = self._get_collection_name(user_id)
+
+        # Parse Qdrant URL (format: http://qdrant:6333)
+        qdrant_url = qdrant_url or getattr(settings, 'qdrant_url', 'http://qdrant:6333')
+        self.qdrant_url = qdrant_url
+
+        # Initialize clients
+        self.client = QdrantClient(url=self.qdrant_url)
+        self.embedding_client = get_embedding_client()
+
+        logger.info(
+            f"Initialized QdrantConversationMemory for user '{user_id}': "
+            f"{self.qdrant_url}/{self.collection_name}"
+        )
+
+        # Ensure user's collection exists
+        self._ensure_collection()
+
+    def _get_collection_name(self, user_id: str) -> str:
+        """
+        Generate collection name for user
+
+        Args:
+            user_id: Sanitized user ID
+
+        Returns:
+            Collection name: {prefix}_{user_id}
+        """
+        # Sanitize user_id for collection name (should already be sanitized, but double-check)
+        sanitized = re.sub(r'[^a-z0-9_]', '_', user_id.lower())
+        return f"{self.collection_prefix}_{sanitized}"
+
+    def _ensure_collection(self) -> None:
+        """Create user's collection if it doesn't exist"""
+        try:
+            collections = self.client.get_collections().collections
+            collection_names = [c.name for c in collections]
+
+            if self.collection_name not in collection_names:
+                logger.info(f"Creating new collection for user '{self.user_id}': {self.collection_name}")
+
+                # Get embedding dimension from settings or default to 768 (nomic-embed-text)
+                embedding_dim = getattr(settings, 'embedding_dimension', 768)
+
+                self.client.create_collection(
+                    collection_name=self.collection_name,
+                    vectors_config=VectorParams(
+                        size=embedding_dim,
+                        distance=Distance.COSINE
+                    )
+                )
+                logger.info(f"✓ Collection created: {self.collection_name}")
+            else:
+                logger.info(f"✓ Collection exists: {self.collection_name}")
+
+        except Exception as e:
+            logger.error(f"Error ensuring collection for user '{self.user_id}': {e}")
+            raise
+
+    async def add_turn(self, conversation_id: str, turn: ConversationTurn) -> None:
+        """
+        Add a conversation turn with its embedding
+
+        Args:
+            conversation_id: Unique conversation identifier
+            turn: The conversation turn to store
+        """
+        # Generate embedding
+        embedding = await self.embedding_client.embed_text(turn.content)
+
+        # Create point ID: deterministic UUID from conversation_id + turn_number
+        point_id_str = f"{conversation_id}_{turn.turn_number}"
+        point_id = str(uuid.uuid5(uuid.NAMESPACE_DNS, point_id_str))
+
+        # Build payload (no user_id needed - collection is already user-specific)
+        payload = {
+            "conversation_id": conversation_id,
+            "turn_number": turn.turn_number,
+            "role": turn.role.value if isinstance(turn.role, MessageRole) else turn.role,
+            "content": turn.content,
+            "timestamp": turn.timestamp.isoformat(),
+            "metadata": turn.metadata,
+        }
+
+        # Add token info if available
+        if turn.tokens:
+            payload["tokens_prompt"] = turn.tokens.prompt
+            payload["tokens_completion"] = turn.tokens.completion
+            payload["tokens_total"] = turn.tokens.total
+
+        # Upsert to user's Qdrant collection
+        try:
+            self.client.upsert(
+                collection_name=self.collection_name,
+                points=[
+                    PointStruct(
+                        id=point_id,
+                        vector=embedding,
+                        payload=payload
+                    )
+                ]
+            )
+            logger.debug(
+                f"Stored turn {turn.turn_number} for conversation {conversation_id} "
+                f"(user: {self.user_id})"
+            )
+
+        except Exception as e:
+            logger.error(f"Error storing turn in Qdrant for user '{self.user_id}': {e}")
+            raise
+
+    async def get_turns(
+        self,
+        conversation_id: str,
+        limit: Optional[int] = None,
+        offset: int = 0
+    ) -> List[ConversationTurn]:
+        """
+        Retrieve turns for a conversation (Tier 2 mode: chronological)
+
+        Args:
+            conversation_id: Unique conversation identifier
+            limit: Maximum number of turns to retrieve
+            offset: Number of turns to skip
+
+        Returns:
+            List of conversation turns
+        """
+        try:
+            # Scroll through all points for this conversation
+            points, _ = self.client.scroll(
+                collection_name=self.collection_name,
+                scroll_filter=Filter(
+                    must=[
+                        FieldCondition(
+                            key="conversation_id",
+                            match=MatchValue(value=conversation_id)
+                        )
+                    ]
+                ),
+                limit=limit or 100,
+                offset=offset,
+                with_payload=True,
+                with_vectors=False
+            )
+
+            # Convert to ConversationTurn objects
+            turns = []
+            for point in points:
+                payload = point.payload
+                turn = ConversationTurn(
+                    role=MessageRole(payload["role"]),
+                    content=payload["content"],
+                    timestamp=datetime.fromisoformat(payload["timestamp"]),
+                    turn_number=payload["turn_number"],
+                    user_id=self.user_id,  # User from collection context
+                    metadata=payload.get("metadata", {})
+                )
+                turns.append(turn)
+
+            # Sort by turn_number
+            turns.sort(key=lambda t: t.turn_number)
+
+            return turns
+
+        except Exception as e:
+            logger.error(f"Error retrieving turns from Qdrant for user '{self.user_id}': {e}")
+            return []
+
+    async def similarity_search(
+        self,
+        query: str,
+        conversation_id: Optional[str] = None,
+        limit: int = 5
+    ) -> List[Dict[str, Any]]:
+        """
+        Semantic search for relevant turns (Tier 3 mode: semantic)
+
+        Args:
+            query: Search query text
+            conversation_id: Optional filter to specific conversation
+            limit: Maximum number of results
+
+        Returns:
+            List of matching turns with scores
+        """
+        try:
+            # Generate query embedding
+            query_embedding = await self.embedding_client.embed_text(query)
+
+            # Build filter if conversation_id specified
+            search_filter = None
+            if conversation_id:
+                search_filter = Filter(
+                    must=[
+                        FieldCondition(
+                            key="conversation_id",
+                            match=MatchValue(value=conversation_id)
+                        )
+                    ]
+                )
+
+            # Search in user's Qdrant collection
+            results = self.client.search(
+                collection_name=self.collection_name,
+                query_vector=query_embedding,
+                query_filter=search_filter,
+                limit=limit,
+                with_payload=True
+            )
+
+            # Convert results
+            matches = []
+            for result in results:
+                payload = result.payload
+                match = {
+                    "conversation_id": payload["conversation_id"],
+                    "turn_number": payload["turn_number"],
+                    "role": payload["role"],
+                    "content": payload["content"],
+                    "timestamp": payload["timestamp"],
+                    "score": result.score,
+                }
+                matches.append(match)
+
+            logger.debug(
+                f"Semantic search found {len(matches)} matches for user '{self.user_id}' "
+                f"query: {query[:50]}..."
+            )
+
+            return matches
+
+        except Exception as e:
+            logger.error(f"Error in semantic search for user '{self.user_id}': {e}")
+            return []
+
+    async def clear_conversation(self, conversation_id: str) -> None:
+        """
+        Clear all turns for a conversation
+
+        Args:
+            conversation_id: Unique conversation identifier
+        """
+        try:
+            # Delete all points with this conversation_id
+            self.client.delete(
+                collection_name=self.collection_name,
+                points_selector=Filter(
+                    must=[
+                        FieldCondition(
+                            key="conversation_id",
+                            match=MatchValue(value=conversation_id)
+                        )
+                    ]
+                )
+            )
+            logger.info(
+                f"Cleared conversation {conversation_id} for user '{self.user_id}' from Qdrant"
+            )
+
+        except Exception as e:
+            logger.error(f"Error clearing conversation for user '{self.user_id}': {e}")
+            raise
+
+    async def clear_all_data(self) -> None:
+        """
+        Clear ALL data for this user (GDPR compliance)
+
+        Deletes the entire collection for this user.
+        """
+        try:
+            self.client.delete_collection(self.collection_name)
+            logger.info(f"Deleted all data for user '{self.user_id}' (collection: {self.collection_name})")
+
+        except Exception as e:
+            logger.error(f"Error deleting user data for '{self.user_id}': {e}")
+            raise
+
+    async def conversation_exists(self, conversation_id: str) -> bool:
+        """
+        Check if a conversation exists
+
+        Args:
+            conversation_id: Unique conversation identifier
+
+        Returns:
+            True if conversation has any turns
+        """
+        try:
+            points, _ = self.client.scroll(
+                collection_name=self.collection_name,
+                scroll_filter=Filter(
+                    must=[
+                        FieldCondition(
+                            key="conversation_id",
+                            match=MatchValue(value=conversation_id)
+                        )
+                    ]
+                ),
+                limit=1,
+                with_payload=False,
+                with_vectors=False
+            )
+            return len(points) > 0
+
+        except Exception as e:
+            logger.error(f"Error checking conversation existence for user '{self.user_id}': {e}")
+            return False
+
+    async def get_conversation_stats(self, conversation_id: str) -> Dict[str, Any]:
+        """
+        Get statistics about a conversation
+
+        Args:
+            conversation_id: Unique conversation identifier
+
+        Returns:
+            Dictionary with stats
+        """
+        try:
+            points, _ = self.client.scroll(
+                collection_name=self.collection_name,
+                scroll_filter=Filter(
+                    must=[
+                        FieldCondition(
+                            key="conversation_id",
+                            match=MatchValue(value=conversation_id)
+                        )
+                    ]
+                ),
+                limit=1000,  # Get all points
+                with_payload=True,
+                with_vectors=False
+            )
+
+            total_turns = len(points)
+            total_tokens = sum(
+                point.payload.get("tokens_total", 0) for point in points
+            )
+
+            return {
+                "user_id": self.user_id,
+                "conversation_id": conversation_id,
+                "total_turns": total_turns,
+                "total_tokens": total_tokens,
+                "exists": total_turns > 0
+            }
+
+        except Exception as e:
+            logger.error(f"Error getting conversation stats for user '{self.user_id}': {e}")
+            return {
+                "user_id": self.user_id,
+                "conversation_id": conversation_id,
+                "total_turns": 0,
+                "total_tokens": 0,
+                "exists": False
+            }
+
+    async def list_conversations(self) -> List[str]:
+        """
+        List all conversation IDs for this user
+
+        Returns:
+            List of conversation IDs
+        """
+        try:
+            # Scroll through all points to collect unique conversation_ids
+            conversation_ids = set()
+            offset = None
+
+            while True:
+                points, next_offset = self.client.scroll(
+                    collection_name=self.collection_name,
+                    limit=100,
+                    offset=offset,
+                    with_payload=True,
+                    with_vectors=False
+                )
+
+                for point in points:
+                    conversation_ids.add(point.payload["conversation_id"])
+
+                if next_offset is None:
+                    break
+                offset = next_offset
+
+            return sorted(list(conversation_ids))
+
+        except Exception as e:
+            logger.error(f"Error listing conversations for user '{self.user_id}': {e}")
+            return []
+
+
+def get_qdrant_memory_for_user(user_id: str) -> QdrantConversationMemory:
+    """
+    Get Qdrant memory instance for a specific user
+
+    Args:
+        user_id: Sanitized user ID (email format: username_at_domain_com)
+
+    Returns:
+        QdrantConversationMemory instance for the user
+    """
+    return QdrantConversationMemory(user_id=user_id)

services/core-ai/src/memory/schemas.py

@@ -0,0 +1,110 @@
+"""
+Pydantic schemas for memory system
+"""
+from pydantic import BaseModel, Field
+from typing import List, Optional, Dict, Any
+from datetime import datetime
+from enum import Enum
+
+
+class MessageRole(str, Enum):
+    """Message role types"""
+    SYSTEM = "system"
+    USER = "user"
+    ASSISTANT = "assistant"
+
+
+class TokenUsage(BaseModel):
+    """Token usage information"""
+    prompt: int = 0
+    completion: int = 0
+    total: int = 0
+
+
+class ConversationTurn(BaseModel):
+    """A single turn in a conversation"""
+    role: MessageRole
+    content: str
+    timestamp: datetime = Field(default_factory=datetime.utcnow)
+    turn_number: int
+    user_id: str = "llmdefault_at_schweitz_net"  # Multi-tenancy: user who owns this turn
+    tokens: Optional[TokenUsage] = None
+    metadata: Dict[str, Any] = Field(default_factory=dict)
+
+
+class ConversationMetadata(BaseModel):
+    """Metadata about a conversation"""
+    conversation_id: str
+    user_id: str = "llmdefault_at_schweitz_net"  # Multi-tenancy: user who owns this conversation
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    last_updated: datetime = Field(default_factory=datetime.utcnow)
+    turn_count: int = 0
+    total_tokens: int = 0
+    status: str = "active"  # active, archived, deleted
+
+
+class ConversationBuffer(BaseModel):
+    """In-memory conversation buffer (Tier 1)"""
+    conversation_id: str
+    turns: List[ConversationTurn] = Field(default_factory=list)
+    metadata: ConversationMetadata
+
+
+class ConversationSummary(BaseModel):
+    """Summarized conversation segment (Tier 2)"""
+    conversation_id: str
+    summary_text: str
+    turn_range_start: int
+    turn_range_end: int
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    token_count: int = 0
+
+
+class MemoryQuery(BaseModel):
+    """Query for memory retrieval"""
+    conversation_id: str
+    query: Optional[str] = None
+    limit: int = Field(default=10, ge=1, le=100)
+    include_tier1: bool = True
+    include_tier2: bool = True
+    include_tier3: bool = True
+
+
+class MemoryResult(BaseModel):
+    """Result from memory retrieval"""
+    conversation_id: str
+    turns: List[ConversationTurn] = Field(default_factory=list)
+    summaries: List[ConversationSummary] = Field(default_factory=list)
+    source_tiers: List[int] = Field(default_factory=list)  # Which tiers contributed
+    total_results: int = 0
+
+
+# API Request/Response Models
+
+class ConversationListResponse(BaseModel):
+    """Response for listing conversations"""
+    conversations: List[ConversationMetadata]
+    total: int
+    page: int = 1
+    page_size: int = 50
+
+
+class ConversationDetailResponse(BaseModel):
+    """Response for conversation details"""
+    metadata: ConversationMetadata
+    recent_turns: List[ConversationTurn]
+    turn_count: int
+
+
+class ConversationSearchRequest(BaseModel):
+    """Request for semantic search in conversation"""
+    query: str
+    limit: int = Field(default=5, ge=1, le=50)
+
+
+class ConversationSearchResponse(BaseModel):
+    """Response for semantic search"""
+    conversation_id: str
+    results: List[ConversationTurn]
+    scores: List[float] = Field(default_factory=list)
+    total_results: int

services/core-ai/src/memory/tier1_buffer.py

@@ -0,0 +1,239 @@
+"""
+Tier 1: ConversationBufferMemory (In-Memory Working Memory)
+
+Fast in-memory storage for recent conversation turns.
+- Stores last N turns in RAM
+- < 1ms access time
+- Ephemeral (lost on restart)
+- Automatic pruning when limit reached
+"""
+import logging
+from typing import Dict, List, Optional
+from datetime import datetime
+from collections import OrderedDict
+
+from .base import Tier1Memory
+from .schemas import (
+    ConversationTurn,
+    ConversationBuffer,
+    ConversationMetadata,
+    MessageRole,
+    TokenUsage
+)
+
+logger = logging.getLogger(__name__)
+
+
+class ConversationBufferMemory(Tier1Memory):
+    """
+    In-memory buffer for recent conversation turns.
+
+    Stores the last N turns of each conversation in RAM for fast access.
+    Automatically prunes old turns when limit is reached.
+    """
+
+    def __init__(self, max_turns: int = 10):
+        """
+        Initialize buffer memory
+
+        Args:
+            max_turns: Maximum number of turns to keep per conversation
+        """
+        self.max_turns = max_turns
+        # Use OrderedDict to maintain insertion order
+        self._buffers: Dict[str, ConversationBuffer] = OrderedDict()
+        logger.info(f"Initialized ConversationBufferMemory with max_turns={max_turns}")
+
+    async def add_turn(self, conversation_id: str, turn: ConversationTurn) -> None:
+        """
+        Add a new turn to the buffer
+
+        Args:
+            conversation_id: Unique conversation identifier
+            turn: The conversation turn to store
+        """
+        # Get or create buffer
+        buffer = await self.get_buffer(conversation_id)
+        if buffer is None:
+            buffer = ConversationBuffer(
+                conversation_id=conversation_id,
+                turns=[],
+                metadata=ConversationMetadata(
+                    conversation_id=conversation_id
+                )
+            )
+            self._buffers[conversation_id] = buffer
+
+        # Add turn
+        buffer.turns.append(turn)
+
+        # Update metadata
+        buffer.metadata.turn_count = len(buffer.turns)
+        buffer.metadata.last_updated = datetime.utcnow()
+
+        if turn.tokens:
+            buffer.metadata.total_tokens += turn.tokens.total
+
+        # Auto-prune if exceeds max turns
+        if len(buffer.turns) > self.max_turns:
+            await self.prune(conversation_id, keep_last=self.max_turns)
+
+        logger.debug(
+            f"Added turn {turn.turn_number} to conversation {conversation_id}. "
+            f"Buffer size: {len(buffer.turns)}"
+        )
+
+    async def get_turns(
+        self,
+        conversation_id: str,
+        limit: Optional[int] = None,
+        offset: int = 0
+    ) -> List[ConversationTurn]:
+        """
+        Retrieve turns from the buffer
+
+        Args:
+            conversation_id: Unique conversation identifier
+            limit: Maximum number of turns to retrieve
+            offset: Number of turns to skip
+
+        Returns:
+            List of conversation turns
+        """
+        buffer = await self.get_buffer(conversation_id)
+        if buffer is None:
+            return []
+
+        turns = buffer.turns[offset:]
+        if limit:
+            turns = turns[:limit]
+
+        return turns
+
+    async def get_recent_turns(
+        self,
+        conversation_id: str,
+        limit: int = 10
+    ) -> List[ConversationTurn]:
+        """
+        Get the most recent N turns
+
+        Args:
+            conversation_id: Unique conversation identifier
+            limit: Number of recent turns to retrieve
+
+        Returns:
+            List of recent turns (most recent last)
+        """
+        buffer = await self.get_buffer(conversation_id)
+        if buffer is None:
+            return []
+
+        return buffer.turns[-limit:] if len(buffer.turns) > limit else buffer.turns
+
+    async def get_buffer(self, conversation_id: str) -> Optional[ConversationBuffer]:
+        """
+        Get the full conversation buffer
+
+        Args:
+            conversation_id: Unique conversation identifier
+
+        Returns:
+            ConversationBuffer or None if not found
+        """
+        return self._buffers.get(conversation_id)
+
+    async def clear_conversation(self, conversation_id: str) -> None:
+        """
+        Clear all turns for a conversation
+
+        Args:
+            conversation_id: Unique conversation identifier
+        """
+        if conversation_id in self._buffers:
+            del self._buffers[conversation_id]
+            logger.info(f"Cleared buffer for conversation {conversation_id}")
+
+    async def conversation_exists(self, conversation_id: str) -> bool:
+        """
+        Check if a conversation exists in the buffer
+
+        Args:
+            conversation_id: Unique conversation identifier
+
+        Returns:
+            True if conversation exists
+        """
+        return conversation_id in self._buffers
+
+    async def prune(self, conversation_id: str, keep_last: int = 5) -> None:
+        """
+        Prune old turns, keeping only the most recent ones
+
+        Args:
+            conversation_id: Unique conversation identifier
+            keep_last: Number of recent turns to keep
+        """
+        buffer = await self.get_buffer(conversation_id)
+        if buffer is None:
+            return
+
+        if len(buffer.turns) > keep_last:
+            removed_count = len(buffer.turns) - keep_last
+            buffer.turns = buffer.turns[-keep_last:]
+            buffer.metadata.turn_count = len(buffer.turns)
+
+            logger.debug(
+                f"Pruned {removed_count} turns from conversation {conversation_id}. "
+                f"Kept last {keep_last} turns."
+            )
+
+    async def get_all_conversation_ids(self) -> List[str]:
+        """
+        Get list of all conversation IDs in memory
+
+        Returns:
+            List of conversation IDs
+        """
+        return list(self._buffers.keys())
+
+    async def get_buffer_stats(self) -> dict:
+        """
+        Get statistics about buffer memory usage
+
+        Returns:
+            Dictionary with stats
+        """
+        total_conversations = len(self._buffers)
+        total_turns = sum(len(buf.turns) for buf in self._buffers.values())
+        total_tokens = sum(buf.metadata.total_tokens for buf in self._buffers.values())
+
+        return {
+            "total_conversations": total_conversations,
+            "total_turns": total_turns,
+            "total_tokens": total_tokens,
+            "max_turns_per_conversation": self.max_turns,
+            "avg_turns_per_conversation": (
+                total_turns / total_conversations if total_conversations > 0 else 0
+            )
+        }
+
+
+# Global instance
+_buffer_memory: Optional[ConversationBufferMemory] = None
+
+
+def get_buffer_memory(max_turns: int = 10) -> ConversationBufferMemory:
+    """
+    Get or create the global buffer memory instance
+
+    Args:
+        max_turns: Maximum turns per conversation
+
+    Returns:
+        ConversationBufferMemory instance
+    """
+    global _buffer_memory
+    if _buffer_memory is None:
+        _buffer_memory = ConversationBufferMemory(max_turns=max_turns)
+    return _buffer_memory

services/core-ai/src/models/__init__.py

@@ -0,0 +1,15 @@
+"""Models for core-ai service"""
+
+from .embeddings_ollama import (
+    OllamaEmbeddingClient,
+    get_embedding_client,
+    embed_text_async,
+    embed_batch_async
+)
+
+__all__ = [
+    "OllamaEmbeddingClient",
+    "get_embedding_client",
+    "embed_text_async",
+    "embed_batch_async",
+]

services/core-ai/src/models/embeddings_ollama.py

@@ -0,0 +1,136 @@
+"""
+Ollama-based embedding client for text vectorization
+
+Uses Ollama's embedding API instead of local sentence-transformers.
+This eliminates the need for PyTorch and heavy ML dependencies.
+"""
+import logging
+import httpx
+from typing import List, Optional
+from src.config import get_settings
+
+logger = logging.getLogger(__name__)
+settings = get_settings()
+
+
+class OllamaEmbeddingClient:
+    """Client for generating text embeddings using Ollama"""
+
+    def __init__(
+        self,
+        model_name: Optional[str] = None,
+        base_url: Optional[str] = None,
+        timeout: int = 30
+    ):
+        """
+        Initialize Ollama embedding client
+
+        Args:
+            model_name: Embedding model name (default: nomic-embed-text)
+            base_url: Ollama base URL (default from settings)
+            timeout: Request timeout in seconds
+        """
+        self.model_name = model_name or settings.embedding_model
+        self.base_url = (base_url or settings.ollama_base_url).rstrip("/")
+        self.timeout = timeout
+        self.dimension = settings.embedding_dimension
+
+        logger.info(f"Initializing OllamaEmbeddingClient with model: {self.model_name}")
+        logger.info(f"Ollama URL: {self.base_url}")
+
+    async def embed_text(self, text: str) -> List[float]:
+        """
+        Generate embedding for a single text using Ollama
+
+        Args:
+            text: Input text to embed
+
+        Returns:
+            List of floats representing the embedding vector
+        """
+        try:
+            async with httpx.AsyncClient(timeout=self.timeout) as client:
+                response = await client.post(
+                    f"{self.base_url}/api/embeddings",
+                    json={
+                        "model": self.model_name,
+                        "prompt": text
+                    }
+                )
+                response.raise_for_status()
+                result = response.json()
+                return result["embedding"]
+
+        except Exception as e:
+            logger.error(f"Error generating embedding via Ollama: {e}")
+            raise
+
+    async def embed_batch(self, texts: List[str]) -> List[List[float]]:
+        """
+        Generate embeddings for multiple texts
+
+        Args:
+            texts: List of input texts
+
+        Returns:
+            List of embedding vectors
+        """
+        embeddings = []
+        for text in texts:
+            embedding = await self.embed_text(text)
+            embeddings.append(embedding)
+        return embeddings
+
+    def get_dimension(self) -> int:
+        """
+        Get embedding dimension
+
+        Returns:
+            Embedding vector dimension
+        """
+        return self.dimension
+
+
+# Global instance
+_embedding_client: Optional[OllamaEmbeddingClient] = None
+
+
+def get_embedding_client() -> OllamaEmbeddingClient:
+    """
+    Get or create global Ollama embedding client instance
+
+    Returns:
+        OllamaEmbeddingClient instance
+    """
+    global _embedding_client
+    if _embedding_client is None:
+        _embedding_client = OllamaEmbeddingClient()
+    return _embedding_client
+
+
+async def embed_text_async(text: str) -> List[float]:
+    """
+    Async wrapper for embedding text
+
+    Args:
+        text: Input text
+
+    Returns:
+        Embedding vector
+    """
+    client = get_embedding_client()
+    return await client.embed_text(text)
+
+
+async def embed_batch_async(texts: List[str]) -> List[List[float]]:
+    """
+    Async wrapper for batch embedding
+
+    Args:
+        texts: List of input texts
+
+    Returns:
+        List of embedding vectors
+    """
+    client = get_embedding_client()
+    return await client.embed_batch(texts)

services/core-ai/src/utils.py

@@ -0,0 +1,102 @@
+"""
+Utility functions for core-ai service.
+"""
+import re
+from typing import Optional
+
+
+# Default user for requests without user_id
+DEFAULT_USER_ID = "llmdefault_at_schweitz.net"
+
+
+def sanitize_email_to_user_id(email: Optional[str] = None) -> str:
+    """
+    Convert email address to standardized user_id format.
+
+    Format: username_at_domain_com (lowercase, @ → _at_)
+
+    Examples:
+        john@example.com → john_at_example_com
+        Alice.Smith@Company.ORG → alice_smith_at_company_org
+        None → llmdefault_at_schweitz.net (default)
+
+    Args:
+        email: Email address to convert (None uses default user)
+
+    Returns:
+        Sanitized user_id string safe for Qdrant collection names
+    """
+    if not email:
+        return DEFAULT_USER_ID
+
+    # Convert to lowercase
+    email = email.lower().strip()
+
+    # Validate email format (basic check)
+    if '@' not in email:
+        # Invalid email, return default
+        return DEFAULT_USER_ID
+
+    # Replace @ with _at_
+    user_id = email.replace('@', '_at_')
+
+    # Replace any non-alphanumeric characters (except underscores) with underscores
+    # This handles dots, hyphens, etc. in email addresses
+    user_id = re.sub(r'[^a-z0-9_]', '_', user_id)
+
+    # Remove any duplicate underscores
+    user_id = re.sub(r'_+', '_', user_id)
+
+    # Remove leading/trailing underscores
+    user_id = user_id.strip('_')
+
+    return user_id
+
+
+def get_collection_name_for_user(user_id: str, prefix: str = "core_ai_user") -> str:
+    """
+    Generate Qdrant collection name for a user.
+
+    Args:
+        user_id: Sanitized user ID (from sanitize_email_to_user_id)
+        prefix: Collection prefix (default: core_ai_user)
+
+    Returns:
+        Full collection name: {prefix}_{user_id}
+
+    Examples:
+        john_at_example_com → core_ai_user_john_at_example_com
+        llmdefault_at_schweitz_net → core_ai_user_llmdefault_at_schweitz_net
+    """
+    return f"{prefix}_{user_id}"
+
+
+def extract_user_id_from_request(data: dict) -> str:
+    """
+    Extract and sanitize user_id from request data.
+
+    Priority:
+    1. data.get("user_id") - if provided, sanitize it
+    2. data.get("user_email") - convert to user_id format
+    3. DEFAULT_USER_ID - fallback to default user
+
+    Args:
+        data: Request JSON data
+
+    Returns:
+        Sanitized user_id string
+    """
+    # Check for explicit user_id
+    if user_id := data.get("user_id"):
+        # If it's already in our format, use it
+        if "_at_" in user_id:
+            return user_id
+        # Otherwise treat it as an email
+        return sanitize_email_to_user_id(user_id)
+
+    # Check for user_email
+    if user_email := data.get("user_email"):
+        return sanitize_email_to_user_id(user_email)
+
+    # Fallback to default
+    return DEFAULT_USER_ID

stacks/core-ai.yml

@@ -14,6 +14,14 @@ services:
       - CORE_API_BASE_URL=http://core-api:8083/v1 # Ensure it can find Core API
       - AGENT_MODEL=mistral-nemo:latest # Better tool calling support
       - SYSTEM_PROMPT_VARIANT=minimal_agent # Match prompts.py definition
+      # Memory system configuration
+      - MEMORY_ENABLED=true
+      - MEMORY_TIER1_SIZE=10  # Keep last 10 turns in RAM
+      - QDRANT_URL=http://qdrant:6333  # Connect to existing Qdrant service
+      - QDRANT_COLLECTION_PREFIX=core_ai_user  # Collection naming: core_ai_user_{user_id}
+      - EMBEDDING_MODEL=nomic-embed-text  # Match Open WebUI embedding model
+      - EMBEDDING_DIMENSION=768  # nomic-embed-text dimension
+      - DEFAULT_USER_ID=llmdefault_at_schweitz_net  # Default user until auth integration
     networks:
       - docker-dataplane