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

"""
Geteiltes Gedächtnis (die „Verfassung") — SQLite aus stdlib, WAL-Mode.
Portiert aus Mission Control v1 (routers/memory.py), DB-Logik als Service isoliert.

5 Kategorien: user · instruction · stable · versioned · ephemeral (7-Tage-TTL).
Dedupe = deterministischer Kurator (exakt/enthalten/ähnlich), KEIN LLM.
"""

import re
import sqlite3
import uuid
from datetime import datetime, timezone
from difflib import SequenceMatcher

from config import MEMORY_DB

CATEGORIES = ("user", "instruction", "stable", "versioned", "ephemeral")

_conn: sqlite3.Connection | None = None


def db() -> sqlite3.Connection:
    global _conn
    if _conn is None:
        MEMORY_DB.parent.mkdir(parents=True, exist_ok=True)
        _conn = sqlite3.connect(str(MEMORY_DB), check_same_thread=False)
        _conn.row_factory = sqlite3.Row
        _conn.execute("PRAGMA journal_mode=WAL")
        _conn.execute("""
            CREATE TABLE IF NOT EXISTS memories (
                id         TEXT PRIMARY KEY,
                content    TEXT NOT NULL,
                category   TEXT NOT NULL DEFAULT 'stable',
                source     TEXT NOT NULL DEFAULT 'manual',
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL
            )
        """)
        _conn.execute(
            "DELETE FROM memories WHERE category='ephemeral'"
            " AND datetime(created_at) < datetime('now','-7 days')"
        )
        _conn.commit()
    return _conn


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def list_memories(q: str = "", category: str = "") -> list[dict]:
    sql, params, conds = "SELECT * FROM memories", [], []
    if q:
        conds.append("content LIKE ?"); params.append(f"%{q}%")
    if category:
        conds.append("category = ?"); params.append(category)
    if conds:
        sql += " WHERE " + " AND ".join(conds)
    sql += " ORDER BY created_at DESC"
    return [dict(r) for r in db().execute(sql, params).fetchall()]


def add_memory(content: str, category: str = "stable", source: str = "manual") -> dict:
    now, mid = _now(), str(uuid.uuid4())
    db().execute(
        "INSERT INTO memories (id,content,category,source,created_at,updated_at) VALUES (?,?,?,?,?,?)",
        (mid, content.strip(), category, source, now, now),
    )
    db().commit()
    return {"id": mid, "content": content.strip(), "category": category,
            "source": source, "created_at": now, "updated_at": now}


def update_memory(mid: str, content: str | None = None, category: str | None = None) -> dict | None:
    row = db().execute("SELECT * FROM memories WHERE id=?", (mid,)).fetchone()
    if not row:
        return None
    now = _now()
    new_content = content.strip() if content is not None else row["content"]
    new_cat = category if category is not None else row["category"]
    db().execute("UPDATE memories SET content=?,category=?,updated_at=? WHERE id=?",
                 (new_content, new_cat, now, mid))
    db().commit()
    return {"id": mid, "content": new_content, "category": new_cat,
            "source": row["source"], "created_at": row["created_at"], "updated_at": now}


def delete_memory(mid: str) -> bool:
    if not db().execute("SELECT id FROM memories WHERE id=?", (mid,)).fetchone():
        return False
    db().execute("DELETE FROM memories WHERE id=?", (mid,))
    db().commit()
    return True


def export_text() -> dict:
    rows = db().execute("SELECT * FROM memories ORDER BY category, updated_at DESC").fetchall()
    lines = ["# Mission Control — Gedächtnis\n"]
    current = ""
    for r in rows:
        if r["category"] != current:
            lines.append(f"\n## {r['category']}\n")
            current = r["category"]
        lines.append(f"- {r['content']}  _(Quelle: {r['source']}, {r['updated_at'][:10]})_")
    return {"text": "\n".join(lines), "count": len(rows)}


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:
    """Deterministischer Kurator: findet Dubletten (exakt/enthalten/ähnlich) je Kategorie,
    behält den vollständigsten (längsten) Eintrag. Konservativ, kein LLM."""
    rows = [dict(r) for r in db().execute(
        "SELECT * FROM memories ORDER BY length(content) DESC, created_at ASC").fetchall()]
    used: set[str] = set()
    groups: list[dict] = []
    for i, a in enumerate(rows):
        if a["id"] in used:
            continue
        na = _norm(a["content"])
        if not na:
            continue
        dups = []
        for b in rows[i + 1:]:
            if b["id"] in used or b["category"] != a["category"]:
                continue
            nb = _norm(b["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["content"], "category": a["category"]},
                "remove": [{"id": d["id"], "content": d["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"]:
                db().execute("DELETE FROM memories WHERE id=?", (d["id"],)); removed += 1
        if removed:
            db().commit()
    return {"groups": groups, "duplicate_count": dup_count, "removed": removed, "applied": apply}