feat: implement hierarchical memory system

backend/app/memory/__init__.py

@@ -0,0 +1,55 @@
+"""Memory module for ScrapeRL agent memory management.
+
+This module provides a multi-layered memory system for RL agents:
+
+- **ShortTermMemory**: Episode-scoped dictionary storage that auto-clears
+- **WorkingMemory**: LRU-based reasoning/scratch space with limited capacity
+- **LongTermMemory**: Persistent vector storage with ChromaDB for semantic search
+- **SharedMemory**: Thread-safe pub/sub and state sharing for multi-agent coordination
+- **MemoryManager**: Unified interface to all memory layers
+
+Example:
+    >>> from app.config import get_settings
+    >>> from app.memory import MemoryManager, MemoryType
+    >>>
+    >>> settings = get_settings()
+    >>> memory = MemoryManager(settings)
+    >>> await memory.initialize()
+    >>>
+    >>> # Store in short-term memory
+    >>> await memory.store("key", "value", MemoryType.SHORT_TERM)
+    >>>
+    >>> # Semantic search in long-term memory
+    >>> results = await memory.search("query", MemoryType.LONG_TERM)
+    >>>
+    >>> # Cleanup
+    >>> await memory.shutdown()
+"""
+
+from app.memory.long_term import Document, LongTermMemory, SearchResult
+from app.memory.manager import MemoryManager, MemoryStats, MemoryType
+from app.memory.shared import Channel, Message, SharedMemory, Subscription
+from app.memory.short_term import MemoryEntry, ShortTermMemory
+from app.memory.working import WorkingMemory, WorkingMemoryItem
+
+__all__ = [
+    # Manager
+    "MemoryManager",
+    "MemoryStats",
+    "MemoryType",
+    # Short-term
+    "ShortTermMemory",
+    "MemoryEntry",
+    # Working
+    "WorkingMemory",
+    "WorkingMemoryItem",
+    # Long-term
+    "LongTermMemory",
+    "Document",
+    "SearchResult",
+    # Shared
+    "SharedMemory",
+    "Channel",
+    "Message",
+    "Subscription",
+]

backend/app/memory/long_term.py

