headroom/proxy/memory_tool_adapter.py

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