mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
Patrick Buckley
723cad24bb
* feat: structured memory system — typed/scoped memories with BM25 relevance and metacognitive prompting Replace flat key-value memories table with structured_memories (migration 014). Four memory types (user/project/feedback/reference), three scopes (global/workstream/user). Consolidate remember/recall/forget into two tools: memory (action-based: save/search/delete/list) and recall (conversation history only). BM25 relevance scoring (extracted to turnstone/core/bm25.py) selects top-5 memories for system message injection based on conversation context. Metacognitive prompting injects ephemeral nudges after corrections, tool denials, workstream resume, and completion signals. Scope isolation enforced: system message injection and nudge counts filtered to visible memories only (global + current workstream + authenticated user). User scope requires authentication. Content capped at 32KB. ILIKE/LIKE metacharacters escaped in both backends. 113 new tests (2053 total). * fix: CI failure + copilot review feedback - Fix time.monotonic() cooldown: use None sentinel instead of 0.0 default (monotonic clock starts at boot, not epoch — fresh CI runners have uptime < 300s so cooldown check always triggered) - Catch sa.exc.IntegrityError specifically in upsert instead of broad Exception (copilot review) - Preserve existing description/type on upsert when caller doesn't explicitly set them (copilot review) - Add last_accessed + access_count columns to schema/migration for future LRU/LFU eviction support
138 lines
5.9 KiB
Python
138 lines
5.9 KiB
Python
"""Tests for structured memory storage backend operations."""
|
|
|
|
import pytest
|
|
|
|
from turnstone.core.storage._sqlite import SQLiteBackend
|
|
|
|
|
|
@pytest.fixture
|
|
def backend(tmp_path):
|
|
return SQLiteBackend(str(tmp_path / "test.db"))
|
|
|
|
|
|
class TestCreateAndGet:
|
|
def test_create_and_get_by_id(self, backend):
|
|
backend.create_structured_memory("m1", "test_key", "desc", "project", "global", "", "data")
|
|
mem = backend.get_structured_memory("m1")
|
|
assert mem is not None
|
|
assert mem["name"] == "test_key"
|
|
assert mem["content"] == "data"
|
|
assert mem["type"] == "project"
|
|
|
|
def test_get_nonexistent(self, backend):
|
|
assert backend.get_structured_memory("nope") is None
|
|
|
|
def test_get_by_name(self, backend):
|
|
backend.create_structured_memory("m1", "mykey", "d", "project", "global", "", "val")
|
|
mem = backend.get_structured_memory_by_name("mykey", "global", "")
|
|
assert mem is not None
|
|
assert mem["memory_id"] == "m1"
|
|
|
|
def test_get_by_name_scoped(self, backend):
|
|
backend.create_structured_memory("m1", "key", "d", "project", "global", "", "g")
|
|
backend.create_structured_memory("m2", "key", "d", "project", "workstream", "ws1", "w")
|
|
g = backend.get_structured_memory_by_name("key", "global", "")
|
|
w = backend.get_structured_memory_by_name("key", "workstream", "ws1")
|
|
assert g["content"] == "g"
|
|
assert w["content"] == "w"
|
|
|
|
|
|
class TestUpdate:
|
|
def test_update_content(self, backend):
|
|
backend.create_structured_memory("m1", "k", "d", "project", "global", "", "old")
|
|
assert backend.update_structured_memory("m1", content="new")
|
|
mem = backend.get_structured_memory("m1")
|
|
assert mem["content"] == "new"
|
|
|
|
def test_update_nonexistent(self, backend):
|
|
assert not backend.update_structured_memory("nope", content="x")
|
|
|
|
def test_update_no_fields(self, backend):
|
|
backend.create_structured_memory("m1", "k", "d", "project", "global", "", "data")
|
|
assert not backend.update_structured_memory("m1", bogus="val")
|
|
|
|
def test_update_bumps_timestamp(self, backend):
|
|
backend.create_structured_memory("m1", "k", "d", "project", "global", "", "data")
|
|
old = backend.get_structured_memory("m1")["updated"]
|
|
import time
|
|
|
|
time.sleep(0.01)
|
|
backend.update_structured_memory("m1", content="new")
|
|
new = backend.get_structured_memory("m1")["updated"]
|
|
assert new >= old
|
|
|
|
|
|
class TestDelete:
|
|
def test_delete_existing(self, backend):
|
|
backend.create_structured_memory("m1", "k", "d", "project", "global", "", "data")
|
|
assert backend.delete_structured_memory("k", "global", "")
|
|
assert backend.get_structured_memory("m1") is None
|
|
|
|
def test_delete_nonexistent(self, backend):
|
|
assert not backend.delete_structured_memory("nope", "global", "")
|
|
|
|
def test_delete_scoped(self, backend):
|
|
backend.create_structured_memory("m1", "k", "d", "project", "workstream", "ws1", "data")
|
|
assert not backend.delete_structured_memory("k", "global", "")
|
|
assert backend.delete_structured_memory("k", "workstream", "ws1")
|
|
|
|
|
|
class TestList:
|
|
def test_list_all(self, backend):
|
|
backend.create_structured_memory("m1", "a", "", "project", "global", "", "1")
|
|
backend.create_structured_memory("m2", "b", "", "user", "global", "", "2")
|
|
mems = backend.list_structured_memories()
|
|
assert len(mems) == 2
|
|
|
|
def test_list_by_type(self, backend):
|
|
backend.create_structured_memory("m1", "a", "", "project", "global", "", "1")
|
|
backend.create_structured_memory("m2", "b", "", "user", "global", "", "2")
|
|
mems = backend.list_structured_memories(mem_type="user")
|
|
assert len(mems) == 1
|
|
assert mems[0]["name"] == "b"
|
|
|
|
def test_list_by_scope(self, backend):
|
|
backend.create_structured_memory("m1", "a", "", "project", "global", "", "1")
|
|
backend.create_structured_memory("m2", "b", "", "project", "workstream", "ws1", "2")
|
|
mems = backend.list_structured_memories(scope="workstream")
|
|
assert len(mems) == 1
|
|
|
|
def test_list_respects_limit(self, backend):
|
|
for i in range(10):
|
|
backend.create_structured_memory(f"m{i}", f"k{i}", "", "project", "global", "", f"{i}")
|
|
mems = backend.list_structured_memories(limit=3)
|
|
assert len(mems) == 3
|
|
|
|
|
|
class TestSearch:
|
|
def test_search_by_name(self, backend):
|
|
backend.create_structured_memory("m1", "database_config", "", "project", "global", "", "pg")
|
|
backend.create_structured_memory("m2", "api_key", "", "project", "global", "", "secret")
|
|
results = backend.search_structured_memories("database")
|
|
assert len(results) == 1
|
|
assert results[0]["name"] == "database_config"
|
|
|
|
def test_search_by_content(self, backend):
|
|
backend.create_structured_memory("m1", "a", "", "project", "global", "", "postgresql host")
|
|
results = backend.search_structured_memories("postgresql")
|
|
assert len(results) == 1
|
|
|
|
def test_search_empty_lists_all(self, backend):
|
|
backend.create_structured_memory("m1", "a", "", "project", "global", "", "1")
|
|
backend.create_structured_memory("m2", "b", "", "project", "global", "", "2")
|
|
results = backend.search_structured_memories("")
|
|
assert len(results) == 2
|
|
|
|
|
|
class TestCount:
|
|
def test_count_all(self, backend):
|
|
backend.create_structured_memory("m1", "a", "", "project", "global", "", "1")
|
|
backend.create_structured_memory("m2", "b", "", "project", "global", "", "2")
|
|
assert backend.count_structured_memories() == 2
|
|
|
|
def test_count_by_scope(self, backend):
|
|
backend.create_structured_memory("m1", "a", "", "project", "global", "", "1")
|
|
backend.create_structured_memory("m2", "b", "", "project", "workstream", "ws1", "2")
|
|
assert backend.count_structured_memories(scope="global") == 1
|
|
assert backend.count_structured_memories(scope="workstream") == 1
|