mission-control-v2/backend/services/memory.py

"""
Geteiltes Gedächtnis (die „Verfassung") — jetzt auto-lernend & semantisch über Mem0.

Dieser Service ist nur noch ein dünner HTTP-Client auf den Mem0-Sidecar (mem0_service/app.py,
läuft im ~/.mem0/venv unter Python 3.12). Die `/api/memory`-API-Form bleibt unverändert, damit
UI und MCP-Server kompatibel bleiben. Neu gegenüber der alten flachen SQLite:

- search (q gesetzt) ist SEMANTISCH (Vektor/Embeddings) statt LIKE-Textsuche, mit Relevanz-Score.
- learn() reicht Gesprächs-Turns durch → Mem0 EXTRAHIERT Fakten selbst (Auto-Lernen).
- Dedup macht Mem0 beim Auto-Lernen selbst; der manuelle Kurator unten bleibt als Komfort.

5 Kategorien (user · instruction · stable · versioned · ephemeral) bleiben als Metadaten erhalten.
"""

import re
from difflib import SequenceMatcher

import httpx

from config import MEM0_SERVICE_URL

CATEGORIES = ("identity", "knowledge", "rules", "events")

_TIMEOUT = httpx.Timeout(60.0, connect=5.0)  # LLM-Extraktion kann ein paar Sekunden dauern


def _get(path: str, **params) -> list | dict:
    r = httpx.get(f"{MEM0_SERVICE_URL}{path}", params=params, timeout=_TIMEOUT)
    r.raise_for_status()
    return r.json()


def _post(path: str, data: dict) -> dict:
    r = httpx.post(f"{MEM0_SERVICE_URL}{path}", json=data, timeout=_TIMEOUT)
    r.raise_for_status()
    return r.json()


def _put(path: str, data: dict) -> dict:
    r = httpx.put(f"{MEM0_SERVICE_URL}{path}", json=data, timeout=_TIMEOUT)
    r.raise_for_status()
    return r.json()


def _delete(path: str) -> dict:
    r = httpx.delete(f"{MEM0_SERVICE_URL}{path}", timeout=_TIMEOUT)
    r.raise_for_status()
    return r.json()


def list_memories(q: str = "", category: str = "") -> list[dict]:
    """Alle Fakten oder — wenn q gesetzt — die semantisch ähnlichsten (mit `score`)."""
    return _get("/memory", **{k: v for k, v in (("q", q), ("category", category)) if v})


def add_memory(content: str, category: str = "stable", source: str = "manual") -> dict:
    """Einen Fakt VERBATIM speichern (keine LLM-Umformung). Auto-Lernen → learn()."""
    return _post("/memory", {"content": content.strip(), "category": category, "source": source})


def update_memory(mid: str, content: str | None = None, category: str | None = None) -> dict | None:
    try:
        return _put(f"/memory/{mid}", {"content": content, "category": category})
    except httpx.HTTPStatusError as exc:
        if exc.response.status_code == 404:
            return None
        raise


def delete_memory(mid: str) -> bool:
    try:
        _delete(f"/memory/{mid}")
        return True
    except httpx.HTTPStatusError as exc:
        if exc.response.status_code == 404:
            return False
        raise


def learn(text: str | None = None, messages: list[dict] | None = None,
          source: str = "auto", category: str = "stable") -> dict:
    """Auto-Lernen: Text/Gesprächs-Turns durchreichen → Mem0 extrahiert die Fakten selbst."""
    return _post("/learn", {"text": text, "messages": messages,
                            "source": source, "category": category})


def graph(min_score: float = 0.45, top_k: int = 3) -> dict:
    """Fakten als Ähnlichkeits-Graph (Knoten + semantische Kanten) für die UI-Visualisierung."""
    return _get("/graph", min_score=min_score, top_k=top_k)


def export_text() -> dict:
    rows = sorted(list_memories(), key=lambda r: (r.get("category", ""), r.get("updated_at", "")))
    lines = ["# Mission Control — Gedächtnis\n"]
    current = ""
    for r in rows:
        if r.get("category") != current:
            current = r.get("category", "")
            lines.append(f"\n## {current}\n")
        src = r.get("source", "")
        when = (r.get("updated_at") or "")[:10]
        lines.append(f"- {r.get('content', '')}  _(Quelle: {src}, {when})_")
    return {"text": "\n".join(lines), "count": len(rows)}


# --- Manueller Kurator (deterministisch, kein LLM) ---------------------------
def _norm(s: str) -> str:
    s = re.sub(r"[^\w\s]", " ", s.lower(), flags=re.UNICODE)
    return re.sub(r"\s+", " ", s).strip()


def dedupe(apply: bool = False, threshold: float = 0.85) -> dict:
    """Findet Dubletten (exakt/enthalten/ähnlich) je Kategorie, behält den längsten
    Eintrag. Mem0 dedupliziert beim Auto-Lernen schon semantisch — das hier ist der
    manuelle Komfort-Knopf fürs UI (z.B. nach vielen Verbatim-Importen)."""
    rows = sorted(list_memories(), key=lambda r: (-len(r.get("content", "")), r.get("created_at", "")))
    used: set[str] = set()
    groups: list[dict] = []
    for i, a in enumerate(rows):
        if a["id"] in used:
            continue
        na = _norm(a.get("content", ""))
        if not na:
            continue
        dups = []
        for b in rows[i + 1:]:
            if b["id"] in used or b.get("category") != a.get("category"):
                continue
            nb = _norm(b.get("content", ""))
            if not nb:
                continue
            if nb in na or na in nb or SequenceMatcher(None, na, nb).ratio() >= threshold:
                dups.append(b); used.add(b["id"])
        if dups:
            used.add(a["id"])
            groups.append({
                "keep": {"id": a["id"], "content": a.get("content"), "category": a.get("category")},
                "remove": [{"id": d["id"], "content": d.get("content")} for d in dups],
            })
    dup_count = sum(len(g["remove"]) for g in groups)
    removed = 0
    if apply:
        for g in groups:
            for d in g["remove"]:
                if delete_memory(d["id"]):
                    removed += 1
    return {"groups": groups, "duplicate_count": dup_count, "removed": removed, "applied": apply}