593621c5e0
Implements the missing pieces of the MemPalace epic (#367): - sovereign_store.py: Self-contained memory store replacing the third-party mempalace CLI and its ONNX dependency. Uses: * SQLite + FTS5 for keyword search (porter stemmer, unicode61) * HRR phase vectors (SHA-256 deterministic, numpy optional) for semantic similarity * Reciprocal Rank Fusion to merge keyword and semantic rankings * Trust scoring with boost/decay lifecycle * Room-based organization matching the existing PalaceRoom model - promotion.py (MP-4, #371): Quality-gated scratchpad-to-palace promotion. Four heuristic gates, no LLM call: 1. Length gate (min 5 words, max 500) 2. Structure gate (rejects fragments and pure code) 3. Duplicate gate (FTS5 + Jaccard overlap detection) 4. Staleness gate (7-day threshold for old notes) Includes force override, batch promotion, and audit logging. - 21 unit tests covering HRR vectors, store operations, search, trust lifecycle, and all promotion gates. Zero external dependencies. Zero API calls. Zero cloud. Refs: #367 #370 #371
256 lines
9.7 KiB
Python
256 lines
9.7 KiB
Python
"""Tests for the Sovereign Memory Store and Promotion system.
|
|
|
|
Zero-API, zero-network — everything runs against an in-memory SQLite DB.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
# Allow imports from parent package
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
from sovereign_store import (
|
|
SovereignStore,
|
|
encode_text,
|
|
cosine_similarity_phase,
|
|
serialize_vector,
|
|
deserialize_vector,
|
|
)
|
|
from promotion import (
|
|
evaluate_for_promotion,
|
|
promote,
|
|
promote_session_batch,
|
|
)
|
|
|
|
|
|
class TestHRRVectors(unittest.TestCase):
|
|
"""Test the HRR encoding and similarity functions."""
|
|
|
|
def test_deterministic_encoding(self):
|
|
"""Same text always produces the same vector."""
|
|
v1 = encode_text("hello world")
|
|
v2 = encode_text("hello world")
|
|
self.assertAlmostEqual(cosine_similarity_phase(v1, v2), 1.0, places=5)
|
|
|
|
def test_similar_texts_higher_similarity(self):
|
|
"""Related texts should be more similar than unrelated ones."""
|
|
v_agent = encode_text("agent memory palace retrieval")
|
|
v_similar = encode_text("agent recall memory search")
|
|
v_unrelated = encode_text("banana strawberry fruit smoothie")
|
|
sim_related = cosine_similarity_phase(v_agent, v_similar)
|
|
sim_unrelated = cosine_similarity_phase(v_agent, v_unrelated)
|
|
self.assertGreater(sim_related, sim_unrelated)
|
|
|
|
def test_serialize_roundtrip(self):
|
|
"""Vectors survive serialization to/from bytes."""
|
|
vec = encode_text("test serialization")
|
|
blob = serialize_vector(vec)
|
|
restored = deserialize_vector(blob)
|
|
sim = cosine_similarity_phase(vec, restored)
|
|
self.assertAlmostEqual(sim, 1.0, places=5)
|
|
|
|
def test_empty_text(self):
|
|
"""Empty text gets a fallback encoding."""
|
|
vec = encode_text("")
|
|
self.assertEqual(len(vec) if hasattr(vec, '__len__') else len(list(vec)), 512)
|
|
|
|
|
|
class TestSovereignStore(unittest.TestCase):
|
|
"""Test the SQLite-backed sovereign store."""
|
|
|
|
def setUp(self):
|
|
self.db_path = os.path.join(tempfile.mkdtemp(), "test.db")
|
|
self.store = SovereignStore(db_path=self.db_path)
|
|
|
|
def tearDown(self):
|
|
self.store.close()
|
|
if os.path.exists(self.db_path):
|
|
os.remove(self.db_path)
|
|
|
|
def test_store_and_retrieve(self):
|
|
"""Store a fact and find it via search."""
|
|
mid = self.store.store("Timmy is a sovereign AI agent on Hermes VPS", room="identity")
|
|
results = self.store.search("sovereign agent", room="identity")
|
|
self.assertTrue(any(r["memory_id"] == mid for r in results))
|
|
|
|
def test_fts_search(self):
|
|
"""FTS5 keyword search works."""
|
|
self.store.store("The beacon game uses paperclips mechanics", room="projects")
|
|
self.store.store("Fleet agents handle delegation and dispatch", room="fleet")
|
|
results = self.store.search("paperclips")
|
|
self.assertTrue(len(results) > 0)
|
|
self.assertIn("paperclips", results[0]["content"].lower())
|
|
|
|
def test_hrr_search_semantic(self):
|
|
"""HRR similarity finds related content even without exact keywords."""
|
|
self.store.store("Memory palace rooms organize facts spatially", room="memory")
|
|
self.store.store("Pizza delivery service runs on weekends", room="unrelated")
|
|
results = self.store.search("organize knowledge rooms", room="memory")
|
|
self.assertTrue(len(results) > 0)
|
|
self.assertIn("palace", results[0]["content"].lower())
|
|
|
|
def test_room_filtering(self):
|
|
"""Room filter restricts search scope."""
|
|
self.store.store("Hermes harness manages tool calls", room="infrastructure")
|
|
self.store.store("Hermes mythology Greek god", room="lore")
|
|
results = self.store.search("Hermes", room="infrastructure")
|
|
self.assertTrue(all(r["room"] == "infrastructure" for r in results))
|
|
|
|
def test_trust_boost(self):
|
|
"""Trust score increases when boosted."""
|
|
mid = self.store.store("fact", trust=0.5)
|
|
self.store.boost_trust(mid, delta=0.1)
|
|
results = self.store.room_contents("general")
|
|
fact = next(r for r in results if r["memory_id"] == mid)
|
|
self.assertAlmostEqual(fact["trust_score"], 0.6, places=2)
|
|
|
|
def test_trust_decay(self):
|
|
"""Trust score decreases when decayed."""
|
|
mid = self.store.store("questionable fact", trust=0.5)
|
|
self.store.decay_trust(mid, delta=0.2)
|
|
results = self.store.room_contents("general")
|
|
fact = next(r for r in results if r["memory_id"] == mid)
|
|
self.assertAlmostEqual(fact["trust_score"], 0.3, places=2)
|
|
|
|
def test_batch_store(self):
|
|
"""Batch store works."""
|
|
ids = self.store.store_batch([
|
|
{"content": "fact one", "room": "test"},
|
|
{"content": "fact two", "room": "test"},
|
|
{"content": "fact three", "room": "test"},
|
|
])
|
|
self.assertEqual(len(ids), 3)
|
|
rooms = self.store.list_rooms()
|
|
test_room = next(r for r in rooms if r["room"] == "test")
|
|
self.assertEqual(test_room["count"], 3)
|
|
|
|
def test_stats(self):
|
|
"""Stats returns correct counts."""
|
|
self.store.store("a fact", room="r1")
|
|
self.store.store("another fact", room="r2")
|
|
s = self.store.stats()
|
|
self.assertEqual(s["total"], 2)
|
|
self.assertEqual(s["room_count"], 2)
|
|
|
|
def test_retrieval_count_increments(self):
|
|
"""Retrieval count goes up when a fact is found via search."""
|
|
self.store.store("unique searchable content xyz123", room="test")
|
|
self.store.search("xyz123")
|
|
results = self.store.room_contents("test")
|
|
self.assertTrue(any(r["retrieval_count"] > 0 for r in results))
|
|
|
|
|
|
class TestPromotion(unittest.TestCase):
|
|
"""Test the quality-gated promotion system."""
|
|
|
|
def setUp(self):
|
|
self.db_path = os.path.join(tempfile.mkdtemp(), "promo_test.db")
|
|
self.store = SovereignStore(db_path=self.db_path)
|
|
|
|
def tearDown(self):
|
|
self.store.close()
|
|
|
|
def test_successful_promotion(self):
|
|
"""Good content passes all gates."""
|
|
result = promote(
|
|
content="Timmy runs on the Hermes VPS at 143.198.27.163 with local Ollama inference",
|
|
store=self.store,
|
|
session_id="test-session-001",
|
|
scratch_key="vps_info",
|
|
room="infrastructure",
|
|
)
|
|
self.assertTrue(result.success)
|
|
self.assertIsNotNone(result.memory_id)
|
|
|
|
def test_reject_too_short(self):
|
|
"""Short fragments get rejected."""
|
|
result = promote(
|
|
content="yes",
|
|
store=self.store,
|
|
session_id="test",
|
|
scratch_key="short",
|
|
)
|
|
self.assertFalse(result.success)
|
|
self.assertIn("Too short", result.reason)
|
|
|
|
def test_reject_duplicate(self):
|
|
"""Duplicate content gets rejected."""
|
|
self.store.store("SOUL.md is the canonical identity document for Timmy", room="identity")
|
|
result = promote(
|
|
content="SOUL.md is the canonical identity document for Timmy",
|
|
store=self.store,
|
|
session_id="test",
|
|
scratch_key="soul",
|
|
room="identity",
|
|
)
|
|
self.assertFalse(result.success)
|
|
self.assertIn("uplicate", result.reason)
|
|
|
|
def test_reject_stale(self):
|
|
"""Old notes get flagged as stale."""
|
|
old_time = time.time() - (86400 * 10)
|
|
result = promote(
|
|
content="This is a note from long ago about something important",
|
|
store=self.store,
|
|
session_id="test",
|
|
scratch_key="old",
|
|
written_at=old_time,
|
|
)
|
|
self.assertFalse(result.success)
|
|
self.assertIn("Stale", result.reason)
|
|
|
|
def test_force_bypasses_gates(self):
|
|
"""Force flag overrides quality gates."""
|
|
result = promote(
|
|
content="ok",
|
|
store=self.store,
|
|
session_id="test",
|
|
scratch_key="forced",
|
|
force=True,
|
|
)
|
|
self.assertTrue(result.success)
|
|
|
|
def test_evaluate_dry_run(self):
|
|
"""Evaluate returns gate details without promoting."""
|
|
eval_result = evaluate_for_promotion(
|
|
content="The fleet uses kimi-k2.5 as the primary model for all agent operations",
|
|
store=self.store,
|
|
room="fleet",
|
|
)
|
|
self.assertTrue(eval_result["eligible"])
|
|
self.assertTrue(all(p for p, _ in eval_result["gates"].values()))
|
|
|
|
def test_batch_promotion(self):
|
|
"""Batch promotion processes all notes."""
|
|
notes = {
|
|
"infra": {"value": "Hermes VPS runs Ubuntu 22.04 with 2 vCPUs and 4GB RAM", "written_at": time.strftime("%Y-%m-%d %H:%M:%S")},
|
|
"short": {"value": "no", "written_at": time.strftime("%Y-%m-%d %H:%M:%S")},
|
|
"model": {"value": "The primary local model is gemma4:latest running on Ollama", "written_at": time.strftime("%Y-%m-%d %H:%M:%S")},
|
|
}
|
|
results = promote_session_batch(self.store, "batch-session", notes, room="config")
|
|
promoted = [r for r in results if r.success]
|
|
rejected = [r for r in results if not r.success]
|
|
self.assertEqual(len(promoted), 2)
|
|
self.assertEqual(len(rejected), 1)
|
|
|
|
def test_promotion_logged(self):
|
|
"""Successful promotions appear in the audit log."""
|
|
promote(
|
|
content="Forge is hosted at forge.alexanderwhitestone.com running Gitea",
|
|
store=self.store,
|
|
session_id="log-test",
|
|
scratch_key="forge",
|
|
room="infrastructure",
|
|
)
|
|
log = self.store.recent_promotions()
|
|
self.assertTrue(len(log) > 0)
|
|
self.assertEqual(log[0]["session_id"], "log-test")
|
|
self.assertEqual(log[0]["scratch_key"], "forge")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|