@@ -0,0 +1,473 @@
+"""Long-term memory with persistent vector storage using ChromaDB."""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import logging
+from datetime import datetime
+from typing import Any
+from uuid import uuid4
+
+from pydantic import BaseModel, Field
+
+logger = logging.getLogger(__name__)
+
+
+class Document(BaseModel):
+    """A document stored in long-term memory."""
+
+    id: str = Field(default_factory=lambda: str(uuid4()))
+    content: str
+    embedding: list[float] | None = None
+    metadata: dict[str, Any] = Field(default_factory=dict)
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    updated_at: datetime = Field(default_factory=datetime.utcnow)
+
+    model_config = {"arbitrary_types_allowed": True}
+
+
+class SearchResult(BaseModel):
+    """A search result from long-term memory."""
+
+    document: Document
+    score: float
+    distance: float | None = None
+
+    model_config = {"arbitrary_types_allowed": True}
+
+
+class LongTermMemory:
+    """
+    Long-term persistent memory using ChromaDB for vector storage.
+
+    This memory layer provides semantic search capabilities using embeddings.
+    It persists across episodes and sessions, storing knowledge that should
+    be retained long-term.
+
+    Attributes:
+        collection_name: Name of the ChromaDB collection.
+        persist_directory: Directory for persistent storage.
+        top_k: Default number of results to return from search.
+    """
+
+    def __init__(
+        self,
+        collection_name: str = "scraperl_memory",
+        persist_directory: str = "./data/chroma",
+        top_k: int = 10,
+        embedding_function: Any | None = None,
+    ) -> None:
+        """
+        Initialize long-term memory.
+
+        Args:
+            collection_name: Name of the ChromaDB collection.
+            persist_directory: Directory for persistent storage.
+            top_k: Default number of results to return from search.
+            embedding_function: Optional custom embedding function.
+        """
+        self.collection_name = collection_name
+        self.persist_directory = persist_directory
+        self.top_k = top_k
+        self._embedding_function = embedding_function
+        self._client: Any = None
+        self._collection: Any = None
+        self._initialized = False
+        self._lock = asyncio.Lock()
+
+    async def initialize(self) -> None:
+        """
+        Initialize ChromaDB client and collection.
+
+        This should be called before using other methods.
+        """
+        if self._initialized:
+            return
+
+        async with self._lock:
+            if self._initialized:
+                return
+
+            try:
+                import chromadb
+                from chromadb.config import Settings
+
+                # Create persistent client
+                self._client = chromadb.Client(
+                    Settings(
+                        chroma_db_impl="duckdb+parquet",
+                        persist_directory=self.persist_directory,
+                        anonymized_telemetry=False,
+                    )
+                )
+
+                # Get or create collection
+                self._collection = self._client.get_or_create_collection(
+                    name=self.collection_name,
+                    embedding_function=self._embedding_function,
+                    metadata={"hnsw:space": "cosine"},
+                )
+
+                self._initialized = True
+                logger.info(
+                    f"Initialized long-term memory: collection={self.collection_name}"
+                )
+
+            except ImportError:
+                logger.warning(
+                    "ChromaDB not available. Long-term memory will use in-memory fallback."
+                )
+                self._use_fallback()
+            except Exception as e:
+                logger.warning(
+                    f"Failed to initialize ChromaDB: {e}. Using in-memory fallback."
+                )
+                self._use_fallback()
+
+    def _use_fallback(self) -> None:
+        """Use in-memory fallback when ChromaDB is unavailable."""
+        self._client = None
+        self._collection = None
+        self._fallback_store: dict[str, Document] = {}
+        self._initialized = True
+
+    @property
+    def is_initialized(self) -> bool:
+        """Check if memory is initialized."""
+        return self._initialized
+
+    @property
+    def _using_fallback(self) -> bool:
+        """Check if using in-memory fallback."""
+        return self._collection is None
+
+    def _generate_id(self, content: str) -> str:
+        """Generate a deterministic ID from content."""
+        return hashlib.sha256(content.encode()).hexdigest()[:16]
+
+    async def store(
+        self,
+        content: str,
+        document_id: str | None = None,
+        metadata: dict[str, Any] | None = None,
+        embedding: list[float] | None = None,
+    ) -> Document:
+        """
+        Store a document in long-term memory.
+
+        Args:
+            content: Text content to store.
+            document_id: Optional custom ID. Generated from content if not provided.
+            metadata: Optional metadata dictionary.
+            embedding: Optional pre-computed embedding vector.
+
+        Returns:
+            The stored document.
+        """
+        if not self._initialized:
+            await self.initialize()
+
+        async with self._lock:
+            doc_id = document_id or self._generate_id(content)
+            now = datetime.utcnow()
+
+            document = Document(
+                id=doc_id,
+                content=content,
+                embedding=embedding,
+                metadata=metadata or {},
+                created_at=now,
+                updated_at=now,
+            )
+
+            if self._using_fallback:
+                self._fallback_store[doc_id] = document
+            else:
+                # Store in ChromaDB
+                try:
+                    self._collection.upsert(
+                        ids=[doc_id],
+                        documents=[content],
+                        metadatas=[
+                            {
+                                **document.metadata,
+                                "created_at": now.isoformat(),
+                                "updated_at": now.isoformat(),
+                            }
+                        ],
+                        embeddings=[embedding] if embedding else None,
+                    )
+                except Exception as e:
+                    logger.error(f"Failed to store document: {e}")
+                    raise
+
+            return document
+
+    async def search(
+        self,
+        query: str,
+        top_k: int | None = None,
+        where: dict[str, Any] | None = None,
+        query_embedding: list[float] | None = None,
+    ) -> list[SearchResult]:
+        """
+        Search for similar documents using semantic search.
+
+        Args:
+            query: Search query text.
+            top_k: Number of results to return. Uses default if not specified.
+            where: Optional metadata filter.
+            query_embedding: Optional pre-computed query embedding.
+
+        Returns:
+            List of search results with scores.
+        """
+        if not self._initialized:
+            await self.initialize()
+
+        k = top_k or self.top_k
+
+        async with self._lock:
+            if self._using_fallback:
+                # Simple substring matching for fallback
+                results = []
+                query_lower = query.lower()
+                for doc in self._fallback_store.values():
+                    if query_lower in doc.content.lower():
+                        results.append(
+                            SearchResult(document=doc, score=1.0, distance=0.0)
+                        )
+                return results[:k]
+
+            try:
+                # Query ChromaDB
+                query_params: dict[str, Any] = {
+                    "n_results": k,
+                }
+
+                if query_embedding:
+                    query_params["query_embeddings"] = [query_embedding]
+                else:
+                    query_params["query_texts"] = [query]
+
+                if where:
+                    query_params["where"] = where
+
+                results = self._collection.query(**query_params)
+
+                # Parse results
+                search_results = []
+                if results and results.get("ids"):
+                    for i, doc_id in enumerate(results["ids"][0]):
+                        content = (
+                            results["documents"][0][i]
+                            if results.get("documents")
+                            else ""
+                        )
+                        metadata = (
+                            results["metadatas"][0][i]
+                            if results.get("metadatas")
+                            else {}
+                        )
+                        distance = (
+                            results["distances"][0][i]
+                            if results.get("distances")
+                            else None
+                        )
+
+                        doc = Document(
+                            id=doc_id,
+                            content=content,
+                            metadata=metadata,
+                        )
+
+                        # Convert distance to score (cosine similarity)
+                        score = 1 - distance if distance is not None else 1.0
+
+                        search_results.append(
+                            SearchResult(
+                                document=doc,
+                                score=score,
+                                distance=distance,
+                            )
+                        )
+
+                return search_results
+
+            except Exception as e:
+                logger.error(f"Search failed: {e}")
+                return []
+
+    async def get(self, document_id: str) -> Document | None:
+        """
+        Retrieve a document by ID.
+
+        Args:
+            document_id: The document ID to retrieve.
+
+        Returns:
+            The document or None if not found.
+        """
+        if not self._initialized:
+            await self.initialize()
+
+        async with self._lock:
+            if self._using_fallback:
+                return self._fallback_store.get(document_id)
+
+            try:
+                result = self._collection.get(ids=[document_id])
+                if result and result["ids"]:
+                    return Document(
+                        id=result["ids"][0],
+                        content=result["documents"][0] if result.get("documents") else "",
+                        metadata=result["metadatas"][0] if result.get("metadatas") else {},
+                    )
+                return None
+            except Exception as e:
+                logger.error(f"Failed to get document: {e}")
+                return None
+
+    async def delete(self, document_id: str) -> bool:
+        """
+        Delete a document from long-term memory.
+
+        Args:
+            document_id: The document ID to delete.
+
+        Returns:
+            True if document was deleted, False otherwise.
+        """
+        if not self._initialized:
+            await self.initialize()
+
+        async with self._lock:
+            if self._using_fallback:
+                if document_id in self._fallback_store:
+                    del self._fallback_store[document_id]
+                    return True
+                return False
+
+            try:
+                self._collection.delete(ids=[document_id])
+                return True
+            except Exception as e:
+                logger.error(f"Failed to delete document: {e}")
+                return False
+
+    async def delete_where(self, where: dict[str, Any]) -> int:
+        """
+        Delete documents matching a metadata filter.
+
+        Args:
+            where: Metadata filter for documents to delete.
+
+        Returns:
+            Number of documents deleted.
+        """
+        if not self._initialized:
+            await self.initialize()
+
+        async with self._lock:
+            if self._using_fallback:
+                to_delete = []
+                for doc_id, doc in self._fallback_store.items():
+                    if all(doc.metadata.get(k) == v for k, v in where.items()):
+                        to_delete.append(doc_id)
+                for doc_id in to_delete:
+                    del self._fallback_store[doc_id]
+                return len(to_delete)
+
+            try:
+                # Get matching IDs first
+                result = self._collection.get(where=where)
+                if result and result["ids"]:
+                    self._collection.delete(ids=result["ids"])
+                    return len(result["ids"])
+                return 0
+            except Exception as e:
+                logger.error(f"Failed to delete documents: {e}")
+                return 0
+
+    async def count(self) -> int:
+        """
+        Get the total number of documents stored.
+
+        Returns:
+            Document count.
+        """
+        if not self._initialized:
+            await self.initialize()
+
+        async with self._lock:
+            if self._using_fallback:
+                return len(self._fallback_store)
+
+            try:
+                return self._collection.count()
+            except Exception as e:
+                logger.error(f"Failed to count documents: {e}")
+                return 0
+
+    async def clear(self) -> int:
+        """
+        Clear all documents from memory.
+
+        Returns:
+            Number of documents that were cleared.
+        """
+        if not self._initialized:
+            await self.initialize()
+
+        async with self._lock:
+            if self._using_fallback:
+                count = len(self._fallback_store)
+                self._fallback_store.clear()
+                return count
+
+            try:
+                count = self._collection.count()
+                # Delete and recreate collection
+                self._client.delete_collection(self.collection_name)
+                self._collection = self._client.create_collection(
+                    name=self.collection_name,
+                    embedding_function=self._embedding_function,
+                    metadata={"hnsw:space": "cosine"},
+                )
+                return count
+            except Exception as e:
+                logger.error(f"Failed to clear memory: {e}")
+                return 0
+
+    async def persist(self) -> None:
+        """Persist changes to disk."""
+        if self._client and hasattr(self._client, "persist"):
+            try:
+                self._client.persist()
+            except Exception as e:
+                logger.error(f"Failed to persist memory: {e}")
+
+    async def shutdown(self) -> None:
+        """Shutdown long-term memory and persist data."""
+        if self._initialized and not self._using_fallback:
+            await self.persist()
+            self._initialized = False
+            logger.info("Long-term memory shutdown complete")
+
+    async def get_stats(self) -> dict[str, Any]:
+        """
+        Get statistics about long-term memory.
+
+        Returns:
+            Dictionary with memory statistics.
+        """
+        count = await self.count()
+        return {
+            "initialized": self._initialized,
+            "using_fallback": self._using_fallback,
+            "collection_name": self.collection_name,
+            "persist_directory": self.persist_directory,
+            "document_count": count,
+            "top_k": self.top_k,
+        }

