Multi-Layer Memory Architecture

{
  "autoMemoryEnabled": true,
  "autoMemoryDirectory": "~/my-custom-memory-dir"
}

from pathlib import Path
import json
from datetime import datetime

MANIFEST = Path(".claude/state/manifest.json")

def write_manifest(modified_files, test_status, plan, current_step, open_todos):
    """Call after EVERY meaningful step in a long-running session."""
    MANIFEST.parent.mkdir(parents=True, exist_ok=True)
    MANIFEST.write_text(json.dumps({
        "updated_at": datetime.utcnow().isoformat(),
        "modified_files": modified_files,   # [{"path": "src/api.ts", "summary": "added /v2/search"}]
        "test_status": test_status,         # "47/50 passing; auth_test.ts:42 failing"
        "plan": plan,                       # ["step1: schema", "step2: route", ...]
        "current_step": current_step,
        "open_todos": open_todos,           # ["fix locale fallback", "add rate-limit test"]
    }, indent=2))

def read_manifest():
    """First action of any resumed session."""
    if not MANIFEST.exists():
        return None
    return json.loads(MANIFEST.read_text())

import json
from datetime import datetime, timezone
from typing import Any


class WorkingMemory:
    """Fast, mutable scratchpad for current task state.

    Stores key-value pairs in memory and provides a formatted
    string for injection into Claude's system prompt.
    """

    def __init__(self, session_id: str):
        self.session_id = session_id
        self.created_at = datetime.now(timezone.utc).isoformat()
        self._state: dict[str, Any] = {}

    def set(self, key: str, value: Any) -> None:
        """Store a key-value pair in working memory."""
        self._state[key] = {
            "value": value,
            "updated_at": datetime.now(timezone.utc).isoformat()
        }

    def get(self, key: str, default: Any = None) -> Any:
        """Retrieve a value by key. Returns default if not found."""
        entry = self._state.get(key)
        return entry["value"] if entry else default

    def delete(self, key: str) -> bool:
        """Remove a key from working memory. Returns True if existed."""
        return self._state.pop(key, None) is not None

    def clear(self) -> None:
        """Clear all working memory state."""
        self._state.clear()

    def to_prompt(self) -> str:
        """Format working memory for injection into system prompt.

        Returns a structured string that Claude can parse to understand
        the current task state.
        """
        if not self._state:
            return "[Working Memory: empty — new task]"

        lines = [
            f"[Working Memory — Session {self.session_id}]",
            f"  Created: {self.created_at}"
        ]
        for key, entry in self._state.items():
            val = json.dumps(entry["value"]) if not isinstance(
                entry["value"], str
            ) else entry["value"]
            lines.append(f"  {key}: {val}")

        return "\n".join(lines)

    def to_dict(self) -> dict:
        """Serialize for persistence (e.g., to Redis or SQLite)."""
        return {
            "session_id": self.session_id,
            "created_at": self.created_at,
            "state": {
                k: v["value"] for k, v in self._state.items()
            }
        }


# --- Usage ---
wm = WorkingMemory(session_id="sess_001")
wm.set("intent", "find_deployment_date")
wm.set("entities", {"topic": "deployment", "timeframe": "last Tuesday"})
wm.set("search_results", [{"session": 42, "summary": "Deploy moved to Friday"}])

print(wm.to_prompt())
# [Working Memory — Session sess_001]
#   Created: 2026-03-29T10:00:00+00:00
#   intent: find_deployment_date
#   entities: {"topic": "deployment", "timeframe": "last Tuesday"}
#   search_results: [{"session": 42, "summary": "Deploy moved to Friday"}]

import chromadb  # pip install chromadb>=0.5.0
import uuid
from datetime import datetime, timezone


