Synapse Memory.py

"""
title: Synapse Memory
author: cloph-dsp
version: 7.0.0
license: MIT
required_open_webui_version: 0.5.0
description: The definitive memory engine for OpenWebUI. LLM-driven lifecycle management with smart
deduplication, conflict resolution, MERGE/EXTENDS consolidation, diversity-aware retrieval, and
natural language vault control — all through OpenWebUI's native APIs.

Highlights:
- UserValves: each user can enable/disable memory and toggle status notifications independently.
- Two-stage retrieval: OpenWebUI native vector search narrows the candidate pool first, then LLM
  reranks. Profile memories (identity/behavior) always enter the candidate set regardless of score.
- Recency-weighted scoring: final = (LLM score × 0.7) + (recency × 0.3) with exponential decay,
  so newer facts beat older ones when relevance is equal.
- Dual-block injection: <user_profile> (who the user is) + <relevant_memories> (what matters now)
  — proven format from mem0 and Supermemory research.
- Five extraction operations: NEW, DELETE, UPDATE, EXTENDS (additive enrichment without overwrite),
  MERGE (consolidate fragments). LLM chooses the right one per the guidelines.
- LLM intent router: keyword pre-screen skips LLM for normal messages; one LLM call handles all
  vault intents ("forget X", "show memories", "find my Python memories", "clear vault").
- Temporal context: current UTC date in every extraction prompt for time-sensitive reasoning.
- Fuzzy deduplication: Jaccard ≥ 0.90 word-overlap blocks near-duplicate facts.
- Correct status events: includes status/done/error fields matching the OpenWebUI event spec.
- Circuit breaker: LLM failures trip after 3 errors; auto-recovers after 30 s.
"""

import asyncio
import json
import logging
import re
import time
from datetime import datetime, timezone
from typing import Any, Awaitable, Callable, Dict, List, Literal, Optional, Set, Tuple

from fastapi import Request
from pydantic import BaseModel, Field

from open_webui.main import app as webui_app
from open_webui.utils.chat import generate_chat_completion
from open_webui.routers.memories import (
    AddMemoryForm,
    QueryMemoryForm,
    Memories,
    add_memory,
    delete_memory_by_id,
    query_memory,
)

logger = logging.getLogger(__name__)

TAG_CATEGORIES: Set[str] = {
    "identity", "behavior", "preference", "goal", "relationship", "possession"
}

MAX_INJECTED_MEMORIES = 12      # Cap on memories injected into conversation context
MAX_MEMORY_CONTEXT = 80         # Memories sent to LLM per prompt (inventory ceiling)
MAX_MEMORY_PROMPT_CHARS = 8000  # Character budget for the inventory block in prompts

# ---------------------------------------------------------------------------
# Keyword pre-screen for intent classification
#
# If the message contains NONE of these words it cannot be a memory management
# intent, so we skip the LLM intent call entirely.  The set is intentionally
# broad — false negatives here mean an unnecessary LLM call for selection,
# false positives are impossible (we only route to LLM intent if a keyword
# fires, then let the LLM decide the actual intent).
# ---------------------------------------------------------------------------

_INTENT_KEYWORDS: frozenset = frozenset({
    "memory", "memories", "remember", "recall", "forget", "forgotten",
    "stored", "saved", "know about me", "clear", "wipe", "erase",
    "delete my", "remove my", "what you know", "what you have",
})


# ---------------------------------------------------------------------------
# Circuit Breaker
# ---------------------------------------------------------------------------


class CircuitBreakerOpenError(Exception):
    pass


class CircuitBreaker:
    """Guards against cascading LLM failures with exponential back-off recovery."""

    def __init__(self, failure_threshold: int = 3, recovery_timeout: int = 30) -> None:
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = 0.0
        self.state = "CLOSED"

    async def call(self, func, *args, **kwargs):
        now = time.monotonic()

        if self.state == "OPEN":
            if now - self.last_failure_time >= self.recovery_timeout:
                self.state = "HALF_OPEN"
                logger.info("Circuit breaker: HALF_OPEN — testing recovery")
            else:
                raise CircuitBreakerOpenError("Circuit breaker OPEN — LLM skipped")

        try:
            result = await func(*args, **kwargs)
            if self.state == "HALF_OPEN":
                logger.info("Circuit breaker: CLOSED — recovery confirmed")
            self.state = "CLOSED"
            self.failure_count = 0
            return result
        except Exception:
            self.failure_count += 1
            self.last_failure_time = time.monotonic()
            if self.failure_count >= self.failure_threshold:
                self.state = "OPEN"
                logger.warning("Circuit breaker: OPEN after %d failures", self.failure_count)
            raise

    def reset(self) -> None:
        self.state = "CLOSED"
        self.failure_count = 0
        self.last_failure_time = 0.0


# ---------------------------------------------------------------------------
# Filter
# ---------------------------------------------------------------------------


