""" Hermes-Agent Control-Plane — read-only Einblicke fuer das Dashboard (Phase 4). Macht das "geschenkte v9" sichtbar: Agent-Status, Cron-Scheduler und Skills des Nous-Hermes-Agent-Frameworks. Quellen (MC laeuft als hitonabi auf dem Bosgame, also direkter Zugriff): - Status : ~/.hermes/gateway_state.json + cron-Heartbeat-Datei (sauberes JSON, kein Parsing). Liveness kommt aus dem Heartbeat, nicht aus dem evtl. beim Start eingefrorenen gateway_state.json. - Cron : `hermes cron list --all` (kanonisch; kein stabiles Storage-Schema). - Skills : `hermes skills list` (kanonisch) + ~/.hermes/skills/.usage.json (Nutzungszaehler). Rich-Tabellen werden mit COLUMNS=400 ohne Truncation erzeugt und ueber die `│`/`┃`-Spalten geparst. Alles mit kurzem TTL-Cache, damit Dashboard-Polling die CLI nicht haemmert. """ import json import os import re import subprocess import time from typing import Callable from config import HERMES_HOME, HERMES_BIN _CACHE_TTL = 10.0 # Sekunden _cache: dict[str, tuple[float, object]] = {} def _cached(key: str, fn: Callable[[], object]) -> object: now = time.monotonic() hit = _cache.get(key) if hit and now - hit[0] < _CACHE_TTL: return hit[1] val = fn() _cache[key] = (now, val) return val def _run_cli(args: list[str], timeout: float = 45.0) -> str: """`hermes ` ausfuehren. COLUMNS=400 verhindert Rich-Truncation.""" env = dict(os.environ, COLUMNS="400", NO_COLOR="1") try: p = subprocess.run( [HERMES_BIN, *args], capture_output=True, text=True, timeout=timeout, env=env, ) return p.stdout or "" except Exception: return "" def _parse_rich_table(text: str) -> list[dict]: """Eine Rich-Box-Tabelle in list[dict] verwandeln. Erste Zeile mit Spaltentrennern (`│` oder `┃`) ist der Header, der Rest sind Datenzeilen. Trenn-/Rahmenzeilen (┏━┓ usw.) enthalten keine Trenner -> ignoriert. """ header: list[str] | None = None rows: list[list[str]] = [] for ln in text.splitlines(): if "│" not in ln and "┃" not in ln: continue cells = [c.strip() for c in re.split(r"[│┃]", ln)] # fuehrende/abschliessende Rahmen-Leerzellen entfernen while cells and cells[0] == "": cells.pop(0) while cells and cells[-1] == "": cells.pop() if not cells: continue if header is None: header = [c.lower() for c in cells] else: rows.append(cells) if not header: return [] out = [] for r in rows: if len(r) != len(header): continue out.append(dict(zip(header, r))) return out # --------------------------------------------------------------------------- # Oeffentliche Reads # --------------------------------------------------------------------------- def agent_status() -> dict: """Live-Status des Hermes-Gateways + Cron-Tickers.""" def _load() -> dict: st = { "gateway_state": "unknown", "pid": None, "active_agents": None, "api_server": "unknown", "cron_running": False, "heartbeat_age_s": None, "version": None, } try: d = json.loads((HERMES_HOME / "gateway_state.json").read_text()) st["gateway_state"] = d.get("gateway_state", "unknown") st["pid"] = d.get("pid") st["active_agents"] = d.get("active_agents") api = (d.get("platforms") or {}).get("api_server") or {} st["api_server"] = api.get("state", "unknown") except Exception: pass # Liveness: Cron-Ticker-Heartbeat (Unix-Timestamp). < 120s = laeuft. try: ts = float((HERMES_HOME / "cron" / "ticker_heartbeat").read_text().strip()) age = max(0, int(time.time() - ts)) st["heartbeat_age_s"] = age st["cron_running"] = age < 120 except Exception: pass try: st["version"] = (HERMES_HOME / "PINNED_VERSION").read_text().strip() or None except Exception: pass return st return _cached("agent", _load) def cron_jobs() -> list[dict]: """Geplante Jobs (inkl. pausierter).""" return _cached("cron", lambda: _parse_rich_table(_run_cli(["cron", "list", "--all"]))) # type: ignore[return-value] def skills() -> list[dict]: """Installierte Skills, angereichert um Nutzungszaehler aus .usage.json.""" def _load() -> list[dict]: rows = _parse_rich_table(_run_cli(["skills", "list"])) usage = {} try: usage = json.loads((HERMES_HOME / "skills" / ".usage.json").read_text()) except Exception: pass for r in rows: u = usage.get(r.get("name", ""), {}) r["use_count"] = u.get("use_count", 0) r["last_used_at"] = u.get("last_used_at") r["pinned"] = bool(u.get("pinned", False)) return rows return _cached("skills", _load) # type: ignore[return-value] def learned_profile() -> dict: """Was Hermes ueber den Nutzer gelernt hat (Phase 8). `~/.hermes/memories/USER.md` ist das vom Curator destillierte Nutzerprofil (Eintraege mit `§` getrennt), `MEMORY.md` allgemeine gelernte Fakten. Direkter Beleg, dass der Agent automatisch mitlernt. """ def _read(name: str) -> str: try: return (HERMES_HOME / "memories" / name).read_text(errors="replace").strip() except Exception: return "" def _load() -> dict: user_raw = _read("USER.md") facts = [s.strip() for s in user_raw.split("§") if s.strip()] return {"user_facts": facts, "memory": _read("MEMORY.md")} return _cached("learned", _load) # type: ignore[return-value] def insights() -> dict: """Aktivitaets-Kennzahlen via `hermes insights` (Phase 8). Die CLI rendert ein Box-Art-Layout (kein --json) -> Headline-Zahlen + Top-Tools per Regex herausziehen (best effort; faellt leer aus, wenn sich das Format aendert). """ def _num(pat: str, text: str): m = re.search(pat, text) return m.group(1).strip() if m else None def _load() -> dict: txt = _run_cli(["insights", "--days", "30"]) out: dict = { "sessions": _num(r"Sessions:\s+([\d,]+)", txt), "messages": _num(r"Messages:\s+([\d,]+)", txt), "tool_calls": _num(r"Tool calls:\s+([\d,]+)", txt), "total_tokens": _num(r"Total tokens:\s+([\d,]+)", txt), "active_time": _num(r"Active time:\s+(~?[\dhm ]+?)\s{2,}", txt), "top_tools": [], } # "Top Tools"-Sektion: Zeilen % sec = txt.split("Top Tools", 1) if len(sec) == 2: for ln in sec[1].splitlines(): m = re.match(r"\s*(\S.*?)\s+(\d+)\s+([\d.]+)%\s*$", ln) if m: out["top_tools"].append( {"name": m.group(1), "calls": int(m.group(2)), "pct": float(m.group(3))} ) return out return _cached("insights", _load) # type: ignore[return-value]