Hitonabi
aa726c01af
`hermes cron list` nutzt bei vorhandenen Jobs kein │-Tabellenformat, sondern "id [status]" + eingerueckte Label:value-Zeilen. Eigener _parse_cron statt _parse_rich_table → Memory-Kurator-Job erscheint jetzt im Cockpit. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
232 lines
8.0 KiB
Python
232 lines
8.0 KiB
Python
"""
|
|
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 <args>` 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 _parse_cron(text: str) -> list[dict]:
|
|
"""`hermes cron list` parsen. Format pro Job (KEINE Rich-Tabelle):
|
|
|
|
7ac183295470 [active]
|
|
Name: Memory-Kurator
|
|
Schedule: 0 4 * * 0
|
|
Next run: 2026-06-28T04:00:00+00:00
|
|
Deliver: local
|
|
"""
|
|
jobs: list[dict] = []
|
|
cur: dict | None = None
|
|
for ln in text.splitlines():
|
|
m = re.match(r"\s*([0-9a-fA-F]{6,})\s+\[(\w+)\]\s*$", ln)
|
|
if m:
|
|
cur = {"id": m.group(1), "status": m.group(2)}
|
|
jobs.append(cur)
|
|
continue
|
|
if cur is not None:
|
|
kv = re.match(r"\s+([A-Za-z][\w ]*?):\s+(.+?)\s*$", ln)
|
|
if kv:
|
|
key = kv.group(1).strip().lower().replace(" ", "_")
|
|
cur[key] = kv.group(2).strip()
|
|
return jobs
|
|
|
|
|
|
def cron_jobs() -> list[dict]:
|
|
"""Geplante Jobs (inkl. pausierter)."""
|
|
return _cached("cron", lambda: _parse_cron(_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 <name> <calls> <pct>%
|
|
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]
|