class EpisodicMemory:
    """Stores and retrieves conversation summaries via semantic search.

    Uses ChromaDB with persistent storage so memories survive restarts.
    """

    def __init__(self, persist_dir: str = "./chroma_db", collection_name: str = "episodes"):
        try:
            self.client = chromadb.PersistentClient(path=persist_dir)
            self.collection = self.client.get_or_create_collection(
                name=collection_name,
                metadata={"hnsw:space": "cosine"}  # cosine similarity
            )
        except Exception as e:
            raise ConnectionError(
                f"Failed to connect to ChromaDB at {persist_dir}: {e}"
            ) from e

    def store_episode(
        self,
        summary: str,
        session_id: str,
        user_id: str = "default",
        topics: list[str] | None = None,
    ) -> str:
        """Store a conversation summary as an episode.

        Returns the episode ID for later reference.
        """
        episode_id = f"ep_{uuid.uuid4().hex[:12]}"
        metadata = {
            "session_id": session_id,
            "user_id": user_id,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "topics": ",".join(topics) if topics else "",
        }

        try:
            self.collection.add(
                documents=[summary],
                metadatas=[metadata],
                ids=[episode_id],
            )
        except Exception as e:
            raise RuntimeError(f"Failed to store episode: {e}") from e

        return episode_id

    def retrieve(
        self,
        query: str,
        n_results: int = 3,
        user_id: str | None = None,
    ) -> list[dict]:
        """Find the most relevant past episodes for a query.

        Returns a list of dicts with 'summary', 'session_id',
        'timestamp', and 'similarity' keys.
        """
        where_filter = {"user_id": user_id} if user_id else None

        try:
            results = self.collection.query(
                query_texts=[query],
                n_results=min(n_results, self.collection.count() or 1),
                where=where_filter,
            )
        except Exception as e:
            # Return empty on search failure — don't crash the agent
            print(f"Warning: episodic search failed: {e}")
            return []

        if not results["documents"] or not results["documents"][0]:
            return []

        episodes = []
        for i, doc in enumerate(results["documents"][0]):
            meta = results["metadatas"][0][i]
            distance = results["distances"][0][i] if results["distances"] else 0
            episodes.append({
                "summary": doc,
                "session_id": meta.get("session_id", ""),
                "timestamp": meta.get("timestamp", ""),
                "topics": meta.get("topics", "").split(","),
                "similarity": round(1 - distance, 3),  # cosine distance → similarity
            })

        return episodes

    def to_prompt(self, query: str, n_results: int = 3) -> str:
        """Retrieve relevant episodes and format for prompt injection."""
        episodes = self.retrieve(query, n_results)
        if not episodes:
            return "[Episodic Memory: no relevant past interactions found]"

        lines = ["[Relevant Past Interactions]"]
        for ep in episodes:
            lines.append(
                f"  - Session {ep['session_id']} ({ep['timestamp'][:10]}): "
                f"{ep['summary']}"
            )
        return "\n".join(lines)


# --- Usage ---
em = EpisodicMemory(persist_dir="./memory_db")

# Store a past conversation summary
em.store_episode(
    summary="User asked about deployment schedule. Decision: deploy moved to Friday due to QA delays.",
    session_id="sess_042",
    user_id="user_alice",
    topics=["deployment", "scheduling"],
)

# Later, retrieve relevant episodes
results = em.retrieve("When is the deployment?", user_id="user_alice")
for r in results:
    print(f"[{r['similarity']}] Session {r['session_id']}: {r['summary']}")

import chromadb
import json
import uuid