backend/app/memory/manager.py

@@ -0,0 +1,463 @@
+"""Unified memory manager providing access to all memory layers."""
+
+from __future__ import annotations
+
+import logging
+from enum import Enum
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+from app.config import Settings
+from app.memory.long_term import Document, LongTermMemory, SearchResult
+from app.memory.shared import Message, SharedMemory
+from app.memory.short_term import MemoryEntry, ShortTermMemory
+from app.memory.working import WorkingMemory, WorkingMemoryItem
+
+logger = logging.getLogger(__name__)
+
+
+class MemoryType(str, Enum):
+    """Types of memory layers."""
+
+    SHORT_TERM = "short_term"
+    WORKING = "working"
+    LONG_TERM = "long_term"
+    SHARED = "shared"
+
+
+class MemoryStats(BaseModel):
+    """Statistics for all memory layers."""
+
+    short_term: dict[str, Any] = Field(default_factory=dict)
+    working: dict[str, Any] = Field(default_factory=dict)
+    long_term: dict[str, Any] = Field(default_factory=dict)
+    shared: dict[str, Any] = Field(default_factory=dict)
+
+
+class MemoryManager:
+    """
+    Unified interface to all memory layers.
+
+    The MemoryManager provides a single entry point for interacting with
+    different types of memory (short-term, working, long-term, shared).
+    It handles initialization, coordination, and lifecycle management.
+
+    Attributes:
+        short_term: Episode-scoped dictionary memory.
+        working: LRU-based reasoning scratch space.
+        long_term: Persistent vector storage.
+        shared: Multi-agent shared state.
+    """
+
+    def __init__(self, settings: Settings) -> None:
+        """
+        Initialize memory manager with settings.
+
+        Args:
+            settings: Application settings.
+        """
+        self._settings = settings
+        self._initialized = False
+
+        # Initialize memory layers
+        self.short_term = ShortTermMemory(
+            max_size=settings.short_term_memory_size,
+        )
+
+        self.working = WorkingMemory(
+            capacity=settings.working_memory_size,
+        )
+
+        self.long_term = LongTermMemory(
+            collection_name=settings.chroma_collection_name,
+            persist_directory=settings.chroma_persist_directory,
+            top_k=settings.long_term_memory_top_k,
+        )
+
+        self.shared = SharedMemory()
+
+    async def initialize(self) -> None:
+        """
+        Initialize all memory layers.
+
+        This should be called during application startup.
+        """
+        if self._initialized:
+            return
+
+        try:
+            # Initialize long-term memory (ChromaDB)
+            await self.long_term.initialize()
+            self._initialized = True
+            logger.info("Memory manager initialized successfully")
+        except Exception as e:
+            logger.error(f"Failed to initialize memory manager: {e}")
+            raise
+
+    async def shutdown(self) -> None:
+        """
+        Shutdown all memory layers gracefully.
+
+        This should be called during application shutdown.
+        """
+        try:
+            # Persist long-term memory
+            await self.long_term.shutdown()
+
+            # Clear working memory
+            await self.working.clear()
+
+            self._initialized = False
+            logger.info("Memory manager shutdown complete")
+        except Exception as e:
+            logger.error(f"Error during memory manager shutdown: {e}")
+            raise
+
+    @property
+    def is_initialized(self) -> bool:
+        """Check if memory manager is initialized."""
+        return self._initialized
+
+    # =========================================================================
+    # Unified Store Interface
+    # =========================================================================
+
+    async def store(
+        self,
+        key: str,
+        value: Any,
+        memory_type: MemoryType = MemoryType.SHORT_TERM,
+        **kwargs: Any,
+    ) -> Any:
+        """
+        Store a value in the specified memory layer.
+
+        Args:
+            key: Key or identifier for the stored value.
+            value: Value to store.
+            memory_type: Which memory layer to use.
+            **kwargs: Additional arguments passed to the specific layer.
+
+        Returns:
+            The created entry/document (varies by memory type).
+
+        Raises:
+            ValueError: If memory_type is invalid.
+        """
+        match memory_type:
+            case MemoryType.SHORT_TERM:
+                tags = kwargs.get("tags")
+                return await self.short_term.set(key, value, tags=tags)
+
+            case MemoryType.WORKING:
+                priority = kwargs.get("priority", 0.0)
+                metadata = kwargs.get("metadata")
+                return await self.working.push(
+                    content=value,
+                    item_id=key,
+                    priority=priority,
+                    metadata=metadata,
+                )
+
+            case MemoryType.LONG_TERM:
+                if not isinstance(value, str):
+                    value = str(value)
+                metadata = kwargs.get("metadata")
+                embedding = kwargs.get("embedding")
+                return await self.long_term.store(
+                    content=value,
+                    document_id=key,
+                    metadata=metadata,
+                    embedding=embedding,
+                )
+
+            case MemoryType.SHARED:
+                await self.shared.set_state(key, value)
+                return value
+
+            case _:
+                raise ValueError(f"Invalid memory type: {memory_type}")
+
+    # =========================================================================
+    # Unified Retrieve Interface
+    # =========================================================================
+
+    async def retrieve(
+        self,
+        key: str,
+        memory_type: MemoryType = MemoryType.SHORT_TERM,
+        default: Any = None,
+    ) -> Any:
+        """
+        Retrieve a value from the specified memory layer.
+
+        Args:
+            key: Key or identifier to look up.
+            memory_type: Which memory layer to query.
+            default: Default value if not found.
+
+        Returns:
+            The stored value or default.
+
+        Raises:
+            ValueError: If memory_type is invalid.
+        """
+        match memory_type:
+            case MemoryType.SHORT_TERM:
+                return await self.short_term.get(key, default=default)
+
+            case MemoryType.WORKING:
+                item = await self.working.peek_by_id(key)
+                return item.content if item else default
+
+            case MemoryType.LONG_TERM:
+                doc = await self.long_term.get(key)
+                return doc.content if doc else default
+
+            case MemoryType.SHARED:
+                return await self.shared.get_state(key, default=default)
+
+            case _:
+                raise ValueError(f"Invalid memory type: {memory_type}")
+
+    # =========================================================================
+    # Unified Search Interface
+    # =========================================================================
+
+    async def search(
+        self,
+        query: str,
+        memory_type: MemoryType = MemoryType.LONG_TERM,
+        top_k: int = 10,
+        **kwargs: Any,
+    ) -> list[Any]:
+        """
+        Search for values in the specified memory layer.
+
+        Args:
+            query: Search query.
+            memory_type: Which memory layer to search.
+            top_k: Maximum number of results.
+            **kwargs: Additional arguments for specific layers.
+
+        Returns:
+            List of matching entries/documents.
+
+        Raises:
+            ValueError: If memory_type is invalid or doesn't support search.
+        """
+        match memory_type:
+            case MemoryType.SHORT_TERM:
+                # Search by tag or return all keys containing query
+                tag = kwargs.get("tag")
+                if tag:
+                    return list((await self.short_term.get_by_tag(tag)).items())[:top_k]
+                keys = await self.short_term.list_keys()
+                matching = [k for k in keys if query.lower() in k.lower()]
+                results = []
+                for key in matching[:top_k]:
+                    value = await self.short_term.get(key)
+                    results.append((key, value))
+                return results
+
+            case MemoryType.WORKING:
+                # Search working memory items
+                def matches(item: WorkingMemoryItem) -> bool:
+                    content_str = str(item.content).lower()
+                    return query.lower() in content_str
+
+                items = await self.working.search(matches)
+                return items[:top_k]
+
+            case MemoryType.LONG_TERM:
+                where = kwargs.get("where")
+                query_embedding = kwargs.get("query_embedding")
+                return await self.long_term.search(
+                    query=query,
+                    top_k=top_k,
+                    where=where,
+                    query_embedding=query_embedding,
+                )
+
+            case MemoryType.SHARED:
+                # Search state keys
+                all_state = await self.shared.get_all_state()
+                matching = [
+                    (k, v)
+                    for k, v in all_state.items()
+                    if query.lower() in k.lower()
+                    or query.lower() in str(v).lower()
+                ]
+                return matching[:top_k]
+
+            case _:
+                raise ValueError(f"Invalid memory type: {memory_type}")
+
+    # =========================================================================
+    # Unified Clear Interface
+    # =========================================================================
+
+    async def clear(
+        self,
+        memory_type: MemoryType | None = None,
+    ) -> dict[str, int]:
+        """
+        Clear memory layers.
+
+        Args:
+            memory_type: Specific layer to clear, or None for all.
+
+        Returns:
+            Dictionary with counts of cleared items per layer.
+        """
+        results: dict[str, int] = {}
+
+        if memory_type is None or memory_type == MemoryType.SHORT_TERM:
+            results["short_term"] = await self.short_term.clear()
+
+        if memory_type is None or memory_type == MemoryType.WORKING:
+            results["working"] = await self.working.clear()
+
+        if memory_type is None or memory_type == MemoryType.LONG_TERM:
+            results["long_term"] = await self.long_term.clear()
+
+        if memory_type is None or memory_type == MemoryType.SHARED:
+            shared_results = await self.shared.clear()
+            results["shared_channels"] = shared_results["channels"]
+            results["shared_state"] = shared_results["state_keys"]
+
+        return results
+
+    # =========================================================================
+    # Episode Management
+    # =========================================================================
+
+    async def start_episode(self, episode_id: str) -> None:
+        """
+        Start a new episode, clearing episode-scoped memory.
+
+        Args:
+            episode_id: Unique identifier for the episode.
+        """
+        await self.short_term.set_episode(episode_id)
+        await self.working.clear()
+        logger.debug(f"Started episode: {episode_id}")
+
+    async def end_episode(self) -> dict[str, int]:
+        """
+        End the current episode, clearing temporary memory.
+
+        Returns:
+            Counts of cleared items.
+        """
+        results = {
+            "short_term": await self.short_term.clear(),
+            "working": await self.working.clear(),
+        }
+        logger.debug(f"Ended episode: {results}")
+        return results
+
+    # =========================================================================
+    # Statistics
+    # =========================================================================
+
+    async def get_stats(self) -> MemoryStats:
+        """
+        Get statistics from all memory layers.
+
+        Returns:
+            MemoryStats with info from each layer.
+        """
+        return MemoryStats(
+            short_term=await self.short_term.get_stats(),
+            working=await self.working.get_stats(),
+            long_term=await self.long_term.get_stats(),
+            shared=await self.shared.get_stats(),
+        )
+
+    # =========================================================================
+    # Convenience Methods
+    # =========================================================================
+
+    async def remember(
+        self,
+        content: str,
+        metadata: dict[str, Any] | None = None,
+    ) -> Document:
+        """
+        Store content in long-term memory for later retrieval.
+
+        This is a convenience method for storing knowledge.
+
+        Args:
+            content: Text content to remember.
+            metadata: Optional metadata.
+
+        Returns:
+            The stored document.
+        """
+        return await self.long_term.store(content=content, metadata=metadata)
+
+    async def recall(
+        self,
+        query: str,
+        top_k: int = 5,
+    ) -> list[SearchResult]:
+        """
+        Recall relevant memories based on a query.
+
+        This is a convenience method for semantic search.
+
+        Args:
+            query: Search query.
+            top_k: Number of results to return.
+
+        Returns:
+            List of relevant search results.
+        """
+        return await self.long_term.search(query=query, top_k=top_k)
+
+    async def think(
+        self,
+        thought: str,
+        priority: float = 0.0,
+    ) -> WorkingMemoryItem:
+        """
+        Add a thought to working memory.
+
+        This is a convenience method for reasoning steps.
+
+        Args:
+            thought: The thought content.
+            priority: Priority score.
+
+        Returns:
+            The working memory item.
+        """
+        return await self.working.push(content=thought, priority=priority)
+
+    async def broadcast(
+        self,
+        channel: str,
+        message: Any,
+        sender: str | None = None,
+    ) -> Message:
+        """
+        Broadcast a message to a shared channel.
+
+        This is a convenience method for multi-agent communication.
+
+        Args:
+            channel: Channel name.
+            message: Message payload.
+            sender: Optional sender identifier.
+
+        Returns:
+            The published message.
+        """
+        return await self.shared.publish(
+            channel=channel,
+            payload=message,
+            sender=sender,
+        )

