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

"""Eigenleben — was die Box von allein gebaut und vorgeschlagen hat.

Beantwortet die User-Frage vom 15.07.2026: „Es gibt keinen Viewpoint, welche Skills
erstellt wurden, was die Box vorgeschlagen hat, welche Skills Lucy hat und wie die
funktionieren." Reine LESE-Schicht, keine Schreib-Operationen:

- Skills aus ~/.hermes/skills (= Lucys Skill-Satz; die Worker-Profile werkstatt/betrieb
  tragen Kopien desselben Satzes). Herkunft dreistufig klassifiziert:
    selbst   → weder mitgeliefert noch aus unserem Repo = die Box/Lucy hat ihn erzeugt
    repo     → liegt in ~/mission-control-v2/deploy/skills (von uns gebaut + deployt)
    bundled  → liegt in ~/.hermes/hermes-agent/skills (kam mit Hermes mit)
- Nutzungszahlen aus ~/.hermes/skills/.usage.json (Hermes' eigener Zähler).
- Bilanz aus Auftragsbuch-Chronik (/srv/models/mc2-auftragsbuch.json, angenommene
  Branches) + Ablehnungs-Lern-Journal (/srv/models/mc2-ablehnungen.jsonl).
"""

from __future__ import annotations

import json
import re
import subprocess
import time
from pathlib import Path

SKILLS_DIR = Path.home() / ".hermes" / "skills"
BUNDLED_DIR = Path.home() / ".hermes" / "hermes-agent" / "skills"
REPO_DIR = Path.home() / "mission-control-v2" / "deploy" / "skills"
HERMES_SRC = Path.home() / ".hermes" / "hermes-agent"
HERMES_PY = HERMES_SRC / "venv" / "bin" / "python"
USAGE_FILE = SKILLS_DIR / ".usage.json"
AUFTRAGSBUCH_FILE = Path("/srv/models/mc2-auftragsbuch.json")
ABLEHNUNGEN_FILE = Path("/srv/models/mc2-ablehnungen.jsonl")

_HERKUNFT_LABEL = {
    "selbst": "Selbst erstellt",
    "repo": "Von uns gebaut",
    "bundled": "Mit Hermes mitgeliefert",
}


def _frontmatter(text: str) -> dict:
    """Sehr kleiner Frontmatter-Leser: nur die flachen key: value-Zeilen des ersten
    ---Blocks (name/description/version/author reichen hier; kein YAML-Import nötig)."""
    out: dict = {}
    if not text.startswith("---"):
        return out
    for line in text.split("\n", 1)[1].split("\n"):
        if line.strip() == "---":
            break
        m = re.match(r"^([A-Za-z_][\w-]*):\s*(.*)$", line)
        if not m:
            continue
        val = m.group(2).strip().strip('"').strip("'")
        if val:
            out[m.group(1)] = val
    return out


def _norm(s: str) -> str:
    return re.sub(r"[\s_-]+", "", (s or "").lower())


# aktiv/inaktiv: dieselbe Wahrheit wie Hermes selbst (und sein eingebautes WebUI) —
# hermes_cli.banner.get_available_skills() prüft Plattform/Voraussetzungen/Config und
# liefert, was der Agent WIRKLICH laden kann. Wir rufen sie im Hermes-venv auf (MC2
# läuft in eigenem Python) und cachen großzügig — der Skill-Satz ändert sich selten.
_aktiv_cache: dict = {"ts": 0.0, "namen": None}
_AKTIV_EVERY = 600.0  # s


def _aktive_namen() -> set[str] | None:
    """Normalisierte Namen aller ladbaren Skills; None = nicht ermittelbar (Dev-Modus)."""
    if not HERMES_PY.is_file():
        return None
    now = time.time()
    if _aktiv_cache["namen"] is not None and now - _aktiv_cache["ts"] < _AKTIV_EVERY:
        return _aktiv_cache["namen"]
    code = ("from hermes_cli.banner import get_available_skills;import json;"
            "print(json.dumps(get_available_skills()))")
    try:
        r = subprocess.run([str(HERMES_PY), "-c", code], capture_output=True,
                           text=True, timeout=60, cwd=str(HERMES_SRC))
        idx = r.stdout.find("{")
        kategorien = json.loads(r.stdout[idx:]) if idx >= 0 else {}
        namen = {_norm(n) for liste in kategorien.values() for n in liste}
    except Exception:
        return _aktiv_cache["namen"]  # alter Stand ist besser als flackerndes None
    _aktiv_cache.update(ts=now, namen=namen)
    return namen