class Filter:
    """Synapse Memory v7 — the definitive OpenWebUI memory engine."""

    class Valves(BaseModel):
        llm_model: str = Field(
            default="llama3.2:latest",
            description="Model used for all reasoning: extraction, selection, and intent detection.",
        )
        status_mode: Literal["silent", "brief", "detailed"] = Field(
            default="brief",
            description=(
                "Status verbosity: "
                "silent = no status updates, "
                "brief = key milestones only, "
                "detailed = step-by-step."
            ),
        )

    class UserValves(BaseModel):
        enabled: bool = Field(
            default=True,
            description="Enable Synapse Memory for your account. Disable to stop all memory activity.",
        )
        show_status: bool = Field(
            default=True,
            description="Show memory status updates in the chat UI.",
        )

    def __init__(self) -> None:
        self.valves = self.Valves()
        self.circuit_breaker = CircuitBreaker()
        logger.info("Synapse Memory v6.0 ready")

    # =========================================================================
    # PUBLIC HOOKS
    # =========================================================================

    async def inlet(
        self,
        body: Dict[str, Any],
        __event_emitter__: Optional[Callable[[Any], Awaitable[None]]] = None,
        __user__: Optional[Dict[str, Any]] = None,
        __request__: Optional[Request] = None,
    ) -> Dict[str, Any]:
        if not __user__ or not __user__.get("id"):
            return body

        user_valves = self._get_user_valves(__user__)
        if not user_valves.enabled:
            return body
        if not user_valves.show_status:
            __event_emitter__ = None  # suppress all status updates for this user

        messages = body.get("messages", [])
        if not messages or messages[-1].get("role") != "user":
            return body

        user_message = (messages[-1].get("content") or "").strip()
        if not user_message:
            return body

        user_id = __user__["id"]

        # ------------------------------------------------------------------
        # Intent classification — skip entirely if no memory keywords present
        # ------------------------------------------------------------------
        msg_lower = user_message.lower()
        has_keyword = any(kw in msg_lower for kw in _INTENT_KEYWORDS)

        if has_keyword:
            intent_result = await self._classify_intent(
                user_message, user_id, __request__, __user__
            )
            intent = intent_result.get("intent", "normal")
            detail = intent_result.get("detail", "")

            if intent == "clear_all":
                return await self._handle_clear_all(
                    body, user_id, __request__, __event_emitter__, __user__
                )
            if intent == "forget":
                return await self._handle_forget(
                    body, user_id, detail or user_message,
                    __request__, __event_emitter__, __user__,
                )
            if intent == "show_all":
                return await self._handle_memory_query(
                    body, user_id, user_message, "show_all",
                    __request__, __event_emitter__, __user__,
                )
            if intent == "search":
                return await self._handle_memory_query(
                    body, user_id, detail or user_message, "search",
                    __request__, __event_emitter__, __user__,
                )

        # ------------------------------------------------------------------
        # Standard conversation: retrieve and inject relevant memories
        # ------------------------------------------------------------------
        memories = await self._fetch_user_memories(user_id)
        if not memories:
            return body

        await self._emit_status(__event_emitter__, "🧠 Retrieving relevant memories…", "detailed")

        conversation = self._format_conversation(messages, last_n=6)
        selected = await self._select_relevant_memories(
            conversation, memories, user_id, __request__, __user__
        )

        if not selected:
            await self._emit_status(
                __event_emitter__, "💤 No memories relevant to this turn", "detailed", done=True
            )
            return body

        injection = self._format_memory_injection(selected)
        if body.get("system"):
            body["system"] = f"{injection}\n\n{body['system']}"
        elif messages:
            messages[0]["content"] = f"{injection}\n\n{messages[0]['content']}"

        n = len(selected)
        await self._emit_status(
            __event_emitter__,
            f"✅ {n} memory{'ies' if n != 1 else ''} injected",
            "brief",
            done=True,
        )
        return body

    async def outlet(
        self,
        body: Dict[str, Any],
        __event_emitter__: Optional[Callable[[Any], Awaitable[None]]] = None,
        __user__: Optional[Dict[str, Any]] = None,
        __request__: Optional[Request] = None,
    ) -> Dict[str, Any]:
        if not __user__ or not __user__.get("id"):
            return body

        user_valves = self._get_user_valves(__user__)
        if not user_valves.enabled:
            return body
        if not user_valves.show_status:
            __event_emitter__ = None  # suppress all status updates for this user

        messages = body.get("messages", [])
        if len(messages) < 2:
            return body

        conversation = self._format_conversation(messages, last_n=8)
        if len(conversation.strip()) < 10:
            return body

        user_id = __user__["id"]
        memories = await self._fetch_user_memories(user_id)

        await self._emit_status(
            __event_emitter__, "🧪 Evaluating conversation for memory operations…", "detailed"
        )

        ops = await self._extract_with_llm(
            conversation, memories, user_id, __request__, __user__
        )
        if not ops:
            await self._emit_status(
                __event_emitter__, "💭 Nothing to remember from this turn", "detailed", done=True
            )
            return body

        stats = {"created": 0, "deleted": 0, "merged": 0}

        for op in ops:
            action = (op.get("action") or "").upper()

            if action == "DELETE":
                target = op.get("target") or {}
                tid = target.get("id")
                if not tid:
                    continue
                try:
                    await delete_memory_by_id(tid)
                    stats["deleted"] += 1
                    memories = [m for m in memories if m.get("id") != tid]
                    await self._emit_status(
                        __event_emitter__,
                        f"🗑️ Cleared: {target.get('content', '')[:70].strip()}",
                        "detailed",
                    )
                except Exception as exc:
                    logger.error("Delete %s failed: %s", tid, exc)

            elif action in {"NEW", "CREATE"}:
                content = (op.get("content") or "").strip()
                tags = op.get("tags") or []
                if not content or self._is_too_short(content):
                    continue
                if self._is_duplicate(content, memories):
                    await self._emit_status(
                        __event_emitter__, f"↪️ Already known: {content[:70]}", "detailed"
                    )
                    continue
                if await self._store_memory(content, tags, __user__, __request__):
                    stats["created"] += 1
                    memories.append(self._build_memory_record(content, tags))
                    await self._emit_status(
                        __event_emitter__, f"✨ Stored: {content[:80]}", "detailed"
                    )

            elif action == "UPDATE":
                target = op.get("target") or {}
                content = (op.get("content") or "").strip()
                tags = op.get("tags") or []
                tid = target.get("id")
                if tid:
                    try:
                        await delete_memory_by_id(tid)
                        memories = [m for m in memories if m.get("id") != tid]
                    except Exception as exc:
                        logger.error("Update-delete %s failed: %s", tid, exc)
                if content and not self._is_too_short(content) and not self._is_duplicate(content, memories):
                    if await self._store_memory(content, tags, __user__, __request__):
                        stats["created"] += 1
                        memories.append(self._build_memory_record(content, tags))
                        old_preview = target.get("content", "")[:40].strip()
                        await self._emit_status(
                            __event_emitter__,
                            f"🔄 Updated: {old_preview} → {content[:40]}",
                            "detailed",
                        )

            elif action == "EXTENDS":
                # Additive enrichment — replace old entry with enriched content
                target = op.get("target") or {}
                content = (op.get("content") or "").strip()
                tags = op.get("tags") or []
                tid = target.get("id")
                if tid:
                    try:
                        await delete_memory_by_id(tid)
                        memories = [m for m in memories if m.get("id") != tid]
                    except Exception as exc:
                        logger.error("Extends-delete %s failed: %s", tid, exc)
                if content and not self._is_too_short(content) and not self._is_duplicate(content, memories):
                    if await self._store_memory(content, tags, __user__, __request__):
                        stats["created"] += 1
                        memories.append(self._build_memory_record(content, tags))
                        old_preview = target.get("content", "")[:40].strip()
                        await self._emit_status(
                            __event_emitter__,
                            f"➕ Enriched: {old_preview} → {content[:40]}",
                            "detailed",
                        )

            elif action == "MERGE":
                sources = op.get("sources") or []
                content = (op.get("content") or "").strip()
                tags = op.get("tags") or []
                if not content or self._is_too_short(content):
                    continue
                deleted_count = 0
                for src in sources:
                    src_id = (src or {}).get("id")
                    if not src_id:
                        continue
                    try:
                        await delete_memory_by_id(src_id)
                        memories = [m for m in memories if m.get("id") != src_id]
                        deleted_count += 1
                    except Exception as exc:
                        logger.error("Merge-delete %s failed: %s", src_id, exc)
                if await self._store_memory(content, tags, __user__, __request__):
                    stats["merged"] += 1
                    memories.append(self._build_memory_record(content, tags))
                    await self._emit_status(
                        __event_emitter__,
                        f"🔗 Merged {deleted_count} → {content[:70]}",
                        "detailed",
                    )

        parts = []
        if stats["created"]:
            parts.append(f"{stats['created']} stored")
        if stats["deleted"]:
            parts.append(f"{stats['deleted']} removed")
        if stats["merged"]:
            parts.append(f"{stats['merged']} merged")

        if parts:
            await self._emit_status(
                __event_emitter__,
                "💾 Memory vault: " + ", ".join(parts),
                "brief",
                done=True,
            )
        else:
            await self._emit_status(
                __event_emitter__, "✅ Memory vault unchanged", "brief", done=True
            )

        return body

    # =========================================================================
    # INTENT CLASSIFICATION
    # =========================================================================

    def _get_user_valves(self, __user__: Optional[Dict[str, Any]]) -> "Filter.UserValves":
        """Extract per-user valves from the __user__ context object."""
        if __user__ is None:
            return self.UserValves()
        uv = __user__.get("valves")
        if isinstance(uv, self.UserValves):
            return uv
        return self.UserValves()

    async def _classify_intent(
        self,
        message: str,
        user_id: str,
        __request__: Optional[Request],
        __user__: Optional[Dict[str, Any]],
    ) -> Dict[str, str]:
        """
        Single LLM call that classifies the user's memory management intent
        and extracts any relevant detail in one shot.

        Returns a dict with:
          intent  — one of: normal, show_all, search, forget, clear_all
          detail  — search topic (for 'search') or forget description (for 'forget'), else ""

        Called only after _INTENT_KEYWORDS pre-screen fires, so the vast
        majority of normal conversational messages never reach this path.
        """
        system = (
            "Classify the user's intent regarding their personal memory vault.\n\n"
            "Intents:\n"
            "  normal    — Regular conversation, not asking about memories.\n"
            "  show_all  — User wants to see all stored memories about them.\n"
            "  search    — User wants to find memories matching a specific topic.\n"
            "  forget    — User wants a specific memory or fact deleted.\n"
            "  clear_all — User wants ALL memories erased.\n\n"
            "For 'search': set detail to the topic the user is searching for.\n"
            "For 'forget': set detail to a concise description of what the user wants forgotten.\n"
            "For all others: set detail to an empty string.\n\n"
            'Return JSON only: {"intent": "<intent>", "detail": "<detail>"}'
        )
        response = await self._query_llm(
            system, f"Message: {message}",
            user_id, __request__, __user__,
            temperature=0.0, max_tokens=60,
        )
        if not response:
            return {"intent": "normal", "detail": ""}
        try:
            data = json.loads(response)
            intent = str(data.get("intent", "normal")).lower().strip()
            if intent not in {"normal", "show_all", "search", "forget", "clear_all"}:
                intent = "normal"
            return {"intent": intent, "detail": str(data.get("detail", "")).strip()}
        except (json.JSONDecodeError, AttributeError):
            return {"intent": "normal", "detail": ""}

    # =========================================================================
    # MEMORY QUERY HANDLERS
    # =========================================================================

    async def _handle_memory_query(
        self,
        body: Dict[str, Any],
        user_id: str,
        user_message: str,
        intent: str,
        __request__: Optional[Request],
        __event_emitter__: Optional[Callable[[Any], Awaitable[None]]],
        __user__: Optional[Dict[str, Any]],
    ) -> Dict[str, Any]:
        memories = await self._fetch_user_memories(user_id)

        if not memories:
            summary = (
                "📝 **Your Memory Vault**\n\n"
                "I don't have any memories stored about you yet. "
                "Just mention important details and I'll remember them automatically!"
            )
        elif intent == "search":
            # user_message here is already the extracted topic from _classify_intent
            summary = self._build_search_summary(memories, user_message)
        else:
            summary = self._build_vault_summary(memories)

        messages = body.get("messages", [])
        if body.get("system"):
            body["system"] = f"{summary}\n\n{body['system']}"
        elif messages:
            messages[0]["content"] = f"{summary}\n\n{messages[0]['content']}"

        await self._emit_status(
            __event_emitter__, "📂 Memory vault details loaded", "brief", done=True
        )
        return body

    async def _handle_forget(
        self,
        body: Dict[str, Any],
        user_id: str,
        forget_description: str,
        __request__: Optional[Request],
        __event_emitter__: Optional[Callable[[Any], Awaitable[None]]],
        __user__: Optional[Dict[str, Any]],
    ) -> Dict[str, Any]:
        memories = await self._fetch_user_memories(user_id)
        if not memories:
            await self._emit_status(
                __event_emitter__, "📭 No memories to forget", "brief", done=True
            )
            return body

        inv, index_map = self._build_memory_inventory(memories)
        system = (
            "The user wants to delete a specific memory. "
            "Identify which stored memory best matches what they want forgotten.\n"
            'Return JSON only: {"memory_number": <int or null>, "reason": "..."}'
        )
        user_p = f"What to forget: {forget_description}\n\nStored memories:\n{inv}"
        response = await self._query_llm(
            system, user_p, user_id, __request__, __user__, temperature=0.0, max_tokens=80
        )

        if response:
            try:
                data = json.loads(response)
                num = data.get("memory_number")
                if num is not None:
                    target = index_map.get(int(num))
                    tid = (target or {}).get("id")
                    if tid:
                        await delete_memory_by_id(tid)
                        snippet = (target or {}).get("content", "")[:80]
                        await self._emit_status(
                            __event_emitter__, f"🗑️ Forgotten: {snippet}", "brief", done=True
                        )
                        return body
            except Exception as exc:
                logger.error("Forget handler failed: %s", exc)

        await self._emit_status(
            __event_emitter__, "🤔 No matching memory found to forget", "brief", done=True
        )
        return body

    async def _handle_clear_all(
        self,
        body: Dict[str, Any],
        user_id: str,
        __request__: Optional[Request],
        __event_emitter__: Optional[Callable[[Any], Awaitable[None]]],
        __user__: Optional[Dict[str, Any]],
    ) -> Dict[str, Any]:
        memories = await self._fetch_user_memories(user_id)
        deleted = 0
        for mem in memories:
            mid = mem.get("id")
            if not mid:
                continue
            try:
                await delete_memory_by_id(mid)
                deleted += 1
            except Exception as exc:
                logger.error("Clear-all delete %s failed: %s", mid, exc)

        msg = (
            f"🗑️ Memory vault cleared ({deleted} memories removed)"
            if deleted
            else "📭 Memory vault was already empty"
        )
        await self._emit_status(__event_emitter__, msg, "brief", done=True)
        return body

    # =========================================================================
    # VAULT DISPLAY HELPERS
    # =========================================================================

    def _build_search_summary(self, memories: List[Dict[str, Any]], topic: str) -> str:
        """Return a keyword-filtered memory summary matching the search topic."""
        topic_words = set(re.split(r"\W+", topic.lower())) - {"", "a", "an", "the", "my", "i"}
        scored: List[Tuple[int, Dict[str, Any]]] = []
        for mem in memories:
            content_words = set(re.split(r"\W+", mem.get("content", "").lower()))
            tag_words = set(" ".join(mem.get("tags", [])).lower().split())
            overlap = len(topic_words & (content_words | tag_words))
            if overlap > 0:
                scored.append((overlap, mem))

        scored.sort(key=lambda x: x[0], reverse=True)
        matched = [m for _, m in scored[:15]]

        if not matched:
            return f"📝 **Memory Search: '{topic}'**\n\nNo memories found matching that topic."

        lines = [f"📝 **Memory Search: '{topic}'**\n", f"Found **{len(matched)}** relevant memories:\n"]
        for i, mem in enumerate(matched, 1):
            tags = mem.get("tags") or []
            tag_str = f"  _{', '.join(tags)}_" if tags else ""
            lines.append(f"{i}. {mem.get('content', '')}{tag_str}")
        return "\n".join(lines)

    def _build_vault_summary(self, memories: List[Dict[str, Any]]) -> str:
        """Build a full categorised memory vault display."""
        emoji_map = {
            "identity": "👤",
            "preference": "❤️",
            "behavior": "🔁",
            "goal": "🎯",
            "relationship": "👥",
            "possession": "🎁",
        }

        lines = [
            "📝 **Your Memory Vault**\n",
            f"**{len(memories)}** memories stored:\n",
        ]

        category_map: Dict[str, List[str]] = {cat: [] for cat in TAG_CATEGORIES}
        untagged: List[str] = []

        for mem in memories:
            content = mem.get("content", "")
            placed = False
            for tag in mem.get("tags") or []:
                if tag in category_map:
                    category_map[tag].append(content)
                    placed = True
                    break
            if not placed:
                untagged.append(content)

        for tag in TAG_CATEGORIES:
            entries = category_map[tag]
            if not entries:
                continue
            lines.append(f"**{emoji_map.get(tag, '📌')} {tag.capitalize()}**")
            for entry in entries[:6]:
                lines.append(f"  • {entry}")
            if len(entries) > 6:
                lines.append(f"  • _…and {len(entries) - 6} more_")
            lines.append("")

        if untagged:
            lines.append("**📌 Other**")
            for entry in untagged[:8]:
                lines.append(f"  • {entry}")
            if len(untagged) > 8:
                lines.append(f"  • _…and {len(untagged) - 8} more_")

        return "\n".join(lines)

    # =========================================================================
    # RETRIEVAL: SELECT RELEVANT MEMORIES
    # =========================================================================

    async def _select_relevant_memories(
        self,
        conversation: str,
        memories: List[Dict[str, Any]],
        user_id: str,
        __request__: Optional[Request],
        __user__: Optional[Dict[str, Any]],
    ) -> List[Dict[str, Any]]:
        # ------------------------------------------------------------------
        # Stage 1: vector search narrows the candidate pool when vault is large.
        # Profile memories (identity/behavior) are always added to the candidates
        # regardless of semantic score — they define who the user is.
        # ------------------------------------------------------------------
        profile_tag_set = {"identity", "behavior"}
        profile_mems = [m for m in memories if set(m.get("tags") or []) & profile_tag_set]

        if len(memories) > 15:
            # Use the last user message as the vector query
            query = ""
            for line in reversed(conversation.split("\n")):
                if line.startswith("User:"):
                    query = line[5:].strip()
                    break
            query = query or conversation[-300:]

            vector_hits = await self._query_memory_vector(
                query, k=25, __user__=__user__, __request__=__request__
            )
            if vector_hits:
                seen_ids: Set[str] = {str(m.get("id")) for m in vector_hits if m.get("id")}
                candidates = list(vector_hits)
                for pm in profile_mems:
                    if str(pm.get("id")) not in seen_ids:
                        candidates.append(pm)
                        seen_ids.add(str(pm.get("id")))
            else:
                # Vector search unavailable — fall back to full list
                candidates = memories[:MAX_MEMORY_CONTEXT]
        else:
            candidates = memories[:MAX_MEMORY_CONTEXT]

        if not candidates:
            return []

        # ------------------------------------------------------------------
        # Stage 2: LLM reranking over the narrowed candidate set
        # ------------------------------------------------------------------
        inv, index_map = self._build_memory_inventory(candidates)

        system = f"""You curate memory context for a conversational AI assistant.
Select the stored memories that would help the assistant give a more personalised, accurate response.

Rules:
1. Only include memories DIRECTLY relevant to the current conversation topic or user intent.
2. Always include identity/behavior memories that describe who the user fundamentally is.
3. Select at most {MAX_INJECTED_MEMORIES} memories. Quality over quantity.
4. Score each 0.0–1.0. Only include scores above 0.50.
5. Do NOT select multiple memories that convey the same information — pick the most specific or recent.
6. Keep reasons brief (under 60 chars).
7. Return an empty list if nothing would meaningfully help.

Respond with valid JSON only:
{{"selected": [{{"number": 3, "score": 0.91, "reason": "User is asking about Python"}}, ...]}}"""

        user_p = f"Conversation:\n{conversation}\n\nStored memories:\n{inv}"

        response = await self._query_llm(
            system, user_p, user_id, __request__, __user__, temperature=0.1, max_tokens=600
        )
        if not response:
            return []

        try:
            data = json.loads(response)
        except json.JSONDecodeError:
            logger.warning("Memory selection: could not parse LLM JSON")
            return []

        selected: List[Dict[str, Any]] = []
        seen: Set[str] = set()

        for entry in data.get("selected", []):
            num = entry.get("number")
            if num is None:
                continue
            record = index_map.get(int(num))
            if not record:
                continue
            uid = str(record.get("id") or record.get("content", ""))
            if uid in seen:
                continue
            seen.add(uid)
            llm_score = float(entry.get("score", 0.0))
            if llm_score < 0.50:
                continue
            # Stage 3: recency-weighted final score
            recency = self._recency_score(record)
            final_score = (llm_score * 0.7) + (recency * 0.3)
            selected.append({
                "record": record,
                "score": final_score,
                "reason": (entry.get("reason") or "").strip(),
            })

        selected.sort(key=lambda x: x["score"], reverse=True)
        return selected[:MAX_INJECTED_MEMORIES]

    # =========================================================================
    # EXTRACTION: IDENTIFY MEMORY OPERATIONS
    # =========================================================================

    async def _extract_with_llm(
        self,
        conversation: str,
        memories: List[Dict[str, Any]],
        user_id: str,
        __request__: Optional[Request],
        __user__: Optional[Dict[str, Any]],
    ) -> List[Dict[str, Any]]:
        limited = memories[:MAX_MEMORY_CONTEXT]
        inv, index_map = self._build_memory_inventory(limited)
        now_str = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")

        system = f"""You manage a personal memory vault for a user.
Current time: {now_str}

Decide what to add, correct, enrich, remove, or consolidate based on this conversation.

Operations (use only when genuinely warranted):
  NEW     — Store a durable, user-specific fact not already present.
  DELETE  — Remove an entry (by number) confirmed incorrect or explicitly asked to forget.
  UPDATE  — Replace an entry (by number) with corrected content (fact changed or was wrong).
  EXTENDS — Enrich an entry (by number) with new details that ADD TO it without contradicting it.
  MERGE   — Combine 2–4 closely related entries (by number list) into one cleaner fact.

Guidelines:
• Skip anything already stored verbatim or near-verbatim.
• Skip transient details: greetings, small-talk, one-off questions, general knowledge.
• Keep each memory concise (under 200 characters) and factual.
• If a new fact CONTRADICTS a stored one → DELETE old + NEW corrected one.
• If a new fact ADDS TO a stored one without contradicting it → use EXTENDS.
• Use tags when helpful: identity, preference, behavior, goal, relationship, possession.
• If 2+ memories cover the same narrow topic, prefer MERGE over adding more fragments.

Return ONLY valid JSON — no prose, no markdown fences:
{{
  "operations": [
    {{"action": "NEW",     "content": "...", "tags": ["preference"]}},
    {{"action": "DELETE",  "memory_number": 3, "reason": "User corrected job title"}},
    {{"action": "UPDATE",  "memory_number": 6, "content": "...", "tags": ["identity"]}},
    {{"action": "EXTENDS", "memory_number": 4, "content": "...", "tags": ["identity"]}},
    {{"action": "MERGE",   "sources": [2, 5], "content": "...", "tags": ["preference"]}}
  ]
}}"""

        user_p = f"Conversation:\n{conversation}\n\nExisting memories:\n{inv}"

        response = await self._query_llm(
            system, user_p, user_id, __request__, __user__, temperature=0.1, max_tokens=1000
        )
        if not response:
            return []

        try:
            data = json.loads(response)
        except json.JSONDecodeError:
            logger.warning("Extraction: could not parse LLM JSON")
            return []

        sanitized: List[Dict[str, Any]] = []

        for op in data.get("operations", []):
            action = (op.get("action") or "").upper()

            if action == "DELETE":
                target = self._resolve_by_number(op.get("memory_number"), index_map)
                if target:
                    sanitized.append({"action": "DELETE", "target": target})

            elif action in {"NEW", "CREATE"}:
                content = (op.get("content") or "").strip()
                if not content or self._is_too_short(content):
                    continue
                tags = self._sanitize_tags(op.get("tags"))
                sanitized.append({"action": "NEW", "content": content, "tags": tags})

            elif action == "UPDATE":
                content = (op.get("content") or "").strip()
                if not content or self._is_too_short(content):
                    continue
                target = self._resolve_by_number(op.get("memory_number"), index_map)
                tags = self._sanitize_tags(op.get("tags"))
                sanitized.append({"action": "UPDATE", "content": content, "tags": tags, "target": target or {}})

            elif action == "EXTENDS":
                # Additive enrichment: replace old with enriched content (same as UPDATE mechanically)
                content = (op.get("content") or "").strip()
                if not content or self._is_too_short(content):
                    continue
                target = self._resolve_by_number(op.get("memory_number"), index_map)
                tags = self._sanitize_tags(op.get("tags"))
                sanitized.append({"action": "EXTENDS", "content": content, "tags": tags, "target": target or {}})

            elif action == "MERGE":
                content = (op.get("content") or "").strip()
                if not content or self._is_too_short(content):
                    continue
                source_nums = op.get("sources") or []
                source_targets = [
                    t for num in source_nums
                    if (t := self._resolve_by_number(num, index_map)) is not None
                ]
                if not source_targets:
                    continue
                tags = self._sanitize_tags(op.get("tags"))
                sanitized.append({"action": "MERGE", "content": content, "tags": tags, "sources": source_targets})

        return sanitized

    # =========================================================================
    # LLM INTERFACE
    # =========================================================================

    async def _query_llm(
        self,
        system_prompt: str,
        user_prompt: str,
        user_id: str,
        __request__: Optional[Request],
        __user__: Optional[Dict[str, Any]],
        temperature: float = 0.1,
        max_tokens: int = 200,
    ) -> Optional[str]:
        try:
            return await self.circuit_breaker.call(
                self._execute_llm,
                system_prompt, user_prompt,
                __request__, __user__,
                temperature, max_tokens,
            )
        except CircuitBreakerOpenError:
            logger.warning("LLM skipped — circuit breaker OPEN")
            return None
        except Exception as exc:
            logger.error("LLM query failed: %s", exc)
            return None

    async def _execute_llm(
        self,
        system_prompt: str,
        user_prompt: str,
        __request__: Optional[Request],
        __user__: Optional[Dict[str, Any]],
        temperature: float,
        max_tokens: int,
    ) -> Optional[str]:
        request_obj = __request__ or Request(scope={"type": "http", "app": webui_app})
        response = await generate_chat_completion(
            request_obj,
            {
                "model": self.valves.llm_model,
                "messages": [
                    {"role": "system", "content": system_prompt},
                    {"role": "user",   "content": user_prompt},
                ],
                "stream": False,
                "temperature": temperature,
                "max_tokens": max_tokens,
            },
            user=__user__,
        )
        raw = response["choices"][0]["message"]["content"]
        return self._extract_json(raw)

    # =========================================================================
    # MEMORY I/O
    # =========================================================================

    async def _fetch_user_memories(self, user_id: str) -> List[Dict[str, Any]]:
        def _load() -> List[Any]:
            return list(Memories.get_memories_by_user_id(user_id) or [])

        records = await asyncio.to_thread(_load)
        snapshot: List[Dict[str, Any]] = []

        for item in records:
            metadata = getattr(item, "metadata", {}) or {}
            created_at = getattr(item, "created_at", None)
            created_iso: Optional[str] = None
            timestamp = 0.0

            if isinstance(created_at, datetime):
                created_iso = created_at.astimezone(timezone.utc).isoformat()
                timestamp = created_at.timestamp()
            elif isinstance(created_at, str):
                created_iso = created_at
                try:
                    timestamp = datetime.fromisoformat(
                        created_at.replace("Z", "+00:00")
                    ).timestamp()
                except ValueError:
                    pass

            snapshot.append({
                "id": getattr(item, "id", None),
                "content": (getattr(item, "content", "") or "").strip(),
                "metadata": metadata,
                "tags": metadata.get("tags") or [],
                "created_at": created_iso,
                "timestamp": timestamp,
            })

        snapshot.sort(key=lambda m: m.get("timestamp", 0.0), reverse=True)
        return snapshot

    async def _store_memory(
        self,
        content: str,
        tags: List[str],
        __user__: Dict[str, Any],
        __request__: Optional[Request],
    ) -> bool:
        try:
            valid_tags = [t for t in tags if t in TAG_CATEGORIES]
            metadata = {"tags": valid_tags} if valid_tags else {}
            form_data = AddMemoryForm(content=content, metadata=metadata)
            request_obj = __request__ or Request(scope={"type": "http", "app": webui_app})
            await add_memory(request=request_obj, form_data=form_data, user=__user__)
            return True
        except Exception as exc:
            logger.error("Store memory failed: %s", exc)
            return False

    # =========================================================================
    # UTILITY HELPERS
    # =========================================================================

    def _recency_score(self, memory: Dict[str, Any]) -> float:
        """
        Exponential decay recency score: 1.0 when brand-new → 0.2 at ≥ 1 year old.
        Used to blend with semantic relevance: final = (semantic × 0.7) + (recency × 0.3).
        """
        ts = memory.get("timestamp", 0.0)
        if not ts:
            return 0.5
        days_old = (time.time() - ts) / 86400.0
        return max(0.2, 1.0 - days_old / 365.0)

    async def _query_memory_vector(
        self,
        query: str,
        k: int,
        __user__: Optional[Dict[str, Any]],
        __request__: Optional[Request],
    ) -> List[Dict[str, Any]]:
        """
        Stage-1 retrieval: use OpenWebUI's native vector search to narrow
        the candidate pool before passing to the LLM reranker.
        Falls back gracefully to an empty list if embeddings are not configured
        or the API call fails.
        """
        try:
            request_obj = __request__ or Request(scope={"type": "http", "app": webui_app})
            response = await query_memory(
                request=request_obj,
                form_data=QueryMemoryForm(content=query, k=k),
                user=__user__,
            )
            raw = response.get("results", response) if isinstance(response, dict) else response
            if not isinstance(raw, list):
                return []
            snapshot: List[Dict[str, Any]] = []
            for item in raw:
                if isinstance(item, dict):
                    mid = item.get("id") or item.get("_id")
                    content = (item.get("content") or "").strip()
                    meta = item.get("metadata") or {}
                    created_at = item.get("created_at")
                else:
                    mid = getattr(item, "id", None)
                    content = (getattr(item, "content", "") or "").strip()
                    meta = getattr(item, "metadata", {}) or {}
                    created_at = getattr(item, "created_at", None)
                if not content:
                    continue
                ts = 0.0
                if isinstance(created_at, datetime):
                    ts = created_at.timestamp()
                elif isinstance(created_at, str):
                    try:
                        ts = datetime.fromisoformat(created_at.replace("Z", "+00:00")).timestamp()
                    except ValueError:
                        pass
                snapshot.append({
                    "id": mid,
                    "content": content,
                    "metadata": meta,
                    "tags": meta.get("tags") or [],
                    "created_at": created_at if isinstance(created_at, str) else None,
                    "timestamp": ts,
                })
            return snapshot
        except Exception as exc:
            logger.debug("Vector search unavailable, using full fetch: %s", exc)
            return []

    def _build_memory_inventory(
        self, memories: List[Dict[str, Any]]
    ) -> Tuple[str, Dict[int, Dict[str, Any]]]:
        lines: List[str] = []
        index_map: Dict[int, Dict[str, Any]] = {}
        total_chars = 0

        for idx, record in enumerate(memories, start=1):
            text = record.get("content", "")
            tags = ", ".join(record.get("tags", [])) or "none"
            date = (record.get("created_at") or "unknown")[:10]
            block = f"{idx}. [id:{record.get('id', 'n/a')}] tags=[{tags}] date={date}\n   {text}\n"
            if total_chars + len(block) > MAX_MEMORY_PROMPT_CHARS:
                break
            total_chars += len(block)
            lines.append(block)
            index_map[idx] = record

        if not lines:
            lines.append("(No stored memories)")

        return "".join(lines), index_map

    def _format_conversation(self, messages: List[Dict[str, Any]], last_n: int) -> str:
        lines: List[str] = []
        for msg in messages[-last_n:]:
            role = msg.get("role", "")
            content = (msg.get("content") or "").strip()
            if role and content:
                lines.append(f"{role.capitalize()}: {content[:1200]}")
        return "\n".join(lines)

    def _format_memory_injection(self, selected: List[Dict[str, Any]]) -> str:
        """
        Dual-block injection format validated by mem0, Supermemory, and MemGPT research:

        <user_profile>   — stable identity/behavior facts always present so the
                           assistant knows who the user is regardless of query.
        <relevant_memories> — query-specific facts surfaced for this turn only.

        Placing this block at the START of the system prompt gives it maximum
        attention weight.
        """
        profile_tag_set = {"identity", "behavior"}
        profile: List[str] = []
        relevant: List[str] = []

        for item in selected:
            content = item["record"].get("content", "")
            tags = set(item["record"].get("tags") or [])
            if tags & profile_tag_set:
                profile.append(content)
            else:
                relevant.append(content)

        parts: List[str] = []
        if profile:
            parts.append("<user_profile>")
            parts.extend(f"- {c}" for c in profile)
            parts.append("</user_profile>")
        if relevant:
            parts.append("<relevant_memories>")
            parts.extend(f"- {c}" for c in relevant)
            parts.append("</relevant_memories>")

        return "\n".join(parts)

    def _is_duplicate(self, content: str, memories: List[Dict[str, Any]]) -> bool:
        """Exact match + Jaccard word-overlap ≥ 0.90 duplicate guard."""
        normalized = content.strip().lower()
        new_words = set(re.split(r"\W+", normalized)) - {""}
        for record in memories:
            existing = (record.get("content") or "").strip().lower()
            if existing == normalized:
                return True
            if len(new_words) >= 4:
                existing_words = set(re.split(r"\W+", existing)) - {""}
                if existing_words:
                    union = new_words | existing_words
                    jaccard = len(new_words & existing_words) / len(union)
                    if jaccard >= 0.90:
                        return True
        return False

    def _is_too_short(self, content: Optional[str]) -> bool:
        return not content or len(content.strip()) < 8

    def _sanitize_tags(self, raw: Any) -> List[str]:
        if not isinstance(raw, list):
            return []
        return [str(t).strip().lower() for t in raw if str(t).strip()]

    def _resolve_by_number(
        self, num: Any, index_map: Dict[int, Dict[str, Any]]
    ) -> Optional[Dict[str, Any]]:
        try:
            return index_map.get(int(num))
        except (TypeError, ValueError):
            return None

    def _build_memory_record(self, content: str, tags: List[str]) -> Dict[str, Any]:
        now = datetime.now(timezone.utc)
        return {
            "id": None,
            "content": content,
            "metadata": {"tags": tags},
            "tags": tags,
            "created_at": now.isoformat(),
            "timestamp": now.timestamp(),
        }

    def _extract_json(self, text: str) -> str:
        """
        Robustly strip prose and markdown fences from LLM output,
        returning just the raw JSON string.
        """
        # ```json ... ``` or ``` ... ```
        m = re.search(r"```(?:json)?\s*(\{[\s\S]*?\}|\[[\s\S]*?\])\s*```", text)
        if m:
            return m.group(1)
        # Bare { ... } or [ ... ] block
        m = re.search(r"(\{[\s\S]*\}|\[[\s\S]*\])", text)
        if m:
            return m.group(1)
        return text

    async def _emit_status(
        self,
        emitter: Optional[Callable[[Any], Awaitable[None]]],
        message: str,
        level: Literal["brief", "detailed"] = "brief",
        done: bool = False,
    ) -> None:
        if emitter is None:
            return
        levels = {"silent": 0, "brief": 1, "detailed": 2}
        if levels.get(self.valves.status_mode, 1) < levels.get(level, 1):
            return
        try:
            await emitter({
                "type": "status",
                "data": {
                    "description": message,
                    "status": "complete" if done else "in_progress",
                    "done": done,
                    "error": False,
                },
            })
        except Exception as exc:
            logger.error("Status emit failed: %s", exc)