"""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, }