mission-control-v2/hermes/plugins/mc2-memory/__init__.py

"""
MC2 Memory — Hermes-Memory-Provider, der ans geteilte **Mem0-Gedächtnis** hängt (über MC2 :9001,
das den Mem0-Sidecar :8765 proxyt). Macht das Gedächtnis für Hermes **hands-off**:

- `sync_turn`  → nach jedem Turn werden die Turn-Messages an `/api/memory/learn` geschickt;
  Mem0 EXTRAHIERT dauerhafte Fakten selbst (infer=True), dedupliziert semantisch. Nicht-blockierend
  (Hintergrund-Worker), damit der Agent nie auf die LLM-Extraktion wartet.
- `prefetch`   → vor jedem Turn semantische Suche gegen die Nutzer-Message; relevante Fakten werden
  als Kontext eingeblendet (automatischer Recall).

**Context-only:** `get_tool_schemas()` liefert `[]` → der Agent bekommt KEINE Memory-Tools. Damit
entfällt das Tool-Loop-Risiko, wegen dem das native Memory-Toolset abgeschaltet ist
(`agent.disabled_toolsets: [..., memory]` bleibt). Single Source of Truth bleibt der MC2-Sidecar
(NoThink-Fix + alleiniger Chroma-Besitzer) — dieser Provider ist nur ein dünner HTTP-Client.

Aktivierung (in ~/.hermes/config.yaml):
    memory:
      memory_enabled: true
      provider: mc2-memory
"""

from __future__ import annotations

import logging
import os
import queue
import re
import threading
import time
from typing import Any, Dict, List, Optional

import httpx

from agent.memory_provider import MemoryProvider

log = logging.getLogger(__name__)

MC_URL = os.environ.get("MC_URL", "http://127.0.0.1:9001").rstrip("/")
MC_TOKEN = os.environ.get("MC_TOKEN", "")
RECALL_LIMIT = int(os.environ.get("MC2_MEMORY_RECALL_LIMIT", "5"))
# 0.5 statt 0.3: nur starke Treffer einblenden → weniger Rausch-Fakten UND ein stabilerer
# Prompt-Prefix (oft leer), damit --cache-reuse über Turns greift.
RECALL_MIN_SCORE = float(os.environ.get("MC2_MEMORY_RECALL_MIN_SCORE", "0.5"))
# 25 statt 12: triviale Kurz-Turns (Begrüßung/Quittung) gar nicht erst zum Lernen schicken.
MIN_USER_LEN = int(os.environ.get("MC2_MEMORY_MIN_USER_LEN", "25"))  # zu Kurzes nicht lernen
# Anti-Memory-Poisoning (Stufe 0 Security): aus Turns mit untrusted Bildschirm-/Web-Inhalt NICHT
# auto-lernen. Lucys Bildschirm-Sicht prependet diesen Marker an die User-Message (voice.py); auf dem
# Schirm sichtbarer Text könnte Injection-Anweisungen tragen, die sonst DAUERHAFT ins Gedächtnis
# wandern und Tage später ungefragt wiederverwendet werden.
SKIP_UNTRUSTED = os.environ.get("MC2_MEMORY_SKIP_UNTRUSTED", "1") not in ("0", "false", "False", "no")
UNTRUSTED_MARKERS = ("[Bildschirm-Sicht",)
# Lern-Buendelung (Phase 2, 19.07.2026): Die LLM-Extraktion lief bisher nach JEDEM Turn und
# konkurrierte dabei mit Lucys Hirn um die Slots (gleiches Modell, --parallel 2). Jetzt werden
# Turns gesammelt und als EIN Gespraechs-Block extrahiert: nach BATCH_TURNS Turns oder wenn
# BATCH_IDLE_S Sekunden Ruhe war. Ein Batch = ein Extraktionslauf statt N.
BATCH_TURNS = max(1, int(os.environ.get("MC2_MEMORY_BATCH_TURNS", "4")))
BATCH_IDLE_S = float(os.environ.get("MC2_MEMORY_BATCH_IDLE", "180"))