class ProceduralMemory:
    """Stores and retrieves reusable action templates (tool-call sequences).

    Trigger conditions are stored as embeddings for semantic matching.
    Execution steps are stored as structured JSON.
    """

    def __init__(self, persist_dir: str = "./chroma_db"):
        try:
            self.client = chromadb.PersistentClient(path=persist_dir)
            self.collection = self.client.get_or_create_collection(
                name="procedures",
                metadata={"hnsw:space": "cosine"},
            )
        except Exception as e:
            raise ConnectionError(f"Failed to connect to ChromaDB: {e}") from e

    def store_procedure(
        self,
        name: str,
        trigger: str,
        steps: list[dict],
        success_count: int = 1,
    ) -> str:
        """Store a reusable procedure.

        Args:
            name: Human-readable procedure name (e.g., 'weekly_report')
            trigger: Description of when this procedure applies
            steps: Ordered list of step dicts with 'tool', 'params', 'description'
            success_count: Number of times this procedure has succeeded
        """
        proc_id = f"proc_{uuid.uuid4().hex[:10]}"
        metadata = {
            "name": name,
            "steps_json": json.dumps(steps),
            "success_count": success_count,
        }

        try:
            self.collection.add(
                documents=[trigger],   # trigger is the searchable text
                metadatas=[metadata],
                ids=[proc_id],
            )
        except Exception as e:
            raise RuntimeError(f"Failed to store procedure: {e}") from e

        return proc_id

    def find_procedure(
        self,
        query: str,
        min_similarity: float = 0.7,
    ) -> dict | None:
        """Find the best matching procedure for a user request.

        Returns None if no procedure matches above the similarity threshold.
        """
        try:
            count = self.collection.count()
            if count == 0:
                return None

            results = self.collection.query(
                query_texts=[query],
                n_results=1,
            )
        except Exception as e:
            print(f"Warning: procedural search failed: {e}")
            return None

        if not results["documents"] or not results["documents"][0]:
            return None

        distance = results["distances"][0][0] if results["distances"] else 1.0
        similarity = 1 - distance

        if similarity < min_similarity:
            return None

        meta = results["metadatas"][0][0]
        return {
            "name": meta["name"],
            "trigger": results["documents"][0][0],
            "steps": json.loads(meta["steps_json"]),
            "success_count": meta.get("success_count", 0),
            "similarity": round(similarity, 3),
        }

    def to_prompt(self, query: str) -> str:
        """Find a matching procedure and format it as a suggested plan."""
        proc = self.find_procedure(query)
        if not proc:
            return "[Procedural Memory: no matching skill template found]"

        lines = [
            f"[Suggested Procedure: {proc['name']} (used {proc['success_count']}x)]",
        ]
        for i, step in enumerate(proc["steps"], 1):
            lines.append(
                f"  Step {i}: {step['description']} "
                f"(tool: {step['tool']})"
            )
        return "\n".join(lines)


# --- Usage ---
pm = ProceduralMemory(persist_dir="./memory_db")

# Store a proven procedure
pm.store_procedure(
    name="weekly_sales_report",
    trigger="Generate a weekly sales report with charts and insights",
    steps=[
        {"tool": "query_db", "params": {"query": "sales_last_7_days"}, "description": "Fetch sales data"},
        {"tool": "aggregate", "params": {"group_by": "region"}, "description": "Aggregate by region"},
        {"tool": "chart_gen", "params": {"type": "bar"}, "description": "Generate bar chart"},
        {"tool": "summarize", "params": {}, "description": "Extract key insights"},
        {"tool": "format_pdf", "params": {}, "description": "Format as PDF report"},
    ],
    success_count=12,
)

# Later, find a matching procedure
result = pm.find_procedure("Can you create the weekly sales report?")
if result:
    print(f"Found: {result['name']} (similarity: {result['similarity']})")
    for step in result["steps"]:
        print(f"  → {step['description']}")

import anthropic  # pip install anthropic>=0.30.0
import json


async def summarize_conversation(
    messages: list[dict],
    client: anthropic.AsyncAnthropic | None = None,
) -> dict:
    """Summarize a conversation into a structured episode record.

    Args:
        messages: List of conversation messages (role + content dicts)
        client: Anthropic client (creates one if not provided)

    Returns:
        Dict with: summary, topics, decisions, user_preferences, action_items
    """
    if client is None:
        client = anthropic.AsyncAnthropic()  # reads ANTHROPIC_API_KEY env var

    # Format the conversation for summarization
    conversation_text = "\n".join(
        f"{msg['role'].upper()}: {msg['content']}" for msg in messages
    )

    try:
        response = await client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=500,
            system=(
                "You are a conversation summarizer. Extract the key information "
                "from the conversation and return a JSON object with these fields:\n"
                '- "summary": 1-2 sentence overview of the conversation\n'
                '- "topics": list of main topics discussed\n'
                '- "decisions": list of decisions made (empty if none)\n'
                '- "user_preferences": any user preferences learned (empty if none)\n'
                '- "action_items": pending items (empty if none)\n'
                "Return ONLY valid JSON, no markdown fences."
            ),
            messages=[{"role": "user", "content": conversation_text}],
        )

        result = json.loads(response.content[0].text)
        return result

    except json.JSONDecodeError:
        # Fallback: return raw summary if Claude didn't produce valid JSON
        return {
            "summary": response.content[0].text[:500],
            "topics": [],
            "decisions": [],
            "user_preferences": [],
            "action_items": [],
        }
    except anthropic.APIError as e:
        raise RuntimeError(f"Summarization API call failed: {e}") from e

import anthropic


