diff --git a/DECISIONS.md b/DECISIONS.md index 0a5876c..b9b5a5e 100644 --- a/DECISIONS.md +++ b/DECISIONS.md @@ -36,3 +36,10 @@ Decisions inside the set architecture. D-NNN, never renumbered. repair path is gone; the whole translate-before-draw step dies in FDB-009 (gpt-image-2 is multilingual). Not worth a config rename for one phase. +- **D-010** — Persistence uses stdlib `sqlite3` via + `asyncio.to_thread`, not aiosqlite: no new dependency, and a + connection-per-operation with WAL is plenty at this message volume. +- **D-011** — `save_history` replaces the channel's rows wholesale + per message instead of appending: histories are capped at + `history-limit` (~200-350 rows) and trims must be reflected; + correctness over micro-optimization. diff --git a/fjerkroa_bot/ai_responder.py b/fjerkroa_bot/ai_responder.py index 2b5130f..6117aab 100644 --- a/fjerkroa_bot/ai_responder.py +++ b/fjerkroa_bot/ai_responder.py @@ -12,6 +12,8 @@ from pathlib import Path from pprint import pformat from typing import Any, Dict, List, Optional, Tuple, Union +from .persistence import PersistentStore + def pp(*args, **kw): if "width" not in kw: @@ -138,17 +140,16 @@ class AIResponder(AIResponderBase): self.history: List[Dict[str, Any]] = [] self.memory: str = "I am an assistant." self.rate_limit_backoff = exponential_backoff() - self.history_file: Optional[Path] = None - self.memory_file: Optional[Path] = None + self.store: Optional[PersistentStore] = None if "history-directory" in self.config: - self.history_file = Path(self.config["history-directory"]).expanduser() / f"{self.channel}.dat" - if self.history_file.exists(): - with open(self.history_file, "rb") as fd: - self.history = pickle.load(fd) - self.memory_file = Path(self.config["history-directory"]).expanduser() / f"{self.channel}.memory" - if self.memory_file.exists(): - with open(self.memory_file, "rb") as fd: - self.memory = pickle.load(fd) + directory = Path(self.config["history-directory"]).expanduser() + self.store = PersistentStore(directory / "bot.db") + # Legacy pickles import once, then live on as *.migrated (PER-03) + self.store.migrate_pickles(self.channel, directory / f"{self.channel}.dat", directory / f"{self.channel}.memory") + self.history = self.store.load_history(self.channel) + stored_memory = self.store.load_memory(self.channel) + if stored_memory is not None: + self.memory = stored_memory logging.info(f"memmory:\n{self.memory}") def message(self, message: AIMessage, limit: Optional[int] = None) -> List[Dict[str, Any]]: @@ -225,9 +226,6 @@ class AIResponder(AIResponderBase): self.history.append({"role": "user", "content": str(message)}) while len(self.history) > limit: self.shrink_history_by_one() - if self.history_file is not None: - with open(self.history_file, "wb") as fd: - pickle.dump(self.history, fd) return True return False @@ -269,15 +267,17 @@ class AIResponder(AIResponderBase): self.history.append(answer) while len(self.history) > limit: self.shrink_history_by_one() - if self.history_file is not None: - with open(self.history_file, "wb") as fd: - pickle.dump(self.history, fd) def update_memory(self, memory) -> None: self.memory = memory - if self.memory_file is not None: - with open(self.memory_file, "wb") as fd: - pickle.dump(self.memory, fd) + + 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)): @@ -299,6 +299,7 @@ class AIResponder(AIResponderBase): 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> ") @@ -312,6 +313,7 @@ class AIResponder(AIResponderBase): # Check if a short path applies, return an empty AIResponse if it does if self.short_path(message, limit): + await self._persist_history() return AIResponse(None, False, None, None, None, False, False) # Number of retries for sending the message; failed attempts are @@ -347,8 +349,9 @@ class AIResponder(AIResponderBase): answer_message = await self.post_process(message, response) answer["content"] = str(answer_message) - # Update message history + # Update message history; persistence runs off the loop (PER-05) self.update_history(messages[-1], answer, limit, message.historise_question) + await self._persist_history() logging.info(f"got this answer:\n{str(answer_message)}") # Update memory diff --git a/fjerkroa_bot/discord_bot.py b/fjerkroa_bot/discord_bot.py index 83c17e3..5d770a4 100644 --- a/fjerkroa_bot/discord_bot.py +++ b/fjerkroa_bot/discord_bot.py @@ -236,14 +236,27 @@ class FjerkroaBot(commands.Bot): ) def on_config_file_modified(self, event): - if event.src_path == self.config_file: - new_config = self.load_config(self.config_file) - if repr(new_config) != repr(self.config): - logging.info(f"config file {self.config_file} changed, reloading.") - self.config = new_config - self.airesponder.config = self.config - for responder in self.aichannels.values(): - responder.config = self.config + # Runs on the watchdog observer thread — the swap itself is + # scheduled onto the event loop so no request reads a + # half-swapped config (CFG-04 / D9) + if event.src_path != self.config_file: + return + new_config = self.load_config(self.config_file) + if repr(new_config) == repr(self.config): + return + logging.info(f"config file {self.config_file} changed, reloading.") + + def apply() -> None: + self.config = new_config + self.airesponder.config = new_config + for responder in self.aichannels.values(): + responder.config = new_config + + try: + self.loop.call_soon_threadsafe(apply) + except (RuntimeError, AttributeError): + # event loop not running yet (startup) — no concurrent readers + apply() @classmethod def load_config(cls, config_file: str = "config.toml"): diff --git a/fjerkroa_bot/persistence.py b/fjerkroa_bot/persistence.py new file mode 100644 index 0000000..dae1ecc --- /dev/null +++ b/fjerkroa_bot/persistence.py @@ -0,0 +1,87 @@ +"""SQLite persistence for history + memory (SPEC-009). + +One database per deployment, stdlib sqlite3 only (D-010). Callers run +the sync methods in a worker thread (`asyncio.to_thread`) so the +event loop never blocks on disk (PER-05 / D6). +""" + +import logging +import os +import pickle +import sqlite3 +from contextlib import closing +from pathlib import Path +from typing import Any, Dict, List, Optional + +SCHEMA_VERSION = 1 + + +class PersistentStore: + def __init__(self, db_path: Path) -> None: + self.db_path = Path(db_path) + self._init_db() + + def _connect(self) -> sqlite3.Connection: + conn = sqlite3.connect(self.db_path) + conn.execute("PRAGMA journal_mode=WAL") + return conn + + def _init_db(self) -> None: + self.db_path.parent.mkdir(parents=True, exist_ok=True) + with closing(self._connect()) as conn, conn: + if conn.execute("PRAGMA user_version").fetchone()[0] < SCHEMA_VERSION: + conn.execute( + "CREATE TABLE IF NOT EXISTS history (id INTEGER PRIMARY KEY, channel TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL)" + ) + conn.execute("CREATE INDEX IF NOT EXISTS history_channel ON history (channel)") + conn.execute("CREATE TABLE IF NOT EXISTS memory (channel TEXT PRIMARY KEY, content TEXT NOT NULL)") + conn.execute(f"PRAGMA user_version = {SCHEMA_VERSION}") + os.chmod(self.db_path, 0o600) # conversation data (PER-04) + + def load_history(self, channel: str) -> List[Dict[str, Any]]: + with closing(self._connect()) as conn: + rows = conn.execute("SELECT role, content FROM history WHERE channel = ? ORDER BY id", (channel,)).fetchall() + return [{"role": role, "content": content} for role, content in rows] + + def save_history(self, channel: str, history: List[Dict[str, Any]]) -> None: + # Full replace per save: histories are small (<= history-limit) + # and trims must be reflected (D-011) + with closing(self._connect()) as conn, conn: + conn.execute("DELETE FROM history WHERE channel = ?", (channel,)) + conn.executemany( + "INSERT INTO history (channel, role, content) VALUES (?, ?, ?)", + [(channel, str(entry.get("role", "user")), str(entry.get("content", ""))) for entry in history], + ) + + def load_memory(self, channel: str) -> Optional[str]: + with closing(self._connect()) as conn: + row = conn.execute("SELECT content FROM memory WHERE channel = ?", (channel,)).fetchone() + return row[0] if row else None + + def save_memory(self, channel: str, content: str) -> None: + with closing(self._connect()) as conn, conn: + conn.execute( + "INSERT INTO memory (channel, content) VALUES (?, ?) ON CONFLICT(channel) DO UPDATE SET content = excluded.content", + (channel, content), + ) + + 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: + return + if history_file.exists(): + try: + with open(history_file, "rb") as fd: + self.save_history(channel, pickle.load(fd)) + history_file.rename(history_file.with_name(history_file.name + ".migrated")) + logging.info(f"migrated legacy history pickle for {channel}") + except Exception as err: + logging.error(f"failed to migrate history pickle {history_file}: {err!r}") + if memory_file.exists(): + try: + with open(memory_file, "rb") as fd: + self.save_memory(channel, str(pickle.load(fd))) + memory_file.rename(memory_file.with_name(memory_file.name + ".migrated")) + logging.info(f"migrated legacy memory pickle for {channel}") + except Exception as err: + logging.error(f"failed to migrate memory pickle {memory_file}: {err!r}") diff --git a/specs/SPEC-008-config.md b/specs/SPEC-008-config.md index 2c8e7ba..d642625 100644 --- a/specs/SPEC-008-config.md +++ b/specs/SPEC-008-config.md @@ -21,3 +21,12 @@ can speak differently per channel. When `news` points to a non-existent file, the `{news}` placeholder stays literal in the system prompt (no crash, no empty substitution). + +### CFG-04 — Config hot-reload applies on the event loop (coverage: test) + +The watchdog observer thread never mutates live config references +itself: a detected change is loaded, then the swap of `bot.config` +and all responder `.config` references is scheduled onto the event +loop (`call_soon_threadsafe`), so no request ever reads a +half-swapped config (D9). Before the loop runs (startup), the swap +applies directly — there are no concurrent readers yet. diff --git a/specs/SPEC-009-persistence.md b/specs/SPEC-009-persistence.md new file mode 100644 index 0000000..e2cc1ef --- /dev/null +++ b/specs/SPEC-009-persistence.md @@ -0,0 +1,38 @@ +# SPEC-009 — Persistence + +One SQLite database per deployment (`/bot.db`), +replacing the per-channel pickle files (`.dat`, +`.memory`). Stdlib `sqlite3` via worker threads — no new +dependency (D-010). Without `history-directory` in config the bot +runs memory-only, as before. + +### PER-01 — History survives restarts (coverage: test) + +History entries written through the store are returned, in order and +per channel, by a fresh store instance on the same database file. + +### PER-02 — Memory survives restarts (coverage: test) + +The per-channel memory string written through the store is returned +by a fresh store instance on the same database file. + +### PER-03 — Existing pickles migrate exactly once (coverage: test) + +On responder start, when the database holds no rows for the channel +and legacy pickle files exist, their content is imported and the +pickle files are renamed to `*.migrated` (kept for rollback). A +second start does not re-import. Users keep their history through +the cutover (review consensus: migration is a decision, not an +accident). + +### PER-04 — Database hygiene (coverage: test) + +The database runs in WAL journal mode, carries `PRAGMA user_version` += schema version (currently 1) for future migrations/rollback +policy, and the file is chmod 0600 (it stores conversation data). + +### PER-05 — Writes run off the event loop (coverage: test) + +History and memory persistence happen in a worker thread +(`asyncio.to_thread`) — a slow disk cannot stall Discord event +handling (D6). diff --git a/tests/test_ai.py b/tests/test_ai.py index 3c92471..cd26aaf 100644 --- a/tests/test_ai.py +++ b/tests/test_ai.py @@ -1,6 +1,3 @@ -import os -import pickle -import tempfile import unittest from unittest.mock import Mock, patch @@ -147,7 +144,6 @@ You always try to say something positive about the current day and the Fjærkroa def test_update_history(self) -> None: updater = self.bot.airesponder updater.history = [] - updater.history_file = None question = {"content": '{"channel": "test_channel", "message": "What is the meaning of life?"}'} answer = {"content": '{"channel": "test_channel", "message": "42"}'} @@ -179,15 +175,7 @@ You always try to say something positive about the current day and the Fjærkroa next_answer2 = {"content": '{"channel": "other_channel", "message": "Tripple Z"}'} updater.update_history(next_question2, next_answer2, 4) self.assertEqual(updater.history, [new_answer, next_answer, next_question2, next_answer2]) - - # Test case 5: Check history file save using mock - with unittest.mock.patch("builtins.open", unittest.mock.mock_open()) as mock_file: - _, temp_path = tempfile.mkstemp() - os.remove(temp_path) - self.bot.airesponder.history_file = temp_path - updater.update_history(question, answer, 2) - mock_file.assert_called_with(temp_path, "wb") - mock_file().write.assert_called_with(pickle.dumps([question, answer])) + # File persistence moved to the SQLite store — covered by PER-01 (SPEC-009) if __name__ == "__mait__": diff --git a/tests/test_main.py b/tests/test_main.py index 74f921b..afd3c9d 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -107,17 +107,13 @@ class TestFunctionality(TestBotBase): ' "channel": "some_channel", "direct": false, "historise_question": true}', ) - @patch("builtins.open", new_callable=mock_open) - def test_update_history_with_file(self, mock_file): + def test_update_history_trims_to_limit(self): self.bot.airesponder.update_history({"content": '{"q": "What\'s your name?"}'}, {"content": '{"a": "AI"}'}, 10) self.assertEqual(len(self.bot.airesponder.history), 2) self.bot.airesponder.update_history({"content": '{"q1": "Q1"}'}, {"content": '{"a1": "A1"}'}, 2) self.bot.airesponder.update_history({"content": '{"q2": "Q2"}'}, {"content": '{"a2": "A2"}'}, 2) self.assertEqual(len(self.bot.airesponder.history), 2) - self.bot.airesponder.history_file = "mock_file.pkl" - self.bot.airesponder.update_history({"content": '{"q": "What\'s your favorite color?"}'}, {"content": '{"a": "Blue"}'}, 10) - mock_file.assert_called_once_with("mock_file.pkl", "wb") - mock_file().write.assert_called_once() + # File persistence moved to the SQLite store — covered by PER-01 (SPEC-009) if __name__ == "__mait__": diff --git a/tests/test_spec_per.py b/tests/test_spec_per.py new file mode 100644 index 0000000..3a0e0db --- /dev/null +++ b/tests/test_spec_per.py @@ -0,0 +1,143 @@ +"""Unit coverage for SPEC-009 persistence (PER-01..05) + CFG-04 (D9).""" + +import pickle +import sqlite3 +import stat +import tempfile +import threading +import unittest +from pathlib import Path +from unittest.mock import MagicMock + +from fjerkroa_bot.ai_responder import AIMessage, AIResponder +from fjerkroa_bot.persistence import PersistentStore + +from .test_bdd_envelope import FakeModelResponder, envelope +from .test_main import TestBotBase + + +class StoreBase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = Path(self.tmp.name) / "bot.db" + + def tearDown(self): + self.tmp.cleanup() + + +class TestHistoryRoundtrip(StoreBase): + def test_history_survives_restart(self): + """PER-01: history rows come back per channel, in order, from a fresh store.""" + store = PersistentStore(self.db_path) + entries = [{"role": "user", "content": "one"}, {"role": "assistant", "content": "two"}] + store.save_history("chat", entries) + store.save_history("other", [{"role": "user", "content": "elsewhere"}]) + reloaded = PersistentStore(self.db_path) + self.assertEqual(reloaded.load_history("chat"), entries) + self.assertEqual(reloaded.load_history("other"), [{"role": "user", "content": "elsewhere"}]) + + def test_save_replaces_previous_state(self): + """PER-01: save_history reflects trims — replaced, not appended.""" + store = PersistentStore(self.db_path) + store.save_history("chat", [{"role": "user", "content": "a"}, {"role": "user", "content": "b"}]) + store.save_history("chat", [{"role": "user", "content": "b"}]) + self.assertEqual(store.load_history("chat"), [{"role": "user", "content": "b"}]) + + +class TestMemoryRoundtrip(StoreBase): + def test_memory_survives_restart(self): + """PER-02: per-channel memory string persists across store instances.""" + store = PersistentStore(self.db_path) + store.save_memory("chat", "remember this") + self.assertEqual(PersistentStore(self.db_path).load_memory("chat"), "remember this") + self.assertIsNone(PersistentStore(self.db_path).load_memory("unknown")) + + +class TestPickleMigration(StoreBase): + def test_pickles_migrate_once(self): + """PER-03: legacy pickles import once, files renamed *.migrated, no re-import.""" + history_file = Path(self.tmp.name) / "chat.dat" + memory_file = Path(self.tmp.name) / "chat.memory" + with open(history_file, "wb") as fd: + pickle.dump([{"role": "user", "content": "old times"}], fd) + with open(memory_file, "wb") as fd: + pickle.dump("old memory", fd) + + config = {"system": "s", "history-limit": 5, "history-directory": self.tmp.name} + responder = AIResponder(config, "chat") + self.assertEqual(responder.history, [{"role": "user", "content": "old times"}]) + self.assertEqual(responder.memory, "old memory") + self.assertFalse(history_file.exists()) + self.assertFalse(memory_file.exists()) + self.assertTrue(history_file.with_suffix(".dat.migrated").exists()) + + # second start reads from the DB, does not re-import + responder2 = AIResponder(config, "chat") + self.assertEqual(responder2.history, [{"role": "user", "content": "old times"}]) + + +class TestDatabaseHygiene(StoreBase): + def test_wal_version_and_permissions(self): + """PER-04: WAL mode, user_version = 1, file mode 0600.""" + PersistentStore(self.db_path) + conn = sqlite3.connect(self.db_path) + try: + self.assertEqual(conn.execute("PRAGMA journal_mode").fetchone()[0], "wal") + self.assertEqual(conn.execute("PRAGMA user_version").fetchone()[0], 1) + finally: + conn.close() + mode = stat.S_IMODE(self.db_path.stat().st_mode) + self.assertEqual(mode, 0o600) + + +class TestWritesOffEventLoop(unittest.IsolatedAsyncioTestCase): + async def test_persistence_runs_in_worker_thread(self): + """PER-05: history writes happen off the event loop (D6).""" + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + config = {"system": "s", "history-limit": 5, "history-directory": tmp.name} + responder = FakeModelResponder(config, "chat") + writing_threads = set() + original_save = responder.store.save_history + + def capture_save(channel, history): + writing_threads.add(threading.current_thread()) + return original_save(channel, history) + + responder.store.save_history = capture_save + responder.scripted.append(envelope(answer="hei", answer_needed=True)) + await responder.send(AIMessage("alice", "hei", "chat")) + self.assertTrue(writing_threads) + self.assertNotIn(threading.main_thread(), writing_threads) + # and the data actually landed + self.assertTrue(PersistentStore(Path(tmp.name) / "bot.db").load_history("chat")) + + +class TestConfigReloadRace(TestBotBase): + async def test_reload_swaps_on_event_loop(self): + """CFG-04: watchdog thread schedules the config swap via call_soon_threadsafe (D9).""" + new_config = dict(self.config_data) + new_config["history-limit"] = 99 + self.bot.load_config = lambda path: new_config + self.bot.loop = MagicMock() + event = MagicMock() + event.src_path = self.bot.config_file + self.bot.on_config_file_modified(event) + self.bot.loop.call_soon_threadsafe.assert_called_once() + apply_fn = self.bot.loop.call_soon_threadsafe.call_args.args[0] + apply_fn() + self.assertEqual(self.bot.config["history-limit"], 99) + self.assertEqual(self.bot.airesponder.config["history-limit"], 99) + + async def test_reload_applies_directly_without_loop(self): + """CFG-04: before the loop runs, the swap applies synchronously.""" + new_config = dict(self.config_data) + new_config["history-limit"] = 42 + self.bot.load_config = lambda path: new_config + loop = MagicMock() + loop.call_soon_threadsafe.side_effect = RuntimeError("no running loop") + self.bot.loop = loop + event = MagicMock() + event.src_path = self.bot.config_file + self.bot.on_config_file_modified(event) + self.assertEqual(self.bot.config["history-limit"], 42)