# Triviale Turns, die keinen dauerhaften Fakt tragen (reine Begrüßung/Quittung, Aufwärm-/Test-
# Kommandos) → gar nicht erst zum Auto-Lernen schicken. Spart Extraktions-Läufe und verhindert
# Müll-Fakten aus Meta-/Ablauf-Turns.
_TRIVIAL_TURN_RE = re.compile(
    r"^\s*(?:(?:hi|hallo|hey|moin|servus|guten\s+(?:morgen|tag|abend)|danke\w*|"
    r"ok(?:ay)?|alles\s+klar|passt|super|top|cool|nice|ja|nein|jo|jup|jap)[\s.,!?]*)+$"
    r"|^\s*(?:aufw(?:ä|ae)rmen|/?test)\b",
    re.IGNORECASE,
)


def _is_trivial_turn(text: str) -> bool:
    return bool(_TRIVIAL_TURN_RE.search(text or ""))


def _headers() -> dict:
    return {"X-MC-Token": MC_TOKEN} if MC_TOKEN else {}


class MC2MemoryProvider(MemoryProvider):
    def __init__(self) -> None:
        self._session_id = ""
        self._agent_context = "primary"
        self._q: "queue.Queue[list]" = queue.Queue(maxsize=200)
        self._worker: Optional[threading.Thread] = None
        self._stop = threading.Event()
        self._recall_cache: Dict[str, str] = {}
        # Turn-Puffer der Lern-Buendelung (siehe BATCH_TURNS/BATCH_IDLE_S).
        self._buf: List[List[Dict[str, str]]] = []
        self._buf_lock = threading.Lock()
        self._buf_last_add = 0.0

    # -- Identität / Verfügbarkeit -------------------------------------------
    def name(self) -> str:
        return "mc2-memory"

    def is_available(self) -> bool:
        try:
            r = httpx.get(f"{MC_URL}/api/health", timeout=3.0)
            return r.status_code == 200
        except Exception:
            return False

    def initialize(self, session_id: str, **kwargs) -> None:
        self._session_id = session_id
        # Nur im primären Agent-Kontext lernen (cron/subagent-Systemprompts nicht einlernen).
        self._agent_context = kwargs.get("agent_context", "primary")
        if self._worker is None or not self._worker.is_alive():
            self._stop.clear()
            self._worker = threading.Thread(target=self._run, name="mc2-memory-learn", daemon=True)
            self._worker.start()

    def system_prompt_block(self) -> str:
        return (
            "Du hast ein dauerhaftes, geteiltes Langzeitgedächtnis (MC2/Mem0). Relevante Fakten "
            "werden dir vor einem Turn automatisch eingeblendet; neue dauerhafte Fakten über den "
            "Nutzer und das Projekt werden nach dem Turn automatisch gelernt — du musst dafür "
            "nichts tun."
        )

    # -- Recall (vor dem Turn) -----------------------------------------------
    def prefetch(self, query: str, *, session_id: str = "") -> str:
        q = (query or "").strip()
        if len(q) < 3:
            return ""
        if q in self._recall_cache:
            return self._recall_cache[q]
        try:
            r = httpx.get(f"{MC_URL}/api/memory", params={"q": q}, headers=_headers(), timeout=5.0)
            r.raise_for_status()
            items = r.json()
        except Exception as exc:  # nie den Turn blockieren/abbrechen
            log.debug("mc2-memory prefetch failed: %s", exc)
            return ""
        facts = [
            i for i in items
            if i.get("score") is None or float(i.get("score", 0)) >= RECALL_MIN_SCORE
        ][:RECALL_LIMIT]
        out = ""
        if facts:
            lines = "\n".join(f"- {f.get('content', '')}" for f in facts)
            out = f"Relevante Fakten aus dem Langzeitgedächtnis:\n{lines}"
        # Cache klein halten
        if len(self._recall_cache) > 64:
            self._recall_cache.clear()
        self._recall_cache[q] = out
        return out

    # -- Lernen (nach dem Turn) ----------------------------------------------
    def sync_turn(self, user_content: str, assistant_content: str, *,
                  session_id: str = "", messages: Optional[List[Dict[str, Any]]] = None) -> None:
        if self._agent_context != "primary":
            return
        u = (user_content or "").strip()
        if len(u) < MIN_USER_LEN:
            return
        if _is_trivial_turn(u):
            log.debug("mc2-memory: trivialer Turn (Begrüßung/Quittung/Aufwärmen) — Auto-Lernen übersprungen")
            return
        if SKIP_UNTRUSTED and any(m in u for m in UNTRUSTED_MARKERS):
            log.debug("mc2-memory: Turn mit untrusted Bildschirm-/Web-Inhalt — Auto-Lernen übersprungen")
            return
        msgs: List[Dict[str, str]] = [{"role": "user", "content": u}]
        a = (assistant_content or "").strip()
        # Hermes ≥0.18 liefert optional den vollen Turn-Kontext (`messages`, inkl. Tool-Calls).
        # Bewusst NUR die Tool-NAMEN mitlernen ("hat X per Tool Y geprüft") — Tool-ERGEBNISSE
        # sind untrusted (Web-Inhalte!) und wären ein Poisoning-Vektor am Junk-Guard vorbei.
        if messages:
            tools = []
            for m in messages:
                for tc in (m.get("tool_calls") or []):
                    name = ((tc.get("function") or {}).get("name") or "").strip()
                    if name and name not in tools:
                        tools.append(name)
            if tools:
                a = (a + f"\n[genutzte Tools: {', '.join(tools[:6])}]").strip()
        if a:
            msgs.append({"role": "assistant", "content": a[:4000]})
        with self._buf_lock:
            self._buf.append(msgs)
            self._buf_last_add = time.time()
            full = len(self._buf) >= BATCH_TURNS
        if full:
            self._drain_buffer()
        # Neue Fakten können gelandet sein → Recall-Cache invalidieren.
        self._recall_cache.clear()

    def _drain_buffer(self) -> None:
        """Gepufferte Turns als EINEN Gespraechs-Block in die Lern-Queue geben."""
        with self._buf_lock:
            if not self._buf:
                return
            merged: List[Dict[str, str]] = []
            for turn in self._buf:
                merged.extend(turn)
            self._buf.clear()
        try:
            self._q.put_nowait(merged)
        except queue.Full:
            log.debug("mc2-memory learn queue full — Batch übersprungen")

    def _post_learn(self, msgs: list) -> None:
        try:
            httpx.post(f"{MC_URL}/api/memory/learn",
                       json={"messages": msgs, "source": "hermes"},
                       headers=_headers(), timeout=90.0)
        except Exception as exc:
            log.debug("mc2-memory learn POST failed: %s", exc)

    def _run(self) -> None:
        while not self._stop.is_set():
            try:
                msgs = self._q.get(timeout=5.0)
            except queue.Empty:
                # Idle-Flush: Ruhe im Gespraech → angesammelte Turns jetzt lernen.
                with self._buf_lock:
                    idle = (self._buf and
                            time.time() - self._buf_last_add >= BATCH_IDLE_S)
                if idle:
                    self._drain_buffer()
                continue
            try:
                self._post_learn(msgs)
            finally:
                self._q.task_done()

    def _flush(self) -> None:
        """Offene Turns garantiert rausschreiben — bei Session-Ende/CLI-Exit, wo der
        Prozess sofort beendet wird (daemon-Worker würde sonst mitten im POST sterben)."""
        self._drain_buffer()
        self._stop.set()
        if self._worker and self._worker.is_alive():
            self._worker.join(timeout=95.0)  # laufenden Worker-POST zu Ende lassen
        while True:                          # vom Worker nicht mehr abgeholte Turns
            try:
                msgs = self._q.get_nowait()
            except queue.Empty:
                break
            self._post_learn(msgs)

    # -- Context-only: keine Agent-Tools → kein Tool-Loop --------------------
    def get_tool_schemas(self) -> List[Dict[str, Any]]:
        return []

    def on_session_end(self, messages: List[Dict[str, Any]]) -> None:
        self._flush()

    def shutdown(self) -> None:
        self._flush()


def register(ctx) -> None:
    ctx.register_memory_provider(MC2MemoryProvider())