fix 3 layer memory ststem

README.md

@@ -18,6 +18,7 @@ hf_oauth_expiration_minutes: 480
 ## Features
 
 - **Explicit ReAct graph** — tool-call dedup, per-tool error feedback, recursion cap, iteration fail-safe
+- **Three-layer persistent memory** — short-term thread checkpoints, long-term semantic facts (LangMem), episodic task experiences; inspired by the Engram memory architecture
 - **Tool belt** — web search, URL fetch, sandboxed Python, file I/O, PDF, audio/video transcription, YouTube frame extraction, vision (Gemini + FAL fallbacks), arXiv, CrossRef, todos
 - **Multi-provider routing** — cheap / strong / extra-strong model tiers with independent provider+model config
 - **Observability** — per-session JSONL trace + rotating log file, optional Arize AX + LangSmith tracing
@@ -88,6 +89,9 @@ The TUI prints the logo, caveman status, and the trace file path. Type your ques
 | Command | Effect |
 | --- | --- |
 | `/clear` | Wipe conversation memory, start a new thread |
+| `/memory list` | Show all stored facts and recent episodic experiences |
+| `/memory forget <id>` | Delete a fact by ID prefix |
+| `/memory reflect` | Manually trigger long-term memory extraction for the current thread |
 | `/caveman` | Toggle caveman on/off |
 | `/caveman off` / `/caveman on` | Explicit on/off |
 | `/caveman lite` | Lightest — keep articles & full sentences, cut fluff |
@@ -136,7 +140,8 @@ All tools live under [src/lilith_agent/tools/](src/lilith_agent/tools/) and are
 | `inspect_pdf` | PDF → text |
 | `inspect_visual_content` | Multimodal vision (Gemini + FAL moondream/llava fallbacks) |
 | `arxiv_search`, `crossref_search`, `count_journal_articles`, `filter_entities` | Academic metadata |
-| `write_todos`, `mark_todo_done` | High-level planning |
+| `todos` | High-level planning |
+| `search_memory` | Query Lilith's long-term memory (facts + episodes) by keyword |
 
 ### Vision fallback chain
 
@@ -156,6 +161,7 @@ src/lilith_agent/
   app.py             # ReAct graph, model routing, caveman prompt wrapping
   tui.py             # interactive loop, slash commands, rich output
   runner.py          # batch runner over GAIA questions
+  memory.py          # three-layer memory: checkpoints, semantic facts, episodic
   config.py          # Config.from_env(), model + API key + feature flags
   observability.py   # logging, Arize setup, JsonlTraceCallback
   models.py          # provider -> chat model builder
@@ -164,11 +170,37 @@ src/lilith_agent/
 scripts/
   dev_run_gaia.py    # CLI to run against real GAIA questions
 .checkpoints/        # per-question answers (gitignored)
