diff --git a/mini_agent/cli.py b/mini_agent/cli.py index f060c9c2..44f3ab3d 100644 --- a/mini_agent/cli.py +++ b/mini_agent/cli.py @@ -35,6 +35,7 @@ from mini_agent.tools.bash_tool import BashKillTool, BashOutputTool, BashTool from mini_agent.tools.file_tools import EditTool, ReadTool, WriteTool from mini_agent.tools.mcp_loader import cleanup_mcp_connections, load_mcp_tools_async, set_mcp_timeout_config +from mini_agent.memory import CoreMemoryTool, EphemeralMemoryTool, LearnedMemoryTool, MemoryManager from mini_agent.tools.note_tool import SessionNoteTool from mini_agent.tools.skill_tool import create_skill_tools from mini_agent.utils import calculate_display_width @@ -427,8 +428,28 @@ async def initialize_base_tools(config: Config): except Exception as e: print(f"{Colors.YELLOW}⚠️ Failed to load MCP tools: {e}{Colors.RESET}") + # 5. Three-layer memory system (CORE / LEARNED / EPHEMERAL) + memory_manager = None + if config.tools.memory.enabled: + print(f"{Colors.BRIGHT_CYAN}Initializing memory system...{Colors.RESET}") + core_path = config.tools.memory.core_memory_path or None + memory_manager = MemoryManager( + core_memory_path=core_path, + learned_confidence_threshold=config.tools.memory.learned_confidence_threshold, + ) + memory_manager.reset_ephemeral() + + tools.append(CoreMemoryTool(memory_manager)) + print(f"{Colors.GREEN}✅ Loaded Core Memory tool{Colors.RESET}") + + tools.append(LearnedMemoryTool(memory_manager)) + print(f"{Colors.GREEN}✅ Loaded Learned Memory tool (threshold: {config.tools.memory.learned_confidence_threshold}){Colors.RESET}") + + tools.append(EphemeralMemoryTool(memory_manager)) + print(f"{Colors.GREEN}✅ Loaded Ephemeral Memory tool{Colors.RESET}") + print() # Empty line separator - return tools, skill_loader + return tools, skill_loader, memory_manager def add_workspace_tools(tools: List[Tool], config: Config, workspace_dir: Path): @@ -572,7 +593,7 @@ def on_retry(exception: Exception, attempt: int): print(f"{Colors.GREEN}✅ LLM retry mechanism enabled (max {config.llm.retry.max_retries} retries){Colors.RESET}") # 3. Initialize base tools (independent of workspace) - tools, skill_loader = await initialize_base_tools(config) + tools, skill_loader, memory_manager = await initialize_base_tools(config) # 4. Add workspace-dependent tools add_workspace_tools(tools, config, workspace_dir) @@ -840,6 +861,10 @@ def esc_key_listener(): # 11. Cleanup MCP connections await _quiet_cleanup() + # 12. Reset ephemeral memory + if memory_manager: + memory_manager.reset_ephemeral() + def main(): """Main entry point for CLI""" diff --git a/mini_agent/config.py b/mini_agent/config.py index bab78f08..7f2f818c 100644 --- a/mini_agent/config.py +++ b/mini_agent/config.py @@ -45,6 +45,14 @@ class MCPConfig(BaseModel): sse_read_timeout: float = 120.0 # SSE read timeout (seconds) +class MemoryConfig(BaseModel): + """Three-layer memory system configuration.""" + + enabled: bool = True + core_memory_path: str = "" # Empty = use default ~/.mini-agent/memory/core.json + learned_confidence_threshold: float = 0.7 + + class ToolsConfig(BaseModel): """Tools configuration""" @@ -62,6 +70,9 @@ class ToolsConfig(BaseModel): mcp_config_path: str = "mcp.json" mcp: MCPConfig = Field(default_factory=MCPConfig) + # Memory system + memory: MemoryConfig = Field(default_factory=MemoryConfig) + class Config(BaseModel): """Main configuration class""" @@ -146,6 +157,14 @@ def from_yaml(cls, config_path: str | Path) -> "Config": sse_read_timeout=mcp_data.get("sse_read_timeout", 120.0), ) + # Parse memory configuration + memory_data = tools_data.get("memory", {}) + memory_config = MemoryConfig( + enabled=memory_data.get("enabled", True), + core_memory_path=memory_data.get("core_memory_path", ""), + learned_confidence_threshold=memory_data.get("learned_confidence_threshold", 0.7), + ) + tools_config = ToolsConfig( enable_file_tools=tools_data.get("enable_file_tools", True), enable_bash=tools_data.get("enable_bash", True), @@ -155,6 +174,7 @@ def from_yaml(cls, config_path: str | Path) -> "Config": enable_mcp=tools_data.get("enable_mcp", True), mcp_config_path=tools_data.get("mcp_config_path", "mcp.json"), mcp=mcp_config, + memory=memory_config, ) return cls( diff --git a/mini_agent/config/config-example.yaml b/mini_agent/config/config-example.yaml index f25bbd83..8474533e 100644 --- a/mini_agent/config/config-example.yaml +++ b/mini_agent/config/config-example.yaml @@ -58,3 +58,9 @@ tools: connect_timeout: 10.0 # Connection timeout in seconds (default: 10) execute_timeout: 60.0 # Tool execution timeout in seconds (default: 60) sse_read_timeout: 120.0 # SSE read timeout in seconds (default: 120) + + # Three-layer memory system + memory: + enabled: true # Enable CORE/LEARNED/EPHEMERAL memory tools + core_memory_path: "" # Path to CORE memory JSON (empty = default ~/.mini-agent/memory/core.json) + learned_confidence_threshold: 0.7 # Minimum confidence to record LEARNED memories (0.0-1.0) diff --git a/mini_agent/config/core_memory.json b/mini_agent/config/core_memory.json new file mode 100644 index 00000000..76c1ef0e --- /dev/null +++ b/mini_agent/config/core_memory.json @@ -0,0 +1,37 @@ +[ + { + "id": "core_001", + "content": "Never expose API keys, passwords, or secrets in code, logs, or output.", + "category": "safety", + "confidence": 1.0, + "source": "core" + }, + { + "id": "core_002", + "content": "Follow the project's existing code style, naming conventions, and architecture patterns. Do not introduce new paradigms without justification.", + "category": "convention", + "confidence": 1.0, + "source": "core" + }, + { + "id": "core_003", + "content": "Use uv for Python environment management: uv venv, uv pip install, uv run.", + "category": "convention", + "confidence": 1.0, + "source": "core" + }, + { + "id": "core_004", + "content": "Before making destructive changes (file deletion, git force push, database modifications), confirm with the user first.", + "category": "safety", + "confidence": 1.0, + "source": "core" + }, + { + "id": "core_005", + "content": "Write tests for all new non-trivial logic. Run the full test suite after changes and verify all tests pass.", + "category": "convention", + "confidence": 1.0, + "source": "core" + } +] diff --git a/mini_agent/config/system_prompt.md b/mini_agent/config/system_prompt.md index 97d3843d..99c0e490 100644 --- a/mini_agent/config/system_prompt.md +++ b/mini_agent/config/system_prompt.md @@ -71,5 +71,58 @@ Skills are loaded dynamically using **Progressive Disclosure**: - **Stay focused** - stop when the task is fulfilled - **Use skills** - leverage specialized knowledge when relevant +## Memory System + +You have access to a three-layer memory system for retaining and recalling information: + +### 1. CORE Memory (`core_memory` tool) +- Immutable project rules, safety constraints, and workspace conventions. +- Read-only at runtime — use `core_memory` to retrieve rules before making critical decisions. +- Loaded at startup from project or user configuration. + +### 2. LEARNED Memory (`learned_memory` tool) +- Cross-session patterns and discoveries that persist across sessions. +- **Recording**: Use `learned_memory` with `action="record"`, provide a `confidence` score (0.0-1.0) reflecting how certain you are about the pattern. Entries below the configured confidence threshold are rejected. +- **Recall**: Use `learned_memory` with `action="recall"` to retrieve past learnings. +- Record here: repeated user preferences, discovered conventions, successful strategies. +- Do NOT record: session-specific context, temporary state, unverified guesses. + +### 3. EPHEMERAL Memory (`ephemeral_memory` tool) +- Session-scoped context, automatically cleared when the session ends. +- Use for: current task state, intermediate results, temporary notes. +- Use `ephemeral_memory` with `action="record"` to store, `action="recall"` to retrieve. + +## Decision Routing + +When handling a request, classify and route your decision process: + +### Simple Queries +- Factual lookups, single-step operations, straightforward questions +- **Route**: Respond directly using available knowledge or a single tool call +- Skip deep analysis — be efficient + +### Complex Multi-Step Tasks +- Multi-file changes, architecture decisions, debugging, planning +- **Route**: Follow the structured think→plan→execute→verify→reflect→learn cycle: + 1. **Analyze**: Understand the request and gather context (use CORE/LEARNED memory) + 2. **Plan**: Break into concrete steps, identify required tools/skills + 3. **Execute**: Run tools systematically, check outputs at each step + 4. **Verify**: Confirm each step succeeded before proceeding + 5. **Reflect**: After completion, assess what worked + 6. **Learn**: Record reusable patterns to LEARNED memory (with appropriate confidence) + +### Conflict or Uncertainty +- Ambiguous requests, conflicting constraints, trade-off decisions +- **Route**: Identify the conflict, weigh alternatives, use confidence scoring +- If uncertain about a LEARNED memory pattern, check against CORE rules first +- When confidence is low (< 0.7), prefer safer/conservative approaches + +### Memory Usage Guidelines +- Check CORE memory for project rules before major decisions +- Recall LEARNED memory for past patterns at the start of complex tasks +- Record to LEARNED only when: (a) pattern is reusable, (b) confidence is high +- Use EPHEMERAL memory for temporary state within a single session +- Never record sensitive information (API keys, passwords) to any memory layer + ## Workspace Context You are working in a workspace directory. All operations are relative to this context unless absolute paths are specified. diff --git a/mini_agent/memory/__init__.py b/mini_agent/memory/__init__.py new file mode 100644 index 00000000..3df57548 --- /dev/null +++ b/mini_agent/memory/__init__.py @@ -0,0 +1,19 @@ +"""Three-layer memory system for Mini-Agent. + +CORE: Immutable project rules and safety constraints. +LEARNED: Confidence-gated cross-session decision patterns. +EPHEMERAL: Session-scoped context, cleared on reset. +""" + +from .manager import MemoryManager +from .schema import MemoryEntry, MemoryStore +from .tools import CoreMemoryTool, EphemeralMemoryTool, LearnedMemoryTool + +__all__ = [ + "MemoryManager", + "MemoryEntry", + "MemoryStore", + "CoreMemoryTool", + "LearnedMemoryTool", + "EphemeralMemoryTool", +] diff --git a/mini_agent/memory/manager.py b/mini_agent/memory/manager.py new file mode 100644 index 00000000..19322534 --- /dev/null +++ b/mini_agent/memory/manager.py @@ -0,0 +1,121 @@ +"""MemoryManager — manages the three-layer memory system. + +Layers: + CORE — Immutable project rules, loaded from JSON file, read-only. + LEARNED — Confidence-gated cross-session patterns, persisted to disk. + EPHEMERAL — Session-scoped context, in-memory only, cleared on reset. +""" + +import uuid +from pathlib import Path + +from .schema import MemoryEntry, MemoryStore + + +class MemoryManager: + """Manages CORE, LEARNED, and EPHEMERAL memory layers.""" + + def __init__( + self, + core_memory_path: str | None = None, + learned_memory_dir: str | None = None, + learned_confidence_threshold: float = 0.7, + ): + self._confidence_threshold = learned_confidence_threshold + + # --- CORE layer --- + if core_memory_path: + self._core_path = Path(core_memory_path).expanduser() + else: + self._core_path = Path.home() / ".mini-agent" / "memory" / "core.json" + self._core_store = MemoryStore.load_from_file(self._core_path) + + # --- LEARNED layer --- + if learned_memory_dir: + self._learned_path = Path(learned_memory_dir).expanduser() / "learned.json" + else: + self._learned_path = Path.home() / ".mini-agent" / "memory" / "learned.json" + self._learned_store = MemoryStore.load_from_file(self._learned_path) + + # --- EPHEMERAL layer --- + self._ephemeral_store = MemoryStore() + + # ------------------------------------------------------------------ + # CORE — read-only + # ------------------------------------------------------------------ + + def get_core_memories(self, category: str | None = None) -> list[MemoryEntry]: + """Retrieve CORE memories, optionally filtered by category.""" + entries = self._core_store.entries + if category: + entries = [e for e in entries if e.category == category] + return entries + + # ------------------------------------------------------------------ + # LEARNED — confidence-gated write + recall + # ------------------------------------------------------------------ + + def get_learned_memories(self, category: str | None = None) -> list[MemoryEntry]: + """Retrieve LEARNED memories, optionally filtered by category.""" + entries = self._learned_store.entries + if category: + entries = [e for e in entries if e.category == category] + return entries + + def record_learned( + self, content: str, confidence: float, category: str = "general" + ) -> tuple[bool, str]: + """Record a LEARNED memory entry with confidence gating. + + Args: + content: The memory content to store. + confidence: Confidence score (0.0-1.0). Must meet threshold. + category: Optional category tag. + + Returns: + Tuple of (accepted: bool, reason: str). + """ + if confidence < self._confidence_threshold: + return ( + False, + f"Confidence {confidence:.2f} below threshold {self._confidence_threshold}. " + "Memory not recorded.", + ) + + entry = MemoryEntry( + id=str(uuid.uuid4())[:8], + content=content, + category=category, + confidence=confidence, + source="learned", + ) + self._learned_store.entries.append(entry) + self._learned_store.save_to_file(self._learned_path) + return True, f"Recorded learned memory (confidence: {confidence:.2f})" + + # ------------------------------------------------------------------ + # EPHEMERAL — session-scoped, in-memory + # ------------------------------------------------------------------ + + def get_ephemeral(self, category: str | None = None) -> list[MemoryEntry]: + """Retrieve EPHEMERAL memories, optionally filtered by category.""" + entries = self._ephemeral_store.entries + if category: + entries = [e for e in entries if e.category == category] + return entries + + def record_ephemeral(self, content: str, category: str = "general") -> MemoryEntry: + """Record a session-scoped EPHEMERAL memory entry.""" + entry = MemoryEntry( + id=str(uuid.uuid4())[:8], + content=content, + category=category, + confidence=1.0, + source="ephemeral", + ) + self._ephemeral_store.entries.append(entry) + return entry + + def reset_ephemeral(self) -> None: + """Clear all EPHEMERAL memories.""" + self._ephemeral_store = MemoryStore() diff --git a/mini_agent/memory/schema.py b/mini_agent/memory/schema.py new file mode 100644 index 00000000..a4e79b97 --- /dev/null +++ b/mini_agent/memory/schema.py @@ -0,0 +1,53 @@ +"""Memory data models for the three-layer memory system.""" + +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Literal + +from pydantic import BaseModel, Field + + +class MemoryEntry(BaseModel): + """A single memory entry with confidence scoring.""" + + id: str + content: str + category: str = "general" + confidence: float = Field(default=1.0, ge=0.0, le=1.0) + timestamp: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat()) + source: Literal["core", "learned", "ephemeral"] = "ephemeral" + + +class MemoryStore(BaseModel): + """Container for a list of memory entries with serialization helpers.""" + + entries: list[MemoryEntry] = Field(default_factory=list) + + def to_dicts(self) -> list[dict]: + """Serialize entries to list of dicts for JSON storage.""" + return [entry.model_dump() for entry in self.entries] + + @classmethod + def from_dicts(cls, data: list[dict]) -> "MemoryStore": + """Deserialize from list of dicts.""" + return cls(entries=[MemoryEntry(**item) for item in data]) + + @classmethod + def load_from_file(cls, filepath: Path) -> "MemoryStore": + """Load memory store from a JSON file. Returns empty store if file doesn't exist.""" + if not filepath.exists(): + return cls() + try: + data = json.loads(filepath.read_text(encoding="utf-8")) + return cls.from_dicts(data) + except (json.JSONDecodeError, KeyError): + return cls() + + def save_to_file(self, filepath: Path) -> None: + """Save memory store to a JSON file, creating parent directories as needed.""" + filepath.parent.mkdir(parents=True, exist_ok=True) + filepath.write_text( + json.dumps(self.to_dicts(), indent=2, ensure_ascii=False), + encoding="utf-8", + ) diff --git a/mini_agent/memory/tools.py b/mini_agent/memory/tools.py new file mode 100644 index 00000000..0de22c95 --- /dev/null +++ b/mini_agent/memory/tools.py @@ -0,0 +1,192 @@ +"""Memory tools for the three-layer memory system. + +CoreMemoryTool — Read-only access to CORE memories. +LearnedMemoryTool — Recall + confidence-gated record to LEARNED. +EphemeralMemoryTool — Session-scoped read/write, auto-cleaned. +""" + +from typing import Any + +from mini_agent.tools.base import Tool, ToolResult + +from .manager import MemoryManager +from .schema import MemoryEntry + + +def _format_entries(entries: list[MemoryEntry]) -> str: + """Format memory entries for tool output.""" + if not entries: + return "No memories found." + lines = [] + for i, entry in enumerate(entries, 1): + line = ( + f"{i}. [{entry.category}] {entry.content}" + + "\n" + + f" (id: {entry.id}, confidence: {entry.confidence:.2f}, " + + f"source: {entry.source}, recorded: {entry.timestamp})" + ) + lines.append(line) + return "\n".join(lines) + + +class CoreMemoryTool(Tool): + """Read-only access to CORE (immutable) project memories.""" + + def __init__(self, manager: MemoryManager): + self._manager = manager + + @property + def name(self) -> str: + return "core_memory" + + @property + def description(self) -> str: + return ( + "Retrieve immutable CORE project memories (rules, safety constraints, " + "workspace conventions). Read-only — these cannot be modified at runtime. " + "Use this to recall project-level rules before making decisions." + ) + + @property + def parameters(self) -> dict[str, Any]: + return { + "type": "object", + "properties": { + "category": { + "type": "string", + "description": "Optional: filter core memories by category (e.g., 'safety', 'convention')", + }, + }, + } + + async def execute(self, category: str | None = None) -> ToolResult: + entries = self._manager.get_core_memories(category=category) + return ToolResult(success=True, content=_format_entries(entries)) + + +class LearnedMemoryTool(Tool): + """Recall and confidence-gated record for LEARNED memories.""" + + def __init__(self, manager: MemoryManager): + self._manager = manager + + @property + def name(self) -> str: + return "learned_memory" + + @property + def description(self) -> str: + return ( + "Recall or record LEARNED memories — cross-session patterns and discoveries. " + "Recording requires a confidence score (0.0-1.0); entries below the confidence " + "threshold are rejected. Use 'recall' to retrieve past learnings, " + "'record' to store a new pattern with a confidence score." + ) + + @property + def parameters(self) -> dict[str, Any]: + return { + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": ["recall", "record"], + "description": "Action: 'recall' to retrieve, 'record' to store (requires content + confidence).", + }, + "content": { + "type": "string", + "description": "For 'record': the memory content. For 'recall': ignored.", + }, + "category": { + "type": "string", + "description": "Optional category/tag for filtering or labeling.", + }, + "confidence": { + "type": "number", + "description": "For 'record': confidence score (0.0-1.0). Below threshold = rejected.", + }, + }, + "required": ["action"], + } + + async def execute( + self, + action: str, + content: str = "", + category: str = "general", + confidence: float = 1.0, + ) -> ToolResult: + if action == "recall": + cat = category if category != "general" else None + entries = self._manager.get_learned_memories(category=cat) + return ToolResult(success=True, content=_format_entries(entries)) + + if action == "record": + if not content: + return ToolResult(success=False, content="", error="'content' is required for 'record' action.") + accepted, reason = self._manager.record_learned( + content=content, confidence=confidence, category=category + ) + return ToolResult(success=accepted, content=reason) + + return ToolResult(success=False, content="", error=f"Unknown action: '{action}'. Use 'recall' or 'record'.") + + +class EphemeralMemoryTool(Tool): + """Session-scoped read/write for EPHEMERAL memories.""" + + def __init__(self, manager: MemoryManager): + self._manager = manager + + @property + def name(self) -> str: + return "ephemeral_memory" + + @property + def description(self) -> str: + return ( + "Recall or record EPHEMERAL (session-scoped) memories. Cleared at session end. " + "Use for temporary context, current task state, and one-off facts. " + "Use 'recall' to retrieve, 'record' to store." + ) + + @property + def parameters(self) -> dict[str, Any]: + return { + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": ["recall", "record"], + "description": "Action: 'recall' to retrieve, 'record' to store.", + }, + "content": { + "type": "string", + "description": "For 'record': the content to store. For 'recall': ignored.", + }, + "category": { + "type": "string", + "description": "Optional category/tag for filtering or labeling.", + }, + }, + "required": ["action"], + } + + async def execute( + self, + action: str, + content: str = "", + category: str = "general", + ) -> ToolResult: + if action == "recall": + cat = category if category != "general" else None + entries = self._manager.get_ephemeral(category=cat) + return ToolResult(success=True, content=_format_entries(entries)) + + if action == "record": + if not content: + return ToolResult(success=False, content="", error="'content' is required for 'record' action.") + entry = self._manager.record_ephemeral(content=content, category=category) + return ToolResult(success=True, content=f"Recorded ephemeral memory (id: {entry.id})") + + return ToolResult(success=False, content="", error=f"Unknown action: '{action}'. Use 'recall' or 'record'.") diff --git a/tests/test_memory.py b/tests/test_memory.py new file mode 100644 index 00000000..7ab3b403 --- /dev/null +++ b/tests/test_memory.py @@ -0,0 +1,246 @@ +"""Test cases for three-layer memory system.""" + +import tempfile +from pathlib import Path + +import pytest + +from mini_agent.memory.manager import MemoryManager +from mini_agent.memory.schema import MemoryEntry, MemoryStore +from mini_agent.memory.tools import CoreMemoryTool, EphemeralMemoryTool, LearnedMemoryTool + + +# --------------------------------------------------------------------------- +# MemoryEntry & MemoryStore +# --------------------------------------------------------------------------- + +def test_memory_entry_creation(): + """Test creating a MemoryEntry with default values.""" + entry = MemoryEntry(id="test_1", content="Hello world", source="ephemeral") + assert entry.id == "test_1" + assert entry.content == "Hello world" + assert entry.category == "general" + assert entry.confidence == 1.0 + assert entry.source == "ephemeral" + +def test_memory_entry_confidence_bounds(): + """Test confidence is clamped to 0.0-1.0.""" + entry = MemoryEntry(id="t", content="x", confidence=0.5, source="learned") + assert entry.confidence == 0.5 + with pytest.raises(Exception): + MemoryEntry(id="t", content="x", confidence=-0.1, source="learned") + with pytest.raises(Exception): + MemoryEntry(id="t", content="x", confidence=1.5, source="learned") + +def test_memory_store_roundtrip(): + """Test serialization round-trip.""" + store = MemoryStore() + store.entries.append(MemoryEntry(id="a", content="one", source="core")) + store.entries.append(MemoryEntry(id="b", content="two", source="learned", confidence=0.8)) + dicts = store.to_dicts() + restored = MemoryStore.from_dicts(dicts) + assert len(restored.entries) == 2 + assert restored.entries[0].content == "one" + assert restored.entries[1].confidence == 0.8 + +def test_memory_store_file_persistence(): + """Test saving and loading from file.""" + with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".json") as f: + filepath = Path(f.name) + + try: + store = MemoryStore() + store.entries.append(MemoryEntry(id="x", content="persisted", source="learned")) + store.save_to_file(filepath) + + reloaded = MemoryStore.load_from_file(filepath) + assert len(reloaded.entries) == 1 + assert reloaded.entries[0].content == "persisted" + finally: + filepath.unlink(missing_ok=True) + +def test_memory_store_load_missing_file(): + """Test loading from a non-existent file returns empty store.""" + store = MemoryStore.load_from_file(Path("/nonexistent/memory.json")) + assert len(store.entries) == 0 + + +# --------------------------------------------------------------------------- +# MemoryManager +# --------------------------------------------------------------------------- + +def test_memory_manager_core_readonly(): + """Test CORE memories can be read but have no write method.""" + with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".json") as f: + f.write('[{"id":"c1","content":"rule","category":"safety","confidence":1.0,"source":"core"}]') + core_path = f.name + + try: + mgr = MemoryManager(core_memory_path=core_path) + entries = mgr.get_core_memories() + assert len(entries) == 1 + assert entries[0].content == "rule" + finally: + Path(core_path).unlink(missing_ok=True) + +def test_memory_manager_core_filter_by_category(): + """Test filtering CORE memories by category.""" + with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".json") as f: + f.write('[{' + '"id":"c1","content":"safety rule","category":"safety","confidence":1.0,"source":"core"},' + '{"id":"c2","content":"convention","category":"convention","confidence":1.0,"source":"core"}' + ']') + core_path = f.name + + try: + mgr = MemoryManager(core_memory_path=core_path) + safety = mgr.get_core_memories(category="safety") + assert len(safety) == 1 + assert safety[0].content == "safety rule" + finally: + Path(core_path).unlink(missing_ok=True) + +def test_memory_manager_learned_confidence_gating(): + """Test LEARNED writes above threshold succeed, below fail.""" + with tempfile.TemporaryDirectory() as tmpdir: + mgr = MemoryManager(learned_memory_dir=tmpdir, learned_confidence_threshold=0.7) + + accepted, reason = mgr.record_learned("high confidence", confidence=0.9, category="test") + assert accepted + assert "Recorded" in reason + + accepted, reason = mgr.record_learned("low confidence", confidence=0.3, category="test") + assert not accepted + assert "below threshold" in reason + + entries = mgr.get_learned_memories() + assert len(entries) == 1 + assert entries[0].content == "high confidence" + +def test_memory_manager_learned_persistence(): + """Test LEARNED memories persist across manager instances.""" + with tempfile.TemporaryDirectory() as tmpdir: + mgr1 = MemoryManager(learned_memory_dir=tmpdir) + mgr1.record_learned("persist", confidence=0.8) + + mgr2 = MemoryManager(learned_memory_dir=tmpdir) + entries = mgr2.get_learned_memories() + assert len(entries) == 1 + assert entries[0].content == "persist" + +def test_memory_manager_ephemeral_scope(): + """Test EPHEMERAL record/recall works, reset clears.""" + mgr = MemoryManager() + mgr.record_ephemeral("temp data", category="session") + mgr.record_ephemeral("more temp", category="session") + + entries = mgr.get_ephemeral() + assert len(entries) == 2 + + mgr.reset_ephemeral() + assert len(mgr.get_ephemeral()) == 0 + +def test_memory_manager_ephemeral_empty(): + """Test EPHEMERAL returns empty when no entries.""" + mgr = MemoryManager() + assert len(mgr.get_ephemeral()) == 0 + + +# --------------------------------------------------------------------------- +# Memory Tools +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_core_memory_tool(): + """Test CoreMemoryTool returns CORE memories.""" + with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".json") as f: + f.write('[{"id":"c1","content":"core rule","category":"safety","confidence":1.0,"source":"core"}]') + core_path = f.name + + try: + mgr = MemoryManager(core_memory_path=core_path) + tool = CoreMemoryTool(mgr) + result = await tool.execute() + assert result.success + assert "core rule" in result.content + finally: + Path(core_path).unlink(missing_ok=True) + +@pytest.mark.asyncio +async def test_learned_memory_tool_recall(): + """Test LearnedMemoryTool recall action.""" + with tempfile.TemporaryDirectory() as tmpdir: + mgr = MemoryManager(learned_memory_dir=tmpdir) + mgr.record_learned("pattern A", confidence=0.85) + mgr.record_learned("pattern B", confidence=0.75) + + tool = LearnedMemoryTool(mgr) + result = await tool.execute(action="recall") + assert result.success + assert "pattern A" in result.content + assert "pattern B" in result.content + +@pytest.mark.asyncio +async def test_learned_memory_tool_record_success(): + """Test LearnedMemoryTool record with high confidence succeeds.""" + with tempfile.TemporaryDirectory() as tmpdir: + mgr = MemoryManager(learned_memory_dir=tmpdir, learned_confidence_threshold=0.7) + tool = LearnedMemoryTool(mgr) + + result = await tool.execute(action="record", content="new pattern", confidence=0.9) + assert result.success + assert "Recorded" in result.content + +@pytest.mark.asyncio +async def test_learned_memory_tool_record_rejected(): + """Test LearnedMemoryTool record with low confidence fails.""" + mgr = MemoryManager(learned_confidence_threshold=0.7) + tool = LearnedMemoryTool(mgr) + + result = await tool.execute(action="record", content="weak pattern", confidence=0.4) + assert not result.success + assert "below threshold" in result.content + +@pytest.mark.asyncio +async def test_learned_memory_tool_record_missing_content(): + """Test LearnedMemoryTool record fails without content.""" + mgr = MemoryManager() + tool = LearnedMemoryTool(mgr) + result = await tool.execute(action="record", content="") + assert not result.success + assert "content" in result.error.lower() + +@pytest.mark.asyncio +async def test_ephemeral_memory_tool(): + """Test EphemeralMemoryTool record and recall.""" + mgr = MemoryManager() + tool = EphemeralMemoryTool(mgr) + + result = await tool.execute(action="record", content="session data") + assert result.success + assert "Recorded" in result.content + + result = await tool.execute(action="recall") + assert result.success + assert "session data" in result.content + +@pytest.mark.asyncio +async def test_ephemeral_memory_tool_reset(): + """Test EphemeralMemoryTool recall after reset returns empty.""" + mgr = MemoryManager() + tool = EphemeralMemoryTool(mgr) + await tool.execute(action="record", content="temp") + mgr.reset_ephemeral() + + result = await tool.execute(action="recall") + assert result.success + assert "No memories found" in result.content + +@pytest.mark.asyncio +async def test_ephemeral_memory_tool_missing_content(): + """Test EphemeralMemoryTool record fails without content.""" + mgr = MemoryManager() + tool = EphemeralMemoryTool(mgr) + result = await tool.execute(action="record", content="") + assert not result.success + assert "content" in result.error.lower()