378 lines
12 KiB
Python
378 lines
12 KiB
Python
"""
|
|
Tests for agent memory — cross-session agent memory via MemPalace.
|
|
|
|
Tests the memory module, hooks, and session mining without requiring
|
|
a live ChromaDB instance. Uses mocking for the MemPalace backend.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from agent.memory import (
|
|
AgentMemory,
|
|
MemoryContext,
|
|
SessionTranscript,
|
|
create_agent_memory,
|
|
)
|
|
from agent.memory_hooks import MemoryHooks
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SessionTranscript tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSessionTranscript:
|
|
def test_create(self):
|
|
t = SessionTranscript(agent_name="test", wing="wing_test")
|
|
assert t.agent_name == "test"
|
|
assert t.wing == "wing_test"
|
|
assert len(t.entries) == 0
|
|
|
|
def test_add_user_turn(self):
|
|
t = SessionTranscript(agent_name="test", wing="wing_test")
|
|
t.add_user_turn("Hello")
|
|
assert len(t.entries) == 1
|
|
assert t.entries[0]["role"] == "user"
|
|
assert t.entries[0]["text"] == "Hello"
|
|
|
|
def test_add_agent_turn(self):
|
|
t = SessionTranscript(agent_name="test", wing="wing_test")
|
|
t.add_agent_turn("Response")
|
|
assert t.entries[0]["role"] == "agent"
|
|
|
|
def test_add_tool_call(self):
|
|
t = SessionTranscript(agent_name="test", wing="wing_test")
|
|
t.add_tool_call("shell", "ls", "file1 file2")
|
|
assert t.entries[0]["role"] == "tool"
|
|
assert t.entries[0]["tool"] == "shell"
|
|
|
|
def test_summary_empty(self):
|
|
t = SessionTranscript(agent_name="test", wing="wing_test")
|
|
assert t.summary() == "Empty session."
|
|
|
|
def test_summary_with_entries(self):
|
|
t = SessionTranscript(agent_name="test", wing="wing_test")
|
|
t.add_user_turn("Do something")
|
|
t.add_agent_turn("Done")
|
|
t.add_tool_call("shell", "ls", "ok")
|
|
|
|
summary = t.summary()
|
|
assert "USER: Do something" in summary
|
|
assert "AGENT: Done" in summary
|
|
assert "TOOL(shell): ok" in summary
|
|
|
|
def test_text_truncation(self):
|
|
t = SessionTranscript(agent_name="test", wing="wing_test")
|
|
long_text = "x" * 5000
|
|
t.add_user_turn(long_text)
|
|
assert len(t.entries[0]["text"]) == 2000
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# MemoryContext tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMemoryContext:
|
|
def test_empty_context(self):
|
|
ctx = MemoryContext()
|
|
assert ctx.to_prompt_block() == ""
|
|
|
|
def test_unloaded_context(self):
|
|
ctx = MemoryContext()
|
|
ctx.loaded = False
|
|
assert ctx.to_prompt_block() == ""
|
|
|
|
def test_loaded_with_data(self):
|
|
ctx = MemoryContext()
|
|
ctx.loaded = True
|
|
ctx.recent_diaries = [
|
|
{"text": "Fixed PR #1386", "timestamp": "2026-04-13T10:00:00Z"}
|
|
]
|
|
ctx.facts = [
|
|
{"text": "Bezalel runs on VPS Beta", "score": 0.95}
|
|
]
|
|
ctx.relevant_memories = [
|
|
{"text": "Changed CI runner", "score": 0.87}
|
|
]
|
|
|
|
block = ctx.to_prompt_block()
|
|
assert "Recent Session Summaries" in block
|
|
assert "Fixed PR #1386" in block
|
|
assert "Known Facts" in block
|
|
assert "Bezalel runs on VPS Beta" in block
|
|
assert "Relevant Past Memories" in block
|
|
|
|
def test_loaded_empty(self):
|
|
ctx = MemoryContext()
|
|
ctx.loaded = True
|
|
# No data — should return empty string
|
|
assert ctx.to_prompt_block() == ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AgentMemory tests (with mocked MemPalace)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAgentMemory:
|
|
def test_create(self):
|
|
mem = AgentMemory(agent_name="bezalel")
|
|
assert mem.agent_name == "bezalel"
|
|
assert mem.wing == "wing_bezalel"
|
|
|
|
def test_custom_wing(self):
|
|
mem = AgentMemory(agent_name="bezalel", wing="custom_wing")
|
|
assert mem.wing == "custom_wing"
|
|
|
|
def test_factory(self):
|
|
mem = create_agent_memory("ezra")
|
|
assert mem.agent_name == "ezra"
|
|
assert mem.wing == "wing_ezra"
|
|
|
|
def test_unavailable_graceful(self):
|
|
"""Test graceful degradation when MemPalace is unavailable."""
|
|
mem = AgentMemory(agent_name="test")
|
|
mem._available = False # Force unavailable
|
|
|
|
# Should not raise
|
|
ctx = mem.recall_context("test query")
|
|
assert ctx.loaded is False
|
|
assert ctx.error == "MemPalace unavailable"
|
|
|
|
# remember returns None
|
|
assert mem.remember("test") is None
|
|
|
|
# search returns empty
|
|
assert mem.search("test") == []
|
|
|
|
def test_start_end_session(self):
|
|
mem = AgentMemory(agent_name="test")
|
|
mem._available = False
|
|
|
|
transcript = mem.start_session()
|
|
assert isinstance(transcript, SessionTranscript)
|
|
assert mem._transcript is not None
|
|
|
|
doc_id = mem.end_session()
|
|
assert mem._transcript is None
|
|
|
|
def test_remember_graceful_when_unavailable(self):
|
|
"""Test remember returns None when MemPalace is unavailable."""
|
|
mem = AgentMemory(agent_name="test")
|
|
mem._available = False
|
|
|
|
doc_id = mem.remember("some important fact")
|
|
assert doc_id is None
|
|
|
|
def test_write_diary_from_transcript(self):
|
|
mem = AgentMemory(agent_name="test")
|
|
mem._available = False
|
|
|
|
transcript = mem.start_session()
|
|
transcript.add_user_turn("Hello")
|
|
transcript.add_agent_turn("Hi there")
|
|
|
|
# Write diary should handle unavailable gracefully
|
|
doc_id = mem.write_diary()
|
|
assert doc_id is None # MemPalace unavailable
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# MemoryHooks tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMemoryHooks:
|
|
def test_create(self):
|
|
hooks = MemoryHooks(agent_name="bezalel")
|
|
assert hooks.agent_name == "bezalel"
|
|
assert hooks.is_active is False
|
|
|
|
def test_session_lifecycle(self):
|
|
hooks = MemoryHooks(agent_name="test")
|
|
|
|
# Force memory unavailable
|
|
hooks._memory = AgentMemory(agent_name="test")
|
|
hooks._memory._available = False
|
|
|
|
# Start session
|
|
block = hooks.on_session_start()
|
|
assert hooks.is_active is True
|
|
assert block == "" # No memory available
|
|
|
|
# Record turns
|
|
hooks.on_user_turn("Hello")
|
|
hooks.on_agent_turn("Hi")
|
|
hooks.on_tool_call("shell", "ls", "ok")
|
|
|
|
# Record decision
|
|
hooks.on_important_decision("Switched to self-hosted CI")
|
|
|
|
# End session
|
|
doc_id = hooks.on_session_end()
|
|
assert hooks.is_active is False
|
|
|
|
def test_hooks_before_session(self):
|
|
"""Hooks before session start should be no-ops."""
|
|
hooks = MemoryHooks(agent_name="test")
|
|
hooks._memory = AgentMemory(agent_name="test")
|
|
hooks._memory._available = False
|
|
|
|
# Should not raise
|
|
hooks.on_user_turn("Hello")
|
|
hooks.on_agent_turn("Response")
|
|
|
|
def test_hooks_after_session_end(self):
|
|
"""Hooks after session end should be no-ops."""
|
|
hooks = MemoryHooks(agent_name="test")
|
|
hooks._memory = AgentMemory(agent_name="test")
|
|
hooks._memory._available = False
|
|
|
|
hooks.on_session_start()
|
|
hooks.on_session_end()
|
|
|
|
# Should not raise
|
|
hooks.on_user_turn("Late message")
|
|
doc_id = hooks.on_session_end()
|
|
assert doc_id is None
|
|
|
|
def test_search_during_session(self):
|
|
hooks = MemoryHooks(agent_name="test")
|
|
hooks._memory = AgentMemory(agent_name="test")
|
|
hooks._memory._available = False
|
|
|
|
results = hooks.search("some query")
|
|
assert results == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Session mining tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSessionMining:
|
|
def test_parse_session_file(self):
|
|
from bin.memory_mine import parse_session_file
|
|
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write('{"role": "user", "content": "Hello"}\n')
|
|
f.write('{"role": "assistant", "content": "Hi there"}\n')
|
|
f.write('{"role": "tool", "name": "shell", "content": "ls output"}\n')
|
|
f.write("\n") # blank line
|
|
f.write("not json\n") # malformed
|
|
path = Path(f.name)
|
|
|
|
turns = parse_session_file(path)
|
|
assert len(turns) == 3
|
|
assert turns[0]["role"] == "user"
|
|
assert turns[1]["role"] == "assistant"
|
|
assert turns[2]["role"] == "tool"
|
|
path.unlink()
|
|
|
|
def test_summarize_session(self):
|
|
from bin.memory_mine import summarize_session
|
|
|
|
turns = [
|
|
{"role": "user", "content": "Check CI"},
|
|
{"role": "assistant", "content": "Running CI check..."},
|
|
{"role": "tool", "name": "shell", "content": "5 tests passed"},
|
|
{"role": "assistant", "content": "CI is healthy"},
|
|
]
|
|
|
|
summary = summarize_session(turns, "bezalel")
|
|
assert "bezalel" in summary
|
|
assert "Check CI" in summary
|
|
assert "shell" in summary
|
|
|
|
def test_summarize_empty(self):
|
|
from bin.memory_mine import summarize_session
|
|
|
|
assert summarize_session([], "test") == "Empty session."
|
|
|
|
def test_find_session_files(self, tmp_path):
|
|
from bin.memory_mine import find_session_files
|
|
|
|
# Create some test files
|
|
(tmp_path / "session1.jsonl").write_text("{}\n")
|
|
(tmp_path / "session2.jsonl").write_text("{}\n")
|
|
(tmp_path / "notes.txt").write_text("not a session")
|
|
|
|
files = find_session_files(tmp_path, days=365)
|
|
assert len(files) == 2
|
|
assert all(f.suffix == ".jsonl" for f in files)
|
|
|
|
def test_find_session_files_missing_dir(self):
|
|
from bin.memory_mine import find_session_files
|
|
|
|
files = find_session_files(Path("/nonexistent/path"), days=7)
|
|
assert files == []
|
|
|
|
def test_mine_session_dry_run(self, tmp_path):
|
|
from bin.memory_mine import mine_session
|
|
|
|
session_file = tmp_path / "test.jsonl"
|
|
session_file.write_text(
|
|
'{"role": "user", "content": "Hello"}\n'
|
|
'{"role": "assistant", "content": "Hi"}\n'
|
|
)
|
|
|
|
result = mine_session(session_file, wing="wing_test", dry_run=True)
|
|
assert result is None # dry run doesn't store
|
|
|
|
def test_mine_session_empty_file(self, tmp_path):
|
|
from bin.memory_mine import mine_session
|
|
|
|
session_file = tmp_path / "empty.jsonl"
|
|
session_file.write_text("")
|
|
|
|
result = mine_session(session_file, wing="wing_test")
|
|
assert result is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Integration test — full lifecycle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFullLifecycle:
|
|
"""Test the full session lifecycle without a real MemPalace backend."""
|
|
|
|
def test_full_session_flow(self):
|
|
hooks = MemoryHooks(agent_name="bezalel")
|
|
|
|
# Force memory unavailable
|
|
hooks._memory = AgentMemory(agent_name="bezalel")
|
|
hooks._memory._available = False
|
|
|
|
# 1. Session start
|
|
context_block = hooks.on_session_start("What CI issues do I have?")
|
|
assert isinstance(context_block, str)
|
|
|
|
# 2. User asks question
|
|
hooks.on_user_turn("Check CI pipeline health")
|
|
|
|
# 3. Agent uses tool
|
|
hooks.on_tool_call("shell", "pytest tests/", "12 passed")
|
|
|
|
# 4. Agent responds
|
|
hooks.on_agent_turn("CI pipeline is healthy. All 12 tests passing.")
|
|
|
|
# 5. Important decision
|
|
hooks.on_important_decision("Decided to keep current CI runner", room="forge")
|
|
|
|
# 6. More interaction
|
|
hooks.on_user_turn("Good, check memory integration next")
|
|
hooks.on_agent_turn("Will test agent.memory module")
|
|
|
|
# 7. Session end
|
|
doc_id = hooks.on_session_end()
|
|
assert hooks.is_active is False
|