Phase 3: SQLite long-term memory + KnowledgeAgent seam

Adds the three-tier memory taxonomy from the 2026 agent-memory literature
(working / episodic / semantic / procedural) backed by stdlib sqlite3 -
zero new deps. Mirrors Mem0's interface so a future swap is one-line.

memory.py
* LongTermMemory(db_path=None) -- in-memory by default, file for persistence.
* Three tables, all keyed by user_id:
- semantic_facts(fact_type, content, source) -- likes/dislikes/allergies
- procedural_records(plan_summary, verdict, issues_json) -- validator history
- episodic_sessions(session_id, payload_json) -- full snapshot for replay
* Filtered recall by fact_type / substring / limit; forget_fact() to support
user-driven correction.

knowledge.py
* KnowledgeAgent.handle_task(task, memory) returns JSON {answer, citations}.
* Default backing is WebSearchTool with per-kind query biasing toward
authoritative domains (USDA FDC for nutrition, WHO/diabetes.org/EFSA/NICE
for guidelines, MedlinePlus/FDA/NIH for drug interactions).
* Always emits a citations list; if empty, appends an "advisory only" note
so the Validator can flag uncited clinical claims.
* Designed as a SEAM: a future phase swaps WebSearch for a real RAG index
over USDA + WHO/ADA/EFSA PDFs without changing the agent's call sites.

nutritionmas.py
* AGENTS dict now exposes 'KnowledgeAgent' so the Coach can call_agent it
for any "what does the literature say about X" question.
* New initialize_long_term_memory(db_path) -> LongTermMemory; module-level
singleton LONG_TERM_MEMORY for the Gradio app (Phase 7) to wire into
per-user sessions.

tests/test_memory.py (9 tests)
* Round-trip for all three tables, user isolation, filter-by-type and
filter-by-substring, recall ordering+limit, forget_fact.
* KnowledgeAgent: extracts USDA URL into citations; appends advisory note
when no URL is present.

46/46 tests green.

Note: full RAG ingestion of USDA FDC + WHO PDFs is intentionally deferred
- it's a large data-engineering task, and the seam is what unblocks it.

knowledge.py

