mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-04-02 17:32:39 +00:00
eaa3b8e3ed
Replace single-stage MemoryConsolidator with a two-stage architecture: - Consolidator: lightweight token-budget triggered summarization, appends to HISTORY.md with cursor-based tracking - Dream: cron-scheduled two-phase processor that analyzes HISTORY.md and updates SOUL.md, USER.md, MEMORY.md via AgentRunner with edit_file tools for surgical, fault-tolerant updates New files: MemoryStore (pure file I/O), Dream class, DreamConfig, /dream and /dream-log commands. 89 tests covering all components.
134 lines
4.5 KiB
Python
134 lines
4.5 KiB
Python
"""Tests for the restructured MemoryStore — pure file I/O layer."""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
from pathlib import Path
|
|
|
|
from nanobot.agent.memory import MemoryStore
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path):
|
|
return MemoryStore(tmp_path)
|
|
|
|
|
|
class TestMemoryStoreBasicIO:
|
|
def test_read_memory_returns_empty_when_missing(self, store):
|
|
assert store.read_memory() == ""
|
|
|
|
def test_write_and_read_memory(self, store):
|
|
store.write_memory("hello")
|
|
assert store.read_memory() == "hello"
|
|
|
|
def test_read_soul_returns_empty_when_missing(self, store):
|
|
assert store.read_soul() == ""
|
|
|
|
def test_write_and_read_soul(self, store):
|
|
store.write_soul("soul content")
|
|
assert store.read_soul() == "soul content"
|
|
|
|
def test_read_user_returns_empty_when_missing(self, store):
|
|
assert store.read_user() == ""
|
|
|
|
def test_write_and_read_user(self, store):
|
|
store.write_user("user content")
|
|
assert store.read_user() == "user content"
|
|
|
|
def test_get_memory_context_returns_empty_when_missing(self, store):
|
|
assert store.get_memory_context() == ""
|
|
|
|
def test_get_memory_context_returns_formatted_content(self, store):
|
|
store.write_memory("important fact")
|
|
ctx = store.get_memory_context()
|
|
assert "Long-term Memory" in ctx
|
|
assert "important fact" in ctx
|
|
|
|
|
|
class TestHistoryWithCursor:
|
|
def test_append_history_returns_cursor(self, store):
|
|
cursor = store.append_history("event 1")
|
|
assert cursor == 1
|
|
cursor2 = store.append_history("event 2")
|
|
assert cursor2 == 2
|
|
|
|
def test_append_history_includes_cursor_in_file(self, store):
|
|
store.append_history("event 1")
|
|
content = store.read_file(store.history_file)
|
|
data = json.loads(content)
|
|
assert data["cursor"] == 1
|
|
|
|
def test_cursor_persists_across_appends(self, store):
|
|
store.append_history("event 1")
|
|
store.append_history("event 2")
|
|
cursor = store.append_history("event 3")
|
|
assert cursor == 3
|
|
|
|
def test_read_unprocessed_history(self, store):
|
|
store.append_history("event 1")
|
|
store.append_history("event 2")
|
|
store.append_history("event 3")
|
|
entries = store.read_unprocessed_history(since_cursor=1)
|
|
assert len(entries) == 2
|
|
assert entries[0]["cursor"] == 2
|
|
|
|
def test_read_unprocessed_history_returns_all_when_cursor_zero(self, store):
|
|
store.append_history("event 1")
|
|
store.append_history("event 2")
|
|
entries = store.read_unprocessed_history(since_cursor=0)
|
|
assert len(entries) == 2
|
|
|
|
def test_compact_history_drops_oldest(self, tmp_path):
|
|
store = MemoryStore(tmp_path, max_history_entries=2)
|
|
store.append_history("event 1")
|
|
store.append_history("event 2")
|
|
store.append_history("event 3")
|
|
store.append_history("event 4")
|
|
store.append_history("event 5")
|
|
store.compact_history()
|
|
entries = store.read_unprocessed_history(since_cursor=0)
|
|
assert len(entries) == 2
|
|
assert entries[0]["cursor"] in {4, 5}
|
|
|
|
|
|
class TestDreamCursor:
|
|
def test_initial_cursor_is_zero(self, store):
|
|
assert store.get_last_dream_cursor() == 0
|
|
|
|
def test_set_and_get_cursor(self, store):
|
|
store.set_last_dream_cursor(5)
|
|
assert store.get_last_dream_cursor() == 5
|
|
|
|
def test_cursor_persists(self, store):
|
|
store.set_last_dream_cursor(3)
|
|
store2 = MemoryStore(store.workspace)
|
|
assert store2.get_last_dream_cursor() == 3
|
|
|
|
|
|
class TestDreamLog:
|
|
def test_read_dream_log_returns_empty_when_missing(self, store):
|
|
assert store.read_dream_log() == ""
|
|
|
|
def test_append_dream_log(self, store):
|
|
store.append_dream_log("## 2026-03-30\nProcessed entries #1-#5")
|
|
log = store.read_dream_log()
|
|
assert "Processed entries #1-#5" in log
|
|
|
|
def test_append_dream_log_is_additive(self, store):
|
|
store.append_dream_log("first run")
|
|
store.append_dream_log("second run")
|
|
log = store.read_dream_log()
|
|
assert "first run" in log
|
|
assert "second run" in log
|
|
|
|
|
|
class TestLegacyHistoryMigration:
|
|
def test_read_unprocessed_history_handles_entries_without_cursor(self, store):
|
|
"""JSONL entries with cursor=1 are correctly parsed and returned."""
|
|
store.history_file.write_text(
|
|
'{"cursor": 1, "timestamp": "2026-03-30 14:30", "content": "Old event"}\n',
|
|
encoding="utf-8")
|
|
entries = store.read_unprocessed_history(since_cursor=0)
|
|
assert len(entries) == 1
|
|
assert entries[0]["cursor"] == 1
|