-.lilith/             # session logs + JSONL traces (gitignored)
+.lilith/             # session logs, JSONL traces, long_term_memory.sqlite (gitignored)
 ```
 
+## Memory system
+
+Lilith uses a three-layer persistent memory architecture loosely inspired by the [Engram memory model](https://arxiv.org/abs/2501.12599):
+
+| Layer | Storage | Role |
+| --- | --- | --- |
+| **Short-term** (thread checkpoints) | `.lilith/threads.sqlite` via LangGraph `SqliteSaver` | Preserves full conversation state across restarts within a thread |
+| **Long-term semantic** (facts) | `.lilith/long_term_memory.sqlite` | Extracts and deduplicates user preferences, names, project details using LangMem; injected into the system prompt on new queries |
+| **Episodic** (task experiences) | `.lilith/long_term_memory.sqlite` | Summarises past tool trajectories — what failed, what worked — so Lilith avoids repeating mistakes |
+
+The semantic layer uses LangMem as a memory governance engine (extraction, conflict resolution, forgetting) while SQLite provides local, auditable, migratable persistence. Long-term memory extraction runs automatically after each conversation and can be triggered manually with `/memory reflect`. The agent can also call `search_memory` during reasoning when the system-prompt injection has been truncated.
+
+For batch GAIA runs each question gets an isolated ephemeral memory store (`MemoryStore(":memory:")`) so questions cannot contaminate each other.
+
 ## Testing
 
 ```bash
 pytest
 ```
+
+Memory tests can use the `ephemeral_memory()` context manager for isolated in-memory stores:
+
+```python
+from lilith_agent.memory import ephemeral_memory
+
+def test_something():
+    with ephemeral_memory() as store:
+        store.add_episode("task", "summary", "success")
+        assert len(store.get_recent_episodes()) == 1
+    # store discarded on exit, no disk writes
+```

requirements.txt

@@ -3,6 +3,7 @@ requests
 httpx
 pandas
 langgraph>=0.2.0
+langmem>=0.0.1
 langchain-core>=0.3.0
 langchain-anthropic>=0.2.0
 langchain-google-genai>=2.0.0

src/lilith_agent/app.py

@@ -4,6 +4,7 @@ import json
 import logging
 import re
 import string
+import ast
 from typing import Callable
 
 from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
@@ -19,6 +20,7 @@ from typing import Annotated, TypedDict
 class AgentState(TypedDict):
     messages: Annotated[list, add_messages]
     iterations: int
+    todos: list[str]
 
 
 from lilith_agent.config import Config
@@ -292,6 +294,7 @@ def _build_tool_node(
         messages = state["messages"]
         last = messages[-1]
         tool_calls = getattr(last, "tool_calls", None) or []
+        todo_state_update = None
         if tool_calls:
             log_tools.info(
                 "[tools] dispatching %d call(s): %s",
@@ -410,13 +413,31 @@ def _build_tool_node(
                 continue
 
             out_str = str(out)
+            if out_str.startswith("SET_TODOS:"):
+                try:
+                    parsed = ast.literal_eval(out_str[len("SET_TODOS:"):].strip())
+                    if isinstance(parsed, list):
+                        todo_state_update = [str(item) for item in parsed]
+                except Exception:
+                    pass
+            elif out_str.startswith("DONE_TODO:"):
+                try:
+                    idx = int(out_str[len("DONE_TODO:"):].strip())
+                    current = list(state.get("todos", []))
+                    if 0 <= idx < len(current):
+                        todo_state_update = current[:idx] + current[idx + 1:]
+                except Exception:
+                    pass
             preview = out_str.replace("\n", " ")
             if len(preview) > _TOOL_RESULT_PREVIEW_CHARS:
                 preview = preview[:_TOOL_RESULT_PREVIEW_CHARS] + "…"
             log_tools.info("[tools] tool result (%d chars): %s", len(out_str), preview)
             results.append(ToolMessage(tool_call_id=tc_id, name=name, content=out_str))
 
-        return {"messages": results}
+        update = {"messages": results}
+        if todo_state_update is not None:
+            update["todos"] = todo_state_update
+        return update
 
     return tool_node
 
@@ -592,10 +613,14 @@ def build_react_agent(cfg: Config):
         return {"messages": [response]}
 
     def extract_memory_node(state):
-        from lilith_agent.memory import extract_and_compress_facts
+        from lilith_agent.memory import extract_and_compress_facts, MIN_MESSAGES_FOR_EXTRACTION
+        messages = state["messages"]
+        if len(messages) < MIN_MESSAGES_FOR_EXTRACTION:
+            log.debug("[memory] skipping extraction: only %d messages", len(messages))
+            return state
         try:
             cheap_model = get_cheap_model(cfg)
-            extract_and_compress_facts(state["messages"], cheap_model)
+            extract_and_compress_facts(messages, cheap_model)
         except Exception as e:
             log.warning("[memory] failed to run extraction: %s", e)
         return state

src/lilith_agent/memory.py

@@ -2,8 +2,10 @@ import os
 import json
 import sqlite3
 import logging
+import uuid
+from contextlib import contextmanager
 from pathlib import Path
-from typing import List, Dict, Any
+from typing import List, Dict, Any, Optional, Union
 from datetime import datetime
 from langchain_core.messages import BaseMessage, AIMessage, HumanMessage
 
@@ -12,6 +14,8 @@ log = logging.getLogger(__name__)
 # Constants
 LILITH_HOME = Path(os.getenv("LILITH_HOME", ".lilith"))
 MEMORY_DB_PATH = LILITH_HOME / "long_term_memory.sqlite"
+MEMORY_CONTEXT_CHAR_BUDGET = 3000
+MIN_MESSAGES_FOR_EXTRACTION = 2
 
 def _content_to_text(content: Any) -> str:
     if content is None:
@@ -42,13 +46,28 @@ def _memory_content_to_text(content: Any) -> str:
     return _content_to_text(content)
 
 class MemoryStore:
-    def __init__(self, db_path: Path = MEMORY_DB_PATH):
-        self.db_path = db_path
+    def __init__(self, db_path: Union[Path, str] = MEMORY_DB_PATH):
+        self._in_memory = (str(db_path) == ":memory:")
+        self.db_path = db_path if self._in_memory else Path(db_path)
+        self._mem_conn: Optional[sqlite3.Connection] = None
         self._init_db()
 
+    def _connect(self) -> sqlite3.Connection:
+        if self._in_memory:
+            if self._mem_conn is None:
+                self._mem_conn = sqlite3.connect(":memory:", check_same_thread=False)
+            return self._mem_conn
+        return sqlite3.connect(str(self.db_path))
+
+    def close(self):
+        if self._mem_conn is not None:
+            self._mem_conn.close()
+            self._mem_conn = None
+
     def _init_db(self):
-        self.db_path.parent.mkdir(parents=True, exist_ok=True)
-        conn = sqlite3.connect(str(self.db_path))
+        if not self._in_memory:
+            Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
+        conn = self._connect()
         with conn:
             conn.execute("""
                 CREATE TABLE IF NOT EXISTS memories (
@@ -68,62 +87,130 @@ class MemoryStore:
                     created_at TEXT NOT NULL
                 )
             """)
-        conn.close()
+        if not self._in_memory:
+            conn.close()
 
     def get_all_memories(self) -> List[Dict[str, Any]]:
-        conn = sqlite3.connect(str(self.db_path))
+        conn = self._connect()
         conn.row_factory = sqlite3.Row
         cur = conn.cursor()
         cur.execute("SELECT * FROM memories ORDER BY updated_at DESC")
         rows = [dict(r) for r in cur.fetchall()]
-        conn.close()
+        if not self._in_memory:
+            conn.close()
         return rows
 
-    def save_memories(self, memories: List[Dict[str, Any]]):
+    def save_memories(self, memories: List[Dict[str, Any]], allow_empty: bool = False):
         """Replaces the memories table with the provided list (active compression)."""
-        conn = sqlite3.connect(str(self.db_path))
+        if not memories:
+            existing = self.get_all_memories()
+            if existing and not allow_empty:
+                log.warning("[memory] save_memories called with empty list while %d facts exist — refusing to wipe",
+                            len(existing))
+                return
+        conn = self._connect()
         with conn:
             conn.execute("DELETE FROM memories")
             for m in memories:
                 now = datetime.now().isoformat()
+                created_at = m.get("created_at", now)
+                updated_at = m.get("updated_at", now)
                 conn.execute(
                     "INSERT INTO memories (id, content, type, created_at, updated_at) VALUES (?, ?, ?, ?, ?)",
-                    (m.get("id", str(hash(m["content"]))), m["content"], m.get("type", "fact"), now, now)
+                    (m.get("id", str(uuid.uuid4())), m["content"], m.get("type", "fact"), created_at, updated_at)
                 )
-        conn.close()
+        if not self._in_memory:
+            conn.close()
+
+    def delete_memory_prefix(self, prefix: str) -> int:
+        if not prefix:
+            return 0
+        matching_ids = [
+            row["id"]
+            for row in self.get_all_memories()
+            if row["id"].startswith(prefix)
+        ]
+        if not matching_ids:
+            return 0
+        conn = self._connect()
+        with conn:
+            for memory_id in matching_ids:
+                conn.execute("DELETE FROM memories WHERE id = ?", (memory_id,))
+        if not self._in_memory:
+            conn.close()
+        return len(matching_ids)
 
     def add_episode(self, task: str, summary: str, outcome: str):
-        conn = sqlite3.connect(str(self.db_path))
+        conn = self._connect()
         with conn:
             now = datetime.now().isoformat()
             conn.execute(
                 "INSERT INTO episodes (id, task, summary, outcome, created_at) VALUES (?, ?, ?, ?, ?)",
-                (str(hash(task + now)), task, summary, outcome, now)
+                (str(uuid.uuid4()), task, summary, outcome, now)
             )
-        conn.close()
+        if not self._in_memory:
+            conn.close()
 
     def get_recent_episodes(self, limit: int = 3) -> List[Dict[str, Any]]:
-        conn = sqlite3.connect(str(self.db_path))
+        conn = self._connect()
         conn.row_factory = sqlite3.Row
         cur = conn.cursor()
         cur.execute("SELECT * FROM episodes ORDER BY created_at DESC LIMIT ?", (limit,))
         rows = [dict(r) for r in cur.fetchall()]
-        conn.close()
+        if not self._in_memory:
+            conn.close()
         return rows
 
 _store = MemoryStore()
 
+
+def _set_store(store: MemoryStore) -> MemoryStore:
+    """Swap the module-level store and return the previous one."""
+    global _store
+    prev, _store = _store, store
+    return prev
+
+
+@contextmanager
+def ephemeral_memory(db_path: Union[str, Path] = ":memory:"):
+    """Context manager that replaces the global _store with a fresh, isolated
+    MemoryStore for the duration of the block. On exit the store is closed and
+    the previous store is restored. Use db_path=':memory:' (default) for tests
+    or GAIA benchmarking where cross-contamination must be prevented.
+
+    Example::
+
+        with ephemeral_memory():
+            result = graph.invoke(state, config)
+            # any memory writes here are discarded on exit
+    """
+    fresh = MemoryStore(db_path)
+    prev = _set_store(fresh)
+    try:
+        yield fresh
+    finally:
+        _set_store(prev)
+        fresh.close()
+
+
+def _is_remove_doc(content: Any) -> bool:
+    return hasattr(content, "__repr_name__") and content.__repr_name__() == "RemoveDoc"
+
+
 def extract_and_compress_facts(messages: List[BaseMessage], model) -> None:
     """
     Extracts new facts and merges them with existing ones using langmem's manager.
     Implements professional reflection and conflict resolution.
     """
     from langmem import create_memory_manager
+    from langmem.knowledge.extraction import Memory
     log.info("[memory] Running langmem memory management...")
     try:
         # 1. Get existing memories from our local store
         existing_rows = _store.get_all_memories()
-        existing_memories = [m["content"] for m in existing_rows]
+        existing_memories = [(m["id"], Memory(content=m["content"])) for m in existing_rows]
+        existing_by_content = {m["content"]: m for m in existing_rows}
+        existing_by_id = {m["id"]: m for m in existing_rows}
 
         # 2. Initialize langmem manager (it's a Runnable)
         # We use the default schema which is basically content strings
@@ -141,14 +228,44 @@ def extract_and_compress_facts(messages: List[BaseMessage], model) -> None:
             # The result is a list of ExtractedMemory objects or tuples depending on version
             # Usually it's (id, content) or just objects. We'll be robust here.
             updated_facts = []
+            removed_ids = set()
             for item in result:
-                # result elements can be ExtractedMemory objects or simple dicts
-                content = getattr(item, "content", None) or (item[1] if isinstance(item, tuple) else item.get("content"))
+                item_id = getattr(item, "id", None)
+                content = getattr(item, "content", None)
+                if isinstance(item, tuple):
+                    if len(item) > 0:
+                        item_id = item[0]
+                    if len(item) > 1 and content is None:
+                        content = item[1]
+                elif isinstance(item, dict):
+                    item_id = item.get("id", item_id)
+                    content = item.get("content", content)
+                stable_id = str(item_id) if item_id else None
+                if _is_remove_doc(content):
+                    if stable_id:
+                        removed_ids.add(stable_id)
+                    continue
                 if content:
-                    updated_facts.append({"content": _memory_content_to_text(content)})
+                    text = _memory_content_to_text(content)
+                    existing = existing_by_id.get(stable_id) if stable_id else existing_by_content.get(text)
+                    fact = {"content": text}
+                    if stable_id:
+                        fact["id"] = stable_id
+                    elif existing:
+                        fact["id"] = existing["id"]
+                    if existing:
+                        fact["created_at"] = existing["created_at"]
+                        if existing["content"] == text:
+                            fact["updated_at"] = existing["updated_at"]
+                    if fact.get("id") not in removed_ids:
+                        updated_facts.append(fact)
             
-            _store.save_memories(updated_facts)
-            log.info(f"[memory] langmem updated store to {len(updated_facts)} facts.")
+            if updated_facts or removed_ids:
+                _store.save_memories(updated_facts, allow_empty=bool(removed_ids))
+                log.info(f"[memory] langmem updated store to {len(updated_facts)} facts.")
+            else:
+                log.info("[memory] langmem returned empty result — keeping existing facts")
+        summarize_episode(messages, model)
 
     except Exception:
         log.exception("[memory] langmem extraction failed")
@@ -193,22 +310,94 @@ Keep it under 150 words.
     except Exception:
         log.exception("[memory] Summarization failed")
 
-def retrieve_relevant_context(query: str) -> str:
-    """Fetches all facts and recent episodes to inject into the prompt."""
+def _relevance_score(text: str, query_tokens: set) -> float:
+    """Simple word-overlap score between a fact and the query (Jaccard-like)."""
+    if not query_tokens:
+        return 0.0
+    fact_tokens = set(text.lower().split())
+    return len(fact_tokens & query_tokens) / (len(fact_tokens | query_tokens) or 1)
+
+
+def retrieve_relevant_context(query: str, char_budget: int = MEMORY_CONTEXT_CHAR_BUDGET) -> str:
+    """Fetches facts and recent episodes ranked by relevance, capped by char_budget.
+    If truncated, appends a note instructing the agent to call search_memory for more."""
     try:
+        query_tokens = set(query.lower().split())
         facts = _store.get_all_memories()
-        episodes = _store.get_recent_episodes(limit=2)
-        
+        episodes = _store.get_recent_episodes(limit=5)
+
+        facts.sort(key=lambda m: _relevance_score(m["content"], query_tokens), reverse=True)
+
         context_parts = []
+        budget_remaining = char_budget
+
         if facts:
-            fact_lines = "\n".join([f"- {m['content']}" for m in facts])
-            context_parts.append(f"<known_facts>\n{fact_lines}\n</known_facts>")
-        
-        if episodes:
-            epi_lines = "\n\n".join([f"Task: {e['task']}\nSummary: {e['summary']}" for e in episodes])
-            context_parts.append(f"<past_experiences>\n{epi_lines}\n</past_experiences>")
-            
+            included, total = [], len(facts)
+            for m in facts:
+                line = f"- {m['content']}"
+                if budget_remaining - len(line) - 1 > 0:
+                    included.append(line)
+                    budget_remaining -= len(line) + 1
+                else:
+                    break
+            fact_block = "<known_facts>\n" + "\n".join(included) + "\n</known_facts>"
+            omitted = total - len(included)
+            if omitted:
+                fact_block += f"\n<!-- {omitted} fact(s) omitted (budget). Call search_memory(query) to retrieve more. -->"
+            context_parts.append(fact_block)
+
+        if episodes and budget_remaining > 0:
+            included_ep = []
+            for e in episodes:
+                line = f"Task: {e['task']}\nSummary: {e['summary']}"
+                if budget_remaining - len(line) - 2 > 0:
+                    included_ep.append(line)
+                    budget_remaining -= len(line) + 2
+                else:
+                    break
+            if included_ep:
+                ep_block = "<past_experiences>\n" + "\n\n".join(included_ep) + "\n</past_experiences>"
+                omitted_ep = len(episodes) - len(included_ep)
+                if omitted_ep:
+                    ep_block += f"\n<!-- {omitted_ep} episode(s) omitted. Call search_memory(query) for more. -->"
+                context_parts.append(ep_block)
+
         return "\n\n".join(context_parts)
     except Exception:
         log.exception("[memory] Retrieval failed")
         return ""
+
+
+def search_memory_store(query: str, max_results: int = 10) -> str:
+    """Keyword search across all facts and episodes. Called by the agent when
+    the system-prompt injection was truncated or the query needs deeper lookup."""
+    try:
+        query_tokens = set(query.lower().split())
+        facts = _store.get_all_memories()
+        episodes = _store.get_recent_episodes(limit=50)
+
+        scored_facts = sorted(
+            [(f, _relevance_score(f["content"], query_tokens)) for f in facts],
+            key=lambda x: x[1], reverse=True
+        )
+        scored_eps = sorted(
+            [(e, _relevance_score(e["task"] + " " + e["summary"], query_tokens)) for e in episodes],
+            key=lambda x: x[1], reverse=True
+        )
+
+        results = []
+        for m, score in scored_facts[:max_results]:
+            if score > 0:
+                results.append((score, f"[fact] {m['content']}"))
+        for e, score in scored_eps[:max_results]:
+            if score > 0:
+                results.append((score, f"[episode] Task: {e['task']}\n  Summary: {e['summary']}"))
+
+        results.sort(key=lambda x: x[0], reverse=True)
+        parts = [text for _, text in results[:max_results]]
+        if not parts:
+            return "No matching memories found."
+        return "\n\n".join(parts)
+    except Exception:
+        log.exception("[memory] search_memory_store failed")
+        return "Memory search failed."

src/lilith_agent/runner.py

@@ -205,8 +205,10 @@ def run_agent_on_questions(graph: Any, questions: list[dict], checkpoint_dir: st
             "iterations": 0
         }
 
+        from lilith_agent.memory import ephemeral_memory
         try:
-            result = graph.invoke(state, {"configurable": {"thread_id": task_id}})
+            with ephemeral_memory():
+                result = graph.invoke(state, {"configurable": {"thread_id": task_id}})
         except Exception as exc:
             log_runner.warning("[runner] task=%s agent error: %s", task_id, exc)
             answers.append(

src/lilith_agent/tools/__init__.py

@@ -14,11 +14,7 @@ from lilith_agent.tools.files import (
     find_files as _find_files,
     write_file as _write_file
 )
-from lilith_agent.tools.todos import (
-    write_todos as _write_todos,
-    mark_todo_done as _mark_todo_done,
-    read_todos as _read_todos
-)
+from lilith_agent.tools.todos import todos as _todos
 from lilith_agent.tools.pdf import inspect_pdf as _inspect_pdf
 from lilith_agent.tools.python_exec import run_python as _run_python
 from lilith_agent.tools.search import web_search as _web_search
@@ -117,14 +113,13 @@ def build_tools(cfg: Config) -> list[BaseTool]:
         return _write_file(path, content)
 
     @tool
-    def write_todos(todos: list[str]) -> str:
-        """Initialize or overwrite the current task list (Todo list). Use for high-level planning."""
-        return _write_todos(todos)
+    def todos(action: str, items: Optional[list[str]] = None, index: Optional[int] = None) -> str:
+        """Manage the planning todo list.
 
-    @tool
-    def mark_todo_done(index: int) -> str:
-        """Mark a specific todo as complete by its position."""
-        return _mark_todo_done(index)
+        action='write': overwrite the list. Requires `items` (list of strings).
+        action='done':  mark a todo complete. Requires `index` (0-based position).
+        """
+        return _todos(action, items=items, index=index)
 
     @tool
     def transcribe_audio(path: str) -> str:
@@ -184,6 +179,15 @@ def build_tools(cfg: Config) -> list[BaseTool]:
         """
         return _filter_entities(entities, keep_conditions=keep_conditions, remove_conditions=remove_conditions)
 
+    @tool
+    def search_memory(query: str) -> str:
+        """Search Lilith's long-term memory (semantic facts and episodic experiences) for information
+        matching the given query. Use this when the system context mentions omitted facts, or when
+        you need to recall specific past experiences, user preferences, or project details that may
+        not be in the current context window."""
+        from lilith_agent.memory import search_memory_store
+        return search_memory_store(query)
+
     return [
         web_search,
         fetch_url,
@@ -203,6 +207,6 @@ def build_tools(cfg: Config) -> list[BaseTool]:
         glob_files,
         find_files,
         write_file,
-        write_todos,
-        mark_todo_done,
+        todos,
+        search_memory,
     ]

src/lilith_agent/tools/todos.py

@@ -1,29 +1,19 @@
 from __future__ import annotations
-from typing import List
+from typing import List, Optional
 
-def write_todos(todos: List[str]) -> str:
-    """
-    Initialize or overwrite the current task list (Todo list).
-    Used for high-level planning and tracking progress.
-    """
-    # This tool will be handled specially by the executor node to update AgentState
-    return f"SET_TODOS: {todos}"
 
-def mark_todo_done(index: int) -> str:
-    """
-    Mark a specific todo as complete by its 0-indexed position.
-    """
-    # This tool will be handled specially by the executor node to update AgentState
-    return f"DONE_TODO: {index}"
-
-def read_todos(todo_list: List[str]) -> str:
-    """
-    Read the current state of the todo list.
-    """
-    if not todo_list:
-        return "Todo list is empty."
-    
-    lines = []
-    for i, todo in enumerate(todo_list):
-        lines.append(f"{i}. {todo}")
-    return "\n".join(lines)
+def todos(
+    action: str,
+    items: Optional[List[str]] = None,
+    index: Optional[int] = None,
+) -> str:
+    """Branching todo tool. Executor node consumes SET_TODOS / DONE_TODO sentinels."""
+    if action == "write":
+        if items is None:
+            return "ERROR: action='write' requires `items` (list of strings)."
+        return f"SET_TODOS: {items}"
+    if action == "done":
+        if index is None:
+            return "ERROR: action='done' requires `index` (0-based position)."
+        return f"DONE_TODO: {index}"
+    return f"ERROR: unknown action '{action}'. Use 'write' or 'done'."

src/lilith_agent/tui.py

@@ -123,7 +123,55 @@ def main_loop(cfg):
         if text.lower() in ("exit", "quit"):
             console.print("[magenta]Goodbye! 🦋[/magenta]")
             break
-            
+
+        if text.lower().startswith("/memory"):
+            from lilith_agent.memory import _store, extract_and_compress_facts
+            from lilith_agent.models import get_cheap_model
+            parts = text.split(maxsplit=1)
+            sub = parts[1].strip() if len(parts) > 1 else "list"
+
+            if sub == "list":
+                facts = _store.get_all_memories()
+                episodes = _store.get_recent_episodes(limit=5)
+                if facts:
+                    console.print("\n[bold cyan]── Semantic Facts ──[/bold cyan]")
+                    for m in facts:
+                        console.print(f"  [dim]{m['id'][:8]}[/dim]  {m['content']}")
+                else:
+                    console.print("[dim]No semantic facts stored.[/dim]")
+                if episodes:
+                    console.print("\n[bold cyan]── Episodic Memory ──[/bold cyan]")
+                    for e in episodes:
+                        console.print(f"  [dim]{e['id'][:8]}[/dim]  [bold]{e['task'][:60]}[/bold]\n           {e['summary'][:120]}...")
+                else:
+                    console.print("[dim]No episodes stored.[/dim]")
+                console.print()
+
+            elif sub.startswith("forget "):
+                target_id = sub[len("forget "):].strip()
+                deleted = _store.delete_memory_prefix(target_id)
+                if deleted:
+                    console.print(f"[dim cyan]Deleted {deleted} fact(s) matching '{target_id}'.[/dim cyan]\n")
+                else:
+                    console.print(f"[yellow]No fact found with id starting with '{target_id}'.[/yellow]\n")
+
+            elif sub == "reflect":
+                console.print("[dim cyan]Running memory reflection...[/dim cyan]")
+                try:
+                    cheap_model = get_cheap_model(cfg)
+                    state = graph.get_state(thread_config)
+                    msgs = state.values.get("messages", []) if state and state.values else []
+                    if msgs:
+                        extract_and_compress_facts(msgs, cheap_model)
+                        console.print("[dim cyan]Reflection complete.[/dim cyan]\n")
+                    else:
+                        console.print("[yellow]No messages in current thread to reflect on.[/yellow]\n")
+                except Exception as exc:
+                    console.print(f"[bold red]Reflection failed: {exc}[/bold red]\n")
+            else:
+                console.print("[dim]Usage: /memory list | /memory forget <id> | /memory reflect[/dim]\n")
+            continue
+
         if text.lower() == "/clear":
             thread_id = str(uuid.uuid4())
             trace_path = log_path.with_name(f"{log_path.stem}-{thread_id[:8]}.jsonl")

tests/test_graph.py

@@ -130,3 +130,27 @@ def test_tool_node_catches_tool_exceptions_and_feeds_back():
     msg = out["messages"][0]
     assert isinstance(msg, ToolMessage)
     assert "kaboom" in msg.content
+
+
+@tool_decorator
+def todo_sentinel_tool(action: str) -> str:
+    """Returns todo sentinel output."""
+    if action == "write":
+        return "SET_TODOS: ['first', 'second']"
+    return "DONE_TODO: 0"
+
+
+def test_tool_node_consumes_todo_sentinels_into_state():
+    node = _build_tool_node([todo_sentinel_tool])
+
+    write_out = node({
+        "messages": [_ai_with_calls([{"id": "1", "name": "todo_sentinel_tool", "args": {"action": "write"}}])],
+        "todos": [],
+    })
+    assert write_out["todos"] == ["first", "second"]
+
+    done_out = node({
+        "messages": [_ai_with_calls([{"id": "2", "name": "todo_sentinel_tool", "args": {"action": "done"}}])],
+        "todos": write_out["todos"],
+    })
+    assert done_out["todos"] == ["second"]

tests/test_memory_persistence.py

@@ -65,7 +65,7 @@ def test_summarize_episode_logs_traceback_on_failure(tmp_path, monkeypatch, capl
     assert record.exc_info is not None
 
 
-def test_extract_and_compress_facts_passes_existing_memories_as_strings(tmp_path, monkeypatch):
+def test_extract_and_compress_facts_passes_existing_memories_with_ids(tmp_path, monkeypatch):
     from lilith_agent import memory
     import langmem
 
@@ -81,6 +81,15 @@ def test_extract_and_compress_facts_passes_existing_memories_as_strings(tmp_path
     monkeypatch.setattr(memory, "_store", store)
     monkeypatch.setattr(langmem, "create_memory_manager", lambda model, enable_deletes: FakeManager())
 
-    memory.extract_and_compress_facts([HumanMessage(content="New fact")], object())
+    class FakeModel:
+        def invoke(self, prompt):
+            class Response:
+                content = "Lesson"
+
+            return Response()
+
+    memory.extract_and_compress_facts([HumanMessage(content="New fact")], FakeModel())
 
-    assert captured["existing"] == ["Existing fact"]
+    existing_id, existing_memory = captured["existing"][0]
+    assert existing_id == "memory-1"
+    assert existing_memory.content == "Existing fact"

tests/test_memory_safety.py

@@ -0,0 +1,207 @@
+"""Tests for memory safety guards: empty-list wipe prevention and ID stability."""
+import pytest
+from langchain_core.messages import HumanMessage
+
+from lilith_agent.memory import MemoryStore, MIN_MESSAGES_FOR_EXTRACTION, ephemeral_memory
+
+
+class _FakeEpisodeModel:
+    def invoke(self, prompt):
+        class Response:
+            content = "Test lesson"
+
+        return Response()
+
+
+class TestSaveMemoriesEmptyGuard:
+    """save_memories([]) must not wipe existing facts."""
+
+    def test_refuses_empty_list_when_facts_exist(self, tmp_path):
+        store = MemoryStore(tmp_path / "test.sqlite")
+        store.save_memories([
+            {"id": "fact-1", "content": "User prefers dark theme"},
+            {"id": "fact-2", "content": "User name is Yujing"},
+        ])
+        assert len(store.get_all_memories()) == 2
+
+        # Empty list should NOT wipe
+        store.save_memories([])
+
+        facts = store.get_all_memories()
+        assert len(facts) == 2, "Empty list wiped all facts!"
+
+    def test_empty_list_on_empty_store_is_noop(self, tmp_path):
+        store = MemoryStore(tmp_path / "test.sqlite")
+        assert len(store.get_all_memories()) == 0
+
+        # Empty save on empty store is fine — no error, still empty
+        store.save_memories([])
+        assert len(store.get_all_memories()) == 0
+
+
+class TestExtractDoesNotWipeOnEmptyResult:
+    """extract_and_compress_facts must not wipe store when LangMem returns []."""
+
+    def test_langmem_empty_result_preserves_existing_facts(self, tmp_path, monkeypatch):
+        from lilith_agent import memory
+        import langmem
+
+        class FakeManager:
+            def invoke(self, payload):
+                return []  # LangMem found nothing new
+
+        store = MemoryStore(tmp_path / "test.sqlite")
+        store.save_memories([
+            {"id": "fact-1", "content": "Important fact"},
+        ])
+        monkeypatch.setattr(memory, "_store", store)
+        monkeypatch.setattr(langmem, "create_memory_manager", lambda model, enable_deletes: FakeManager())
+
+        memory.extract_and_compress_facts(
+            [HumanMessage(content="Hello"), HumanMessage(content="How are you?")],
+            _FakeEpisodeModel(),
+        )
+
+        facts = store.get_all_memories()
+        assert len(facts) == 1, "LangMem empty result wiped existing facts!"
+        assert facts[0]["content"] == "Important fact"
+
+
+class TestMemoryIdStability:
+    def test_passes_existing_memories_with_stable_ids_to_langmem(self, tmp_path, monkeypatch):
+        from lilith_agent import memory
+        import langmem
+
+        captured = {}
+
+        class FakeManager:
+            def invoke(self, payload):
+                captured["existing"] = payload["existing"]
+                return [
+                    {"id": "memory-1", "content": "Existing fact"},
+                    {"content": "New fact"},
+                ]
+
+        store = MemoryStore(tmp_path / "test.sqlite")
+        store.save_memories([
+            {"id": "memory-1", "content": "Existing fact"},
+        ])
+        original = store.get_all_memories()[0]
+        monkeypatch.setattr(memory, "_store", store)
+        monkeypatch.setattr(langmem, "create_memory_manager", lambda model, enable_deletes: FakeManager())
+
+        memory.extract_and_compress_facts(
+            [HumanMessage(content="Remember a new fact"), HumanMessage(content="New fact")],
+            _FakeEpisodeModel(),
+        )
+
+        facts_by_content = {fact["content"]: fact for fact in store.get_all_memories()}
+        existing_payload = captured["existing"][0]
+        assert existing_payload[0] == "memory-1"
+        assert getattr(existing_payload[1], "content") == "Existing fact"
+        assert facts_by_content["Existing fact"]["id"] == "memory-1"
+        assert facts_by_content["Existing fact"]["created_at"] == original["created_at"]
+        assert facts_by_content["New fact"]["id"] != "memory-1"
+
+    def test_langmem_remove_doc_deletes_fact_instead_of_persisting_marker(self, tmp_path, monkeypatch):
+        from lilith_agent import memory
+        import langmem
+
+        class RemoveDocLike:
+            def __repr_name__(self):
+                return "RemoveDoc"
+
+        class FakeExtractedMemory:
+            def __init__(self, id, content):
+                self.id = id
+                self.content = content
+
+        class FakeMemory:
+            content = "Keep fact"
+
+        class FakeManager:
+            def invoke(self, payload):
+                return [
+                    FakeExtractedMemory("fact-1", RemoveDocLike()),
+                    FakeExtractedMemory("fact-2", FakeMemory()),
+                ]
+
+        store = MemoryStore(tmp_path / "test.sqlite")
+        store.save_memories([
+            {"id": "fact-1", "content": "Delete fact"},
+            {"id": "fact-2", "content": "Keep fact"},
+        ])
+        monkeypatch.setattr(memory, "_store", store)
+        monkeypatch.setattr(langmem, "create_memory_manager", lambda model, enable_deletes: FakeManager())
+
+        memory.extract_and_compress_facts(
+            [HumanMessage(content="Forget delete fact"), HumanMessage(content="Keep fact")],
+            _FakeEpisodeModel(),
+        )
+
+        facts_by_id = {fact["id"]: fact for fact in store.get_all_memories()}
+        assert "fact-1" not in facts_by_id
+        assert facts_by_id["fact-2"]["content"] == "Keep fact"
+
+    def test_successful_fact_extraction_also_records_episode(self, tmp_path, monkeypatch):
+        from lilith_agent import memory
+        import langmem
+
+        class FakeManager:
+            def invoke(self, payload):
+                return [{"content": "New fact"}]
+
+        class FakeModel:
+            def invoke(self, prompt):
+                class Response:
+                    content = "Successful lesson"
+
+                return Response()
+
+        store = MemoryStore(tmp_path / "test.sqlite")
+        monkeypatch.setattr(memory, "_store", store)
+        monkeypatch.setattr(langmem, "create_memory_manager", lambda model, enable_deletes: FakeManager())
+
+        memory.extract_and_compress_facts(
+            [HumanMessage(content="Remember useful fact"), HumanMessage(content="New fact")],
+            FakeModel(),
+        )
+
+        episodes = store.get_recent_episodes()
+        assert episodes[0]["summary"] == "Successful lesson"
+
+    def test_one_turn_exchange_is_eligible_for_memory_extraction(self):
+        assert MIN_MESSAGES_FOR_EXTRACTION <= 2
+
+
+class TestForgetSafety:
+    def test_delete_memory_prefix_does_not_treat_sql_wildcards_as_patterns(self, tmp_path):
+        store = MemoryStore(tmp_path / "test.sqlite")
+        store.save_memories([
+            {"id": "abc-1", "content": "First fact"},
+            {"id": "def-1", "content": "Second fact"},
+        ])
+
+        assert store.delete_memory_prefix("%") == 0
+        assert len(store.get_all_memories()) == 2
+        assert store.delete_memory_prefix("abc") == 1
+        assert [fact["id"] for fact in store.get_all_memories()] == ["def-1"]
+
+
+class TestSearchMemoryLimits:
+    def test_max_results_applies_to_combined_facts_and_episodes(self, tmp_path, monkeypatch):
+        from lilith_agent import memory
+
+        store = MemoryStore(tmp_path / "test.sqlite")
+        store.save_memories([
+            {"id": "fact-1", "content": "alpha fact one"},
+            {"id": "fact-2", "content": "alpha fact two"},
+        ])
+        store.add_episode("alpha task one", "alpha summary one", "success")
+        store.add_episode("alpha task two", "alpha summary two", "success")
+        monkeypatch.setattr(memory, "_store", store)
+
+        result = memory.search_memory_store("alpha", max_results=3)
+
+        result_count = result.count("[fact]") + result.count("[episode]")
+        assert result_count == 3