@@ -0,0 +1,127 @@
+"""KnowledgeAgent — citation-first retrieval over authoritative sources.
+
+Phase 3 wires the *interface* and a default WebSearch-backed implementation.
+Full RAG over USDA FoodData Central and WHO/ADA/EFSA PDFs is intentionally
+left as a follow-up (it requires bulk data ingestion + an embedding store).
+The seam is here so a later phase can drop in a real index without changing
+the agents that call it.
+
+The contract is: every query returns a synthesised answer **with at least
+one citation**. The Validator will reject medical recommendations that lack
+a citation, so this agent is the safety story for clinical content.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime
+from typing import Any, Dict, List, Optional
+
+from logging_setup import get_logger
+from utils import save_to_json
+
+_logger = get_logger("agents.knowledge")
+
+
+class KnowledgeAgent:
+    """Default Knowledge implementation backed by WebSearchTool.
+
+    Future drop-in: replace ``self.web`` with a RAG retriever that walks an
+    embedded USDA/WHO/ADA/EFSA index and returns citation tuples.
+    """
+
+    SUPPORTED_KINDS = {"nutrition", "guideline", "drug_interaction", "general"}
+
+    def __init__(self, web_search_tool, llm_instance: Optional[Any] = None) -> None:
+        self.web = web_search_tool
+        self.llm = llm_instance  # reserved for the RAG-backed variant
+
+    def handle_task(self, task: str, memory: Dict[str, Any]) -> str:  # noqa: ARG002
+        """Answer ``task`` and return JSON ``{answer, citations}``.
+
+        ``task`` is a free-text question. Optional structured form:
+            ``{"kind": "nutrition" | "guideline" | ...,
+               "query": "...",
+               "context": "..."}``
+        """
+        kind, query, context = self._parse_task(task)
+        _logger.info("📚 KNOWLEDGE: kind=%s query=%r", kind, query[:80])
+
+        # Bias the query toward citation-rich sources.
+        biased_query = self._bias_query(kind, query)
+        web_answer = self.web.handle_task(biased_query)
+
+        citations = self._extract_citations(web_answer)
+        answer = web_answer  # WebSearch already synthesises; we just append a citation note.
+        if not citations:
+            answer += (
+                "\n\n[Note] This answer comes from a generalist web search; "
+                "no authoritative clinical citation was found. Treat as advisory only."
+            )
+
+        payload = {"kind": kind, "answer": answer, "citations": citations}
+        save_to_json(
+            {
+                "task": task,
+                "kind": kind,
+                "query": query,
+                "context": context,
+                "biased_query": biased_query,
+                "answer": answer,
+                "citations": citations,
+                "timestamp": datetime.now().isoformat(),
+            },
+            f"knowledge_{datetime.now().isoformat()}.json",
+            subdirectory="KnowledgeAgent",
+        )
+        return json.dumps(payload)
+
+    # ------------------------------------------------------------------
+    @staticmethod
+    def _parse_task(task: str) -> tuple[str, str, str]:
+        try:
+            data = json.loads(task)
+            if isinstance(data, dict):
+                kind = data.get("kind", "general")
+                if kind not in KnowledgeAgent.SUPPORTED_KINDS:
+                    kind = "general"
+                return kind, data.get("query", ""), data.get("context", "")
+        except (json.JSONDecodeError, TypeError):
+            pass
+        return "general", task, ""
+
+    @staticmethod
+    def _bias_query(kind: str, query: str) -> str:
+        """Steer the search toward authoritative domains per kind."""
+        if kind == "nutrition":
+            return (
+                f"{query} site:fdc.nal.usda.gov OR site:nutritionsource.hsph.harvard.edu "
+                "OR site:who.int"
+            )
+        if kind == "guideline":
+            return (
+                f"{query} site:who.int OR site:diabetes.org OR site:efsa.europa.eu "
+                "OR site:nice.org.uk"
+            )
+        if kind == "drug_interaction":
+            return f"{query} site:medlineplus.gov OR site:fda.gov OR site:nih.gov"
+        return query
+
+    @staticmethod
+    def _extract_citations(text: str) -> List[str]:
+        """Pull URL-looking tokens out of the synthesised answer."""
+        import re
+
+        urls = re.findall(r"https?://[^\s)\]]+", text)
+        # De-duplicate while preserving order.
+        seen = set()
+        out: List[str] = []
+        for u in urls:
+            u_clean = u.rstrip(".,);")
+            if u_clean not in seen:
+                seen.add(u_clean)
+                out.append(u_clean)
+        return out
+
+
+__all__ = ["KnowledgeAgent"]

memory.py

@@ -0,0 +1,198 @@
+"""Long-term memory layer (semantic / procedural / episodic).
+
+Phase 3 deliberately uses stdlib ``sqlite3`` rather than Mem0 / Letta /
+sqlite-vec so the demo ships with zero extra dependencies. The interface,
+however, mirrors the modern three-tier taxonomy from the 2026 agent-memory
+literature so a later phase can swap the backend without touching call sites.
+
+Tiers
+-----
+* **Working** — kept in the LangGraph state (untouched by this module).
+* **Semantic** — atomic facts about the user (likes, dislikes, hard
+  constraints, lab results). Survives across sessions.
+* **Procedural** — verdicts the validator produced. Lets the system learn
+  "this user rejected high-carb breakfasts twice" without re-asking.
+* **Episodic** — JSON snapshot of past sessions for replay / audit.
+
+Schema is intentionally tiny — three tables, one row per fact / verdict /
+session. Vector search is *not* needed for this demo; SQL ``LIKE`` over
+short text is good enough and adds zero dependencies. Phase 6 evals will
+make the case for upgrading.
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+import threading
+from datetime import datetime
+from typing import Any, Dict, List, Optional
+
+
+_SCHEMA = """
+CREATE TABLE IF NOT EXISTS semantic_facts (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    user_id TEXT NOT NULL,
+    fact_type TEXT NOT NULL,         -- e.g. 'dislike', 'allergy', 'preference'
+    content TEXT NOT NULL,
+    source TEXT NOT NULL DEFAULT '',  -- e.g. 'user_stated', 'inferred', 'validator'
+    created_at TEXT NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_facts_user ON semantic_facts(user_id, fact_type);
+
+CREATE TABLE IF NOT EXISTS procedural_records (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    user_id TEXT NOT NULL,
+    plan_summary TEXT NOT NULL,
+    verdict TEXT NOT NULL,            -- 'pass' | 'revise' | 'reject'
+    issues_json TEXT NOT NULL,        -- JSON list of ValidationIssue
+    created_at TEXT NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_proc_user ON procedural_records(user_id, created_at);
+
+CREATE TABLE IF NOT EXISTS episodic_sessions (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    user_id TEXT NOT NULL,
+    session_id TEXT NOT NULL,
+    payload_json TEXT NOT NULL,       -- JSON snapshot of session state
+    created_at TEXT NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_episodic_user ON episodic_sessions(user_id, created_at);
+"""
+
+
+class LongTermMemory:
+    """SQLite-backed three-tier long-term memory.
+
+    Pass a file path for persistence across runs, or ``None`` (default) for an
+    in-memory database useful in tests / ephemeral demos.
+    """
+
+    def __init__(self, db_path: Optional[str] = None) -> None:
+        self.db_path = db_path or ":memory:"
+        # SQLite connections are not thread-safe by default; one connection per
+        # thread is the standard pattern. The demo is single-process so a single
+        # connection + lock is enough.
+        self.conn = sqlite3.connect(self.db_path, check_same_thread=False)
+        self.conn.row_factory = sqlite3.Row
+        self._lock = threading.Lock()
+        self._init_schema()
+
+    def _init_schema(self) -> None:
+        with self._lock:
+            self.conn.executescript(_SCHEMA)
+            self.conn.commit()
+
+    def close(self) -> None:
+        with self._lock:
+            self.conn.close()
+
+    # ------------------------------------------------------------------
+    # Semantic facts
+    # ------------------------------------------------------------------
+    def remember_fact(
+        self,
+        user_id: str,
+        fact_type: str,
+        content: str,
+        source: str = "user_stated",
+    ) -> int:
+        """Insert a semantic fact. Returns the row id."""
+        now = datetime.utcnow().isoformat()
+        with self._lock:
+            cur = self.conn.execute(
+                "INSERT INTO semantic_facts (user_id, fact_type, content, source, created_at) "
+                "VALUES (?, ?, ?, ?, ?)",
+                (user_id, fact_type, content, source, now),
+            )
+            self.conn.commit()
+            return int(cur.lastrowid or 0)
+
+    def recall_facts(
+        self,
+        user_id: str,
+        fact_type: Optional[str] = None,
+        contains: Optional[str] = None,
+        limit: int = 50,
+    ) -> List[Dict[str, Any]]:
+        """List facts for a user, optionally filtered by type / substring."""
+        sql = "SELECT * FROM semantic_facts WHERE user_id = ?"
+        params: List[Any] = [user_id]
+        if fact_type:
+            sql += " AND fact_type = ?"
+            params.append(fact_type)
+        if contains:
+            sql += " AND content LIKE ?"
+            params.append(f"%{contains}%")
+        sql += " ORDER BY created_at DESC LIMIT ?"
+        params.append(limit)
+        with self._lock:
+            cur = self.conn.execute(sql, params)
+            return [dict(row) for row in cur.fetchall()]
+
+    def forget_fact(self, fact_id: int) -> None:
+        with self._lock:
+            self.conn.execute("DELETE FROM semantic_facts WHERE id = ?", (fact_id,))
+            self.conn.commit()
+
+    # ------------------------------------------------------------------
+    # Procedural records (validator history)
+    # ------------------------------------------------------------------
+    def remember_validation(
+        self,
+        user_id: str,
+        plan_summary: str,
+        verdict: str,
+        issues: List[Dict[str, Any]],
+    ) -> int:
+        now = datetime.utcnow().isoformat()
+        with self._lock:
+            cur = self.conn.execute(
+                "INSERT INTO procedural_records (user_id, plan_summary, verdict, issues_json, created_at) "
+                "VALUES (?, ?, ?, ?, ?)",
+                (user_id, plan_summary, verdict, json.dumps(issues), now),
+            )
+            self.conn.commit()
+            return int(cur.lastrowid or 0)
+
+    def recall_validations(self, user_id: str, limit: int = 10) -> List[Dict[str, Any]]:
+        with self._lock:
+            cur = self.conn.execute(
+                "SELECT * FROM procedural_records WHERE user_id = ? ORDER BY created_at DESC LIMIT ?",
+                (user_id, limit),
+            )
+            return [
+                {**dict(row), "issues": json.loads(row["issues_json"])}
+                for row in cur.fetchall()
+            ]
+
+    # ------------------------------------------------------------------
+    # Episodic sessions
+    # ------------------------------------------------------------------
+    def remember_session(self, user_id: str, session_id: str, payload: Dict[str, Any]) -> int:
+        now = datetime.utcnow().isoformat()
+        with self._lock:
+            cur = self.conn.execute(
+                "INSERT INTO episodic_sessions (user_id, session_id, payload_json, created_at) "
+                "VALUES (?, ?, ?, ?)",
+                (user_id, session_id, json.dumps(payload, default=str), now),
+            )
+            self.conn.commit()
+            return int(cur.lastrowid or 0)
+
+    def recall_sessions(self, user_id: str, limit: int = 5) -> List[Dict[str, Any]]:
+        with self._lock:
+            cur = self.conn.execute(
+                "SELECT * FROM episodic_sessions WHERE user_id = ? ORDER BY created_at DESC LIMIT ?",
+                (user_id, limit),
+            )
+            return [
+                {**dict(row), "payload": json.loads(row["payload_json"])}
+                for row in cur.fetchall()
+            ]
+
+
+__all__ = ["LongTermMemory"]

