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/memory_tool_adapter.py

@@ -0,0 +1,1273 @@
+"""Memory tool adapter for multi-provider support.
+
+This module provides a unified adapter for memory tools across different LLM providers.
+It handles provider detection, tool injection, and tool call execution with appropriate
+format conversions for each provider.
+
+Supported providers:
+- Anthropic: Native memory_20250818 tool and custom tools
+- OpenAI: Function calling format
+- Gemini: Function calling format
+- Generic: Fallback for unknown providers
+
+Usage:
+    config = MemoryToolAdapterConfig(enabled=True)
+    adapter = MemoryToolAdapter(config)
+
+    # Detect provider from request
+    provider = adapter.detect_provider(request_headers, model_name)
+
+    # Inject tools
+    tools, beta_headers = adapter.inject_tools(existing_tools, provider)
+
+    # Handle tool calls in response
+    if adapter.has_memory_tool_calls(response, provider):
+        results = await adapter.handle_tool_calls(response, user_id, provider)
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any, Literal
+
+if TYPE_CHECKING:
+    from headroom.memory.backends.local import LocalBackend
+
+logger = logging.getLogger(__name__)
+
+# =============================================================================
+# Provider Types
+# =============================================================================
+
+Provider = Literal["anthropic", "openai", "gemini", "generic"]
+
+# =============================================================================
+# Tool Names
+# =============================================================================
+
+# Custom memory tool names (Headroom's tools)
+MEMORY_TOOL_NAMES = {"memory_save", "memory_search", "memory_update", "memory_delete"}
+
+# Anthropic's native memory tool
+NATIVE_MEMORY_TOOL_NAME = "memory"
+NATIVE_MEMORY_TOOL_TYPE = "memory_20250818"
+
+# Beta header for Anthropic's native memory tool
+ANTHROPIC_BETA_HEADER = "context-management-2025-06-27"
+
+# =============================================================================
+# Tool Schemas - Anthropic Native Tool
+# =============================================================================
+
+ANTHROPIC_NATIVE_TOOL: dict[str, Any] = {
+    "type": NATIVE_MEMORY_TOOL_TYPE,
+    "name": NATIVE_MEMORY_TOOL_NAME,
+}
+
+# =============================================================================
+# Tool Schemas - Anthropic Custom Tools
+# =============================================================================
+
+ANTHROPIC_CUSTOM_TOOLS: list[dict[str, Any]] = [
+    {
+        "name": "memory_save",
+        "description": """Save important information to long-term memory for future reference.
+
+Use this tool when you encounter information that should be remembered across conversations:
+- User preferences (e.g., "prefers Python over JavaScript")
+- Personal facts (e.g., "works at Acme Corp", "has a dog named Max")
+- Project context (e.g., "working on a CLI tool", "using React 18")
+- Decisions made (e.g., "chose PostgreSQL for the database")
+- Important relationships (e.g., "Alice is Bob's manager")
+
+DO NOT save: transient info, sensitive data (passwords, keys), redundant info.""",
+        "input_schema": {
+            "type": "object",
+            "properties": {
+                "content": {
+                    "type": "string",
+                    "description": "The information to remember. Be specific and self-contained.",
+                },
+                "importance": {
+                    "type": "number",
+                    "minimum": 0.0,
+                    "maximum": 1.0,
+                    "description": "Importance score from 0.0 (low) to 1.0 (critical).",
+                },
+                "facts": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "Pre-extracted discrete facts for efficient storage.",
+                },
+                "entities": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "Entity names referenced in this memory.",
+                },
+                "extracted_entities": {
+                    "type": "array",
+                    "items": {
+                        "type": "object",
+                        "properties": {
+                            "entity": {"type": "string"},
+                            "entity_type": {"type": "string"},
+                        },
+                        "required": ["entity", "entity_type"],
+                    },
+                    "description": "Pre-extracted entities with types.",
+                },
+                "extracted_relationships": {
+                    "type": "array",
+                    "items": {
+                        "type": "object",
+                        "properties": {
+                            "source": {"type": "string"},
+                            "relationship": {"type": "string"},
+                            "destination": {"type": "string"},
+                        },
+                        "required": ["source", "relationship", "destination"],
+                    },
+                    "description": "Pre-extracted relationships for graph storage.",
+                },
+            },
+            "required": ["content", "importance"],
+        },
+    },
+    {
+        "name": "memory_search",
+        "description": """Search stored memories to recall relevant information.
+
+Use this tool to retrieve previously saved information before responding to questions about:
+- User preferences or past decisions
+- Personal or professional context
+- Previously discussed topics or projects
+- Relationships between people, systems, or concepts
+
+Search BEFORE saving to avoid duplicates.""",
+        "input_schema": {
+            "type": "object",
+            "properties": {
+                "query": {
+                    "type": "string",
+                    "description": "Natural language search query.",
+                },
+                "entities": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "Filter to memories mentioning these entities.",
+                },
+                "include_related": {
+                    "type": "boolean",
+                    "description": "Also retrieve connected memories.",
+                },
+                "top_k": {
+                    "type": "integer",
+                    "minimum": 1,
+                    "maximum": 50,
+                    "description": "Maximum number of memories to retrieve (default 10).",
+                },
+            },
+            "required": ["query"],
+        },
+    },
+    {
+        "name": "memory_update",
+        "description": """Update an existing memory with corrected or evolved information.
+
+Use when:
+- User provides a correction to stored information
+- Information has changed over time
+- Adding detail or clarification to an existing memory""",
+        "input_schema": {
+            "type": "object",
+            "properties": {
+                "memory_id": {
+                    "type": "string",
+                    "description": "The unique ID of the memory to update.",
+                },
+                "new_content": {
+                    "type": "string",
+                    "description": "The updated content.",
+                },
+                "reason": {
+                    "type": "string",
+                    "description": "Explanation for the update.",
+                },
+            },
+            "required": ["memory_id", "new_content"],
+        },
+    },
+    {
+        "name": "memory_delete",
+        "description": """Delete a memory that is no longer relevant or was stored in error.
+
+Use when:
+- User explicitly asks to forget something
+- Information is outdated and no longer applicable
+- A memory was saved in error""",
+        "input_schema": {
+            "type": "object",
+            "properties": {
+                "memory_id": {
+                    "type": "string",
+                    "description": "The unique ID of the memory to delete.",
+                },
+                "reason": {
+                    "type": "string",
+                    "description": "Explanation for the deletion.",
+                },
+            },
+            "required": ["memory_id"],
+        },
+    },
+]
+
+# =============================================================================
+# Tool Schemas - OpenAI Function Calling Format
+# =============================================================================
+
+OPENAI_TOOLS: list[dict[str, Any]] = [
+    {
+        "type": "function",
+        "function": {
+            "name": "memory_save",
+            "description": """Save important information to long-term memory for future reference.
+
+Use this tool when you encounter information that should be remembered across conversations:
+- User preferences, personal facts, project context, decisions, relationships
+
+DO NOT save: transient info, sensitive data (passwords, keys), redundant info.""",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "content": {
+                        "type": "string",
+                        "description": "The information to remember. Be specific and self-contained.",
+                    },
+                    "importance": {
+                        "type": "number",
+                        "minimum": 0.0,
+                        "maximum": 1.0,
+                        "description": "Importance score from 0.0 (low) to 1.0 (critical).",
+                    },
+                    "facts": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": "Pre-extracted discrete facts.",
+                    },
+                    "entities": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": "Entity names referenced in this memory.",
+                    },
+                    "extracted_entities": {
+                        "type": "array",
+                        "items": {
+                            "type": "object",
+                            "properties": {
+                                "entity": {"type": "string"},
+                                "entity_type": {"type": "string"},
+                            },
+                            "required": ["entity", "entity_type"],
+                        },
+                        "description": "Pre-extracted entities with types.",
+                    },
+                    "extracted_relationships": {
+                        "type": "array",
+                        "items": {
+                            "type": "object",
+                            "properties": {
+                                "source": {"type": "string"},
+                                "relationship": {"type": "string"},
+                                "destination": {"type": "string"},
+                            },
+                            "required": ["source", "relationship", "destination"],
+                        },
+                        "description": "Pre-extracted relationships.",
+                    },
+                },
+                "required": ["content", "importance"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "memory_search",
+            "description": "Search stored memories to recall relevant information.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "query": {
+                        "type": "string",
+                        "description": "Natural language search query.",
+                    },
+                    "entities": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": "Filter to memories mentioning these entities.",
+                    },
+                    "include_related": {
+                        "type": "boolean",
+                        "description": "Also retrieve connected memories.",
+                    },
+                    "top_k": {
+                        "type": "integer",
+                        "minimum": 1,
+                        "maximum": 50,
+                        "description": "Maximum number of memories to retrieve.",
+                    },
+                },
+                "required": ["query"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "memory_update",
+            "description": "Update an existing memory with corrected or evolved information.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "memory_id": {
+                        "type": "string",
+                        "description": "The unique ID of the memory to update.",
+                    },
+                    "new_content": {
+                        "type": "string",
+                        "description": "The updated content.",
+                    },
+                    "reason": {
+                        "type": "string",
+                        "description": "Explanation for the update.",
+                    },
+                },
+                "required": ["memory_id", "new_content"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "memory_delete",
+            "description": "Delete a memory that is no longer relevant.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "memory_id": {
+                        "type": "string",
+                        "description": "The unique ID of the memory to delete.",
+                    },
+                    "reason": {
+                        "type": "string",
+                        "description": "Explanation for the deletion.",
+                    },
+                },
+                "required": ["memory_id"],
+            },
+        },
+    },
+]
+
+# =============================================================================
+# Tool Schemas - Gemini Function Calling Format
+# =============================================================================
+
+# Gemini uses a similar format to OpenAI but with slight differences
+GEMINI_TOOLS: list[dict[str, Any]] = [
+    {
+        "name": "memory_save",
+        "description": """Save important information to long-term memory for future reference.
+
+Use this tool when you encounter information that should be remembered across conversations:
+- User preferences, personal facts, project context, decisions, relationships
+
+DO NOT save: transient info, sensitive data (passwords, keys), redundant info.""",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "content": {
+                    "type": "string",
+                    "description": "The information to remember. Be specific and self-contained.",
+                },
+                "importance": {
+                    "type": "number",
+                    "description": "Importance score from 0.0 (low) to 1.0 (critical).",
+                },
+                "facts": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "Pre-extracted discrete facts.",
+                },
+                "entities": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "Entity names referenced in this memory.",
+                },
+            },
+            "required": ["content", "importance"],
+        },
+    },
+    {
+        "name": "memory_search",
+        "description": "Search stored memories to recall relevant information.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "query": {
+                    "type": "string",
+                    "description": "Natural language search query.",
+                },
+                "entities": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "Filter to memories mentioning these entities.",
+                },
+                "include_related": {
+                    "type": "boolean",
+                    "description": "Also retrieve connected memories.",
+                },
+                "top_k": {
+                    "type": "integer",
+                    "description": "Maximum number of memories to retrieve.",
+                },
+            },
+            "required": ["query"],
+        },
+    },
+    {
+        "name": "memory_update",
+        "description": "Update an existing memory with corrected or evolved information.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "memory_id": {
+                    "type": "string",
+                    "description": "The unique ID of the memory to update.",
+                },
+                "new_content": {
+                    "type": "string",
+                    "description": "The updated content.",
+                },
+                "reason": {
+                    "type": "string",
+                    "description": "Explanation for the update.",
+                },
+            },
+            "required": ["memory_id", "new_content"],
+        },
+    },
+    {
+        "name": "memory_delete",
+        "description": "Delete a memory that is no longer relevant.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "memory_id": {
+                    "type": "string",
+                    "description": "The unique ID of the memory to delete.",
+                },
+                "reason": {
+                    "type": "string",
+                    "description": "Explanation for the deletion.",
+                },
+            },
+            "required": ["memory_id"],
+        },
+    },
+]
+
+
+# =============================================================================
+# Configuration
+# =============================================================================
+
+
+@dataclass
+class MemoryToolAdapterConfig:
+    """Configuration for the memory tool adapter.
+
+    Attributes:
+        enabled: Whether memory features are enabled.
+        use_native_tool: Use Anthropic's native memory_20250818 tool (Anthropic only).
+        inject_tools: Whether to inject memory tools into requests.
+        inject_context: Whether to inject memory context into requests.
+        db_path: Path to the local memory database.
+        top_k: Number of memories to retrieve in searches.
+        min_similarity: Minimum similarity score for memory retrieval.
+    """
+
+    enabled: bool = False
+    use_native_tool: bool = True  # Default to native for Anthropic (subscription-safe)
+    inject_tools: bool = True
+    inject_context: bool = True
+    db_path: str = "headroom_memory.db"
+    top_k: int = 10
+    min_similarity: float = 0.3
+
+
+# =============================================================================
+# Memory Tool Adapter
+# =============================================================================
+
+
+class MemoryToolAdapter:
+    """Adapter for memory tools across different LLM providers.
+
+    This adapter provides a unified interface for:
+    1. Detecting the LLM provider from requests
+    2. Injecting memory tools in provider-specific formats
+    3. Providing required beta headers
+    4. Detecting memory tool calls in responses
+    5. Handling tool calls with the semantic backend
+
+    Example:
+        adapter = MemoryToolAdapter(config)
+        provider = adapter.detect_provider(headers, model)
+        tools, headers = adapter.inject_tools(existing_tools, provider)
+
+        # Later, when processing response
+        if adapter.has_memory_tool_calls(response, provider):
+            results = await adapter.handle_tool_calls(response, user_id, provider)
+    """
+
+    def __init__(self, config: MemoryToolAdapterConfig) -> None:
+        """Initialize the adapter.
+
+        Args:
+            config: Configuration for the adapter.
+        """
+        self.config = config
+        self._backend: LocalBackend | Any = None
+        self._initialized = False
+
+    async def _ensure_initialized(self) -> None:
+        """Lazy initialization of the semantic backend.
+
+        Imports and initializes the LocalBackend from memory_handler
+        to provide semantic search and storage capabilities.
+        """
+        if self._initialized:
+            return
+
+        if not self.config.enabled:
+            return
+
+        from headroom.memory.backends.local import LocalBackend, LocalBackendConfig
+
+        backend_config = LocalBackendConfig(db_path=self.config.db_path)
+        self._backend = LocalBackend(backend_config)
+        await self._backend._ensure_initialized()
+
+        self._initialized = True
+        logger.info(f"MemoryToolAdapter: Initialized backend at {self.config.db_path}")
+
+    def detect_provider(
+        self,
+        request_headers: dict[str, str] | None = None,
+        model_name: str | None = None,
+    ) -> Provider:
+        """Detect the LLM provider from request headers and model name.
+
+        Detection priority:
+        1. Explicit headers (x-api-key for Anthropic, authorization for OpenAI)
+        2. Model name patterns (claude-*, gpt-*, gemini-*)
+        3. Fallback to generic
+
+        Args:
+            request_headers: HTTP headers from the request (optional).
+            model_name: Name of the model being used (optional).
+
+        Returns:
+            The detected provider.
+        """
+        headers = request_headers or {}
+        model = (model_name or "").lower()
+
+        # Check headers for provider hints
+        if "x-api-key" in headers or "anthropic-version" in headers:
+            return "anthropic"
+
+        if headers.get("authorization", "").startswith("Bearer sk-"):
+            # OpenAI uses sk-* API keys
+            return "openai"
+
+        # Check model name patterns
+        if model.startswith("claude"):
+            return "anthropic"
+
+        if model.startswith("gpt") or model.startswith("o1") or model.startswith("o3"):
+            return "openai"
+
+        if model.startswith("gemini") or "gemma" in model:
+            return "gemini"
+
+        # Fallback to generic
+        return "generic"
+
+    def inject_tools(
+        self,
+        tools: list[dict[str, Any]] | None,
+        provider: Provider,
+    ) -> tuple[list[dict[str, Any]], dict[str, str]]:
+        """Inject memory tools into the tools list for the given provider.
+
+        Args:
+            tools: Existing tools list (may be None).
+            provider: The LLM provider to format tools for.
+
+        Returns:
+            Tuple of (updated_tools, beta_headers).
+            beta_headers contains any required headers (e.g., anthropic-beta).
+        """
+        if not self.config.inject_tools:
+            return tools or [], {}
+
+        tools = list(tools) if tools else []
+        beta_headers: dict[str, str] = {}
+
+        # Get existing tool names
+        existing_names = self._get_existing_tool_names(tools)
+
+        # Handle Anthropic native tool
+        if provider == "anthropic" and self.config.use_native_tool:
+            if NATIVE_MEMORY_TOOL_NAME not in existing_names:
+                tools.append(ANTHROPIC_NATIVE_TOOL.copy())
+                beta_headers["anthropic-beta"] = ANTHROPIC_BETA_HEADER
+                logger.info("MemoryToolAdapter: Injected native memory tool for Anthropic")
+            return tools, beta_headers
+
+        # Handle custom tools by provider
+        if provider == "anthropic":
+            tools, was_injected = self._inject_anthropic_tools(tools, existing_names)
+        elif provider == "openai":
+            tools, was_injected = self._inject_openai_tools(tools, existing_names)
+        elif provider == "gemini":
+            tools, was_injected = self._inject_gemini_tools(tools, existing_names)
+        else:
+            # Generic fallback uses OpenAI format
+            tools, was_injected = self._inject_openai_tools(tools, existing_names)
+
+        if was_injected:
+            logger.info(f"MemoryToolAdapter: Injected custom tools for {provider}")
+
+        return tools, beta_headers
+
+    def _get_existing_tool_names(self, tools: list[dict[str, Any]]) -> set[str]:
+        """Extract tool names from existing tools list."""
+        names: set[str] = set()
+        for tool in tools:
+            # Anthropic format
+            if "name" in tool:
+                names.add(tool["name"])
+            # OpenAI format
+            if "function" in tool and "name" in tool["function"]:
+                names.add(tool["function"]["name"])
+        return names
+
+    def _inject_anthropic_tools(
+        self,
+        tools: list[dict[str, Any]],
+        existing_names: set[str],
+    ) -> tuple[list[dict[str, Any]], bool]:
+        """Inject Anthropic-formatted custom memory tools."""
+        was_injected = False
+        for memory_tool in ANTHROPIC_CUSTOM_TOOLS:
+            if memory_tool["name"] not in existing_names:
+                tools.append(memory_tool.copy())
+                was_injected = True
+        return tools, was_injected
+
+    def _inject_openai_tools(
+        self,
+        tools: list[dict[str, Any]],
+        existing_names: set[str],
+    ) -> tuple[list[dict[str, Any]], bool]:
+        """Inject OpenAI-formatted memory tools."""
+        was_injected = False
+        for memory_tool in OPENAI_TOOLS:
+            tool_name = memory_tool["function"]["name"]
+            if tool_name not in existing_names:
+                tools.append(memory_tool.copy())
+                was_injected = True
+        return tools, was_injected
+
+    def _inject_gemini_tools(
+        self,
+        tools: list[dict[str, Any]],
+        existing_names: set[str],
+    ) -> tuple[list[dict[str, Any]], bool]:
+        """Inject Gemini-formatted memory tools."""
+        was_injected = False
+        for memory_tool in GEMINI_TOOLS:
+            if memory_tool["name"] not in existing_names:
+                tools.append(memory_tool.copy())
+                was_injected = True
+        return tools, was_injected
+
+    def get_beta_headers(self, provider: Provider) -> dict[str, str]:
+        """Get any required beta headers for the provider.
+
+        Args:
+            provider: The LLM provider.
+
+        Returns:
+            Dict of header name -> value for any required beta headers.
+        """
+        if provider == "anthropic" and self.config.use_native_tool:
+            return {"anthropic-beta": ANTHROPIC_BETA_HEADER}
+        return {}
+
+    def has_memory_tool_calls(
+        self,
+        response: dict[str, Any],
+        provider: Provider,
+    ) -> bool:
+        """Check if the response contains memory tool calls.
+
+        Args:
+            response: The API response from the LLM.
+            provider: The LLM provider.
+
+        Returns:
+            True if response contains memory tool calls.
+        """
+        tool_calls = self._extract_tool_calls(response, provider)
+        for tc in tool_calls:
+            name = self._get_tool_name(tc, provider)
+            if name in MEMORY_TOOL_NAMES or name == NATIVE_MEMORY_TOOL_NAME:
+                return True
+        return False
+
+    def _extract_tool_calls(
+        self,
+        response: dict[str, Any],
+        provider: Provider,
+    ) -> list[dict[str, Any]]:
+        """Extract tool calls from response based on provider format."""
+        if provider == "anthropic":
+            content = response.get("content", [])
+            if isinstance(content, list):
+                return [block for block in content if block.get("type") == "tool_use"]
+            return []
+
+        elif provider == "openai":
+            choices = response.get("choices", [])
+            if choices:
+                message = choices[0].get("message", {})
+                return list(message.get("tool_calls", []) or [])
+            return []
+
+        elif provider == "gemini":
+            # Gemini format: candidates[0].content.parts[*].functionCall
+            candidates = response.get("candidates", [])
+            if candidates:
+                content = candidates[0].get("content", {})
+                parts = content.get("parts", [])
+                return [p for p in parts if "functionCall" in p]
+            return []
+
+        # Generic fallback - try both formats
+        tool_calls = []
+
+        # Try Anthropic format
+        content = response.get("content", [])
+        if isinstance(content, list):
+            tool_calls.extend([block for block in content if block.get("type") == "tool_use"])
+
+        # Try OpenAI format
+        choices = response.get("choices", [])
+        if choices:
+            message = choices[0].get("message", {})
+            tool_calls.extend(list(message.get("tool_calls", []) or []))
+
+        return tool_calls
+
+    def _get_tool_name(self, tool_call: dict[str, Any], provider: Provider) -> str:
+        """Get the tool name from a tool call."""
+        if provider == "anthropic":
+            return str(tool_call.get("name", ""))
+        elif provider == "openai":
+            return str(tool_call.get("function", {}).get("name", ""))
+        elif provider == "gemini":
+            func_call = tool_call.get("functionCall", {})
+            return str(func_call.get("name", ""))
+        else:
+            # Generic - try both
+            return str(tool_call.get("name", "") or tool_call.get("function", {}).get("name", ""))
+
+    def _get_tool_id(self, tool_call: dict[str, Any], provider: Provider) -> str:
+        """Get the tool call ID."""
+        if provider == "anthropic":
+            return str(tool_call.get("id", ""))
+        elif provider == "openai":
+            return str(tool_call.get("id", ""))
+        elif provider == "gemini":
+            # Gemini doesn't use IDs in the same way
+            return str(tool_call.get("functionCall", {}).get("name", ""))
+        else:
+            return str(tool_call.get("id", ""))
+
+    def _get_tool_input(
+        self,
+        tool_call: dict[str, Any],
+        provider: Provider,
+    ) -> dict[str, Any]:
+        """Get the tool input/arguments from a tool call."""
+        if provider == "anthropic":
+            result = tool_call.get("input", {})
+            return dict(result) if isinstance(result, dict) else {}
+        elif provider == "openai":
+            args_str = tool_call.get("function", {}).get("arguments", "{}")
+            try:
+                parsed = json.loads(args_str)
+                return dict(parsed) if isinstance(parsed, dict) else {}
+            except json.JSONDecodeError:
+                return {}
+        elif provider == "gemini":
+            result = tool_call.get("functionCall", {}).get("args", {})
+            return dict(result) if isinstance(result, dict) else {}
+        else:
+            # Generic - try both
+            if "input" in tool_call:
+                result = tool_call["input"]
+                return dict(result) if isinstance(result, dict) else {}
+            args_str = tool_call.get("function", {}).get("arguments", "{}")
+            try:
+                parsed = json.loads(args_str)
+                return dict(parsed) if isinstance(parsed, dict) else {}
+            except json.JSONDecodeError:
+                return {}
+
+    async def handle_tool_calls(
+        self,
+        response: dict[str, Any],
+        user_id: str,
+        provider: Provider,
+    ) -> list[dict[str, Any]]:
+        """Handle memory tool calls and return results in provider format.
+
+        Args:
+            response: The API response containing tool calls.
+            user_id: User identifier for memory operations.
+            provider: The LLM provider.
+
+        Returns:
+            List of tool results in provider-appropriate format.
+        """
+        await self._ensure_initialized()
+
+        tool_calls = self._extract_tool_calls(response, provider)
+        results: list[dict[str, Any]] = []
+
+        for tc in tool_calls:
+            tool_name = self._get_tool_name(tc, provider)
+            tool_id = self._get_tool_id(tc, provider)
+            input_data = self._get_tool_input(tc, provider)
+
+            # Skip non-memory tools
+            if tool_name not in MEMORY_TOOL_NAMES and tool_name != NATIVE_MEMORY_TOOL_NAME:
+                continue
+
+            # Execute the tool
+            if tool_name == NATIVE_MEMORY_TOOL_NAME:
+                result_content = await self._execute_native_tool(input_data, user_id)
+            else:
+                result_content = await self._execute_custom_tool(tool_name, input_data, user_id)
+
+            # Format result for provider
+            result = self._format_tool_result(tool_id, result_content, provider)
+            results.append(result)
+
+            logger.info(f"MemoryToolAdapter: Executed {tool_name} for user {user_id}")
+
+        return results
+
+    def _format_tool_result(
+        self,
+        tool_id: str,
+        content: str,
+        provider: Provider,
+    ) -> dict[str, Any]:
+        """Format a tool result for the given provider."""
+        if provider == "anthropic":
+            return {
+                "type": "tool_result",
+                "tool_use_id": tool_id,
+                "content": content,
+            }
+        elif provider == "openai":
+            return {
+                "role": "tool",
+                "tool_call_id": tool_id,
+                "content": content,
+            }
+        elif provider == "gemini":
+            return {
+                "functionResponse": {
+                    "name": tool_id,
+                    "response": {"result": content},
+                }
+            }
+        else:
+            # Generic uses OpenAI format
+            return {
+                "role": "tool",
+                "tool_call_id": tool_id,
+                "content": content,
+            }
+
+    async def _execute_native_tool(
+        self,
+        input_data: dict[str, Any],
+        user_id: str,
+    ) -> str:
+        """Execute Anthropic's native memory tool.
+
+        This translates native memory commands to our semantic backend:
+        - view: semantic search or list memories
+        - create: save to vector store
+        - str_replace: update memory
+        - delete: remove from vector store
+        """
+        if not self._backend:
+            return "Error: Memory backend not initialized"
+
+        command = input_data.get("command", "")
+
+        try:
+            if command == "view":
+                return await self._native_view(input_data, user_id)
+            elif command == "create":
+                return await self._native_create(input_data, user_id)
+            elif command == "str_replace":
+                return await self._native_update(input_data, user_id)
+            elif command == "delete":
+                return await self._native_delete(input_data, user_id)
+            else:
+                return f"Error: Unknown command '{command}'"
+        except Exception as e:
+            logger.error(f"MemoryToolAdapter: Native tool error: {e}")
+            return f"Error: {e}"
+
+    async def _native_view(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Handle VIEW command - semantic search or list memories."""
+        path = input_data.get("path", "/memories")
+
+        # Normalize path
+        if path.startswith("/memories"):
+            subpath = path[len("/memories") :].lstrip("/")
+        else:
+            subpath = path.lstrip("/")
+
+        # Search pattern: /memories/search/<query>
+        if subpath.startswith("search/"):
+            query = subpath[len("search/") :]
+            if not query:
+                return "Error: Please provide a search query"
+            return await self._semantic_search(query, user_id)
+
+        # Recent: /memories/recent
+        if subpath == "recent":
+            return await self._semantic_search("recent memories", user_id, top_k=10)
+
+        # Root: /memories
+        if not subpath:
+            return await self._get_memory_overview(user_id)
+
+        # Treat path as search topic
+        return await self._semantic_search(
+            subpath.replace("/", " ").replace("_", " "),
+            user_id,
+        )
+
+    async def _native_create(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Handle CREATE command - save to vector store."""
+        path = input_data.get("path", "")
+        file_text = input_data.get("file_text", "")
+
+        if not file_text:
+            return "Error: file_text is required"
+
+        topic = path.replace("/memories/", "").replace("/", "_").replace(".txt", "")
+
+        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"MemoryToolAdapter: Created memory {memory.id} for {user_id}")
+        return f"File created successfully at: {path}"
+
+    async def _native_update(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Handle STR_REPLACE command - update memory content."""
+        old_str = input_data.get("old_str", "")
+        new_str = input_data.get("new_str", "")
+
+        if not old_str:
+            return "Error: old_str is required"
+
+        # 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 "No replacement performed, old_str not found in memories"
+
+        # Perform replacement
+        new_content = matching_memory.content.replace(old_str, new_str, 1)
+
+        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,
+            )
+
+        return "The memory has been edited."
+
+    async def _native_delete(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Handle DELETE command - remove from vector store."""
+        path = input_data.get("path", "")
+        topic = path.replace("/memories/", "").replace("/", " ").replace("_", " ")
+
+        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"
+
+        deleted_count = 0
+        for r in results:
+            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"
+
+        return f"Successfully deleted {path}"
+
+    async def _semantic_search(
+        self,
+        query: str,
+        user_id: str,
+        top_k: int = 5,
+    ) -> str:
+        """Perform semantic search and format results."""
+        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}'"
+
+        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}. [{score_pct}% match] {content_preview}")
+
+        return "\n".join(lines)
+
+    async def _get_memory_overview(self, user_id: str) -> str:
+        """Get memory overview with search instructions."""
+        results = await self._backend.search_memories(
+            query="*",
+            user_id=user_id,
+            top_k=100,
+        )
+        count = len(results) if results else 0
+
+        return f"""Memory System ({count} memories stored)
+
+To SEARCH: view /memories/search/<query>
+To see RECENT: view /memories/recent
+To SAVE: create /memories/<topic>.txt "content"
+"""
+
+    async def _execute_custom_tool(
+        self,
+        tool_name: str,
+        input_data: dict[str, Any],
+        user_id: str,
+    ) -> str:
+        """Execute a custom memory tool."""
+        if not self._backend:
+            return json.dumps({"error": "Memory backend not initialized"})
+
+        try:
+            if tool_name == "memory_save":
+                return await self._execute_save(input_data, user_id)
+            elif tool_name == "memory_search":
+                return await self._execute_search(input_data, user_id)
+            elif tool_name == "memory_update":
+                return await self._execute_update(input_data, user_id)
+            elif tool_name == "memory_delete":
+                return await self._execute_delete(input_data, user_id)
+            else:
+                return json.dumps({"error": f"Unknown tool: {tool_name}"})
+        except Exception as e:
+            logger.error(f"MemoryToolAdapter: Tool {tool_name} failed: {e}")
+            return json.dumps({"status": "error", "error": str(e)})
+
+    async def _execute_save(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Execute memory_save tool."""
+        content = input_data.get("content", "")
+        if not content:
+            return json.dumps({"status": "error", "error": "content is required"})
+
+        importance = input_data.get("importance", 0.5)
+        facts = input_data.get("facts")
+        entities = input_data.get("entities")
+        extracted_entities = input_data.get("extracted_entities")
+        extracted_relationships = input_data.get("extracted_relationships")
+
+        memory = await self._backend.save_memory(
+            content=content,
+            user_id=user_id,
+            importance=importance,
+            facts=facts,
+            entities=entities,
+            extracted_entities=extracted_entities,
+            relationships=extracted_relationships,
+            extracted_relationships=extracted_relationships,
+        )
+
+        return json.dumps(
+            {
+                "status": "saved",
+                "memory_id": memory.id,
+                "content": memory.content[:100] + "..."
+                if len(memory.content) > 100
+                else memory.content,
+            }
+        )
+
+    async def _execute_search(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Execute memory_search tool."""
+        query = input_data.get("query", "")
+        if not query:
+            return json.dumps({"status": "error", "error": "query is required"})
+
+        top_k = input_data.get("top_k", self.config.top_k)
+        include_related = input_data.get("include_related", True)
+        entities_filter = input_data.get("entities")
+
+        results = await self._backend.search_memories(
+            query=query,
+            user_id=user_id,
+            top_k=top_k,
+            include_related=include_related,
+            entities=entities_filter,
+        )
+
+        return json.dumps(
+            {
+                "status": "found",
+                "count": len(results),
+                "memories": [
+                    {
+                        "id": r.memory.id,
+                        "content": r.memory.content,
+                        "score": round(r.score, 3),
+                        "entities": (
+                            r.related_entities[:5]
+                            if hasattr(r, "related_entities") and r.related_entities
+                            else []
+                        ),
+                    }
+                    for r in results
+                ],
+            }
+        )
+
+    async def _execute_update(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Execute memory_update tool."""
+        memory_id = input_data.get("memory_id", "")
+        new_content = input_data.get("new_content", "")
+
+        if not memory_id:
+            return json.dumps({"status": "error", "error": "memory_id is required"})
+        if not new_content:
+            return json.dumps({"status": "error", "error": "new_content is required"})
+
+        reason = input_data.get("reason")
+
+        if hasattr(self._backend, "update_memory"):
+            memory = await self._backend.update_memory(
+                memory_id=memory_id,
+                new_content=new_content,
+                reason=reason,
+                user_id=user_id,
+            )
+            return json.dumps({"status": "updated", "memory_id": memory.id})
+        else:
+            # Fallback: delete old, save new
+            await self._backend.delete_memory(memory_id)
+            memory = await self._backend.save_memory(
+                content=new_content,
+                user_id=user_id,
+                importance=0.5,
+            )
+            return json.dumps(
+                {
+                    "status": "updated",
+                    "memory_id": memory.id,
+                    "note": "Replaced via delete+save",
+                }
+            )
+
+    async def _execute_delete(self, input_data: dict[str, Any], user_id: str) -> str:
+        """Execute memory_delete tool."""
+        memory_id = input_data.get("memory_id", "")
+        if not memory_id:
+            return json.dumps({"status": "error", "error": "memory_id is required"})
+
+        deleted = await self._backend.delete_memory(memory_id)
+
+        return json.dumps(
+            {
+                "status": "deleted" if deleted else "not_found",
+                "memory_id": memory_id,
+            }
+        )
+
+    async def close(self) -> None:
+        """Close the backend connection."""
+        if self._backend and hasattr(self._backend, "close"):
+            await self._backend.close()
+        self._backend = None
+        self._initialized = False
+        logger.info("MemoryToolAdapter: Closed")

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)