structured memory: facts/pinned/episodes, batched consolidation, participant-scoped recall

fjerkroa_bot/ai_responder.py

@@ -12,6 +12,7 @@ from pathlib import Path
 from pprint import pformat
 from typing import Any, Dict, List, Optional, Tuple, Union
 
+from .memory import MemoryManager
 from .persistence import PersistentStore
 
 
@@ -150,6 +151,7 @@ class AIResponder(AIResponderBase):
             stored_memory = self.store.load_memory(self.channel)
             if stored_memory is not None:
                 self.memory = stored_memory
+        self.memory_manager = MemoryManager(self.store, lambda: self.config, self.consolidate, self.channel)
         logging.info(f"memmory:\n{self.memory}")
 
     def message(self, message: AIMessage, limit: Optional[int] = None) -> List[Dict[str, Any]]:
@@ -161,7 +163,8 @@ class AIResponder(AIResponderBase):
             with open(news_feed) as fd:
                 news_feed = fd.read().strip()
             system = system.replace("{news}", sanitize_external_text(news_feed))
-        system = system.replace("{memory}", self.memory)
+        participants = [message.user] + [entry_user for entry_user in self._history_users(20)]
+        system = system.replace("{memory}", self.memory_manager.memory_block(participants, self.memory))
         messages.append({"role": "system", "content": system})
         if limit is not None:
             while len(self.history) > limit:
@@ -232,7 +235,7 @@ class AIResponder(AIResponderBase):
     async def chat(self, messages: List[Dict[str, Any]], limit: int) -> Tuple[Optional[Dict[str, Any]], int]:
         raise NotImplementedError()
 