nutritionmas.py

@@ -7,7 +7,9 @@ from IPython.display import Markdown, display
 
 from agents import CoachAgent, MedicalAssessmentAgent, PlannerAgent
 from config import set_settings
+from knowledge import KnowledgeAgent
 from logging_setup import get_logger, refresh_level
+from memory import LongTermMemory
 from state import initialize_empty_memory
 from tools import ComputationTool, QuantitiesFinder, WebSearchTool
 from utils import APIPoolManager, create_llm
@@ -187,8 +189,30 @@ def initialize_agents():
             PLANNER_LLM, TOOLS["ComputationTool"], TOOLS["WebSearchTool"], TOOLS["QuantitiesFinder"]
         ),
         "ValidationAgent": ValidationAgent(VALIDATION_LLM),
+        # KnowledgeAgent is the citation-first retrieval seam; defaults to
+        # WebSearch backing. Phase 3+ can swap in a RAG-backed implementation
+        # over USDA / WHO / ADA / EFSA without touching Coach call-sites.
+        "KnowledgeAgent": KnowledgeAgent(TOOLS["WebSearchTool"]),
     }
 
+
+# ---------------------------------------------------------------------------
+# Long-term memory singleton (Phase 3)
+# ---------------------------------------------------------------------------
+LONG_TERM_MEMORY: Optional[LongTermMemory] = None
+
+
+def initialize_long_term_memory(db_path: Optional[str] = None) -> LongTermMemory:
+    """Initialise the SQLite-backed three-tier memory.
+
+    Pass a file path for cross-session persistence, or omit for an in-memory
+    DB (default; tests / ephemeral demos).
+    """
+    global LONG_TERM_MEMORY
+    LONG_TERM_MEMORY = LongTermMemory(db_path=db_path)
+    _logger.info("Long-term memory initialised at %s", db_path or ":memory:")
+    return LONG_TERM_MEMORY
+
 def setup_workflow():
     global APP
     if AGENTS is None or TOOLS is None:

