shira-hermes/api/services/memory.py

"""Per-lawyer and per-case memory store — bounded markdown files with atomic writes.

The filesystem files (under /opt/data/) remain the canonical write target for
update_case_memory and update_user_profile. After each successful filesystem
write, update_user_profile also syncs the result up to the EspoCRM UserProfile
entity (write-through) so an admin browsing UserProfile in the UI sees the
same content that's in the prompt. The CRM is the read source for the prompt
(context['userProfile'] populated by UserProfileContextProvider).
"""

from __future__ import annotations

import asyncio
import logging
import os
import re
import tempfile
from pathlib import Path

import httpx

logger = logging.getLogger("shira.memory")

DATA_DIR = os.environ.get("SHIRA_DATA_DIR", "/opt/data")
PROFILES_DIR = f"{DATA_DIR}/profiles"
CASES_DIR = f"{DATA_DIR}/cases"

PROFILE_MAX_CHARS = 1500
CASE_MEMORY_MAX_CHARS = 2500
ENTRY_DELIMITER = "\n§\n"

_SAFE_ID_RE = re.compile(r"^[a-zA-Z0-9_-]+$")

# Per-path locks for thread-safe writes
_locks: dict[str, asyncio.Lock] = {}


def _safe_id(value: str) -> str:
    """Validate that an ID is safe for use in file paths."""
    if not value or not _SAFE_ID_RE.match(value):
        raise ValueError(f"Invalid ID for file path: {value!r}")
    return value


def _get_lock(path: str) -> asyncio.Lock:
    return _locks.setdefault(path, asyncio.Lock())


def _read_file(path: str) -> str:
    p = Path(path)
    if not p.exists():
        return ""
    return p.read_text(encoding="utf-8")


def _write_file_atomic(path: str, content: str) -> None:
    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=str(p.parent), suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write(content)
        os.rename(tmp, str(p))
    except Exception:
        try:
            os.unlink(tmp)
        except OSError:
            pass
        raise


def _entries(content: str) -> list[str]:
    if not content.strip():
        return []
    return [e.strip() for e in content.split(ENTRY_DELIMITER) if e.strip()]


def _join_entries(entries: list[str]) -> str:
    return ENTRY_DELIMITER.join(entries)


def _truncate(content: str, max_chars: int) -> str:
    """Remove oldest entries (from the top) until content fits."""
    if len(content) <= max_chars:
        return content
    entries = _entries(content)
    while entries and len(_join_entries(entries)) > max_chars:
        entries.pop(0)
    return _join_entries(entries)


# ---------- CRM write-through ----------

async def _sync_user_profile_to_crm(user_id: str, content: str, default_name: str | None = None) -> None:
    """Write-through to UserProfile entity in EspoCRM. Best-effort; logs and swallows errors."""
    base = (os.environ.get("ESPOCRM_URL") or "").rstrip("/")
    key = os.environ.get("ESPOCRM_API_KEY") or ""
    if not base or not key:
        logger.debug("ESPOCRM_URL or API key missing — skipping UserProfile sync")
        return

    name = (default_name or f"Profile {user_id[:8]}")[:255]
    headers = {"X-Api-Key": key, "Content-Type": "application/json"}
    try:
        async with httpx.AsyncClient(timeout=15.0) as client:
            # Find existing row for this user
            r = await client.get(
                f"{base}/api/v1/UserProfile",
                headers=headers,
                params={
                    "where[0][type]": "equals",
                    "where[0][attribute]": "userId",
                    "where[0][value]": user_id,
                    "maxSize": 1,
                },
            )
            existing = (r.json().get("list", []) if r.status_code == 200 else [])
            row = existing[0] if existing else None

            payload = {"content": content, "isActive": True}
            if row:
                resp = await client.put(f"{base}/api/v1/UserProfile/{row['id']}", headers=headers, json=payload)
            else:
                payload.update({"userId": user_id, "name": name})
                resp = await client.post(f"{base}/api/v1/UserProfile", headers=headers, json=payload)
            if resp.status_code >= 400:
                logger.warning("UserProfile sync HTTP %s: %s", resp.status_code, resp.text[:200])
    except Exception as e:
        logger.warning("UserProfile sync failed: %s", e)


# ---------- Public API ----------

def load_user_profile(user_id: str) -> str:
    """Load the per-lawyer profile. Returns empty string if none exists."""
    return _read_file(f"{PROFILES_DIR}/{_safe_id(user_id)}.md")


def load_case_memory(case_id: str) -> str:
    """Load the per-case memory. Returns empty string if none exists."""
    return _read_file(f"{CASES_DIR}/{_safe_id(case_id)}/memory.md")


