Oleksandr Kozachuk
187 lines
9.4 KiB
Python
187 lines
9.4 KiB
Python
"""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")
|