backend/app/memory/shared.py

@@ -0,0 +1,480 @@
+"""Shared memory for multi-agent communication and state sharing."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from collections import defaultdict
+from datetime import datetime
+from typing import Any, Callable, Awaitable
+from uuid import uuid4
+
+from pydantic import BaseModel, Field
+
+logger = logging.getLogger(__name__)
+
+# Type alias for async callback functions
+MessageCallback = Callable[[Any], Awaitable[None]]
+
+
+class Message(BaseModel):
+    """A message published to a channel."""
+
+    id: str = Field(default_factory=lambda: str(uuid4()))
+    channel: str
+    payload: Any
+    sender: str | None = None
+    timestamp: datetime = Field(default_factory=datetime.utcnow)
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+    model_config = {"arbitrary_types_allowed": True}
+
+
+class Subscription(BaseModel):
+    """A subscription to a channel."""
+
+    id: str = Field(default_factory=lambda: str(uuid4()))
+    channel: str
+    subscriber_id: str
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+
+    model_config = {"arbitrary_types_allowed": True}
+
+
+class Channel:
+    """A named channel for pub/sub communication."""
+
+    def __init__(self, name: str, max_history: int = 100) -> None:
+        """
+        Initialize a channel.
+
+        Args:
+            name: Channel name.
+            max_history: Maximum number of messages to retain in history.
+        """
+        self.name = name
+        self.max_history = max_history
+        self._subscribers: dict[str, MessageCallback] = {}
+        self._history: list[Message] = []
+        self._lock = asyncio.Lock()
+
+    @property
+    def subscriber_count(self) -> int:
+        """Get the number of subscribers."""
+        return len(self._subscribers)
+
+    async def publish(self, message: Message) -> int:
+        """
+        Publish a message to all subscribers.
+
+        Args:
+            message: Message to publish.
+
+        Returns:
+            Number of subscribers that received the message.
+        """
+        async with self._lock:
+            # Add to history
+            self._history.append(message)
+            if len(self._history) > self.max_history:
+                self._history = self._history[-self.max_history:]
+
+            # Notify subscribers
+            notified = 0
+            for sub_id, callback in list(self._subscribers.items()):
+                try:
+                    await callback(message)
+                    notified += 1
+                except Exception as e:
+                    logger.error(f"Error notifying subscriber {sub_id}: {e}")
+
+            return notified
+
+    async def subscribe(
+        self,
+        subscriber_id: str,
+        callback: MessageCallback,
+    ) -> Subscription:
+        """
+        Subscribe to the channel.
+
+        Args:
+            subscriber_id: Unique identifier for the subscriber.
+            callback: Async callback function to receive messages.
+
+        Returns:
+            Subscription object.
+        """
+        async with self._lock:
+            self._subscribers[subscriber_id] = callback
+            return Subscription(
+                channel=self.name,
+                subscriber_id=subscriber_id,
+            )
+
+    async def unsubscribe(self, subscriber_id: str) -> bool:
+        """
+        Unsubscribe from the channel.
+
+        Args:
+            subscriber_id: Subscriber to remove.
+
+        Returns:
+            True if subscriber was found and removed.
+        """
+        async with self._lock:
+            if subscriber_id in self._subscribers:
+                del self._subscribers[subscriber_id]
+                return True
+            return False
+
+    async def get_history(
+        self,
+        limit: int | None = None,
+        since: datetime | None = None,
+    ) -> list[Message]:
+        """
+        Get channel message history.
+
+        Args:
+            limit: Maximum number of messages to return.
+            since: Only return messages after this timestamp.
+
+        Returns:
+            List of historical messages.
+        """
+        async with self._lock:
+            messages = self._history
+
+            if since:
+                messages = [m for m in messages if m.timestamp > since]
+
+            if limit:
+                messages = messages[-limit:]
+
+            return messages
+
+    async def clear_history(self) -> int:
+        """
+        Clear the channel's message history.
+
+        Returns:
+            Number of messages cleared.
+        """
+        async with self._lock:
+            count = len(self._history)
+            self._history.clear()
+            return count
+
+
+class SharedMemory:
+    """
+    Thread-safe shared memory for multi-agent coordination.
+
+    This memory layer provides pub/sub messaging and shared state storage
+    for coordination between multiple agents. All operations are thread-safe.
+
+    Attributes:
+        _channels: Dictionary of channels by name.
+        _state: Shared key-value state store.
+    """
+
+    def __init__(self, max_channel_history: int = 100) -> None:
+        """
+        Initialize shared memory.
+
+        Args:
+            max_channel_history: Maximum history per channel.
+        """
+        self.max_channel_history = max_channel_history
+        self._channels: dict[str, Channel] = {}
+        self._state: dict[str, Any] = {}
+        self._state_lock = asyncio.Lock()
+        self._channel_lock = asyncio.Lock()
+        self._queues: dict[str, dict[str, asyncio.Queue]] = defaultdict(dict)
+
+    async def get_channel(self, name: str) -> Channel:
+        """
+        Get or create a channel by name.
+
+        Args:
+            name: Channel name.
+
+        Returns:
+            The channel object.
+        """
+        async with self._channel_lock:
+            if name not in self._channels:
+                self._channels[name] = Channel(
+                    name=name,
+                    max_history=self.max_channel_history,
+                )
+            return self._channels[name]
+
+    async def publish(
+        self,
+        channel: str,
+        payload: Any,
+        sender: str | None = None,
+        metadata: dict[str, Any] | None = None,
+    ) -> Message:
+        """
+        Publish a message to a channel.
+
+        Args:
+            channel: Channel name to publish to.
+            payload: Message payload.
+            sender: Optional sender identifier.
+            metadata: Optional message metadata.
+
+        Returns:
+            The published message.
+        """
+        ch = await self.get_channel(channel)
+
+        message = Message(
+            channel=channel,
+            payload=payload,
+            sender=sender,
+            metadata=metadata or {},
+        )
+
+        await ch.publish(message)
+
+        # Also put in subscriber queues
+        async with self._channel_lock:
+            if channel in self._queues:
+                for queue in self._queues[channel].values():
+                    try:
+                        queue.put_nowait(message)
+                    except asyncio.QueueFull:
+                        # Remove oldest and add new
+                        try:
+                            queue.get_nowait()
+                            queue.put_nowait(message)
+                        except asyncio.QueueEmpty:
+                            pass
+
+        return message
+
+    async def subscribe(
+        self,
+        channel: str,
+        subscriber_id: str,
+        callback: MessageCallback,
+    ) -> Subscription:
+        """
+        Subscribe to a channel with a callback.
+
+        Args:
+            channel: Channel name to subscribe to.
+            subscriber_id: Unique subscriber identifier.
+            callback: Async callback for received messages.
+
+        Returns:
+            Subscription object.
+        """
+        ch = await self.get_channel(channel)
+        return await ch.subscribe(subscriber_id, callback)
+
+    async def subscribe_queue(
+        self,
+        channel: str,
+        subscriber_id: str,
+        max_size: int = 100,
+    ) -> asyncio.Queue[Message]:
+        """
+        Subscribe to a channel and receive messages via a queue.
+
+        This is an alternative to callback-based subscriptions.
+
+        Args:
+            channel: Channel name to subscribe to.
+            subscriber_id: Unique subscriber identifier.
+            max_size: Maximum queue size.
+
+        Returns:
+            Queue that will receive messages.
+        """
+        async with self._channel_lock:
+            if subscriber_id not in self._queues[channel]:
+                self._queues[channel][subscriber_id] = asyncio.Queue(maxsize=max_size)
+            return self._queues[channel][subscriber_id]
+
+    async def unsubscribe(self, channel: str, subscriber_id: str) -> bool:
+        """
+        Unsubscribe from a channel.
+
+        Args:
+            channel: Channel name.
+            subscriber_id: Subscriber to remove.
+
+        Returns:
+            True if subscriber was found and removed.
+        """
+        async with self._channel_lock:
+            # Remove from callback subscriptions
+            if channel in self._channels:
+                await self._channels[channel].unsubscribe(subscriber_id)
+
+            # Remove from queue subscriptions
+            if channel in self._queues and subscriber_id in self._queues[channel]:
+                del self._queues[channel][subscriber_id]
+                return True
+
+            return False
+
+    async def set_state(self, key: str, value: Any) -> None:
+        """
+        Set a shared state value.
+
+        Args:
+            key: State key.
+            value: Value to store.
+        """
+        async with self._state_lock:
+            self._state[key] = value
+
+    async def get_state(self, key: str, default: Any = None) -> Any:
+        """
+        Get a shared state value.
+
+        Args:
+            key: State key.
+            default: Default value if key not found.
+
+        Returns:
+            The stored value or default.
+        """
+        async with self._state_lock:
+            return self._state.get(key, default)
+
+    async def delete_state(self, key: str) -> bool:
+        """
+        Delete a shared state value.
+
+        Args:
+            key: State key to delete.
+
+        Returns:
+            True if key was found and deleted.
+        """
+        async with self._state_lock:
+            if key in self._state:
+                del self._state[key]
+                return True
+            return False
+
+    async def update_state(self, key: str, updater: Callable[[Any], Any]) -> Any:
+        """
+        Atomically update a state value.
+
+        Args:
+            key: State key.
+            updater: Function that takes current value and returns new value.
+
+        Returns:
+            The new value after update.
+        """
+        async with self._state_lock:
+            current = self._state.get(key)
+            new_value = updater(current)
+            self._state[key] = new_value
+            return new_value
+
+    async def get_all_state(self) -> dict[str, Any]:
+        """
+        Get all shared state values.
+
+        Returns:
+            Copy of the state dictionary.
+        """
+        async with self._state_lock:
+            return dict(self._state)
+
+    async def clear_state(self) -> int:
+        """
+        Clear all shared state.
+
+        Returns:
+            Number of keys cleared.
+        """
+        async with self._state_lock:
+            count = len(self._state)
+            self._state.clear()
+            return count
+
+    async def list_channels(self) -> list[str]:
+        """
+        List all active channels.
+
+        Returns:
+            List of channel names.
+        """
+        async with self._channel_lock:
+            return list(self._channels.keys())
+
+    async def delete_channel(self, name: str) -> bool:
+        """
+        Delete a channel and all its subscriptions.
+
+        Args:
+            name: Channel name to delete.
+
+        Returns:
+            True if channel was found and deleted.
+        """
+        async with self._channel_lock:
+            if name in self._channels:
+                del self._channels[name]
+                if name in self._queues:
+                    del self._queues[name]
+                return True
+            return False
+
+    async def clear(self) -> dict[str, int]:
+        """
+        Clear all channels and state.
+
+        Returns:
+            Dictionary with counts of cleared items.
+        """
+        async with self._channel_lock:
+            channel_count = len(self._channels)
+            self._channels.clear()
+            self._queues.clear()
+
+        async with self._state_lock:
+            state_count = len(self._state)
+            self._state.clear()
+
+        return {
+            "channels": channel_count,
+            "state_keys": state_count,
+        }
+
+    async def get_stats(self) -> dict[str, Any]:
+        """
+        Get statistics about shared memory.
+
+        Returns:
+            Dictionary with memory statistics.
+        """
+        async with self._channel_lock:
+            channel_stats = {}
+            for name, channel in self._channels.items():
+                channel_stats[name] = {
+                    "subscribers": channel.subscriber_count,
+                    "history_size": len(channel._history),
+                }
+
+        async with self._state_lock:
+            state_keys = list(self._state.keys())
+
+        return {
+            "channel_count": len(channel_stats),
+            "channels": channel_stats,
+            "state_key_count": len(state_keys),
+            "state_keys": state_keys,
+            "max_channel_history": self.max_channel_history,
+        }

backend/app/memory/short_term.py

@@ -0,0 +1,243 @@
+"""Short-term memory for episode-scoped data storage."""
+
+from __future__ import annotations
+
+import asyncio
+from collections import OrderedDict
+from datetime import datetime
+from typing import Any, Generic, TypeVar
+
+from pydantic import BaseModel, Field
+
+T = TypeVar("T")
+
+
+class MemoryEntry(BaseModel, Generic[T]):
+    """A single memory entry with metadata."""
+
+    key: str
+    value: Any
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    updated_at: datetime = Field(default_factory=datetime.utcnow)
+    access_count: int = 0
+    tags: list[str] = Field(default_factory=list)
+
+    model_config = {"arbitrary_types_allowed": True}
+
+
+class ShortTermMemory:
+    """
+    Episode-scoped memory using dictionary-based storage.
+
+    This memory layer is designed for transient data that should persist
+    only within a single episode. It automatically clears when the episode
+    resets.
+
+    Attributes:
+        max_size: Maximum number of entries allowed.
+        _store: Internal storage dictionary.
+        _episode_id: Current episode identifier.
+    """
+
+    def __init__(self, max_size: int = 100) -> None:
+        """
+        Initialize short-term memory.
+
+        Args:
+            max_size: Maximum number of entries to store. Defaults to 100.
+        """
+        self.max_size = max_size
+        self._store: OrderedDict[str, MemoryEntry] = OrderedDict()
+        self._episode_id: str | None = None
+        self._lock = asyncio.Lock()
+
+    @property
+    def episode_id(self) -> str | None:
+        """Get the current episode ID."""
+        return self._episode_id
+
+    @property
+    def size(self) -> int:
+        """Get the current number of entries."""
+        return len(self._store)
+
+    async def set_episode(self, episode_id: str) -> None:
+        """
+        Set the current episode ID and clear existing memory.
+
+        Args:
+            episode_id: Unique identifier for the new episode.
+        """
+        async with self._lock:
+            if self._episode_id != episode_id:
+                self._store.clear()
+                self._episode_id = episode_id
+
+    async def set(
+        self,
+        key: str,
+        value: Any,
+        tags: list[str] | None = None,
+    ) -> MemoryEntry:
+        """
+        Store a value in short-term memory.
+
+        Args:
+            key: Unique key for the entry.
+            value: Value to store.
+            tags: Optional tags for categorization.
+
+        Returns:
+            The created or updated memory entry.
+
+        Raises:
+            ValueError: If max_size would be exceeded for a new key.
+        """
+        async with self._lock:
+            now = datetime.utcnow()
+
+            if key in self._store:
+                entry = self._store[key]
+                entry.value = value
+                entry.updated_at = now
+                if tags is not None:
+                    entry.tags = tags
+                # Move to end (most recent)
+                self._store.move_to_end(key)
+            else:
+                # Check capacity
+                if len(self._store) >= self.max_size:
+                    # Remove oldest entry
+                    self._store.popitem(last=False)
+
+                entry = MemoryEntry(
+                    key=key,
+                    value=value,
+                    created_at=now,
+                    updated_at=now,
+                    tags=tags or [],
+                )
+                self._store[key] = entry
+
+            return entry
+
+    async def get(self, key: str, default: Any = None) -> Any:
+        """
+        Retrieve a value from short-term memory.
+
+        Args:
+            key: Key to look up.
+            default: Default value if key not found.
+
+        Returns:
+            The stored value or default.
+        """
+        async with self._lock:
+            entry = self._store.get(key)
+            if entry is None:
+                return default
+            entry.access_count += 1
+            return entry.value
+
+    async def get_entry(self, key: str) -> MemoryEntry | None:
+        """
+        Retrieve a full memory entry with metadata.
+
+        Args:
+            key: Key to look up.
+
+        Returns:
+            The memory entry or None if not found.
+        """
+        async with self._lock:
+            entry = self._store.get(key)
+            if entry:
+                entry.access_count += 1
+            return entry
+
+    async def delete(self, key: str) -> bool:
+        """
+        Delete an entry from memory.
+
+        Args:
+            key: Key to delete.
+
+        Returns:
+            True if the key was found and deleted, False otherwise.
+        """
+        async with self._lock:
+            if key in self._store:
+                del self._store[key]
+                return True
+            return False
+
+    async def clear(self) -> int:
+        """
+        Clear all entries from memory.
+
+        Returns:
+            Number of entries that were cleared.
+        """
+        async with self._lock:
+            count = len(self._store)
+            self._store.clear()
+            return count
+
+    async def list_keys(self, tag: str | None = None) -> list[str]:
+        """
+        List all keys in memory, optionally filtered by tag.
+
+        Args:
+            tag: Optional tag to filter by.
+
+        Returns:
+            List of matching keys.
+        """
+        async with self._lock:
+            if tag is None:
+                return list(self._store.keys())
+            return [k for k, v in self._store.items() if tag in v.tags]
+
+    async def get_by_tag(self, tag: str) -> dict[str, Any]:
+        """
+        Retrieve all entries with a specific tag.
+
+        Args:
+            tag: Tag to filter by.
+
+        Returns:
+            Dictionary of key-value pairs matching the tag.
+        """
+        async with self._lock:
+            return {
+                k: v.value for k, v in self._store.items() if tag in v.tags
+            }
+
+    async def exists(self, key: str) -> bool:
+        """
+        Check if a key exists in memory.
+
+        Args:
+            key: Key to check.
+
+        Returns:
+            True if key exists, False otherwise.
+        """
+        async with self._lock:
+            return key in self._store
+
+    async def get_stats(self) -> dict[str, Any]:
+        """
+        Get statistics about the memory store.
+
+        Returns:
+            Dictionary with memory statistics.
+        """
+        async with self._lock:
+            return {
+                "size": len(self._store),
+                "max_size": self.max_size,
+                "episode_id": self._episode_id,
+                "keys": list(self._store.keys()),
+                "utilization": len(self._store) / self.max_size if self.max_size > 0 else 0,
+            }

backend/app/memory/working.py

@@ -0,0 +1,277 @@
+"""Working memory for reasoning and scratch space with LRU eviction."""
+
+from __future__ import annotations
+
+import asyncio
+from collections import OrderedDict
+from datetime import datetime
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+
+class WorkingMemoryItem(BaseModel):
+    """A single item in working memory."""
+
+    id: str
+    content: Any
+    priority: float = 0.0
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    last_accessed: datetime = Field(default_factory=datetime.utcnow)
+    access_count: int = 0
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+    model_config = {"arbitrary_types_allowed": True}
+
+
+class WorkingMemory:
+    """
+    Working memory for reasoning and scratch computations.
+
+    This memory layer provides a limited-capacity buffer with LRU (Least Recently Used)
+    eviction policy. It's designed for temporary reasoning steps, intermediate results,
+    and scratch space during agent deliberation.
+
+    Attributes:
+        capacity: Maximum number of items in working memory.
+        _items: Internal LRU-ordered storage.
+    """
+
+    def __init__(self, capacity: int = 20) -> None:
+        """
+        Initialize working memory.
+
+        Args:
+            capacity: Maximum number of items to store. Defaults to 20.
+        """
+        self.capacity = capacity
+        self._items: OrderedDict[str, WorkingMemoryItem] = OrderedDict()
+        self._counter = 0
+        self._lock = asyncio.Lock()
+
+    @property
+    def size(self) -> int:
+        """Get current number of items in memory."""
+        return len(self._items)
+
+    @property
+    def is_full(self) -> bool:
+        """Check if memory is at capacity."""
+        return len(self._items) >= self.capacity
+
+    async def push(
+        self,
+        content: Any,
+        item_id: str | None = None,
+        priority: float = 0.0,
+        metadata: dict[str, Any] | None = None,
+    ) -> WorkingMemoryItem:
+        """
+        Push a new item into working memory.
+
+        If capacity is reached, the least recently used item is evicted.
+
+        Args:
+            content: The content to store.
+            item_id: Optional custom ID. Auto-generated if not provided.
+            priority: Priority score for potential prioritized eviction.
+            metadata: Optional metadata dictionary.
+
+        Returns:
+            The created working memory item.
+        """
+        async with self._lock:
+            # Generate ID if not provided
+            if item_id is None:
+                self._counter += 1
+                item_id = f"wm_{self._counter}"
+
+            now = datetime.utcnow()
+
+            # Check if item already exists (update it)
+            if item_id in self._items:
+                item = self._items[item_id]
+                item.content = content
+                item.last_accessed = now
+                item.access_count += 1
+                if metadata:
+                    item.metadata.update(metadata)
+                if priority != 0.0:
+                    item.priority = priority
+                # Move to end (most recent)
+                self._items.move_to_end(item_id)
+                return item
+
+            # Evict LRU item if at capacity
+            if len(self._items) >= self.capacity:
+                self._evict_lru()
+
+            # Create new item
+            item = WorkingMemoryItem(
+                id=item_id,
+                content=content,
+                priority=priority,
+                created_at=now,
+                last_accessed=now,
+                metadata=metadata or {},
+            )
+            self._items[item_id] = item
+            return item
+
+    def _evict_lru(self) -> WorkingMemoryItem | None:
+        """
+        Evict the least recently used item.
+
+        Returns:
+            The evicted item, or None if memory was empty.
+        """
+        if not self._items:
+            return None
+        # Pop first item (least recently used)
+        _, item = self._items.popitem(last=False)
+        return item
+
+    async def pop(self) -> WorkingMemoryItem | None:
+        """
+        Remove and return the most recently used item.
+
+        Returns:
+            The most recent item, or None if memory is empty.
+        """
+        async with self._lock:
+            if not self._items:
+                return None
+            _, item = self._items.popitem(last=True)
+            return item
+
+    async def pop_by_id(self, item_id: str) -> WorkingMemoryItem | None:
+        """
+        Remove and return an item by its ID.
+
+        Args:
+            item_id: The ID of the item to remove.
+
+        Returns:
+            The removed item, or None if not found.
+        """
+        async with self._lock:
+            return self._items.pop(item_id, None)
+
+    async def peek(self) -> WorkingMemoryItem | None:
+        """
+        Return the most recently used item without removing it.
+
+        Returns:
+            The most recent item, or None if memory is empty.
+        """
+        async with self._lock:
+            if not self._items:
+                return None
+            # Get last item
+            item_id = next(reversed(self._items))
+            item = self._items[item_id]
+            item.last_accessed = datetime.utcnow()
+            item.access_count += 1
+            return item
+
+    async def peek_by_id(self, item_id: str) -> WorkingMemoryItem | None:
+        """
+        Return an item by ID without removing it.
+
+        Args:
+            item_id: The ID of the item to peek.
+
+        Returns:
+            The item, or None if not found.
+        """
+        async with self._lock:
+            item = self._items.get(item_id)
+            if item:
+                item.last_accessed = datetime.utcnow()
+                item.access_count += 1
+                # Move to end (mark as recently accessed)
+                self._items.move_to_end(item_id)
+            return item
+
+    async def get_all(self) -> list[WorkingMemoryItem]:
+        """
+        Get all items in memory, ordered by recency.
+
+        Returns:
+            List of items from least to most recent.
+        """
+        async with self._lock:
+            return list(self._items.values())
+
+    async def get_recent(self, n: int = 5) -> list[WorkingMemoryItem]:
+        """
+        Get the N most recently accessed items.
+
+        Args:
+            n: Number of items to return.
+
+        Returns:
+            List of most recent items.
+        """
+        async with self._lock:
+            items = list(self._items.values())
+            return items[-n:] if n < len(items) else items
+
+    async def clear(self) -> int:
+        """
+        Clear all items from working memory.
+
+        Returns:
+            Number of items that were cleared.
+        """
+        async with self._lock:
+            count = len(self._items)
+            self._items.clear()
+            self._counter = 0
+            return count
+
+    async def search(self, predicate: Any) -> list[WorkingMemoryItem]:
+        """
+        Search items using a predicate function.
+
+        Args:
+            predicate: Callable that takes a WorkingMemoryItem and returns bool.
+
+        Returns:
+            List of matching items.
+        """
+        async with self._lock:
+            return [item for item in self._items.values() if predicate(item)]
+
+    async def update_priority(self, item_id: str, priority: float) -> bool:
+        """
+        Update the priority of an item.
+
+        Args:
+            item_id: ID of the item to update.
+            priority: New priority value.
+
+        Returns:
+            True if item was found and updated, False otherwise.
+        """
+        async with self._lock:
+            if item_id in self._items:
+                self._items[item_id].priority = priority
+                return True
+            return False
+
+    async def get_stats(self) -> dict[str, Any]:
+        """
+        Get statistics about working memory.
+
+        Returns:
+            Dictionary with memory statistics.
+        """
+        async with self._lock:
+            return {
+                "size": len(self._items),
+                "capacity": self.capacity,
+                "is_full": len(self._items) >= self.capacity,
+                "utilization": len(self._items) / self.capacity if self.capacity > 0 else 0,
+                "item_ids": list(self._items.keys()),
+            }