""" Memory Layer fuer Mission Control. Persistentes Gedaechtnis fuer lokale LLM-Agenten (Cline, Claude Code, OpenCode). SQLite aus stdlib — kein Vektor-Overhead, keine neue Abhaengigkeit. Alle MCP-Tools teilen denselben Speicher via mcp_memory.py-Wrapper. """ import re import sqlite3 import uuid from datetime import datetime, timezone from difflib import SequenceMatcher from enum import Enum from typing import Optional from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from auth import auth from config import MEMORY_DB router = APIRouter(prefix="/api", dependencies=[Depends(auth)]) _db_conn: Optional[sqlite3.Connection] = None def _db() -> sqlite3.Connection: global _db_conn if _db_conn is None: MEMORY_DB.parent.mkdir(parents=True, exist_ok=True) _db_conn = sqlite3.connect(str(MEMORY_DB), check_same_thread=False) _db_conn.row_factory = sqlite3.Row _db_conn.execute("PRAGMA journal_mode=WAL") # sicherer bei concurrent FastAPI-Threads _db_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 ) """) # Temporaere Eintraege nach 7 Tagen automatisch loeschen _db_conn.execute( "DELETE FROM memories WHERE category = 'ephemeral'" " AND datetime(created_at) < datetime('now', '-7 days')" ) _db_conn.commit() return _db_conn class _MemCategory(str, Enum): user = "user" instruction = "instruction" stable = "stable" versioned = "versioned" ephemeral = "ephemeral" class _MemIn(BaseModel): content: str category: _MemCategory = _MemCategory.stable source: str = "manual" class _MemUp(BaseModel): content: Optional[str] = None category: Optional[_MemCategory] = None def _row(r: sqlite3.Row) -> dict: return dict(r) # Export muss vor /{mid} stehen, sonst wuerde FastAPI "export" als ID behandeln @router.get("/memory/export") def export_memories(): """Alle Eintraege als strukturierter Plain-Text — direkt als System-Prompt-Kontext nutzbar.""" db = _db() rows = db.execute( "SELECT * FROM memories ORDER BY category, updated_at DESC" ).fetchall() cat_names = { "stable": "Stabile Fakten", "versioned": "Versionierte Infos", "ephemeral": "Temporaerer Kontext", } lines = ["# Mission Control — Gedaechtnis\n"] current = "" for r in rows: cat = cat_names.get(r["category"], r["category"]) if cat != current: lines.append(f"\n## {cat}\n") current = cat 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: """Normalisiert Text fuer Dubletten-Vergleich: klein, ohne Satzzeichen, Whitespace kollabiert.""" s = re.sub(r"[^\w\s]", " ", s.lower(), flags=re.UNICODE) return re.sub(r"\s+", " ", s).strip() class _DedupeIn(BaseModel): apply: bool = False # False = Dry-Run (nur Vorschau), True = wirklich loeschen threshold: float = 0.85 # Aehnlichkeits-Schwelle (SequenceMatcher) @router.post("/memory/dedupe") def dedupe_memories(body: _DedupeIn): """Deterministischer Kurator: findet Dubletten (exakt / enthalten / aehnlich) je Kategorie, behaelt den vollstaendigsten (laengsten) Eintrag, entfernt die uebrigen. KEIN LLM. Konservativ: nur innerhalb derselben Kategorie, Containment braucht vollstaendige Teilstring-Deckung.""" db = _db() # laengste zuerst -> der vollstaendigste Eintrag einer Gruppe wird zum Kanon 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 contained = nb in na or na in nb ratio = SequenceMatcher(None, na, nb).ratio() if contained or ratio >= body.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 body.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": body.apply, } @router.get("/memory") def list_memories(q: str = "", category: str = ""): db = _db() sql = "SELECT * FROM memories" params: list = [] conds: list = [] 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 [_row(r) for r in db.execute(sql, params).fetchall()] @router.post("/memory", status_code=201) def add_memory(body: _MemIn): now = datetime.now(timezone.utc).isoformat() mid = str(uuid.uuid4()) db = _db() db.execute( "INSERT INTO memories (id, content, category, source, created_at, updated_at)" " VALUES (?,?,?,?,?,?)", (mid, body.content.strip(), body.category, body.source, now, now), ) db.commit() return { "id": mid, "content": body.content.strip(), "category": body.category, "source": body.source, "created_at": now, "updated_at": now, } @router.put("/memory/{mid}") def update_memory(mid: str, body: _MemUp): db = _db() row = db.execute("SELECT * FROM memories WHERE id = ?", (mid,)).fetchone() if not row: raise HTTPException(404, "Eintrag nicht gefunden") now = datetime.now(timezone.utc).isoformat() content = body.content.strip() if body.content is not None else row["content"] category = body.category if body.category is not None else row["category"] db.execute( "UPDATE memories SET content=?, category=?, updated_at=? WHERE id=?", (content, category, now, mid), ) db.commit() return { "id": mid, "content": content, "category": category, "source": row["source"], "created_at": row["created_at"], "updated_at": now, } @router.delete("/memory/{mid}") def delete_memory(mid: str): db = _db() if not db.execute("SELECT id FROM memories WHERE id = ?", (mid,)).fetchone(): raise HTTPException(404, "Eintrag nicht gefunden") db.execute("DELETE FROM memories WHERE id = ?", (mid,)) db.commit() return {"ok": True}