"""Memory tool — search, save, and forget facts about the user."""
from datetime import datetime, timedelta, timezone
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"}
_VALID_SOURCES = {"conversation", "tool_call", "auto_discovery", "user_explicit"}
class MemoryTool(Tool):
name = "memory"
description = (
"Manage long-term memory about the user — facts that survive across sessions. "
"Actions: save (upsert a fact), search (find facts by query), forget (delete by key), list (all facts). "
"When saving system facts (IPs, hosts, services), set source='tool_call', confidence=95, and expires_days=7."
)
parameters = {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["save", "search", "forget", "list"],
"description": (
"save — upsert a fact (overwrites existing key). "
"search — find facts by keyword query. "
"forget — delete a fact by key. "
"list — return all stored facts."
),
},
"query": {
"type": "string",
"description": "search only: keywords describing what to look for.",
},
"category": {
"type": "string",
"enum": ["profile", "preferences", "technical", "projects", "other"],
"description": (
"save/forget: fact category. "
"profile=who they are, preferences=likes/dislikes, "
"technical=OS/tools/servers, projects=ongoing work, other=anything else."
),
},
"key": {
"type": "string",
"description": (
"save/forget: snake_case identifier unique within the category. "
"Examples: name, primary_os, home_server_ip, response_language."
),
},
"value": {
"type": "string",
"description": "save only: the fact as a concise plain-text statement.",
},
"source": {
"type": "string",
"enum": ["conversation", "tool_call", "auto_discovery", "user_explicit"],
"description": (
"save only: how the fact was obtained. "
"conversation=extracted from chat, tool_call=found via tool execution, "
"auto_discovery=system scan, user_explicit=user told me directly."
),
},
"confidence": {
"type": "integer",
"description": "save only: 0-100. Tool output=95, user statement=80, web=50, guess=30.",
"minimum": 0,
"maximum": 100,
},
"expires_days": {
"type": "integer",
"description": "save only: how many days this fact stays valid. Null = never expires.",
},
"source_context": {
"type": "string",
"description": "save only: provenance — 'found via ip addr on localhost', 'user said in session X'.",
},
},
"required": ["action"],
}
def __init__(self, memory_store: MemoryStore) -> None:
self._store = memory_store
async def execute(self, params: dict) -> ToolResult:
action = params.get("action", "")
if action == "save":
return await self._save(params)
if action == "search":
return await self._search(params)
if action == "forget":
return await self._forget(params)
if action == "list":
return await self._list()
return ToolResult(success=False, output=f"Unknown action '{action}'.", error="invalid action")
async def _save(self, params: dict) -> ToolResult:
category = (params.get("category") or "").strip().lower()
key = (params.get("key") or "").strip()
value = (params.get("value") or "").strip()
source = (params.get("source") or "conversation").strip().lower()
confidence = params.get("confidence", 70)
expires_days = params.get("expires_days")
source_context = (params.get("source_context") or "").strip()
if not category:
return ToolResult(success=False, output="category is required for save.", 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 for save.", error="missing key")
if not value:
return ToolResult(success=False, output="value is required for save.", error="missing value")
if source not in _VALID_SOURCES:
source = "conversation"
try:
confidence = int(confidence)
except Exception:
confidence = 70
confidence = max(0, min(100, confidence))
expires_at = None
if expires_days is not None:
try:
expires_at = datetime.now(timezone.utc) + timedelta(days=int(expires_days))
except Exception:
pass
session_id = current_session_id.get(None)
user_id = current_user_id.get(None)
await self._store.upsert_fact(
category=category,
key=key,
value=value,
user_id=user_id,
source_session_id=session_id,
source=source,
confidence=confidence,
expires_at=expires_at,
source_context=source_context,
)
return ToolResult(success=True, output=f"Saved [{category}] {key}: {value}")
async def _search(self, params: dict) -> ToolResult:
query = (params.get("query") or "").strip()
if not query:
return ToolResult(success=False, output="query is required for search.", error="missing query")
user_id = current_user_id.get(None)
facts = await self._store.search_facts(query, user_id=user_id, limit=15)
if not facts:
return ToolResult(success=True, output="No matching facts found in memory.")
lines = []
for f in facts:
prov = ""
if f.get("source"):
prov += f" (src: {f['source']}"
if f.get("confidence"):
prov += f", conf: {f['confidence']}"
if f.get("source_context"):
prov += f", ctx: {f['source_context']}"
prov += ")"
lines.append(f"[{f['category']}] {f['key']}: {f['value']}{prov}")
return ToolResult(success=True, output=f"Found {len(facts)} fact(s):\n" + "\n".join(lines))
async def _forget(self, params: dict) -> ToolResult:
key = (params.get("key") or "").strip()
category = (params.get("category") or "").strip() or None
if not key:
return ToolResult(success=False, output="key is required for forget.", error="missing key")
user_id = current_user_id.get(None)
deleted = await self._store.delete_fact(key, category, user_id=user_id)
if deleted == 0:
return ToolResult(success=False, output=f"No fact found with key '{key}'.", error="not found")
noun = "fact" if deleted == 1 else "facts"
return ToolResult(success=True, output=f"Deleted {deleted} {noun} with key '{key}'.")
async def _list(self) -> ToolResult:
user_id = current_user_id.get(None)
facts = await self._store.get_all_facts(user_id=user_id)
if not facts:
return ToolResult(success=True, output="Memory is empty.")
lines = []
for f in facts:
prov = ""
if f.get("source"):
prov += f" (src: {f['source']}"
if f.get("confidence"):
prov += f", conf: {f['confidence']}"
prov += ")"
lines.append(f"[{f['category']}] {f['key']}: {f['value']}{prov}")
return ToolResult(success=True, output=f"{len(facts)} fact(s) in memory:\n" + "\n".join(lines))
