"""Test persistent memory, goal memory, and agents.""" import sys import os import tempfile sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from singularity_llm.memory.persistent import PersistentMemory from singularity_llm.memory.goal_memory import GoalMemory, Goal from singularity_llm.agents.agent_manager import AgentManager from singularity_llm.agents.planner_agent import PlannerAgent from singularity_llm.agents.coder_agent import CoderAgent from singularity_llm.agents.researcher_agent import ResearcherAgent from singularity_llm.agents.reviewer_agent import ReviewerAgent from singularity_llm.agents.executor_agent import ExecutorAgent def test_persistent_memory(): """Test persistent memory storage and recall.""" with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: db_path = f.name mem = PersistentMemory(db_path=db_path) mem.set_session("test-session") # Store episodic mem.add_episodic("user", "What is Python?", importance=0.7) mem.add_episodic("assistant", "Python is a programming language.", importance=0.8) # Store semantic mem.add_semantic("Python is an interpreted programming language", confidence=0.9) # Recall episodic = mem.recall_episodic("Python", max_results=3) assert len(episodic) >= 1, "No episodic memories recalled" print(f" Episodic recalled: {len(episodic)}") semantic = mem.recall_semantic("Python", max_results=3) assert len(semantic) >= 1, "No semantic memories recalled" print(f" Semantic recalled: {len(semantic)}") # Context injection context = mem.get_context("Python programming") assert "Python" in context or "programming" in context, f"Context empty: {context}" print(f" Context: {context[:80]}") # Persistence test mem2 = PersistentMemory(db_path=db_path) stats = mem2.get_stats() assert stats["episodic_total"] >= 2, f"Episodic not persisted: {stats}" assert stats["semantic_total"] >= 1, f"Semantic not persisted: {stats}" print(f" Persisted: {stats['episodic_total']} episodic, {stats['semantic_total']} semantic") def test_goal_memory(): """Test goal creation, planning, and execution.""" with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: db_path = f.name gm = GoalMemory(db_path=db_path) # Create goal goal = gm.create_goal("Build a web app", "Create a simple Flask web application", priority="high") assert goal.status == "pending" print(f" Created goal: {goal.title} ({goal.id[:8]})") # Plan goal result = gm.plan_goal(goal.id, steps=[ {"title": "Setup", "description": "Create project structure"}, {"title": "Code", "description": "Write the Flask app"}, {"title": "Test", "description": "Test the application"}, ]) assert result["success"] goal = gm.get_goal(goal.id) assert goal.status == "in_progress" assert len(goal.steps) == 3 print(f" Planned: {len(goal.steps)} steps, status: {goal.status}") # Execute steps r1 = gm.execute_step(goal.id, "Project structure created", success=True) assert r1["success"] r2 = gm.execute_step(goal.id, "Flask app written", success=True) assert r2["success"] r3 = gm.execute_step(goal.id, "Tests passed", success=True) assert r3["success"] goal = gm.get_goal(goal.id) assert goal.status == "completed" assert goal.progress() == 1.0 print(f" Completed: {goal.progress():.0%}") # Persistence gm2 = GoalMemory(db_path=db_path) stats = gm2.get_stats() assert stats["total"] == 1 assert stats["completed"] == 1 print(f" Persisted: {stats}") def test_agent_manager(): """Test agent manager with 5 agents.""" with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: goal_db = f.name with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: mem_db = f.name gm = GoalMemory(db_path=goal_db) pm = PersistentMemory(db_path=mem_db) # Mock generate function def mock_generate(prompt): if "planning" in prompt.lower(): return '{"steps": [{"title": "Step 1", "description": "Do thing 1"}], "sub_goals": []}' return "Mock response for: " + prompt[:50] manager = AgentManager(gm, pm, generate_fn=mock_generate) assert len(manager.agents) == 5 print(f" Agents: {list(manager.agents.keys())}") # Create a project goal = manager.create_project("Test project", "A test project for validation", priority="high") print(f" Created project: {goal.id[:8]}") # Get status status = manager.get_agent_status() assert len(status) == 5 for name, info in status.items(): assert info["name"] == name assert info["running"] == False # not started yet print(f" All 5 agents present: {list(status.keys())}") # Get stats stats = manager.get_stats() assert "agents" in stats assert "goals" in stats print(f" Stats: {stats['goals']}") def test_agent_specialization(): """Test that agents have correct specializations.""" with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: db_path = f.name gm = GoalMemory(db_path=db_path) planner = PlannerAgent(gm) coder = CoderAgent(gm) researcher = ResearcherAgent(gm) reviewer = ReviewerAgent(gm) executor = ExecutorAgent(gm) # Test coder can handle code goals code_goal = gm.create_goal("Write code", "Implement a Python function to sort data") assert coder._can_handle(code_goal), "Coder should handle code goals" assert not researcher._can_handle(code_goal), "Researcher should not handle code goals" print(" Coder handles code goals: OK") # Test researcher can handle research goals research_goal = gm.create_goal("Research topic", "Research the best sorting algorithms") assert researcher._can_handle(research_goal), "Researcher should handle research goals" print(" Researcher handles research goals: OK") # Test executor can handle execution goals exec_goal = gm.create_goal("Run tests", "Execute the test suite and deploy") assert executor._can_handle(exec_goal), "Executor should handle execution goals" print(" Executor handles execution goals: OK") if __name__ == "__main__": print("Running memory & agents tests...") test_persistent_memory() print(" ✓ test_persistent_memory") test_goal_memory() print(" ✓ test_goal_memory") test_agent_manager() print(" ✓ test_agent_manager") test_agent_specialization() print(" ✓ test_agent_specialization") print("\nAll memory & agents tests passed!")