async def update_user_profile(user_id: str, action: str, content: str, match: str = "") -> str:
    """Add, replace, or remove an entry in the lawyer's profile."""
    path = f"{PROFILES_DIR}/{_safe_id(user_id)}.md"
    async with _get_lock(path):
        current = _read_file(path)
        entries = _entries(current)

        if action == "add":
            entries.append(content)
            result = _truncate(_join_entries(entries), PROFILE_MAX_CHARS)
            _write_file_atomic(path, result)
            await _sync_user_profile_to_crm(user_id, result)
            return f'✅ נוסף לפרופיל: "{content[:60]}"'

        elif action == "replace":
            if not match:
                return "❌ צריך לספק match כדי למצוא את הרשומה להחלפה."
            replaced = False
            for i, e in enumerate(entries):
                if match in e:
                    entries[i] = content
                    replaced = True
                    break
            if not replaced:
                return f'❌ לא נמצאה רשומה עם "{match}"'
            result = _truncate(_join_entries(entries), PROFILE_MAX_CHARS)
            _write_file_atomic(path, result)
            await _sync_user_profile_to_crm(user_id, result)
            return f'✅ רשומה עודכנה בפרופיל.'

        elif action == "remove":
            if not match:
                return "❌ צריך לספק match כדי למצוא את הרשומה למחיקה."
            original_len = len(entries)
            entries = [e for e in entries if match not in e]
            if len(entries) == original_len:
                return f'❌ לא נמצאה רשומה עם "{match}"'
            result = _join_entries(entries)
            _write_file_atomic(path, result)
            await _sync_user_profile_to_crm(user_id, result)
            return "✅ רשומה הוסרה מהפרופיל."

        elif action == "read":
            return current if current else "הפרופיל ריק."

        return f"❌ פעולה לא מוכרת: {action}"


async def update_case_memory(case_id: str, action: str, content: str, match: str = "") -> str:
    """Add, replace, or remove an entry in the case memory."""
    path = f"{CASES_DIR}/{_safe_id(case_id)}/memory.md"
    async with _get_lock(path):
        current = _read_file(path)
        entries = _entries(current)

        if action == "add":
            entries.append(content)
            result = _truncate(_join_entries(entries), CASE_MEMORY_MAX_CHARS)
            _write_file_atomic(path, result)
            return f'✅ נשמר בזיכרון התיק: "{content[:60]}"'

        elif action == "replace":
            if not match:
                return "❌ צריך לספק match כדי למצוא את הרשומה להחלפה."
            replaced = False
            for i, e in enumerate(entries):
                if match in e:
                    entries[i] = content
                    replaced = True
                    break
            if not replaced:
                return f'❌ לא נמצאה רשומה עם "{match}"'
            result = _truncate(_join_entries(entries), CASE_MEMORY_MAX_CHARS)
            _write_file_atomic(path, result)
            return "✅ רשומה עודכנה בזיכרון התיק."

        elif action == "remove":
            if not match:
                return "❌ צריך לספק match כדי למצוא את הרשומה למחיקה."
            original_len = len(entries)
            entries = [e for e in entries if match not in e]
            if len(entries) == original_len:
                return f'❌ לא נמצאה רשומה עם "{match}"'
            _write_file_atomic(path, _join_entries(entries))
            return "✅ רשומה הוסרה מזיכרון התיק."

        elif action == "read":
            return current if current else "זיכרון התיק ריק."

        return f"❌ פעולה לא מוכרת: {action}"


def register_memory_tools(tools: dict, case_id: str | None, user_id: str | None) -> None:
    """Register memory tools into the tools dict."""

    async def _update_user_profile(action: str, content: str = "", match: str = "") -> str:
        if not user_id:
            return "❌ לא ניתן לזהות את המשתמש."
        return await update_user_profile(user_id, action, content, match)

    async def _update_case_memory(action: str, content: str = "", match: str = "") -> str:
        if not case_id:
            return "❌ צריך להיות בתוך תיק כדי לעדכן את הזיכרון."
        return await update_case_memory(case_id, action, content, match)

    tools["update_user_profile"] = {
        "description": (
            "Manage the lawyer's personal profile — preferences, rules, communication style, common case types. "
            "This persists across conversations. Use proactively when you learn about the user's preferences. "
            "Also use this when the user says 'from now on always...', 'I prefer...', 'remember that I...' — "
            "save their preference here so you remember it in future conversations."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "action": {
                    "type": "string",
                    "enum": ["add", "replace", "remove", "read"],
                    "description": "add=append entry, replace=find by match and replace, remove=find by match and delete, read=show current profile",
                },
                "content": {"type": "string", "description": "The content to add or replace with"},
                "match": {"type": "string", "description": "Substring to match when replacing or removing"},
            },
            "required": ["action"],
        },
        "handler": _update_user_profile,
    }

    tools["update_case_memory"] = {
        "description": (
            "Manage the case's persistent memory — key facts, strategy, decisions, timeline. "
            "This persists across conversations about this case. Use alongside save_memory (which writes to CRM notes)."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "action": {
                    "type": "string",
                    "enum": ["add", "replace", "remove", "read"],
                    "description": "add=append entry, replace=find by match and replace, remove=find by match and delete, read=show current memory",
                },
                "content": {"type": "string", "description": "The content to add or replace with"},
                "match": {"type": "string", "description": "Substring to match when replacing or removing"},
            },
            "required": ["action"],
        },
        "handler": _update_case_memory,
    }