def list_skills() -> list[dict]:
    """Alle Skills mit Beschreibung, Herkunft und Nutzungszahlen (leer im Dev-Modus)."""
    if not SKILLS_DIR.is_dir():
        return []
    try:
        usage_raw = json.loads(USAGE_FILE.read_text(encoding="utf-8"))
    except Exception:
        usage_raw = {}
    usage = {_norm(k): v for k, v in usage_raw.items()}
    aktive = _aktive_namen()

    def eintrag(md: Path, rel: str, kategorie: str | None) -> dict:
        try:
            fm = _frontmatter(md.read_text(encoding="utf-8", errors="replace"))
        except Exception:
            fm = {}
        if (BUNDLED_DIR / rel).is_dir():
            herkunft = "bundled"
        elif (REPO_DIR / rel).is_dir():
            herkunft = "repo"
        else:
            herkunft = "selbst"
        kurz = rel.split("/")[-1]
        u = usage.get(_norm(kurz)) or usage.get(_norm(fm.get("name", ""))) or {}
        aktiv = None
        if aktive is not None:
            aktiv = _norm(kurz) in aktive or _norm(fm.get("name", "")) in aktive
        return {
            "id": rel,
            "name": fm.get("name") or kurz,
            "kategorie": kategorie,
            "aktiv": aktiv,
            "beschreibung": fm.get("description") or "",
            "version": fm.get("version") or "",
            "autor": fm.get("author") or "",
            "herkunft": herkunft,
            "herkunft_label": _HERKUNFT_LABEL[herkunft],
            "genutzt": int(u.get("use_count") or 0),
            "zuletzt_genutzt": u.get("last_used_at") or None,
            "erstellt": u.get("created_at") or None,
            "status": u.get("state") or "active",
            "geaendert": md.stat().st_mtime,
        }

    skills: list[dict] = []
    for d in sorted(SKILLS_DIR.iterdir()):
        if not d.is_dir() or d.name.startswith("."):
            continue  # .archive (Curator) & Co. sind keine aktiven Skills
        md = d / "SKILL.md"
        if md.is_file():
            skills.append(eintrag(md, d.name, None))
            continue
        # Kategorie-Ordner (mitgelieferte Sammlungen wie research/, apple/): eine Ebene
        # tiefer liegen die echten Skills (research/blogwatcher/SKILL.md) — je einzeln zeigen.
        for sub in sorted(d.iterdir()):
            smd = sub / "SKILL.md"
            if sub.is_dir() and not sub.name.startswith(".") and smd.is_file():
                skills.append(eintrag(smd, f"{d.name}/{sub.name}", d.name))
    # Selbst erstellte zuerst — das ist der Star der Ansicht.
    order = {"selbst": 0, "repo": 1, "bundled": 2}
    skills.sort(key=lambda s: (order[s["herkunft"]], -s["genutzt"], s["name"]))
    return skills


def skill_text(skill_id: str) -> str | None:
    """Voller SKILL.md-Text („wie funktioniert der Skill") — max. Kategorie/Name, keine Pfad-Tricks."""
    if not re.fullmatch(r"[\w.-]+(/[\w.-]+)?", skill_id or ""):
        return None
    md = SKILLS_DIR / skill_id / "SKILL.md"
    if not md.is_file():
        return None
    try:
        return md.read_text(encoding="utf-8", errors="replace")
    except Exception:
        return None


def bilanz() -> dict:
    """Vorschlags-Bilanz: was die Box vorschlug und was daraus wurde (eine Timeline)."""
    eintraege: list[dict] = []
    try:
        branches = json.loads(AUFTRAGSBUCH_FILE.read_text(encoding="utf-8")).get("branches") or {}
        for branch, info in branches.items():
            eintraege.append({
                "art": "angenommen" if info.get("state") == "eingespielt" else info.get("state", "?"),
                "titel": branch,
                "detail": info.get("detail") or "",
                "ts": float(info.get("ts") or 0),
            })
    except Exception:
        pass
    try:
        for line in ABLEHNUNGEN_FILE.read_text(encoding="utf-8").splitlines():
            if not line.strip():
                continue
            try:
                j = json.loads(line)
            except Exception:
                continue
            eintraege.append({
                "art": "abgelehnt",
                "titel": j.get("subject") or j.get("branch") or "?",
                "detail": j.get("grund") or "",
                "ts": float(j.get("ts") or 0),
            })
    except Exception:
        pass
    eintraege.sort(key=lambda e: -e["ts"])
    return {
        "eintraege": eintraege,
        "angenommen": sum(1 for e in eintraege if e["art"] == "angenommen"),
        "abgelehnt": sum(1 for e in eintraege if e["art"] == "abgelehnt"),
    }


def overview() -> dict:
    skills = list_skills()
    b = bilanz()
    return {
        "available": SKILLS_DIR.is_dir(),
        "skills": skills,
        "selbst_erstellt": sum(1 for s in skills if s["herkunft"] == "selbst"),
        "bilanz": b,
    }