tests/test_memory.py

@@ -0,0 +1,115 @@
+"""Tests for the SQLite-backed three-tier long-term memory."""
+
+from __future__ import annotations
+
+from memory import LongTermMemory
+
+
+def test_round_trip_semantic_facts() -> None:
+    mem = LongTermMemory()
+    fid = mem.remember_fact("user1", "dislike", "okra", source="user_stated")
+    assert fid > 0
+    facts = mem.recall_facts("user1")
+    assert len(facts) == 1
+    assert facts[0]["content"] == "okra"
+    assert facts[0]["fact_type"] == "dislike"
+
+
+def test_filter_facts_by_type_and_substring() -> None:
+    mem = LongTermMemory()
+    mem.remember_fact("u1", "dislike", "okra")
+    mem.remember_fact("u1", "dislike", "kale")
+    mem.remember_fact("u1", "preference", "high-protein")
+    mem.remember_fact("u2", "dislike", "okra")  # different user
+
+    dislikes = mem.recall_facts("u1", fact_type="dislike")
+    assert {f["content"] for f in dislikes} == {"okra", "kale"}
+
+    prefs = mem.recall_facts("u1", fact_type="preference")
+    assert prefs[0]["content"] == "high-protein"
+
+    okra_only = mem.recall_facts("u1", contains="okra")
+    assert len(okra_only) == 1
+
+
+def test_user_isolation() -> None:
+    mem = LongTermMemory()
+    mem.remember_fact("alice", "allergy", "peanut")
+    mem.remember_fact("bob", "allergy", "shellfish")
+
+    assert {f["content"] for f in mem.recall_facts("alice")} == {"peanut"}
+    assert {f["content"] for f in mem.recall_facts("bob")} == {"shellfish"}
+
+
+def test_forget_fact() -> None:
+    mem = LongTermMemory()
+    fid = mem.remember_fact("u", "dislike", "okra")
+    mem.forget_fact(fid)
+    assert mem.recall_facts("u") == []
+
+
+def test_procedural_records_round_trip() -> None:
+    mem = LongTermMemory()
+    issues = [{"code": "calorie_deviation", "description": "x", "severity": "medium"}]
+    mem.remember_validation("u1", "1500 kcal plan", "revise", issues)
+
+    history = mem.recall_validations("u1")
+    assert len(history) == 1
+    assert history[0]["verdict"] == "revise"
+    assert history[0]["issues"] == issues
+
+
+def test_episodic_session_round_trip() -> None:
+    mem = LongTermMemory()
+    payload = {"messages": [{"role": "user", "content": "hi"}], "memory": {"x": 1}}
+    mem.remember_session("u1", "session-A", payload)
+
+    sessions = mem.recall_sessions("u1")
+    assert len(sessions) == 1
+    assert sessions[0]["payload"] == payload
+    assert sessions[0]["session_id"] == "session-A"
+
+
+def test_recall_limit_and_order() -> None:
+    mem = LongTermMemory()
+    for i in range(15):
+        mem.remember_fact("u", "note", f"fact-{i}")
+    facts = mem.recall_facts("u", limit=5)
+    assert len(facts) == 5
+    # newest first
+    assert facts[0]["content"] == "fact-14"
+
+
+def test_knowledge_agent_returns_citations() -> None:
+    """KnowledgeAgent should always return a citations list (possibly empty)."""
+    from knowledge import KnowledgeAgent
+
+    class StubWebSearch:
+        def handle_task(self, query: str) -> str:
+            return (
+                "Calories per 100g of chicken breast: 165 kcal (USDA FDC).\n"
+                "Source: https://fdc.nal.usda.gov/food-details/171477/nutrients"
+            )
+
+    agent = KnowledgeAgent(StubWebSearch())
+    import json
+    raw = agent.handle_task('{"kind": "nutrition", "query": "chicken breast"}', memory={})
+    payload = json.loads(raw)
+    assert payload["kind"] == "nutrition"
+    assert payload["citations"], "Should extract at least one URL"
+    assert any("fdc.nal.usda.gov" in c for c in payload["citations"])
+
+
+def test_knowledge_agent_marks_uncited_answers() -> None:
+    """When no citations are found, agent appends an advisory note."""
+    from knowledge import KnowledgeAgent
+    import json
+
+    class NoCitationStub:
+        def handle_task(self, query: str) -> str:
+            return "Generic answer without any URL."
+
+    agent = KnowledgeAgent(NoCitationStub())
+    payload = json.loads(agent.handle_task("how many calories in an apple?", memory={}))
+    assert payload["citations"] == []
+    assert "advisory only" in payload["answer"]