Newer
Older
navi-1 / navi / tools / memory_save.py
"""Memory save tool — persist a fact about the user to long-term memory."""

from navi.memory.store import MemoryStore
from navi.tools.base import current_session_id, current_user_id

from .base import Tool, ToolResult

_VALID_CATEGORIES = {"profile", "preferences", "technical", "projects", "other"}


class MemorySaveTool(Tool):
    name = "memory_save"
    description = (
        "Save or update a fact about the user in long-term memory. "
        "Use this when the user tells you something stable and reusable — name, location, "
        "preferences, ongoing projects, technical environment, recurring workflows, etc. "
        "Also use it to correct an outdated fact you know is wrong. "
        "Facts survive across sessions and are available via memory_search."
    )
    parameters = {
        "type": "object",
        "properties": {
            "category": {
                "type": "string",
                "enum": ["profile", "preferences", "technical", "projects", "other"],
                "description": (
                    "Broad category for the fact. "
                    "profile=who they are, preferences=what they like/dislike, "
                    "technical=OS/tools/servers/languages, projects=ongoing work, other=anything else."
                ),
            },
            "key": {
                "type": "string",
                "description": (
                    "Short snake_case identifier, unique within the category. "
                    "Examples: 'name', 'primary_os', 'home_server_ip', 'preferred_language', "
                    "'current_project', 'response_language'."
                ),
            },
            "value": {
                "type": "string",
                "description": "The fact itself, written as a concise plain-text statement.",
            },
        },
        "required": ["category", "key", "value"],
    }

    def __init__(self, memory_store: MemoryStore) -> None:
        self._store = memory_store

    async def execute(self, params: dict) -> ToolResult:
        category = params.get("category", "").strip().lower()
        key = params.get("key", "").strip()
        value = params.get("value", "").strip()

        if not category:
            return ToolResult(success=False, output="category is required.", error="missing category")
        if category not in _VALID_CATEGORIES:
            return ToolResult(
                success=False,
                output=f"Invalid category '{category}'. Must be one of: {', '.join(sorted(_VALID_CATEGORIES))}",
                error="invalid category",
            )
        if not key:
            return ToolResult(success=False, output="key is required.", error="missing key")
        if not value:
            return ToolResult(success=False, output="value is required.", error="missing value")

        session_id = current_session_id.get(None)
        user_id = current_user_id.get(None)
        await self._store.upsert_fact(category, key, value, user_id=user_id, source_session_id=session_id)
        return ToolResult(success=True, output=f"Saved [{category}] {key}: {value}")