class MemoryManager:
    """Orchestrates all three memory tiers for a conversational agent.

    Lifecycle:
      1. start_session() — creates working memory, loads relevant episodes + procedures
      2. During conversation — update working memory via set/get
      3. end_session() — summarizes conversation, stores episode, clears working memory
    """

    def __init__(self, persist_dir: str = "./memory_db"):
        self.episodic = EpisodicMemory(persist_dir=persist_dir)
        self.procedural = ProceduralMemory(persist_dir=persist_dir)
        self.working: WorkingMemory | None = None
        self.client = anthropic.AsyncAnthropic()
        self.conversation_log: list[dict] = []

    def start_session(self, session_id: str, user_id: str = "default") -> str:
        """Initialize a new session with fresh working memory.

        Returns the formatted memory context for the first prompt.
        """
        self.working = WorkingMemory(session_id=session_id)
        self.working.set("user_id", user_id)
        self.conversation_log = []
        return self.working.to_prompt()

    def build_memory_context(self, user_message: str) -> str:
        """Build the full memory context to inject into Claude's system prompt.

        Combines all three tiers into a single formatted string.
        """
        if not self.working:
            raise RuntimeError("No active session. Call start_session() first.")

        sections = [
            self.working.to_prompt(),
            self.episodic.to_prompt(user_message, n_results=3),
            self.procedural.to_prompt(user_message),
        ]
        return "\n\n".join(sections)

    def log_turn(self, role: str, content: str) -> None:
        """Log a conversation turn for later summarization."""
        self.conversation_log.append({"role": role, "content": content})

    async def end_session(self) -> str | None:
        """End the session: summarize, store episode, clear working memory.

        Returns the episode ID if a summary was stored, None otherwise.
        """
        if not self.working or not self.conversation_log:
            return None

        session_id = self.working.session_id
        user_id = self.working.get("user_id", "default")

        # Summarize the conversation
        try:
            summary_record = await summarize_conversation(
                self.conversation_log, self.client
            )
        except Exception as e:
            print(f"Warning: summarization failed: {e}")
            summary_record = {
                "summary": f"Session {session_id} — summarization failed",
                "topics": [],
            }

        # Store in episodic memory
        episode_id = self.episodic.store_episode(
            summary=summary_record["summary"],
            session_id=session_id,
            user_id=user_id,
            topics=summary_record.get("topics", []),
        )

        # Clear working memory
        self.working.clear()
        self.working = None
        self.conversation_log = []

        return episode_id


# --- Usage: Full agent loop ---
async def agent_loop():
    """Demonstrates a full conversation with memory management."""
    manager = MemoryManager(persist_dir="./memory_db")
    client = anthropic.AsyncAnthropic()

    # Start session
    session_id = "sess_044"
    manager.start_session(session_id, user_id="user_alice")

    # Simulate a user message
    user_msg = "What did we decide about the deployment schedule?"
    manager.log_turn("user", user_msg)

    # Build memory-enriched system prompt
    memory_context = manager.build_memory_context(user_msg)
    manager.working.set("intent", "recall_decision")
    manager.working.set("topic", "deployment schedule")

    system_prompt = f"""You are a helpful assistant with memory.

{memory_context}

Use the memory context above to provide informed, personalized responses.
If you find relevant past interactions, reference them naturally."""

    # Call Claude with memory context
    response = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=system_prompt,
        messages=[{"role": "user", "content": user_msg}],
    )

    assistant_msg = response.content[0].text
    manager.log_turn("assistant", assistant_msg)
    print(f"Agent: {assistant_msg}")

    # End session — summarize and persist
    episode_id = await manager.end_session()
    print(f"Session saved as episode: {episode_id}")

mkdir memory-lab && cd memory-lab
python -m venv venv && source venv/bin/activate   # Windows: venv\Scripts\activate
pip install "anthropic>=0.30.0" "chromadb>=0.4.0"
export ANTHROPIC_API_KEY=your-key-here             # Windows: set ANTHROPIC_API_KEY=your-key-here

import json
import time
import uuid
import chromadb
import anthropic

client = anthropic.Anthropic()

