From 3d2289496b1d6ae1b2fba89de930cf1253b9306f Mon Sep 17 00:00:00 2001 From: Oleksandr Kozachuk Date: Mon, 13 Jul 2026 14:12:21 +0200 Subject: [PATCH] structured memory: facts/pinned/episodes, batched consolidation, participant-scoped recall --- DECISIONS.md | 12 +- config.toml | 6 + fjerkroa_bot/ai_responder.py | 43 +++--- fjerkroa_bot/discord_bot.py | 54 +++++--- fjerkroa_bot/memory.py | 95 +++++++++++++ fjerkroa_bot/openai_responder.py | 68 +++++++--- fjerkroa_bot/persistence.py | 109 ++++++++++++++- specs/SPEC-001-responder-envelope.md | 14 +- specs/SPEC-002-memory.md | 79 +++++++++++ specs/SPEC-003-safety.md | 5 +- tests/test_bdd_envelope.py | 4 +- tests/test_openai_responder_simple.py | 11 +- tests/test_spec_defects.py | 12 +- tests/test_spec_mem.py | 186 ++++++++++++++++++++++++++ 14 files changed, 608 insertions(+), 90 deletions(-) create mode 100644 fjerkroa_bot/memory.py create mode 100644 specs/SPEC-002-memory.md create mode 100644 tests/test_spec_mem.py diff --git a/DECISIONS.md b/DECISIONS.md index 1749038..107e8f3 100644 --- a/DECISIONS.md +++ b/DECISIONS.md @@ -52,4 +52,14 @@ Decisions inside the set architecture. D-NNN, never renumbered. - **D-013** — `!forgetme` v1 purges history rows only; the single-string channel memory cannot be selectively cleaned. Full fact-level erasure ships with FDB-007 structured memory — stated in - the user-facing confirmation, not hidden. + the user-facing confirmation, not hidden. (Superseded by D-014: + erasure now covers facts, observations and episode traces.) +- **D-014** — Structured memory (FDB-007): observations are the only + consolidation feed (independent of history trimming); consolidation + returns NEW facts only (no wholesale rewrite — the lossiness of the + old memoize path is exactly what we're removing); self-authorship + is enforced in code (fact subject must be an observation author), + not just in the prompt; memory reads run on the loop (small indexed + SQLite queries), writes off-loop. Legacy memory strings survive as + episodes; the memory table stays as a read-only legacy fallback for + deployments without memory-model. diff --git a/config.toml b/config.toml index b4605cf..c8fe204 100644 --- a/config.toml +++ b/config.toml @@ -36,3 +36,9 @@ enable-game-info = true # user-daily-images = 10 # Privacy (SAF-08/09): users can always run !forgetme and !privacy # privacy-notice = "I keep recent messages and a summary. !forgetme deletes yours." +# Structured memory (SPEC-002) — active only when memory-model is set: +# memory-model = "gpt-5.6-luna" +# memory-consolidate-every = 20 # observations per consolidation batch +# memory-episodes-per-channel = 10 # episode decay cap +# memory-fact-retention-days = 180 # GDPR storage limitation +# Staff: !bot memory | forget-fact | pin | unpin diff --git a/fjerkroa_bot/ai_responder.py b/fjerkroa_bot/ai_responder.py index 6117aab..12ab211 100644 --- a/fjerkroa_bot/ai_responder.py +++ b/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 diff --git a/fjerkroa_bot/discord_bot.py b/fjerkroa_bot/discord_bot.py index eccb0eb..a00f89b 100644 --- a/fjerkroa_bot/discord_bot.py +++ b/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 , 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 , status, spend, memory , forget-fact , pin , unpin " 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 diff --git a/fjerkroa_bot/memory.py b/fjerkroa_bot/memory.py new file mode 100644 index 0000000..d775a0c --- /dev/null +++ b/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) diff --git a/fjerkroa_bot/openai_responder.py b/fjerkroa_bot/openai_responder.py index 7da9999..4c54148 100644 --- a/fjerkroa_bot/openai_responder.py +++ b/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]]: """ diff --git a/fjerkroa_bot/persistence.py b/fjerkroa_bot/persistence.py index 213676c..044d3a3 100644 --- a/fjerkroa_bot/persistence.py +++ b/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: diff --git a/specs/SPEC-001-responder-envelope.md b/specs/SPEC-001-responder-envelope.md index 5f5ad30..11c82d3 100644 --- a/specs/SPEC-001-responder-envelope.md +++ b/specs/SPEC-001-responder-envelope.md @@ -69,9 +69,10 @@ is the channel the message came from. ### ENV-10 — System prompt template substitution (coverage: test) `message()` substitutes `{date}` (YYYY-MM-DD), `{time}`, `{memory}` -(current memory string) in the system prompt; `{news}` is replaced -with the news file content when the configured file exists and stays -literal when it does not. +(the assembled memory block — legacy memory string while the +structured memory is inactive, see MEM-10) in the system prompt; +`{news}` is replaced with the news file content when the configured +file exists and stays literal when it does not. ### ENV-11 — Per-channel history shrink prefers busy channels (coverage: test) @@ -92,10 +93,11 @@ back-to-back (D1). records the clearing in the channel's memory. (D7: the previous handler declared `(reaction, user)` and crashed on dispatch.) -### ENV-14 — update_memory persists its argument (coverage: test) +### ENV-14 — update_memory persists its argument (coverage: withdrawn — successor SPEC-002) -`update_memory(memory)` sets the responder's memory to the passed -value and persists that value. (D12: the argument was ignored.) +Withdrawn 2026-07-13 with FDB-007: the per-answer memory rewrite +(`update_memory`/`memoize`) is deleted; structured memory (MEM-01+) +replaces it. The D12 defect died with the code. ### ENV-15 — retry-model is used after a rate limit (coverage: test) diff --git a/specs/SPEC-002-memory.md b/specs/SPEC-002-memory.md new file mode 100644 index 0000000..5a57a7a --- /dev/null +++ b/specs/SPEC-002-memory.md @@ -0,0 +1,79 @@ +# SPEC-002 — Structured memory + +Replaces the single-string per-answer LLM rewrite (lossy, O(convo) +cost — the old `memoize` path). Layers, simplified per plan v4: +**user facts** (durable, provenance-tracked), **pinned facts** +(operator-set, global or per channel), **episodes** (rolling channel +summaries with decay). Channel-facts deferred until recall proves +insufficient. Raw feed = **observations** (messages, reactions, +edits, deletes); an async consolidation pass turns observations into +facts + episodes. The memory system is active only when both a store +(`history-directory`) and a `memory-model` are configured — otherwise +the legacy memory string is used read-only (MEM-10). + +### MEM-01 — Events become observation rows (coverage: test) + +User messages, bot answers, reactions (add/remove/clear), edits and +deletes are recorded as observation rows (channel, user, kind, +content excerpt) — cheap writes, no LLM call per event. + +### MEM-02 — Consolidation is batched, never per message (coverage: test) + +Consolidation triggers when `memory-consolidate-every` (default 20) +unconsumed observations have accumulated for a channel; it runs as a +background task guarded by a lock (no overlapping runs), consumes the +observations it processed, and leaves them in place when the model +call fails (retry next trigger). + +### MEM-03 — Only self-authored facts persist (coverage: test) + +The consolidator stores a fact only when its subject user is among +the authors of the consumed observations, with provenance +`source='self'`; facts the model attributes to absent third parties +are discarded and logged. Operator pins carry `source='operator'`. +"Bob says Alice likes X" must never become Alice's profile +(review consensus C2). + +### MEM-04 — Recall is participant-scoped (coverage: test) + +The memory block assembled into the system prompt contains: pinned +facts (global + this channel), user facts of **conversation +participants only** (current author + authors in the recent history +tail), and the channel's recent episodes. Facts of non-participants +never enter the prompt — the model cannot leak what it cannot see. + +### MEM-05 — Episodes decay (coverage: test) + +At most `memory-episodes-per-channel` (default 10) episodes are kept +per channel; consolidation drops the oldest beyond the cap. + +### MEM-06 — User facts have a retention limit (coverage: test) + +Facts not updated within `memory-fact-retention-days` (default 180) +are purged during consolidation. Durable is not indefinite (GDPR +storage limitation). + +### MEM-07 — Staff review and edit memory (coverage: test) + +Staff commands: `!bot memory ` lists the user's facts with ids; +`!bot forget-fact ` deletes one; `!bot pin +` adds an operator pin; `!bot unpin ` removes one. + +### MEM-08 — Legacy memory strings migrate to episodes (coverage: test) + +Schema v3 migration copies existing per-channel memory strings into +an episode row each; pickle migration does the same. Deployments keep +their accumulated context through the upgrade. + +### MEM-09 — !forgetme erases facts, observations and episode traces (coverage: test) + +`!forgetme` now deletes the user's facts, their observation rows, and +episodes mentioning the user's name — in addition to the SAF-08 +history purge. This completes the erasure that SAF-08 v1 could not. + +### MEM-10 — Memory system off degrades gracefully (coverage: test) + +Without `memory-model` (or without a store) no observations are +written, no consolidation runs, and `{memory}` falls back to the +legacy memory string — no crash, no behavior change for +unconfigured deployments. diff --git a/specs/SPEC-003-safety.md b/specs/SPEC-003-safety.md index f3c8e93..fcc2ef2 100644 --- a/specs/SPEC-003-safety.md +++ b/specs/SPEC-003-safety.md @@ -66,9 +66,8 @@ still goes out). `!forgetme` removes the requesting user's messages from all live responder histories and from the store, then confirms in-channel. -v1 limitation, stated in the confirmation: the single-string channel -memory may still contain summarized traces — full fact-level erasure -arrives with the structured memory (FDB-007). +Since FDB-007 the purge extends to memory itself — facts, +observations and episode traces (MEM-09). ### SAF-09 — !privacy states the data practice (coverage: test) diff --git a/tests/test_bdd_envelope.py b/tests/test_bdd_envelope.py index ee41e44..87cfe7f 100644 --- a/tests/test_bdd_envelope.py +++ b/tests/test_bdd_envelope.py @@ -43,8 +43,8 @@ class FakeModelResponder(AIResponder): return None, limit return {"role": "assistant", "content": self.scripted.pop(0)}, limit - async def memory_rewrite(self, memory: str, message_user: str, answer_user: str, question: str, answer: str) -> str: - return memory + async def consolidate(self, observations, known_facts): + return {"facts": [], "episode": None} async def translate(self, text: str, language: str = "english") -> str: return text diff --git a/tests/test_openai_responder_simple.py b/tests/test_openai_responder_simple.py index 5d889d8..402ed5c 100644 --- a/tests/test_openai_responder_simple.py +++ b/tests/test_openai_responder_simple.py @@ -40,15 +40,12 @@ class TestOpenAIResponderSimple(unittest.IsolatedAsyncioTestCase): self.assertEqual(result, original_text) - async def test_memory_rewrite_no_memory_model(self): - """Test memory rewrite when no memory-model is configured.""" + async def test_consolidate_no_memory_model(self): + """MEM-10: without memory-model, consolidation is a no-op returning None.""" config_no_memory = {"openai-key": "test", "model": "gpt-4"} responder = OpenAIResponder(config_no_memory) - - original_memory = "Old memory" - result = await responder.memory_rewrite(original_memory, "user1", "assistant", "question", "answer") - - self.assertEqual(result, original_memory) + result = await responder.consolidate([{"id": 1, "user": "u", "kind": "message", "content": "x"}], []) + self.assertIsNone(result) if __name__ == "__main__": diff --git a/tests/test_spec_defects.py b/tests/test_spec_defects.py index ebfed0c..e26b0b1 100644 --- a/tests/test_spec_defects.py +++ b/tests/test_spec_defects.py @@ -40,17 +40,9 @@ class TestReactionClear(TestBotBase): message.content = "Some message text" message.author.name = "alice" message.channel = MagicMock(spec=TextChannel) - self.bot.airesponder.memoize = AsyncMock() + self.bot.airesponder.observe_event = AsyncMock() await self.bot.on_reaction_clear(message, [Mock()]) - self.bot.airesponder.memoize.assert_awaited_once() - - -class TestUpdateMemory(unittest.TestCase): - def test_update_memory_uses_argument(self): - """ENV-14: update_memory persists the passed value, not stale state (D12).""" - responder = AIResponder({"system": "s", "history-limit": 5}, "chat") - responder.update_memory("NEW MEMORY") - self.assertEqual(responder.memory, "NEW MEMORY") + self.bot.airesponder.observe_event.assert_awaited_once() class TestRetryModel(unittest.IsolatedAsyncioTestCase): diff --git a/tests/test_spec_mem.py b/tests/test_spec_mem.py new file mode 100644 index 0000000..35c46ad --- /dev/null +++ b/tests/test_spec_mem.py @@ -0,0 +1,186 @@ +"""Unit coverage for SPEC-002 structured memory (MEM-01..10).""" + +import sqlite3 +import tempfile +import unittest +from pathlib import Path +from unittest.mock import AsyncMock + +from fjerkroa_bot.memory import MemoryManager +from fjerkroa_bot.persistence import PersistentStore + +from .test_bdd_envelope import FakeModelResponder +from .test_spec_ops import OpsBase + + +class MemBase(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.store = PersistentStore(Path(self.tmp.name) / "bot.db") + self.config = {"memory-model": "luna", "memory-consolidate-every": 3, "memory-episodes-per-channel": 2} + self.consolidator = AsyncMock(return_value={"facts": [], "episode": None}) + self.manager = MemoryManager(self.store, lambda: self.config, self.consolidator, "chat") + + +class TestObservations(MemBase): + async def test_events_become_observation_rows(self): + """MEM-01: observe() writes channel/user/kind/content rows.""" + await self.manager.observe("alice", "message", "hei") + await self.manager.observe("bob", "reaction-adding", "👍 on alice: hei") + rows = self.store.peek_observations("chat") + self.assertEqual([(row["user"], row["kind"]) for row in rows], [("alice", "message"), ("bob", "reaction-adding")]) + + +class TestConsolidationBatching(MemBase): + async def test_triggers_at_batch_size_and_consumes(self): + """MEM-02: consolidation fires at the configured batch size and consumes rows.""" + self.consolidator.return_value = {"facts": [], "episode": "they said hi"} + for i in range(3): + await self.manager.observe("alice", "message", f"msg {i}") + await self.manager.consolidate_now() + self.consolidator.assert_awaited() + self.assertEqual(self.store.peek_observations("chat"), []) + self.assertEqual(self.store.recent_episodes("chat", 5), ["they said hi"]) + + async def test_failed_model_call_keeps_observations(self): + """MEM-02: consolidator returning None leaves observations for the next run.""" + self.consolidator.return_value = None + await self.manager.observe("alice", "message", "hei") + await self.manager.consolidate_now() + self.assertEqual(len(self.store.peek_observations("chat")), 1) + + +class TestSelfAuthoredOnly(MemBase): + async def test_third_party_facts_dropped(self): + """MEM-03: facts about users absent from the observations are discarded.""" + self.consolidator.return_value = { + "facts": [{"user": "alice", "fact": "likes espresso"}, {"user": "charlie", "fact": "owes bob money"}], + "episode": None, + } + await self.manager.observe("alice", "message", "I love espresso") + await self.manager.consolidate_now() + facts = self.store.facts_for(["alice", "charlie"]) + self.assertEqual(len(facts), 1) + self.assertEqual((facts[0]["user"], facts[0]["fact"]), ("alice", "likes espresso")) + + +class TestRecallScope(MemBase): + async def test_block_is_participant_scoped(self): + """MEM-04: only participants' facts + pinned + episodes enter the block.""" + self.store.add_user_fact("alice", "likes espresso", "self") + self.store.add_user_fact("mallory", "secret fact", "self") + self.store.add_pinned(None, "Fjerkroa opens at 10") + self.store.add_episode("chat", "yesterday they planned a trip") + block = self.manager.memory_block(["alice", "bob"], "LEGACY") + self.assertIn("likes espresso", block) + self.assertIn("Fjerkroa opens at 10", block) + self.assertIn("planned a trip", block) + self.assertNotIn("secret fact", block) + self.assertNotIn("LEGACY", block) + + +class TestEpisodeDecay(MemBase): + async def test_episodes_capped(self): + """MEM-05: oldest episodes beyond the per-channel cap are dropped.""" + self.consolidator.return_value = {"facts": [], "episode": "ep-final"} + for i in range(4): + self.store.add_episode("chat", f"ep-{i}") + await self.manager.observe("alice", "message", "hei") + await self.manager.consolidate_now() + episodes = self.store.recent_episodes("chat", 10) + self.assertEqual(len(episodes), 2) # memory-episodes-per-channel = 2 + self.assertEqual(episodes[-1], "ep-final") + + +class TestFactRetention(MemBase): + async def test_old_facts_purged(self): + """MEM-06: facts older than the retention window die at consolidation.""" + self.store.add_user_fact("alice", "fresh", "self") + with sqlite3.connect(self.store.db_path) as conn: + conn.execute( + "INSERT INTO user_facts (user, fact, source, updated_at) VALUES ('alice', 'ancient', 'self', datetime('now', '-400 days'))" + ) + self.config["memory-fact-retention-days"] = 180 + self.consolidator.return_value = {"facts": [], "episode": None} + await self.manager.observe("alice", "message", "hei") + await self.manager.consolidate_now() + facts = [fact["fact"] for fact in self.store.facts_for(["alice"])] + self.assertIn("fresh", facts) + self.assertNotIn("ancient", facts) + + +class TestStaffMemoryCommands(OpsBase): + async def test_pin_list_forget(self): + """MEM-07: !bot memory/forget-fact/pin/unpin work from the staff channel.""" + with tempfile.TemporaryDirectory() as tmp: + store = PersistentStore(Path(tmp) / "bot.db") + self.bot.airesponder.store = store + store.add_user_fact("alice", "likes espresso", "self") + await self.bot.on_message(self.staff_msg("!bot memory alice")) + listing = self.bot.staff_channel.send.await_args.args[0] + self.assertIn("likes espresso", listing) + fact_id = listing.split(":")[0] + await self.bot.on_message(self.staff_msg(f"!bot forget-fact {fact_id}")) + self.assertEqual(store.facts_for(["alice"]), []) + await self.bot.on_message(self.staff_msg("!bot pin global Opening hours 10-22")) + self.assertEqual(len(store.pinned_for("chat")), 1) + pin_id = store.pinned_for("chat")[0]["id"] + await self.bot.on_message(self.staff_msg(f"!bot unpin {pin_id}")) + self.assertEqual(store.pinned_for("chat"), []) + + +class TestLegacyMigration(unittest.TestCase): + def test_v2_memory_strings_become_episodes(self): + """MEM-08: schema v3 migration copies memory strings into episodes.""" + with tempfile.TemporaryDirectory() as tmp: + db_path = Path(tmp) / "bot.db" + conn = sqlite3.connect(db_path) + conn.execute("CREATE TABLE history (id INTEGER PRIMARY KEY, channel TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL)") + conn.execute("CREATE TABLE memory (channel TEXT PRIMARY KEY, content TEXT NOT NULL)") + conn.execute("CREATE TABLE usage (day TEXT NOT NULL, key TEXT NOT NULL, value REAL NOT NULL, PRIMARY KEY (day, key))") + conn.execute("INSERT INTO memory (channel, content) VALUES ('chat', 'old accumulated context')") + conn.execute("PRAGMA user_version = 2") + conn.commit() + conn.close() + store = PersistentStore(db_path) + self.assertEqual(store.recent_episodes("chat", 5), ["old accumulated context"]) + + +class TestForgetmeErasesMemory(OpsBase): + async def test_forgetme_purges_facts_observations_episodes(self): + """MEM-09: !forgetme removes facts, observations and episode traces.""" + with tempfile.TemporaryDirectory() as tmp: + store = PersistentStore(Path(tmp) / "bot.db") + self.bot.airesponder.store = store + store.add_user_fact("alice", "likes espresso", "self") + store.add_observation("chat", "alice", "message", "hei") + store.add_episode("chat", "alice planned a trip with bob") + store.add_episode("chat", "quiet evening, nothing happened") + message = self.public_msg("!forgetme") + message.author.name = "alice" + await self.bot.on_message(message) + self.assertEqual(store.facts_for(["alice"]), []) + self.assertEqual(store.peek_observations("chat"), []) + self.assertEqual(store.recent_episodes("chat", 5), ["quiet evening, nothing happened"]) + + +class TestInactiveMemory(unittest.IsolatedAsyncioTestCase): + async def test_no_memory_model_means_legacy_passthrough(self): + """MEM-10: without memory-model nothing is written and legacy string is used.""" + responder = FakeModelResponder({"system": "s {memory}", "history-limit": 5}, "chat") + responder.memory = "LEGACY STRING" + await responder.observe_event("alice", "message", "hei") # no store, no crash + from fjerkroa_bot.ai_responder import AIMessage + + system = responder.message(AIMessage("alice", "hei", "chat"))[0]["content"] + self.assertIn("LEGACY STRING", system) + + async def test_store_without_memory_model_stays_silent(self): + """MEM-10: store configured but no memory-model -> no observations written.""" + with tempfile.TemporaryDirectory() as tmp: + store = PersistentStore(Path(tmp) / "bot.db") + manager = MemoryManager(store, lambda: {}, AsyncMock(), "chat") + await manager.observe("alice", "message", "hei") + self.assertEqual(store.peek_observations("chat"), []) + self.assertEqual(manager.memory_block(["alice"], "LEGACY"), "LEGACY")