-    async def memory_rewrite(self, memory: str, message_user: str, answer_user: str, question: str, answer: str) -> str:
+    async def consolidate(self, observations: List[Dict[str, Any]], known_facts: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
         raise NotImplementedError()
 
     async def translate(self, text: str, language: str = "english") -> str:
@@ -245,6 +248,17 @@ class AIResponder(AIResponderBase):
         except Exception:
             return None
 
+    def _history_users(self, tail: int) -> List[str]:
+        users = []
+        for item in self.history[-tail:]:
+            try:
+                user = parse_json(item["content"]).get("user")
+            except Exception:
+                user = None
+            if user:
+                users.append(str(user))
+        return users
+
     def shrink_history_by_one(self) -> None:
         if not self.history:
             return
@@ -268,17 +282,10 @@ class AIResponder(AIResponderBase):
         while len(self.history) > limit:
             self.shrink_history_by_one()
 
-    def update_memory(self, memory) -> None:
-        self.memory = memory
-
     async def _persist_history(self) -> None:
         if self.store is not None:
             await asyncio.to_thread(self.store.save_history, self.channel, list(self.history))
 
-    async def _persist_memory(self) -> None:
-        if self.store is not None:
-            await asyncio.to_thread(self.store.save_memory, self.channel, self.memory)
-
     async def handle_picture(self, response: Dict) -> bool:
         if not isinstance(response.get("picture"), (type(None), str)):
             logging.warning(f"picture key is wrong in response: {pp(response)}")
@@ -296,16 +303,9 @@ class AIResponder(AIResponderBase):
             logging.error(f"failed to parse the answer: {pp(err)}\n{repr(answer['content'])}")
             return None
 
-    async def memoize(self, message_user: str, answer_user: str, message: str, answer: str) -> None:
-        self.memory = await self.memory_rewrite(self.memory, message_user, answer_user, message, answer)
-        self.update_memory(self.memory)
-        await self._persist_memory()
-
-    async def memoize_reaction(self, message_user: str, reaction_user: str, operation: str, reaction: str, message: str) -> None:
-        quoted_message = message.replace("\n", "\n> ")
-        await self.memoize(
-            message_user, "assistant", f"\n> {quoted_message}", f"User {reaction_user} has {operation} this raction: {reaction}"
-        )
+    async def observe_event(self, user: str, kind: str, content: str) -> None:
+        """Feed a Discord event into the observation stream (MEM-01)."""
+        await self.memory_manager.observe(user, kind, content)
 
     async def send(self, message: AIMessage) -> AIResponse:
         # Get the history limit from the configuration
@@ -354,9 +354,10 @@ class AIResponder(AIResponderBase):
             await self._persist_history()
             logging.info(f"got this answer:\n{str(answer_message)}")
 
-            # Update memory
+            # Feed the observation stream — consolidation is batched (MEM-01/02)
+            await self.observe_event(message.user, "message", message.message)
             if answer_message.answer is not None:
-                await self.memoize(message.user, "assistant", message.message, answer_message.answer)
+                await self.observe_event("assistant", "message", answer_message.answer)
 
             # Return the updated answer message
             return answer_message

fjerkroa_bot/discord_bot.py

@@ -156,10 +156,11 @@ class FjerkroaBot(commands.Bot):
             await responder._persist_history()
         if self.airesponder.store is not None:
             removed += self.airesponder.store.delete_history_of_user(user)
-        logging.info(f"forgetme: removed {removed} history entries for {user}")
+            # facts + observations + episode traces (MEM-09)
+            removed += self.airesponder.store.purge_user_memory(user)
+        logging.info(f"forgetme: removed {removed} entries for {user}")
         await message.channel.send(
-            f"Removed your messages from my history ({removed} entries). "
-            "Note: channel memory summaries may still hold traces until the structured-memory upgrade.",
+            f"Removed your messages, facts and memory traces ({removed} entries).",
             suppress_embeds=True,
         )
 
@@ -174,10 +175,34 @@ class FjerkroaBot(commands.Bot):
         # Gate for boreness today, the FDB-011 scheduler later (OPS-09)
         return self.tasks_enabled and self.replies_allowed()
 
+    def _memory_command(self, args) -> Optional[str]:
+        """Staff memory review/edit (MEM-07)."""
+        if args[:1] not in (["memory"], ["forget-fact"], ["pin"], ["unpin"]):
+            return None
+        store = self.airesponder.store
+        if store is None:
+            return "No store configured - memory commands unavailable."
+        if args[:1] == ["memory"] and args[1:2]:
+            facts = store.facts_for([args[1]])
+            return "\n".join(f"{fact['id']}: {fact['fact']}" for fact in facts) or f"No facts stored for {args[1]}."
+        if args[:1] == ["forget-fact"] and args[1:2] and args[1].isdigit():
+            return f"Deleted {store.delete_fact(int(args[1]))} fact(s)."
+        if args[:1] == ["pin"] and len(args) >= 3:
+            channel = None if args[1] == "global" else args[1]
+            store.add_pinned(channel, " ".join(args[2:]))
+            return f"Pinned for {args[1]}."
+        if args[:1] == ["unpin"] and args[1:2] and args[1].isdigit():
+            return f"Removed {store.delete_pinned(int(args[1]))} pin(s)."
+        return None
+
     async def handle_staff_command(self, message: Message) -> None:
-        """Operator kill-switches, staff channel only (OPS-01..05, OPS-09)."""
+        """Operator kill-switches, staff channel only (OPS-01..05, OPS-09, MEM-07)."""
         args = str(message.content).split()[1:]
-        reply = "Commands: pause, resume, images on|off, tasks on|off, quiet <minutes>, status"
+        memory_reply = self._memory_command(args)
+        if memory_reply is not None:
+            await message.channel.send(memory_reply, suppress_embeds=True)
+            return
+        reply = "Commands: pause, resume, images on|off, tasks on|off, quiet <minutes>, status, spend, memory <user>, forget-fact <id>, pin <channel|global> <fact>, unpin <id>"
         if args[:1] == ["pause"]:
             self.replies_enabled = False
             reply = "Replies paused."
@@ -243,7 +268,9 @@ class FjerkroaBot(commands.Bot):
         airesponder = self.get_ai_responder(self.get_channel_name(reaction.message.channel))
         message = str(reaction.message.content) if reaction.message.content else ""
         if len(message) > 1:
-            await airesponder.memoize_reaction(reaction.message.author.name, user.name, operation, str(reaction.emoji), message)
+            await airesponder.observe_event(
+                user.name, f"reaction-{operation}", f"{reaction.emoji} on {reaction.message.author.name}: {message}"
+            )
 
     async def on_reaction_add(self, reaction, user):
         await self.on_reaction_operation(reaction, user, "adding")
@@ -256,26 +283,17 @@ class FjerkroaBot(commands.Bot):
         airesponder = self.get_ai_responder(self.get_channel_name(message.channel))
         content = str(message.content) if message.content else ""
         if len(content) > 1:
-            await airesponder.memoize(
-                message.author.name, "assistant", "\n> " + content.replace("\n", "\n> "), "All reactions were removed from this message."
-            )
+            await airesponder.observe_event(message.author.name, "reaction-clear", f"all reactions removed from: {content}")
 
     async def on_message_edit(self, before, after):
         if before.author.bot or before.content == after.content:
             return
         airesponder = self.get_ai_responder(self.get_channel_name(before.channel))
-        await airesponder.memoize(
-            before.author.name,
-            "assistant",
-            "\n> " + before.content.replace("\n", "\n> "),
-            "User changed this message to:\n> " + after.content.replace("\n", "\n> "),
-        )
+        await airesponder.observe_event(before.author.name, "edit", f"changed {before.content!r} to {after.content!r}")
 
     async def on_message_delete(self, message):
         airesponder = self.get_ai_responder(self.get_channel_name(message.channel))
-        await airesponder.memoize(
-            message.author.name, "assistant", "\n> " + message.content.replace("\n", "\n> "), "User deleted this message."
-        )
+        await airesponder.observe_event(message.author.name, "delete", f"deleted: {message.content}")
 
     def on_config_file_modified(self, event):
         # Runs on the watchdog observer thread — the swap itself is

fjerkroa_bot/memory.py

@@ -0,0 +1,95 @@
+"""Structured memory manager (SPEC-002).
+
+Observations in, facts + episodes out — via a batched, lock-guarded
+consolidation pass on `memory-model`. Recall assembly is
+participant-scoped (MEM-04): the model never sees facts of users who
+are not part of the conversation.
+"""
+
+import asyncio
+import logging
+from typing import Any, Awaitable, Callable, Dict, List, Optional
+
+from .persistence import PersistentStore
+
+Consolidator = Callable[[List[Dict[str, Any]], List[Dict[str, Any]]], Awaitable[Optional[Dict[str, Any]]]]
+
+DEFAULT_CONSOLIDATE_EVERY = 20
+DEFAULT_EPISODES_PER_CHANNEL = 10
+DEFAULT_FACT_RETENTION_DAYS = 180
+OBSERVATION_EXCERPT = 500
+
+
+class MemoryManager:
+    def __init__(
+        self,
+        store: Optional[PersistentStore],
+        config_getter: Callable[[], Dict[str, Any]],
+        consolidator: Consolidator,
+        channel: str,
+    ) -> None:
+        self.store = store
+        self._config = config_getter
+        self._consolidator = consolidator
+        self.channel = channel
+        self._lock = asyncio.Lock()
+
+    def active(self) -> bool:
+        return self.store is not None and "memory-model" in self._config()
+
+    async def observe(self, user: str, kind: str, content: str) -> None:
+        """Record one event; trigger consolidation when the batch is full (MEM-01/02)."""
+        if not self.active():
+            return
+        assert self.store is not None
+        await asyncio.to_thread(self.store.add_observation, self.channel, user, kind, str(content)[:OBSERVATION_EXCERPT])
+        every = int(self._config().get("memory-consolidate-every", DEFAULT_CONSOLIDATE_EVERY))
+        if await asyncio.to_thread(self.store.unconsumed_observations, self.channel) >= every:
+            asyncio.get_running_loop().create_task(self.consolidate_now())
+
+    async def consolidate_now(self) -> None:
+        """One batched pass: observations -> self-authored facts + episode (MEM-02/03/05/06)."""
+        if not self.active() or self._lock.locked():
+            return
+        assert self.store is not None
+        async with self._lock:
+            observations = await asyncio.to_thread(self.store.peek_observations, self.channel)
+            if not observations:
+                return
+            authors = {observation["user"] for observation in observations}
+            known_facts = await asyncio.to_thread(self.store.facts_for, sorted(authors))
+            result = await self._consolidator(observations, known_facts)
+            if result is None:
+                return  # model call failed — observations stay for the next trigger
+            for fact in result.get("facts", []):
+                if fact.get("user") in authors:
+                    await asyncio.to_thread(self.store.add_user_fact, fact["user"], str(fact["fact"]), "self")
+                else:
+                    logging.warning(f"memory: dropped third-party fact about {fact.get('user')!r} (MEM-03)")
+            episode = result.get("episode")
+            if episode:
+                await asyncio.to_thread(self.store.add_episode, self.channel, str(episode))
+            await asyncio.to_thread(self.store.consume_observations, self.channel, observations[-1]["id"])
+            config = self._config()
+            await asyncio.to_thread(
+                self.store.trim_episodes, self.channel, int(config.get("memory-episodes-per-channel", DEFAULT_EPISODES_PER_CHANNEL))
+            )
+            await asyncio.to_thread(self.store.purge_old_facts, int(config.get("memory-fact-retention-days", DEFAULT_FACT_RETENTION_DAYS)))
+
+    def memory_block(self, participants: List[str], legacy: str) -> str:
+        """Assemble the {memory} block, participant-scoped (MEM-04/10)."""
+        if not self.active():
+            return legacy
+        assert self.store is not None
+        sections: List[str] = []
+        pinned = self.store.pinned_for(self.channel)
+        if pinned:
+            sections.append("Operator notes:\n" + "\n".join(f"- {pin['fact']}" for pin in pinned))
+        facts = self.store.facts_for(sorted(set(participants)))
+        if facts:
+            sections.append("What users told about themselves:\n" + "\n".join(f"- {fact['user']}: {fact['fact']}" for fact in facts))
+        config = self._config()
+        episodes = self.store.recent_episodes(self.channel, int(config.get("memory-episodes-per-channel", DEFAULT_EPISODES_PER_CHANNEL)))
+        if episodes:
+            sections.append("Recent conversation summaries:\n" + "\n".join(f"- {episode}" for episode in episodes))
+        return "\n\n".join(sections)

fjerkroa_bot/openai_responder.py

@@ -31,6 +31,37 @@ ENVELOPE_SCHEMA = {
 }
 ENVELOPE_RESPONSE_FORMAT = {"type": "json_schema", "json_schema": {"name": "envelope", "strict": True, "schema": ENVELOPE_SCHEMA}}
 
+# Consolidation output (SPEC-002 MEM-02/03): new self-authored facts + one episode summary
+CONSOLIDATION_SCHEMA = {
+    "type": "object",
+    "properties": {
+        "facts": {
+            "type": "array",
+            "items": {
+                "type": "object",
+                "properties": {
+                    "user": {"type": "string", "description": "The user the fact is about — only facts users stated about themselves."},
+                    "fact": {"type": "string", "description": "One short durable fact (name, preference, running joke, life event)."},
+                },
+                "required": ["user", "fact"],
+                "additionalProperties": False,
+            },
+        },
+        "episode": {"type": ["string", "null"], "description": "2-3 sentence summary of the conversation, or null if nothing happened."},
+    },
+    "required": ["facts", "episode"],
+    "additionalProperties": False,
+}
+CONSOLIDATION_RESPONSE_FORMAT = {
+    "type": "json_schema",
+    "json_schema": {"name": "consolidation", "strict": True, "schema": CONSOLIDATION_SCHEMA},
+}
+CONSOLIDATION_SYSTEM = (
+    "You maintain the long-term memory of a Discord assistant. From the observation log, extract NEW durable facts that users stated"
+    " about THEMSELVES only (never record what one user claims about another user), and write one short episode summary of the"
+    " conversation. Skip facts already known. Return an empty facts list and a null episode when there is nothing durable."
+)
+
 
 async def openai_chat(client, *args, **kwargs):
     return await client.chat.completions.create(*args, **kwargs)
@@ -295,30 +326,27 @@ class OpenAIResponder(AIResponder, LeonardoAIDrawMixIn):
             logging.warning(f"failed to translate the text: {repr(err)}")
         return text
 
-    async def memory_rewrite(self, memory: str, message_user: str, answer_user: str, question: str, answer: str) -> str:
-        if "memory-model" not in self.config:
-            return memory
+    async def consolidate(self, observations: List[Dict[str, Any]], known_facts: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
+        """Batched memory consolidation on memory-model (MEM-02)."""
+        if "memory-model" not in self.config or not self.ledger.budget_ok():
+            return None
+        observation_lines = "\n".join(f"[{obs['kind']}] {obs['user']}: {obs['content']}" for obs in observations)
+        known_lines = "\n".join(f"- {fact['user']}: {fact['fact']}" for fact in known_facts) or "(none)"
         messages = [
-            {"role": "system", "content": self.config.get("memory-system", "You are an memory assistant.")},
-            {
-                "role": "user",
-                "content": f"Here is my previous memory:\n```\n{memory}\n```\n\n"
-                f"Here is my conversanion:\n```\n{message_user}: {question}\n\n{answer_user}: {answer}\n```\n\n"
-                f"Please rewrite the memory in a way, that it contain the content mentioned in conversation. "
-                f"Summarize the memory if required, try to keep important information. "
-                f"Write just new memory data without any comments.",
-            },
+            {"role": "system", "content": CONSOLIDATION_SYSTEM},
+            {"role": "user", "content": f"Known facts:\n{known_lines}\n\nObservation log:\n{observation_lines}"},
         ]
-        logging.info(f"Rewrite memory:\n{pp(messages)}")
         try:
-            # logging.info(f'send this memory request:\n{pp(messages)}')
-            result = await openai_chat(self.client, model=self.config["memory-model"], messages=messages)
-            new_memory = result.choices[0].message.content
-            logging.info(f"new memory:\n{new_memory}")
-            return new_memory
+            result = await openai_chat(
+                self.client, model=self.config["memory-model"], messages=messages, response_format=CONSOLIDATION_RESPONSE_FORMAT
+            )
+            self._record_usage(result)
+            parsed = json.loads(result.choices[0].message.content)
+            logging.info(f"memory consolidation: {len(parsed.get('facts', []))} new facts, episode={bool(parsed.get('episode'))}")
+            return parsed
         except Exception as err:
-            logging.warning(f"failed to create new memory: {repr(err)}")
-        return memory
+            logging.warning(f"memory consolidation failed: {repr(err)}")
+        return None
 
     async def _execute_igdb_function(self, function_name: str, function_args: Dict[str, Any]) -> Optional[Dict[str, Any]]:
         """

fjerkroa_bot/persistence.py

@@ -13,7 +13,7 @@ from contextlib import closing
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
-SCHEMA_VERSION = 2
+SCHEMA_VERSION = 3
 
 
 class PersistentStore:
@@ -41,6 +41,25 @@ class PersistentStore:
                 conn.execute(
                     "CREATE TABLE IF NOT EXISTS usage (day TEXT NOT NULL, key TEXT NOT NULL, value REAL NOT NULL, PRIMARY KEY (day, key))"
                 )
+            if version < 3:
+                conn.execute(
+                    "CREATE TABLE IF NOT EXISTS user_facts (id INTEGER PRIMARY KEY, user TEXT NOT NULL, fact TEXT NOT NULL,"
+                    " source TEXT NOT NULL DEFAULT 'self', updated_at TEXT NOT NULL DEFAULT (datetime('now')))"
+                )
+                conn.execute(
+                    "CREATE TABLE IF NOT EXISTS pinned_facts (id INTEGER PRIMARY KEY, channel TEXT, fact TEXT NOT NULL,"
+                    " created_at TEXT NOT NULL DEFAULT (datetime('now')))"
+                )
+                conn.execute(
+                    "CREATE TABLE IF NOT EXISTS episodes (id INTEGER PRIMARY KEY, channel TEXT NOT NULL, summary TEXT NOT NULL,"
+                    " created_at TEXT NOT NULL DEFAULT (datetime('now')))"
+                )
+                conn.execute(
+                    "CREATE TABLE IF NOT EXISTS observations (id INTEGER PRIMARY KEY, channel TEXT NOT NULL, user TEXT NOT NULL,"
+                    " kind TEXT NOT NULL, content TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT (datetime('now')))"
+                )
+                # Legacy single-string memories carry over as one episode each (MEM-08)
+                conn.execute("INSERT INTO episodes (channel, summary) SELECT channel, content FROM memory")
             if version < SCHEMA_VERSION:
                 conn.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
         os.chmod(self.db_path, 0o600)  # conversation data (PER-04)
@@ -90,6 +109,90 @@ class PersistentStore:
             cursor = conn.execute("DELETE FROM history WHERE content LIKE ?", (f'%"user": "{user}"%',))
         return cursor.rowcount
 
+    # --- structured memory (SPEC-002) ---
+
+    def add_observation(self, channel: str, user: str, kind: str, content: str) -> None:
+        with closing(self._connect()) as conn, conn:
+            conn.execute("INSERT INTO observations (channel, user, kind, content) VALUES (?, ?, ?, ?)", (channel, user, kind, content))
+
+    def unconsumed_observations(self, channel: str) -> int:
+        with closing(self._connect()) as conn:
+            return int(conn.execute("SELECT COUNT(*) FROM observations WHERE channel = ?", (channel,)).fetchone()[0])
+
+    def peek_observations(self, channel: str) -> List[Dict[str, Any]]:
+        with closing(self._connect()) as conn:
+            rows = conn.execute("SELECT id, user, kind, content FROM observations WHERE channel = ? ORDER BY id", (channel,)).fetchall()
+        return [{"id": row[0], "user": row[1], "kind": row[2], "content": row[3]} for row in rows]
+
+    def consume_observations(self, channel: str, up_to_id: int) -> None:
+        with closing(self._connect()) as conn, conn:
+            conn.execute("DELETE FROM observations WHERE channel = ? AND id <= ?", (channel, up_to_id))
+
+    def add_user_fact(self, user: str, fact: str, source: str = "self") -> None:
+        with closing(self._connect()) as conn, conn:
+            conn.execute("INSERT INTO user_facts (user, fact, source) VALUES (?, ?, ?)", (user, fact, source))
+
+    def facts_for(self, users: List[str]) -> List[Dict[str, Any]]:
+        if not users:
+            return []
+        marks = ",".join("?" for _ in users)
+        with closing(self._connect()) as conn:
+            # marks is only "?" placeholders; user values stay parameterized
+            rows = conn.execute(
+                f"SELECT id, user, fact FROM user_facts WHERE user IN ({marks}) ORDER BY id", tuple(users)  # nosec B608
+            ).fetchall()
+        return [{"id": row[0], "user": row[1], "fact": row[2]} for row in rows]
+
+    def delete_fact(self, fact_id: int) -> int:
+        with closing(self._connect()) as conn, conn:
+            return conn.execute("DELETE FROM user_facts WHERE id = ?", (fact_id,)).rowcount
+
+    def purge_old_facts(self, retention_days: int) -> int:
+        with closing(self._connect()) as conn, conn:
+            cursor = conn.execute("DELETE FROM user_facts WHERE updated_at < datetime('now', ?)", (f"-{int(retention_days)} days",))
+        return cursor.rowcount
+
+    def add_pinned(self, channel: Optional[str], fact: str) -> None:
+        with closing(self._connect()) as conn, conn:
+            conn.execute("INSERT INTO pinned_facts (channel, fact) VALUES (?, ?)", (channel, fact))
+
+    def pinned_for(self, channel: str) -> List[Dict[str, Any]]:
+        with closing(self._connect()) as conn:
+            rows = conn.execute(
+                "SELECT id, channel, fact FROM pinned_facts WHERE channel IS NULL OR channel = ? ORDER BY id", (channel,)
+            ).fetchall()
+        return [{"id": row[0], "channel": row[1], "fact": row[2]} for row in rows]
+
+    def delete_pinned(self, pin_id: int) -> int:
+        with closing(self._connect()) as conn, conn:
+            return conn.execute("DELETE FROM pinned_facts WHERE id = ?", (pin_id,)).rowcount
+
+    def add_episode(self, channel: str, summary: str) -> None:
+        with closing(self._connect()) as conn, conn:
+            conn.execute("INSERT INTO episodes (channel, summary) VALUES (?, ?)", (channel, summary))
+
+    def recent_episodes(self, channel: str, count: int) -> List[str]:
+        with closing(self._connect()) as conn:
+            rows = conn.execute("SELECT summary FROM episodes WHERE channel = ? ORDER BY id DESC LIMIT ?", (channel, count)).fetchall()
+        return [row[0] for row in reversed(rows)]
+
+    def trim_episodes(self, channel: str, keep: int) -> int:
+        with closing(self._connect()) as conn, conn:
+            cursor = conn.execute(
+                "DELETE FROM episodes WHERE channel = ? AND id NOT IN (SELECT id FROM episodes WHERE channel = ? ORDER BY id DESC LIMIT ?)",
+                (channel, channel, keep),
+            )
+        return cursor.rowcount
+
+    def purge_user_memory(self, user: str) -> int:
+        """Facts, observations and episode traces of one user (MEM-09)."""
+        removed = 0
+        with closing(self._connect()) as conn, conn:
+            removed += conn.execute("DELETE FROM user_facts WHERE user = ?", (user,)).rowcount
+            removed += conn.execute("DELETE FROM observations WHERE user = ?", (user,)).rowcount
+            removed += conn.execute("DELETE FROM episodes WHERE summary LIKE ?", (f"%{user}%",)).rowcount
+        return removed
+
     def migrate_pickles(self, channel: str, history_file: Path, memory_file: Path) -> None:
         """Import legacy pickles once; rename them *.migrated (PER-03)."""
         if self.load_history(channel) or self.load_memory(channel) is not None:
@@ -105,7 +208,9 @@ class PersistentStore:
         if memory_file.exists():
             try:
                 with open(memory_file, "rb") as fd:
-                    self.save_memory(channel, str(pickle.load(fd)))
+                    legacy_memory = str(pickle.load(fd))
+                self.save_memory(channel, legacy_memory)
+                self.add_episode(channel, legacy_memory)  # MEM-08
                 memory_file.rename(memory_file.with_name(memory_file.name + ".migrated"))
                 logging.info(f"migrated legacy memory pickle for {channel}")
             except Exception as err: