Add multi-provider memory system with auto-detection

- Add MemoryToolAdapter for unified memory across providers
- Anthropic: Uses native memory tool (memory_20250818) for subscription safety
- OpenAI/Gemini/Others: Uses function calling format
- All providers share the same semantic vector store backend
- Simplify CLI to single --memory flag with auto-detection
- Add proper resource cleanup (close methods) to fix test isolation
- Update README with memory documentation

README.md

@@ -256,6 +256,19 @@ ANTHROPIC_BASE_URL=http://localhost:8787 claude
 OPENAI_BASE_URL=http://localhost:8787/v1 cursor
 ```
 
+**Enable Persistent Memory** - Claude remembers across conversations:
+
+```bash
+headroom proxy --memory
+```
+
+Memory auto-detects your provider (Anthropic, OpenAI, Gemini) and uses the appropriate format:
+- **Anthropic**: Uses native memory tool (`memory_20250818`) - works with Claude Code subscriptions
+- **OpenAI/Gemini/Others**: Uses function calling format
+- All providers share the same semantic vector store for search
+
+Set `x-headroom-user-id` header for per-user memory isolation (defaults to 'default').
+
 **Using AWS Bedrock, Google Vertex, or Azure?** Route through Headroom:
 
 ```bash

headroom/cli/proxy.py

@@ -45,22 +45,17 @@ from .main import main
     is_flag=True,
     help="Disable trying deeper compression before dropping messages",
 )
-# Memory System
+# Memory System (Multi-Provider Support)
 @click.option(
     "--memory",
     is_flag=True,
-    help="Enable persistent user memory (uses x-headroom-user-id header if set, otherwise 'default')",
-)
-@click.option(
-    "--memory-backend",
-    type=click.Choice(["local", "qdrant-neo4j"]),
-    default="local",
-    help="Memory storage backend: local (SQLite+HNSW) or qdrant-neo4j (default: local)",
+    help="Enable persistent user memory. Auto-detects provider and uses appropriate tool format. "
+    "Set x-headroom-user-id header for per-user memory (defaults to 'default').",
 )
 @click.option(
     "--memory-db-path",
     default="headroom_memory.db",
-    help="Path to memory database file for local backend (default: headroom_memory.db)",
+    help="Path to memory database file (default: headroom_memory.db)",
 )
 @click.option("--no-memory-tools", is_flag=True, help="Disable automatic memory tool injection")
 @click.option(
@@ -114,7 +109,6 @@ def proxy(
     no_intelligent_scoring: bool,
     no_compress_first: bool,
     memory: bool,
-    memory_backend: str,
     memory_db_path: str,
     no_memory_tools: bool,
     no_memory_context: bool,
@@ -166,9 +160,8 @@ def proxy(
         intelligent_context=not no_intelligent_context,
         intelligent_context_scoring=not no_intelligent_scoring,
         intelligent_context_compress_first=not no_compress_first,
-        # Memory System
+        # Memory System (Multi-Provider with auto-detection)
         memory_enabled=memory,
-        memory_backend=memory_backend,  # type: ignore[arg-type]
         memory_db_path=memory_db_path,
         memory_inject_tools=not no_memory_tools,
         memory_inject_context=not no_memory_context,
@@ -181,7 +174,7 @@ def proxy(
 
     memory_status = "DISABLED"
     if config.memory_enabled:
-        memory_status = f"ENABLED ({config.memory_backend})"
+        memory_status = "ENABLED (multi-provider)"
 
     effective_region = bedrock_region or region
     backend_status = "Anthropic (direct API)"
@@ -220,12 +213,16 @@ IMPORTANT for {provider_config.display_name} users:
     memory_section = ""
     if config.memory_enabled:
         memory_section = f"""
-Memory:
-  - Memories are scoped per user. Set x-headroom-user-id header (defaults to 'default').
-  - Tools: {"ENABLED" if config.memory_inject_tools else "DISABLED"}  Context: {"ENABLED" if config.memory_inject_context else "DISABLED"}
+Memory (Multi-Provider):
+  - Auto-detects provider from request (Anthropic, OpenAI, Gemini, etc.)
+  - Anthropic: Uses native memory tool (memory_20250818) - subscription safe
+  - OpenAI/Gemini/Others: Uses function calling format
+  - All providers share the same semantic vector store backend
+  - Set x-headroom-user-id header for per-user memory (defaults to 'default')
+  - Tools: {"ENABLED" if config.memory_inject_tools else "DISABLED"}
+  - Context injection: {"ENABLED" if config.memory_inject_context else "DISABLED"}
+  - Database: {config.memory_db_path}
 """
-        if config.memory_inject_tools:
-            memory_section += "  - NOTE: Memory tools require ANTHROPIC_API_KEY.\n"
 
     click.echo(f"""
 ╔═══════════════════════════════════════════════════════════════════════╗

headroom/memory/adapters/embedders.py

@@ -259,6 +259,11 @@ class LocalEmbedder:
         """Return the maximum number of tokens the model can process."""
         return self.DEFAULT_MAX_TOKENS
 
+    async def close(self) -> None:
+        """Close resources (no-op for local embedder)."""
+        # LocalEmbedder doesn't hold persistent connections
+        pass
+
 
 # =============================================================================
 # OpenAIEmbedder - OpenAI API
@@ -465,6 +470,13 @@ class OpenAIEmbedder:
         """Return the maximum number of tokens the model can process."""
         return self.DEFAULT_MAX_TOKENS
 
+    async def close(self) -> None:
+        """Close the OpenAI async client and its underlying httpx connection."""
+        if "_async_client" in self.__dict__:
+            await self._async_client.close()
+            # Remove from cache to allow re-creation if needed
+            del self.__dict__["_async_client"]
+
 
 # =============================================================================
 # OllamaEmbedder - Ollama API

headroom/memory/backends/local.py

@@ -639,6 +639,9 @@ class LocalBackend:
 
     async def close(self) -> None:
         """Close the backend and release resources."""
+        # Close HierarchicalMemory to release httpx clients in embedders
+        if self._hierarchical_memory is not None:
+            await self._hierarchical_memory.close()
         self._hierarchical_memory = None
         self._graph = None
         self._initialized = False

headroom/memory/core.py

@@ -859,3 +859,43 @@ class HierarchicalMemory:
     def config(self) -> MemoryConfig:
         """Access the configuration."""
         return self._config
+
+    # =========================================================================
+    # Lifecycle
+    # =========================================================================
+
+    async def close(self) -> None:
+        """Close all resources held by the memory system.
+
+        This should be called when done using the memory system to properly
+        clean up resources like HTTP clients used by embedders.
+        """
+        # Close embedder if it has a close method (e.g., API-based embedders)
+        if hasattr(self._embedder, "close"):
+            await self._embedder.close()
+
+        # Close store if it has a close method
+        if hasattr(self._store, "close"):
+            await self._store.close()
+
+        # Close vector index if it has a close method
+        if hasattr(self._vector_index, "close"):
+            await self._vector_index.close()
+
+        # Close text index if it has a close method
+        if hasattr(self._text_index, "close"):
+            await self._text_index.close()
+
+        # Close cache if it has a close method
+        if self._cache is not None and hasattr(self._cache, "close"):
+            await self._cache.close()
+
+        logger.debug("HierarchicalMemory closed")
+
+    async def __aenter__(self) -> HierarchicalMemory:
+        """Async context manager entry."""
+        return self
+
+    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
+        """Async context manager exit - closes resources."""
+        await self.close()

headroom/proxy/memory_handler.py

@@ -27,6 +27,7 @@ from __future__ import annotations
 import json
 import logging
 from dataclasses import dataclass
+from pathlib import Path
 from typing import TYPE_CHECKING, Any, Literal
 
 if TYPE_CHECKING:
@@ -34,9 +35,18 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
-# Memory tool names for detection
+# Memory tool names for detection (Headroom's custom tools)
 MEMORY_TOOL_NAMES = {"memory_save", "memory_search", "memory_update", "memory_delete"}
 
+# Anthropic's native memory tool name
+NATIVE_MEMORY_TOOL_NAME = "memory"
+
+# Beta header required for native memory tool
+NATIVE_MEMORY_BETA_HEADER = "context-management-2025-06-27"
+
+# Native memory tool type
+NATIVE_MEMORY_TOOL_TYPE = "memory_20250818"
+
 
 @dataclass
 class MemoryConfig:
@@ -49,6 +59,9 @@ class MemoryConfig:
     inject_context: bool = True
     top_k: int = 10
     min_similarity: float = 0.3
+    # Native memory tool (Anthropic's built-in memory_20250818)
+    use_native_tool: bool = False
+    native_memory_dir: str = ""  # Directory for native memory files (default: ~/.headroom/memories)
     # Qdrant+Neo4j config
     qdrant_host: str = "localhost"
     qdrant_port: int = 6333
@@ -65,6 +78,10 @@ class MemoryHandler:
     2. Inject memory tools into requests
     3. Search and inject relevant memories as context
     4. Handle memory tool calls in responses
+
+    Supports two modes:
+    - Custom tools: Headroom's memory_save, memory_search, etc. (default)
+    - Native tool: Anthropic's memory_20250818 built-in tool (experimental)
     """
 
     def __init__(self, config: MemoryConfig) -> None:
@@ -72,6 +89,32 @@ class MemoryHandler:
         self._backend: LocalBackend | Any = None
         self._initialized = False
         self._memory_tools: list[dict[str, Any]] | None = None
+        # Native memory tool directory
+        self._native_memory_dir: Path | None = None
+        if config.use_native_tool:
+            self._init_native_memory_dir()
+
+    def _init_native_memory_dir(self) -> None:
+        """Initialize native memory directory."""
+        if self.config.native_memory_dir:
+            self._native_memory_dir = Path(self.config.native_memory_dir)
+        else:
+            # Default: ~/.headroom/memories
+            self._native_memory_dir = Path.home() / ".headroom" / "memories"
+
+        # Create directory if it doesn't exist
+        self._native_memory_dir.mkdir(parents=True, exist_ok=True)
+        logger.info(f"Memory: Native memory directory: {self._native_memory_dir}")
+
+    def get_beta_headers(self) -> dict[str, str]:
+        """Get beta headers required for native memory tool.
+
+        Returns:
+            Dict with beta headers to add to request, or empty dict.
+        """
+        if self.config.use_native_tool and self.config.inject_tools:
+            return {"anthropic-beta": NATIVE_MEMORY_BETA_HEADER}
+        return {}
 
     async def _ensure_initialized(self) -> None:
         """Lazy initialization of memory backend."""
@@ -147,6 +190,10 @@ class MemoryHandler:
 
         tools = list(tools) if tools else []
 
+        # Use native memory tool if configured
+        if self.config.use_native_tool:
+            return self._inject_native_tool(tools)
+
         # Check which tools are already present
         existing_names: set[str] = set()
         for tool in tools:
@@ -178,6 +225,35 @@ class MemoryHandler:
 
         return tools, was_injected
 
+    def _inject_native_tool(self, tools: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], bool]:
+        """Inject Anthropic's native memory tool (memory_20250818).
+
+        This uses Anthropic's built-in memory tool format which may be
+        allowed by Claude Code subscription credentials (unlike custom tools).
+
+        Returns:
+            Tuple of (updated_tools, was_injected).
+        """
+        # Check if native memory tool already present
+        for tool in tools:
+            if tool.get("type") == NATIVE_MEMORY_TOOL_TYPE:
+                return tools, False
+            if tool.get("name") == NATIVE_MEMORY_TOOL_NAME:
+                return tools, False
+
+        # Add native memory tool
+        native_tool = {
+            "type": NATIVE_MEMORY_TOOL_TYPE,
+            "name": NATIVE_MEMORY_TOOL_NAME,
+        }
+        tools.append(native_tool)
+
+        logger.info(
+            f"Memory: Injected native memory tool ({NATIVE_MEMORY_TOOL_TYPE}). "
+            f"Beta header required: {NATIVE_MEMORY_BETA_HEADER}"
+        )
+        return tools, True
+
     async def search_and_format_context(
         self,
         user_id: str,
@@ -283,7 +359,8 @@ Use this context to provide personalized and contextually relevant responses."""
         tool_calls = self._extract_tool_calls(response, provider)
         for tc in tool_calls:
             name = tc.get("name") or tc.get("function", {}).get("name")
-            if name in MEMORY_TOOL_NAMES:
+            # Check for both custom and native memory tools
+            if name in MEMORY_TOOL_NAMES or name == NATIVE_MEMORY_TOOL_NAME:
                 return True
         return False
 
@@ -324,18 +401,11 @@ Use this context to provide personalized and contextually relevant responses."""
         Returns:
             List of tool results in provider format.
         """
-        await self._ensure_initialized()
-        if not self._backend:
-            return []
-
         tool_calls = self._extract_tool_calls(response, provider)
         results: list[dict[str, Any]] = []
 
         for tc in tool_calls:
             tool_name = tc.get("name") or tc.get("function", {}).get("name")
-            if tool_name not in MEMORY_TOOL_NAMES:
-                continue
-
             tool_id = tc.get("id", "")
 
             # Parse input data
@@ -348,8 +418,17 @@ Use this context to provide personalized and contextually relevant responses."""
                 except json.JSONDecodeError:
                     input_data = {}
 
-            # Execute the tool
-            result_content = await self._execute_memory_tool(tool_name, input_data, user_id)
+            # Handle native memory tool
+            if tool_name == NATIVE_MEMORY_TOOL_NAME:
+                result_content = await self._execute_native_memory_tool(input_data, user_id)
+            elif tool_name in MEMORY_TOOL_NAMES:
+                # Custom memory tools need backend
+                await self._ensure_initialized()
+                if not self._backend:
+                    continue
+                result_content = await self._execute_memory_tool(tool_name, input_data, user_id)
+            else:
+                continue
 
             # Format result based on provider
             if provider == "anthropic":
@@ -522,6 +601,780 @@ Use this context to provide personalized and contextually relevant responses."""
             }
         )
 
+    # =========================================================================
+    # Native Memory Tool (Anthropic's memory_20250818)
+    # =========================================================================
+    #
+    # HYBRID ARCHITECTURE:
+    # Claude uses Anthropic's native memory tool interface (file operations),
+    # but we translate these to our semantic vector store backend.
+    #
+    # This gives us:
+    # - Native tool format (subscription-safe, approved by Anthropic)
+    # - Semantic search (our vector embeddings under the hood)
+    # - Best of both worlds
+    #
+    # Translation mapping:
+    #   view /memories              → Show overview + search instructions
+    #   view /memories/search/X     → Semantic search for X
+    #   view /memories/recent       → Recent memories
+    #   view /memories/<path>       → Find memory by path/topic
+    #   create /memories/<path>     → Save to vector store (path as tag)
+    #   delete /memories/<path>     → Delete from vector store
+    #   str_replace                 → Update memory content
+    # =========================================================================
+
+    async def _execute_native_memory_tool(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Execute Anthropic's native memory tool with semantic backend.
+
+        This is a TRANSLATION LAYER: Claude thinks it's doing file operations,
+        but we're actually using our semantic vector store.
+
+        Commands:
+        - view: Semantic search or list memories
+        - create: Save to vector store
+        - str_replace: Update memory content
+        - insert: Append to memory
+        - delete: Remove from vector store
+        - rename: Update memory tags/path
+        """
+        # Ensure our semantic backend is initialized
+        await self._ensure_initialized()
+
+        command = input_data.get("command", "")
+
+        try:
+            if command == "view":
+                return await self._native_view_semantic(input_data, user_id)
+            elif command == "create":
+                return await self._native_create_semantic(input_data, user_id)
+            elif command == "str_replace":
+                return await self._native_update_semantic(input_data, user_id)
+            elif command == "insert":
+                return await self._native_append_semantic(input_data, user_id)
+            elif command == "delete":
+                return await self._native_delete_semantic(input_data, user_id)
+            elif command == "rename":
+                return await self._native_rename_semantic(input_data, user_id)
+            else:
+                return f"Error: Unknown command '{command}'"
+        except Exception as e:
+            logger.error(f"Memory: Native tool error: {e}")
+            return f"Error: {e}"
+
+    def _resolve_native_path(self, path: str, user_id: str) -> Path:
+        """Resolve path within user's memory directory safely.
+
+        Prevents path traversal attacks by ensuring path stays within
+        the user's memory directory.
+        """
+        assert self._native_memory_dir is not None
+
+        # User-scoped memory directory
+        user_dir = self._native_memory_dir / user_id
+        user_dir.mkdir(parents=True, exist_ok=True)
+
+        # Normalize path (remove /memories prefix if present)
+        if path.startswith("/memories"):
+            path = path[len("/memories") :]
+        if path.startswith("/"):
+            path = path[1:]
+
+        # Resolve and validate
+        resolved = (user_dir / path).resolve()
+
+        # Security: ensure path is within user directory
+        try:
+            resolved.relative_to(user_dir.resolve())
+        except ValueError:
+            raise ValueError(f"Path traversal detected: {path}") from None
+
+        return resolved
+
+    def _native_view(self, input_data: dict[str, Any], user_id: str) -> str:
+        """View directory contents or file contents."""
+        path = input_data.get("path", "/memories")
+        view_range = input_data.get("view_range")
+
+        resolved = self._resolve_native_path(path, user_id)
+
+        if not resolved.exists():
+            return f"The path {path} does not exist. Please provide a valid path."
+
+        if resolved.is_dir():
+            # List directory contents
+            lines = [
+                f"Here're the files and directories up to 2 levels deep in {path}, "
+                "excluding hidden items and node_modules:"
+            ]
+
+            def get_size(p: Path) -> str:
+                if p.is_file():
+                    size = p.stat().st_size
+                    if size < 1024:
+                        return f"{size}B"
+                    elif size < 1024 * 1024:
+                        return f"{size / 1024:.1f}K"
+                    else:
+                        return f"{size / (1024 * 1024):.1f}M"
+                return "4.0K"  # Default for directories
+
+            def list_recursive(p: Path, rel_path: str, depth: int) -> None:
+                if depth > 2:
+                    return
+                if p.name.startswith(".") or p.name == "node_modules":
+                    return
+
+                lines.append(f"{get_size(p)}\t{rel_path}")
+
+                if p.is_dir() and depth < 2:
+                    try:
+                        for child in sorted(p.iterdir()):
+                            child_rel = (
+                                f"{rel_path}/{child.name}"
+                                if rel_path != path
+                                else f"{path}/{child.name}"
+                            )
+                            list_recursive(child, child_rel, depth + 1)
+                    except PermissionError:
+                        pass
+
+            list_recursive(resolved, path, 0)
+            return "\n".join(lines)
+
+        else:
+            # Read file contents with line numbers
+            try:
+                content = resolved.read_text(encoding="utf-8")
+            except UnicodeDecodeError:
+                content = resolved.read_text(encoding="latin-1")
+
+            lines_content = content.split("\n")
+
+            if len(lines_content) > 999999:
+                return f"File {path} exceeds maximum line limit of 999,999 lines."
+
+            # Apply view_range if specified
+            start_line = 1
+            end_line = len(lines_content)
+            if view_range and len(view_range) >= 2:
+                start_line = max(1, view_range[0])
+                end_line = min(len(lines_content), view_range[1])
+
+            result_lines = [f"Here's the content of {path} with line numbers:"]
+            for i, line in enumerate(lines_content[start_line - 1 : end_line], start=start_line):
+                result_lines.append(f"{i:6d}\t{line}")
+
+            return "\n".join(result_lines)
+
+    def _native_create(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Create a new file."""
+        path = input_data.get("path", "")
+        file_text = input_data.get("file_text", "")
+
+        if not path:
+            return "Error: path is required"
+
+        resolved = self._resolve_native_path(path, user_id)
+
+        if resolved.exists():
+            return f"Error: File {path} already exists"
+
+        # Create parent directories if needed
+        resolved.parent.mkdir(parents=True, exist_ok=True)
+
+        resolved.write_text(file_text, encoding="utf-8")
+        logger.info(f"Memory: Native create: {path} for user {user_id}")
+
+        return f"File created successfully at: {path}"
+
+    def _native_str_replace(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Replace text in a file."""
+        path = input_data.get("path", "")
+        old_str = input_data.get("old_str", "")
+        new_str = input_data.get("new_str", "")
+
+        if not path:
+            return "Error: path is required"
+        if not old_str:
+            return "Error: old_str is required"
+
+        resolved = self._resolve_native_path(path, user_id)
+
+        if not resolved.exists():
+            return f"Error: The path {path} does not exist. Please provide a valid path."
+
+        if resolved.is_dir():
+            return f"Error: The path {path} does not exist. Please provide a valid path."
+
+        content = resolved.read_text(encoding="utf-8")
+
+        # Check for occurrences
+        occurrences = content.count(old_str)
+        if occurrences == 0:
+            return f"No replacement was performed, old_str `{old_str}` did not appear verbatim in {path}."
+        if occurrences > 1:
+            # Find line numbers
+            lines = content.split("\n")
+            found_lines = []
+            for i, line in enumerate(lines, 1):
+                if old_str in line:
+                    found_lines.append(str(i))
+            return (
+                f"No replacement was performed. Multiple occurrences of old_str `{old_str}` "
+                f"in lines: {', '.join(found_lines)}. Please ensure it is unique"
+            )
+
+        # Perform replacement
+        new_content = content.replace(old_str, new_str, 1)
+        resolved.write_text(new_content, encoding="utf-8")
+
+        # Show snippet around the change
+        lines = new_content.split("\n")
+        for i, line in enumerate(lines):
+            if new_str in line:
+                start = max(0, i - 2)
+                end = min(len(lines), i + 3)
+                snippet_lines = ["The memory file has been edited."]
+                for j in range(start, end):
+                    snippet_lines.append(f"{j + 1:6d}\t{lines[j]}")
+                return "\n".join(snippet_lines)
+
+        return "The memory file has been edited."
+
+    def _native_insert(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Insert text at a specific line."""
+        path = input_data.get("path", "")
+        insert_line = input_data.get("insert_line", 0)
+        insert_text = input_data.get("insert_text", "")
+
+        if not path:
+            return "Error: path is required"
+
+        resolved = self._resolve_native_path(path, user_id)
+
+        if not resolved.exists():
+            return f"Error: The path {path} does not exist"
+
+        if resolved.is_dir():
+            return f"Error: The path {path} does not exist"
+
+        content = resolved.read_text(encoding="utf-8")
+        lines = content.split("\n")
+        n_lines = len(lines)
+
+        if insert_line < 0 or insert_line > n_lines:
+            return (
+                f"Error: Invalid `insert_line` parameter: {insert_line}. "
+                f"It should be within the range of lines of the file: [0, {n_lines}]"
+            )
+
+        # Insert at specified line
+        lines.insert(insert_line, insert_text.rstrip("\n"))
+
+        resolved.write_text("\n".join(lines), encoding="utf-8")
+
+        return f"The file {path} has been edited."
+
+    def _native_delete_file(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Delete a file or directory."""
+        path = input_data.get("path", "")
+
+        if not path:
+            return "Error: path is required"
+
+        resolved = self._resolve_native_path(path, user_id)
+
+        if not resolved.exists():
+            return f"Error: The path {path} does not exist"
+
+        import shutil
+
+        if resolved.is_dir():
+            shutil.rmtree(resolved)
+        else:
+            resolved.unlink()
+
+        logger.info(f"Memory: Native delete: {path} for user {user_id}")
+        return f"Successfully deleted {path}"
+
+    def _native_rename(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Rename or move a file/directory."""
+        old_path = input_data.get("old_path", "")
+        new_path = input_data.get("new_path", "")
+
+        if not old_path:
+            return "Error: old_path is required"
+        if not new_path:
+            return "Error: new_path is required"
+
+        resolved_old = self._resolve_native_path(old_path, user_id)
+        resolved_new = self._resolve_native_path(new_path, user_id)
+
+        if not resolved_old.exists():
+            return f"Error: The path {old_path} does not exist"
+
+        if resolved_new.exists():
+            return f"Error: The destination {new_path} already exists"
+
+        # Create parent directory if needed
+        resolved_new.parent.mkdir(parents=True, exist_ok=True)
+
+        resolved_old.rename(resolved_new)
+
+        logger.info(f"Memory: Native rename: {old_path} -> {new_path} for user {user_id}")
+        return f"Successfully renamed {old_path} to {new_path}"
+
+    # =========================================================================
+    # Semantic Translation Methods (Native Tool → Vector Store)
+    # =========================================================================
+
+    async def _native_view_semantic(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Handle VIEW command with semantic search capabilities.
+
+        Path patterns:
+        - /memories              → Overview + search instructions
+        - /memories/search/X     → Semantic search for X
+        - /memories/recent       → Recent memories (last 10)
+        - /memories/all          → List all memories (paginated)
+        - /memories/<topic>      → Search by topic/path
+        """
+        path = input_data.get("path", "/memories")
+
+        # Normalize path
+        if path.startswith("/memories"):
+            subpath = path[len("/memories") :].lstrip("/")
+        else:
+            subpath = path.lstrip("/")
+
+        # CASE 1: /memories/search/<query> → Semantic search
+        if subpath.startswith("search/"):
+            query = subpath[len("search/") :]
+            if not query:
+                return "Error: Please provide a search query. Example: view /memories/search/food preferences"
+            return await self._semantic_search(query, user_id)
+
+        # CASE 2: /memories/recent → Recent memories
+        if subpath == "recent":
+            return await self._get_recent_memories(user_id, limit=10)
+
+        # CASE 3: /memories/all → List all (paginated)
+        if subpath == "all":
+            return await self._list_all_memories(user_id, limit=20)
+
+        # CASE 4: /memories (root) → Overview with instructions
+        if not subpath or subpath == "":
+            return await self._get_memory_overview(user_id)
+
+        # CASE 5: /memories/<something> → Search by topic
+        # Treat the path as a search query
+        return await self._semantic_search(subpath.replace("/", " ").replace("_", " "), user_id)
+
+    async def _semantic_search(self, query: str, user_id: str, top_k: int = 5) -> str:
+        """Perform semantic search and format results."""
+        if not self._backend:
+            return "Error: Memory backend not initialized"
+
+        try:
+            results = await self._backend.search_memories(
+                query=query,
+                user_id=user_id,
+                top_k=top_k,
+                include_related=True,
+            )
+
+            if not results:
+                return f"No memories found matching '{query}'.\n\nTip: Try a broader search term, or use 'view /memories/recent' to see recent memories."
+
+            lines = [f"Found {len(results)} memories matching '{query}':\n"]
+            for i, r in enumerate(results, 1):
+                score_pct = int(r.score * 100)
+                content_preview = r.memory.content[:200]
+                if len(r.memory.content) > 200:
+                    content_preview += "..."
+
+                lines.append(f"{i:6d}\t[{score_pct}% match] {content_preview}")
+
+                # Show related entities if available
+                if hasattr(r, "related_entities") and r.related_entities:
+                    entities = ", ".join(r.related_entities[:3])
+                    lines.append(f"      \t   Related: {entities}")
+                lines.append("")
+
+            return "\n".join(lines)
+
+        except Exception as e:
+            logger.error(f"Memory: Semantic search failed: {e}")
+            return f"Error searching memories: {e}"
+
+    async def _get_recent_memories(self, user_id: str, limit: int = 10) -> str:
+        """Get most recent memories."""
+        if not self._backend:
+            return "Error: Memory backend not initialized"
+
+        try:
+            # Use a generic query to get recent items
+            # Most backends will return by recency when query is broad
+            results = await self._backend.search_memories(
+                query="recent memories",
+                user_id=user_id,
+                top_k=limit,
+            )
+
+            if not results:
+                return "No memories stored yet.\n\nTo save a memory, use: create /memories/<topic>.txt with your content"
+
+            lines = ["Recent memories:\n"]
+            for i, r in enumerate(results, 1):
+                content_preview = r.memory.content[:150]
+                if len(r.memory.content) > 150:
+                    content_preview += "..."
+                # Format timestamp if available
+                timestamp = ""
+                if hasattr(r.memory, "created_at") and r.memory.created_at:
+                    timestamp = f" ({r.memory.created_at})"
+                lines.append(f"{i:6d}\t{content_preview}{timestamp}")
+            lines.append("")
+
+            return "\n".join(lines)
+
+        except Exception as e:
+            logger.error(f"Memory: Get recent failed: {e}")
+            return f"Error getting recent memories: {e}"
+
+    async def _list_all_memories(self, user_id: str, limit: int = 20) -> str:
+        """List all memories (paginated)."""
+        if not self._backend:
+            return "Error: Memory backend not initialized"
+
+        try:
+            # Get all memories with a broad search
+            results = await self._backend.search_memories(
+                query="*",  # Broad query
+                user_id=user_id,
+                top_k=limit,
+            )
+
+            if not results:
+                return "No memories stored yet."
+
+            lines = [f"Showing up to {limit} memories:\n"]
+            for i, r in enumerate(results, 1):
+                content_preview = r.memory.content[:100]
+                if len(r.memory.content) > 100:
+                    content_preview += "..."
+                lines.append(f"{i:6d}\t{content_preview}")
+
+            if len(results) >= limit:
+                lines.append(f"\n(Showing first {limit}. Use search to find specific memories.)")
+
+            return "\n".join(lines)
+
+        except Exception as e:
+            logger.error(f"Memory: List all failed: {e}")
+            return f"Error listing memories: {e}"
+
+    async def _get_memory_overview(self, user_id: str) -> str:
+        """Get memory directory overview with search instructions."""
+        if not self._backend:
+            return "Error: Memory backend not initialized"
+
+        try:
+            # Get count of memories
+            results = await self._backend.search_memories(
+                query="*",
+                user_id=user_id,
+                top_k=100,  # Just to get a count
+            )
+            count = len(results) if results else 0
+
+            # Get a few recent as preview
+            preview_lines = []
+            if results:
+                for r in results[:3]:
+                    preview = r.memory.content[:60]
+                    if len(r.memory.content) > 60:
+                        preview += "..."
+                    preview_lines.append(f"  • {preview}")
+
+            overview = f"""Here're the files and directories up to 2 levels deep in /memories:
+4.0K\t/memories
+
+📁 Memory System ({count} memories stored)
+
+To SEARCH memories (semantic):
+  view /memories/search/<your query>
+  Example: view /memories/search/food preferences
+  Example: view /memories/search/work projects
+
+To see RECENT memories:
+  view /memories/recent
+
+To see ALL memories:
+  view /memories/all
+
+To SAVE a new memory:
+  create /memories/<topic>.txt "your content here"
+  Example: create /memories/preferences.txt "User likes pizza"
+"""
+
+            if preview_lines:
+                overview += "\nRecent memories:\n" + "\n".join(preview_lines)
+
+            return overview
+
+        except Exception as e:
+            logger.error(f"Memory: Overview failed: {e}")
+            # Return basic help even on error
+            return """📁 Memory System
+
+To SEARCH memories: view /memories/search/<query>
+To see RECENT: view /memories/recent
+To SAVE: create /memories/<topic>.txt "content"
+"""
+
+    async def _native_create_semantic(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Handle CREATE command - save to semantic vector store."""
+        path = input_data.get("path", "")
+        file_text = input_data.get("file_text", "")
+
+        if not path:
+            return "Error: path is required"
+        if not file_text:
+            return "Error: file_text is required (the memory content)"
+
+        if not self._backend:
+            return "Error: Memory backend not initialized"
+
+        try:
+            # Extract topic from path for metadata
+            topic = (
+                path.replace("/memories/", "")
+                .replace("/", "_")
+                .replace(".txt", "")
+                .replace(".md", "")
+            )
+
+            # Save to our semantic backend
+            memory = await self._backend.save_memory(
+                content=file_text,
+                user_id=user_id,
+                importance=0.5,
+                metadata={"virtual_path": path, "topic": topic},
+            )
+
+            logger.info(f"Memory: Semantic create: {path} -> id={memory.id} for user {user_id}")
+            return f"File created successfully at: {path}"
+
+        except Exception as e:
+            logger.error(f"Memory: Semantic create failed: {e}")
+            return f"Error: {e}"
+
+    async def _native_update_semantic(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Handle STR_REPLACE command - update memory content."""
+        path = input_data.get("path", "")
+        old_str = input_data.get("old_str", "")
+        new_str = input_data.get("new_str", "")
+
+        if not path:
+            return "Error: path is required"
+        if not old_str:
+            return "Error: old_str is required"
+
+        if not self._backend:
+            return "Error: Memory backend not initialized"
+
+        try:
+            # Search for memory containing old_str
+            results = await self._backend.search_memories(
+                query=old_str,
+                user_id=user_id,
+                top_k=5,
+            )
+
+            # Find exact match
+            matching_memory = None
+            for r in results:
+                if old_str in r.memory.content:
+                    matching_memory = r.memory
+                    break
+
+            if not matching_memory:
+                return f"No replacement was performed, old_str `{old_str}` did not appear verbatim in memories."
+
+            # Check for multiple occurrences
+            if matching_memory.content.count(old_str) > 1:
+                return f"No replacement was performed. Multiple occurrences of old_str `{old_str}`. Please ensure it is unique."
+
+            # Perform replacement
+            new_content = matching_memory.content.replace(old_str, new_str, 1)
+
+            # Update via delete + create (or update if backend supports it)
+            if hasattr(self._backend, "update_memory"):
+                await self._backend.update_memory(
+                    memory_id=matching_memory.id,
+                    new_content=new_content,
+                    user_id=user_id,
+                )
+            else:
+                await self._backend.delete_memory(matching_memory.id)
+                await self._backend.save_memory(
+                    content=new_content,
+                    user_id=user_id,
+                    importance=0.5,
+                )
+
+            # Show snippet around the change
+            lines = new_content.split("\n")
+            snippet = "\n".join(f"{i + 1:6d}\t{line}" for i, line in enumerate(lines[:5]))
+
+            logger.info(f"Memory: Semantic update for user {user_id}")
+            return f"The memory file has been edited.\n{snippet}"
+
+        except Exception as e:
+            logger.error(f"Memory: Semantic update failed: {e}")
+            return f"Error: {e}"
+
+    async def _native_append_semantic(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Handle INSERT command - append to memory or create new."""
+        path = input_data.get("path", "")
+        insert_text = input_data.get("insert_text", "")
+        _insert_line = input_data.get("insert_line", 0)  # Unused in semantic mode
+
+        if not path:
+            return "Error: path is required"
+        if not insert_text:
+            return "Error: insert_text is required"
+
+        if not self._backend:
+            return "Error: Memory backend not initialized"
+
+        try:
+            # For semantic backend, append is just creating a new memory
+            # with the additional context
+            topic = path.replace("/memories/", "").replace("/", "_").replace(".txt", "")
+
+            await self._backend.save_memory(
+                content=insert_text,
+                user_id=user_id,
+                importance=0.5,
+                metadata={"virtual_path": path, "topic": topic, "appended": True},
+            )
+
+            logger.info(f"Memory: Semantic append: {path} for user {user_id}")
+            return f"The file {path} has been edited."
+
+        except Exception as e:
+            logger.error(f"Memory: Semantic append failed: {e}")
+            return f"Error: {e}"
+
+    async def _native_delete_semantic(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Handle DELETE command - remove from vector store."""
+        path = input_data.get("path", "")
+
+        if not path:
+            return "Error: path is required"
+
+        if not self._backend:
+            return "Error: Memory backend not initialized"
+
+        try:
+            # Search for memories with this path
+            topic = (
+                path.replace("/memories/", "")
+                .replace("/", " ")
+                .replace("_", " ")
+                .replace(".txt", "")
+            )
+
+            results = await self._backend.search_memories(
+                query=topic,
+                user_id=user_id,
+                top_k=10,
+            )
+
+            if not results:
+                return f"Error: The path {path} does not exist"
+
+            # Delete matching memories
+            deleted_count = 0
+            for r in results:
+                # Check if metadata matches path
+                metadata = getattr(r.memory, "metadata", {}) or {}
+                if metadata.get("virtual_path") == path or r.score > 0.8:
+                    await self._backend.delete_memory(r.memory.id)
+                    deleted_count += 1
+
+            if deleted_count == 0:
+                return f"Error: The path {path} does not exist"
+
+            logger.info(
+                f"Memory: Semantic delete: {path} ({deleted_count} memories) for user {user_id}"
+            )
+            return f"Successfully deleted {path}"
+
+        except Exception as e:
+            logger.error(f"Memory: Semantic delete failed: {e}")
+            return f"Error: {e}"
+
+    async def _native_rename_semantic(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Handle RENAME command - update memory path/topic."""
+        old_path = input_data.get("old_path", "")
+        new_path = input_data.get("new_path", "")
+
+        if not old_path:
+            return "Error: old_path is required"
+        if not new_path:
+            return "Error: new_path is required"
+
+        if not self._backend:
+            return "Error: Memory backend not initialized"
+
+        try:
+            # Search for memories with old path
+            old_topic = (
+                old_path.replace("/memories/", "")
+                .replace("/", " ")
+                .replace("_", " ")
+                .replace(".txt", "")
+            )
+
+            results = await self._backend.search_memories(
+                query=old_topic,
+                user_id=user_id,
+                top_k=10,
+            )
+
+            if not results:
+                return f"Error: The path {old_path} does not exist"
+
+            # Update metadata for matching memories (re-save with new path)
+            new_topic = new_path.replace("/memories/", "").replace("/", "_").replace(".txt", "")
+            renamed_count = 0
+
+            for r in results:
+                metadata = getattr(r.memory, "metadata", {}) or {}
+                if metadata.get("virtual_path") == old_path or r.score > 0.8:
+                    # Delete old and create with new path
+                    await self._backend.delete_memory(r.memory.id)
+                    await self._backend.save_memory(
+                        content=r.memory.content,
+                        user_id=user_id,
+                        importance=getattr(r.memory, "importance", 0.5),
+                        metadata={"virtual_path": new_path, "topic": new_topic},
+                    )
+                    renamed_count += 1
+
+            if renamed_count == 0:
+                return f"Error: The path {old_path} does not exist"
+
+            logger.info(f"Memory: Semantic rename: {old_path} -> {new_path} for user {user_id}")
+            return f"Successfully renamed {old_path} to {new_path}"
+
+        except Exception as e:
+            logger.error(f"Memory: Semantic rename failed: {e}")
+            return f"Error: {e}"
+
     async def close(self) -> None:
         """Close the memory backend."""
         if self._backend and hasattr(self._backend, "close"):

headroom/proxy/server.py

@@ -296,6 +296,7 @@ class ProxyConfig:
     memory_backend: Literal["local", "qdrant-neo4j"] = "local"  # Backend type
     memory_db_path: str = "headroom_memory.db"  # Path for local backend
     memory_inject_tools: bool = True  # Auto-inject memory tools
+    memory_use_native_tool: bool = False  # Use Anthropic's native memory_20250818 tool
     memory_inject_context: bool = True  # Inject searched memories into context
     memory_top_k: int = 10  # Number of memories to inject
     memory_min_similarity: float = 0.3  # Minimum similarity threshold
@@ -1082,6 +1083,7 @@ class HeadroomProxy:
                 backend=config.memory_backend,
                 db_path=config.memory_db_path,
                 inject_tools=config.memory_inject_tools,
+                use_native_tool=config.memory_use_native_tool,
                 inject_context=config.memory_inject_context,
                 top_k=config.memory_top_k,
                 min_similarity=config.memory_min_similarity,
@@ -1623,10 +1625,27 @@ class HeadroomProxy:
                 tools, mem_tools_injected = self.memory_handler.inject_tools(tools, "anthropic")
                 if mem_tools_injected:
                     tool_names = [
-                        t.get("name") for t in tools if t.get("name", "").startswith("memory_")
+                        t.get("name") or t.get("type", "")
+                        for t in tools
+                        if t.get("name", "").startswith("memory")
+                        or t.get("type", "").startswith("memory")
                     ]
                     logger.info(f"[{request_id}] Memory: Injected tools: {tool_names}")
 
+                    # Add beta headers for native memory tool
+                    beta_headers = self.memory_handler.get_beta_headers()
+                    if beta_headers:
+                        for key, value in beta_headers.items():
+                            # Merge with existing beta header if present
+                            existing = headers.get(key, "")
+                            if existing and value not in existing:
+                                headers[key] = f"{existing},{value}"
+                            else:
+                                headers[key] = value
+                            logger.info(
+                                f"[{request_id}] Memory: Added beta header: {key}={headers[key]}"
+                            )
+
         # Update body
         body["messages"] = optimized_messages
         if tools is not None:

tests/test_memory/test_core_operations.py

@@ -78,6 +78,8 @@ async def memory_system(temp_db_path):
     config = MemoryConfig(db_path=str(temp_db_path))
     system = await HierarchicalMemory.create(config)
     yield system
+    # Properly close to release httpx clients
+    await system.close()
 
 
 # =============================================================================

tests/test_parser.py

@@ -569,17 +569,23 @@ class TestFindToolUnits:
             # OpenAI format
             {
                 "role": "assistant",
-                "tool_calls": [{"id": "call_1", "function": {"name": "openai_tool", "arguments": "{}"}}],
+                "tool_calls": [
+                    {"id": "call_1", "function": {"name": "openai_tool", "arguments": "{}"}}
+                ],
             },
             {"role": "tool", "tool_call_id": "call_1", "content": "openai result"},
             # Anthropic format
             {
                 "role": "assistant",
-                "content": [{"type": "tool_use", "id": "toolu_2", "name": "anthropic_tool", "input": {}}],
+                "content": [
+                    {"type": "tool_use", "id": "toolu_2", "name": "anthropic_tool", "input": {}}
+                ],
             },
             {
                 "role": "user",
-                "content": [{"type": "tool_result", "tool_use_id": "toolu_2", "content": "anthropic result"}],
+                "content": [
+                    {"type": "tool_result", "tool_use_id": "toolu_2", "content": "anthropic result"}
+                ],
             },
         ]
         units = find_tool_units(messages)