feat(memory): Add three-layer memory system with decision routing

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"""

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(

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)

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"
+  }
+]

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.

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",
+]

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()