# ── Tier 1: Working Memory (key-value scratchpad) ───────────
class WorkingMemory:
    """Fast, mutable scratchpad for current task state."""

    def __init__(self):
        self._store: dict = {}
        self.session_id = f"sess_{uuid.uuid4().hex[:8]}"
        self.created_at = time.time()

    def set(self, key: str, value) -> None:
        self._store[key] = value

    def get(self, key: str, default=None):
        return self._store.get(key, default)

    def delete(self, key: str) -> None:
        self._store.pop(key, None)

    def clear(self) -> None:
        self._store.clear()

    def to_prompt(self) -> str:
        if not self._store:
            return "[Working Memory: empty]"
        lines = [f"[Working Memory — Session {self.session_id}]"]
        for k, v in self._store.items():
            lines.append(f"  {k}: {v}")
        return "\n".join(lines)

    def to_dict(self) -> dict:
        return dict(self._store)


# ── Tier 2: Episodic Memory (past interactions via ChromaDB) ─
class EpisodicMemory:
    """Searchable archive of past conversation summaries."""

    def __init__(self, persist_dir: str = "./memory_db"):
        self._client = chromadb.PersistentClient(path=persist_dir)
        self._collection = self._client.get_or_create_collection(
            name="episodes", metadata={"hnsw:space": "cosine"}
        )

    def store_episode(self, summary: str, session_id: str, metadata: dict = None) -> str:
        episode_id = f"ep_{uuid.uuid4().hex[:12]}"
        meta = {
            "session_id": session_id,
            "timestamp": time.time(),
            **(metadata or {}),
        }
        self._collection.add(
            ids=[episode_id],
            documents=[summary],
            metadatas=[meta],
        )
        return episode_id

    def recall(self, query: str, top_k: int = 3) -> list[dict]:
        if self._collection.count() == 0:
            return []
        results = self._collection.query(query_texts=[query], n_results=min(top_k, self._collection.count()))
        episodes = []
        for doc, meta in zip(results["documents"][0], results["metadatas"][0]):
            episodes.append({"summary": doc, **meta})
        return episodes

    def to_prompt(self, query: str) -> str:
        episodes = self.recall(query, top_k=2)
        if not episodes:
            return "[Past Interactions: none yet]"
        lines = ["[Relevant Past Interactions]"]
        for ep in episodes:
            ts = time.strftime("%Y-%m-%d", time.localtime(ep.get("timestamp", 0)))
            lines.append(f"  - Session {ep.get('session_id', '?')} ({ts}): {ep['summary']}")
        return "\n".join(lines)

    @property
    def count(self) -> int:
        return self._collection.count()


# ── Tier 3: Procedural Memory (reusable action templates) ────
class ProceduralMemory:
    """Library of proven tool-call sequences."""

    def __init__(self):
        self._procedures: dict[str, dict] = {}

    def store_procedure(self, name: str, description: str, steps: list[str]) -> None:
        self._procedures[name] = {
            "description": description,
            "steps": steps,
            "usage_count": 0,
        }

    def find_procedure(self, query: str) -> dict | None:
        """Simple keyword matching (production: use embeddings)."""
        query_lower = query.lower()
        best_match, best_score = None, 0
        for name, proc in self._procedures.items():
            score = sum(1 for word in query_lower.split()
                       if word in name.lower() or word in proc["description"].lower())
            if score > best_score:
                best_match, best_score = (name, proc), score
        if best_match and best_score > 0:
            best_match[1]["usage_count"] += 1
            return {"name": best_match[0], **best_match[1]}
        return None

    def to_prompt(self, query: str) -> str:
        proc = self.find_procedure(query)
        if not proc:
            return "[Procedures: no matching template]"
        steps_str = "\n".join(f"    Step {i+1}: {s}" for i, s in enumerate(proc["steps"]))
        return f"[Suggested Procedure: {proc['name']} (used {proc['usage_count']}x)]\n{steps_str}"


# ── MemoryManager (wires all 3 tiers together) ───────────────
class MemoryManager:
    """Orchestrates all three memory tiers."""

    def __init__(self, persist_dir: str = "./memory_db"):
        self.working = WorkingMemory()
        self.episodic = EpisodicMemory(persist_dir=persist_dir)
        self.procedural = ProceduralMemory()
        self.conversation_log: list[dict] = []

    def build_context(self, current_query: str) -> str:
        """Build combined memory context for injection into the LLM prompt."""
        parts = [
            self.working.to_prompt(),
            self.episodic.to_prompt(current_query),
            self.procedural.to_prompt(current_query),
        ]
        return "\n\n".join(parts)

    def log_turn(self, role: str, content: str) -> None:
        self.conversation_log.append({"role": role, "content": content})

    def end_session(self, summary: str = None) -> str:
        """Summarize and archive the session to episodic memory."""
        if not summary and self.conversation_log:
            # Auto-summarize with Claude
            transcript = "\n".join(f"{m['role']}: {m['content']}" for m in self.conversation_log)
            try:
                response = client.messages.create(
                    model="claude-sonnet-4-6", max_tokens=300,
                    system="Summarize this conversation in 2-3 sentences. Preserve key decisions, preferences, and facts.",
                    messages=[{"role": "user", "content": transcript}],
                )
                summary = response.content[0].text
            except Exception:
                summary = f"Session with {len(self.conversation_log)} turns."

        episode_id = self.episodic.store_episode(
            summary=summary,
            session_id=self.working.session_id,
            metadata={"turns": len(self.conversation_log)},
        )
        # Reset for next session
        self.working.clear()
        self.working = WorkingMemory()  # new session_id
        self.conversation_log = []
        return episode_id

    def chat(self, user_message: str) -> str:
        """Send a message with full memory context."""
        self.log_turn("user", user_message)
        memory_context = self.build_context(user_message)

        response = client.messages.create(
            model="claude-sonnet-4-6", max_tokens=1024,
            system=(
                "You are a helpful assistant with multi-session memory. "
                "Use the memory context below to inform your responses.\n\n"
                f"{memory_context}"
            ),
            messages=[{"role": m["role"], "content": m["content"]}
                      for m in self.conversation_log],
        )
        reply = response.content[0].text
        self.log_turn("assistant", reply)
        return reply


# ── Tests ────────────────────────────────────────────────────
if __name__ == "__main__":
    import shutil
    # Clean up any previous test data
    shutil.rmtree("./memory_db", ignore_errors=True)

    print("═" * 55)
    print("TEST 1: Individual Memory Tiers")
    print("═" * 55)

    # Working Memory
    wm = WorkingMemory()
    wm.set("intent", "schedule_meeting")
    wm.set("date", "2026-04-05")
    wm.set("attendees", ["Alice", "Bob"])
    print("\n  Working Memory:")
    print("  " + wm.to_prompt().replace("\n", "\n  "))

    # Episodic Memory
    em = EpisodicMemory(persist_dir="./memory_db")
    em.store_episode("User prefers email over Slack for notifications.", "sess_001")
    em.store_episode("Discussed deployment schedule. Decision: deploy Friday.", "sess_002")
    em.store_episode("Reviewed API rate limits. Set max to 1000 req/min.", "sess_003")
    print(f"\n  Episodic Memory: {em.count} episodes stored")
    print("  " + em.to_prompt("deployment schedule").replace("\n", "\n  "))

    # Procedural Memory
    pm = ProceduralMemory()
    pm.store_procedure("generate_report", "Generate a formatted report from data",
                       ["Gather data from sources", "Analyze key metrics", "Format as markdown", "Send to user"])
    pm.store_procedure("debug_api_error", "Debug and fix API errors",
                       ["Check error code and message", "Search logs for context", "Identify root cause", "Suggest fix"])
    print(f"\n  Procedural Memory:")
    print("  " + pm.to_prompt("generate a report").replace("\n", "\n  "))

    print(f"\n{'═' * 55}")
    print("TEST 2: Cross-Session Memory (with Claude)")
    print("═" * 55)

    # Session 1
    mm = MemoryManager(persist_dir="./memory_db")
    mm.working.set("intent", "setup_preferences")

    print("\n  Session 1:")
    reply1 = mm.chat("Hi! I prefer getting notifications via email, not Slack.")
    print(f"    Turn 1: {reply1[:120]}...")
    reply2 = mm.chat("Also, let's plan to deploy the new API on Friday.")
    print(f"    Turn 2: {reply2[:120]}...")

    ep_id = mm.end_session()
    print(f"    Session ended → episode {ep_id}")

    # Session 2 — should remember preferences from Session 1
    print("\n  Session 2:")
    reply3 = mm.chat("Do you remember my notification preference?")
    print(f"    Turn 1: {reply3[:150]}...")
    print(f"\n  Memory context that was injected:")
    print("    " + mm.build_context("notification preference").replace("\n", "\n    "))

    # Cleanup
    shutil.rmtree("./memory_db", ignore_errors=True)
    print(f"\n{'═' * 55}")
    print("✓ All tests complete.")
    print("═" * 55)