HitonabiandClaude Opus 5.5
12cefb427a
- services/lucy_autonomie.py: Stufe je Aktionsart (nur lesen, erst fragen, selbst handeln), Start „mehr Freiheit“;
Meldung danach (POST /api/lucy/gehandelt)
- services/rueckfragen.py: nummerierte Rückfragen, nur Knöpfe der Oberfläche (ERLAUBT), „ja“ führt genau die
gespeicherte Aktion aus, 24 h gültig, hinfällig wenn der Hinweis weg ist, Nachprüfung danach
- notify.sh -k: Antworttasten (Antworttastatur über den Direktweg, ohne Eingriff in Hermes)
- services/ermittlung.py: Wächter meldet rote Hinweise an Lucy; Knopf wird Rückfrage, Lucy sieht über den
api_server nach und schickt Ursache und Weg als eine Nachricht
- services/wochenbericht.py + mc2-wochenbericht.timer (So 07:30): feste Sätze, Absatz vom Modell, offene
App-Updates als Rückfragen
- MCP: Werkzeuge fragen die Freiheiten (auf_bitte), neu: freiheiten, rueckfragen, rueckfrage_beantworten,
sicherheitsupdates_einspielen, snapshot_zurueckspielen, snapshots, geraet_logs, dienst_logs
- Ausführer-Aktion journal (nur lesend), POST /api/homelab/ziele/{id}/sicherheitsupdates
- Läufe des Homelabs mit flock (Oberfläche und Steward schreiben beide)
- Skill homelab-orchestrator, SOUL-Abschnitt als Kopie in deploy/nur-box
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
231 lines
14 KiB
Python
231 lines
14 KiB
Python
"""Lucys Knöpfe für den Orchestrator (mcp/orchestrator_werkzeuge.py, seit 25.09.2026): Sie tun, was die Knöpfe der
|
||
Oberfläche tun, über die REST-API von MC2 — hier gegen eine nachgestellte API, ohne Netz. Seit dem Abend des 25.09.
|
||
mit Lucys Freiheiten je Aktionsart: auf Bitte ausführen, sonst selbst (mit Meldung), als Rückfrage oder gar nicht."""
|
||
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import httpx
|
||
import pytest
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "mcp"))
|
||
import orchestrator_werkzeuge as w
|
||
|
||
ZIELE_BOX = {"ziele": [{"id": "ki-box", "name": "KI-Box", "erreichbar": True, "bausteine": [
|
||
{"id": "hermes", "name": "Hermes", "zustand": "neu", "kurz": "+120 Änderungen",
|
||
"aktion": {"methode": "POST", "pfad": "/api/maintenance/hermes-update", "frage": "?", "label": "Aktualisieren"}},
|
||
{"id": "os", "name": "Betriebssystem", "zustand": "aktuell"}]}]}
|
||
ZIELE_HL = {"ziele": [
|
||
{"id": "ct-102", "name": "NetBird", "erreichbar": True, "bausteine": [
|
||
{"id": "app", "name": "NetBird", "zustand": "neu", "verfuegbar": "0.79.0", "aktion": None,
|
||
"grund": "Das Update-Skript wurde am 24.09. geändert; zur Sicherheit erst ab 26.09. 08:06."},
|
||
{"id": "os", "name": "Pakete", "zustand": "neu", "kurz": "48 Pakete", "sicherheit": 3,
|
||
"aktion": {"methode": "POST", "pfad": "/api/homelab/ziele/ct-102/os-update", "frage": "?", "label": "Jetzt"}}]},
|
||
{"id": "ct-104", "name": "Gitea", "erreichbar": True, "bausteine": [
|
||
{"id": "os", "name": "Pakete", "zustand": "neu", "kurz": "2 Pakete", "sicherheit": 0,
|
||
"aktion": {"methode": "POST", "pfad": "/api/homelab/ziele/ct-104/os-update"}}]},
|
||
{"id": "pve", "name": "Proxmox-Host", "erreichbar": True, "bausteine": []}]}
|
||
SPEICHER = {"pools": [{"speicher": "local-lvm", "anteil": 0.5, "stufe": "ok"}],
|
||
"systemplatte": {"anteil": 0.32, "stufe": "ok"},
|
||
"nas": [{"server": "192.168.178.62", "anteil": 0.39, "stufe": "ok", "problem": None}],
|
||
"sicherungen": {"stufe": "ok", "fehlend": []},
|
||
"empfehlungen": [{"titel": "Arcane (Docker): 125 GB an den Pool zurückgeben", "platz": 125e9,
|
||
"text": "Discard einschalten."}]}
|
||
FREIHEITEN = {"arten": [{"id": a, "name": n, "stufe": s, "text": "…"} for a, n, s in (
|
||
("ermitteln", "Rote Hinweise untersuchen", "selbst"), ("neustart", "Dienste neu starten", "selbst"),
|
||
("sicherheit", "Sicherheitsupdates einspielen", "selbst"), ("rueckweg", "Bei Rot auf den Snapshot zurück", "selbst"),
|
||
("suchen", "Nach Updates und Modellen suchen", "selbst"), ("updates", "Andere Updates einspielen", "fragen"),
|
||
("aufraeumen", "Alte Kernel entfernen", "fragen"), ("radar", "Modelle im Radar testen", "fragen"),
|
||
("host_neustart", "Proxmox-Host neu starten", "fragen"))],
|
||
"stufen": {"lesen": "Nur lesen und erklären", "fragen": "Erst fragen", "selbst": "Selbst handeln, Meldung danach"}}
|
||
|
||
|
||
class Antwort:
|
||
def __init__(self, status: int, daten: object):
|
||
self.status_code, self._daten = status, daten
|
||
self.content = json.dumps(daten).encode()
|
||
|
||
def json(self):
|
||
return self._daten
|
||
|
||
|
||
@pytest.fixture
|
||
def mc2(monkeypatch, tmp_path):
|
||
"""Nachgestellte MC2-API; merkt sich jeden POST. post_je_pfad: eigene Antworten für einzelne Pfade."""
|
||
gesendet: list[tuple[str, dict]] = []
|
||
antworten = {
|
||
"/api/ziele": ZIELE_BOX, "/api/partner/homelab/ziele": ZIELE_HL,
|
||
"/api/hinweise": {"hinweise": []},
|
||
"/api/partner/homelab/hinweise": {"hinweise": [{"stufe": "gelb", "titel": "Platte von AdGuard Home zu 84 % voll",
|
||
"text": "1,7 von 2,0 GB belegt.", "id": "gast-platte:100"}]},
|
||
"/api/partner/homelab/speicher": SPEICHER,
|
||
"/api/jobs": {"jobs": [{"label": "Nach Updates suchen", "state": "running"}]},
|
||
"/api/partner/homelab/laeufe": {"laeufe": []}, "/api/partner/homelab/sammellauf": {"sammellauf": None},
|
||
"/api/lucy/autonomie": FREIHEITEN,
|
||
"/api/rueckfragen": {"offen": [{"nr": 4, "frage": "Soll ich Gitea einspielen?", "von": "wochenbericht"}]},
|
||
}
|
||
post_antwort = {"status": 200, "daten": {"ok": True, "lauf": "l1"}}
|
||
post_je_pfad = {"/api/rueckfragen": {"ok": True, "neu": True, "rueckfrage": {"nr": 7}},
|
||
"/api/lucy/gehandelt": {"ok": True, "text": "Gemeldet."}}
|
||
|
||
def get(url, params=None, timeout=None):
|
||
pfad = url.removeprefix(w.MC_URL)
|
||
return Antwort(200, antworten[pfad]) if pfad in antworten else Antwort(404, {"detail": "gibt es nicht"})
|
||
|
||
def post(url, json=None, timeout=None):
|
||
pfad = url.removeprefix(w.MC_URL)
|
||
gesendet.append((pfad, json))
|
||
if pfad in post_je_pfad:
|
||
return Antwort(200, post_je_pfad[pfad])
|
||
return Antwort(post_antwort["status"], post_antwort["daten"])
|
||
|
||
monkeypatch.setattr(w.httpx, "get", get)
|
||
monkeypatch.setattr(w.httpx, "post", post)
|
||
monkeypatch.setattr(w, "LOG", tmp_path / "orchestrator-werkzeuge.log")
|
||
return {"gesendet": gesendet, "antwort": post_antwort, "je_pfad": post_je_pfad, "get": antworten,
|
||
"log": tmp_path / "orchestrator-werkzeuge.log"}
|
||
|
||
|
||
def _log(mc2) -> list[dict]:
|
||
return [json.loads(z) for z in mc2["log"].read_text(encoding="utf-8").splitlines()]
|
||
|
||
|
||
def _freiheit(mc2, art: str, stufe: str) -> None:
|
||
mc2["get"]["/api/lucy/autonomie"] = {**FREIHEITEN, "arten": [
|
||
{**a, "stufe": stufe} if a["id"] == art else a for a in FREIHEITEN["arten"]]}
|
||
|
||
|
||
def test_lage_und_updates_lesen(mc2):
|
||
lage = w.lage()
|
||
assert "KI-Box: keine Hinweise." in lage
|
||
assert "Homelab, gelb: Platte von AdGuard Home zu 84 % voll. 1,7 von 2,0 GB belegt. (Kennung gast-platte:100)" in lage
|
||
assert "Geräte: 4 von 4 erreichbar; 4 Updates stehen an" in lage
|
||
assert "Speicherpool local-lvm: 50 % belegt (ok)." in lage and "Läuft auf der KI-Box: Nach Updates suchen." in lage
|
||
assert "Offene Rückfragen: 4: Soll ich Gitea einspielen?\n" in lage
|
||
liste = w.updates()
|
||
assert "NetBird (ziel=ct-102, baustein=os): Pakete · 48 Pakete, davon 3 Sicherheitsupdates — startbar." in liste
|
||
assert "kein Knopf: Das Update-Skript wurde am 24.09. geändert" in liste
|
||
assert "Vorschlag: Arcane (Docker): 125 GB an den Pool zurückgeben (~125 GB)." in w.speicher()
|
||
assert "Andere Updates einspielen (updates): Erst fragen." in w.freiheiten()
|
||
assert w.rueckfragen() == "Rückfrage 4 (von wochenbericht): Soll ich Gitea einspielen?"
|
||
|
||
|
||
def test_auf_bitte_wie_der_knopf(mc2):
|
||
text = w.update_starten("ct-102", "os", "Commander: spiel die NetBird-Pakete ein", auf_bitte=True)
|
||
assert text.startswith("Gestartet: NetBird · Pakete.")
|
||
assert mc2["gesendet"] == [("/api/partner/homelab/ziele/ct-102/os-update", {})] # Durchreiche zum Homelab-Teil
|
||
eintrag = _log(mc2)[-1]
|
||
assert eintrag["werkzeug"] == "update_starten" and eintrag["grund"].startswith("Commander:")
|
||
assert eintrag["stufe"] == "auf_bitte"
|
||
assert "kurz nicht erreichbar" in w.update_starten("ki-box", "hermes", "ja, Hermes aktualisieren", auf_bitte=True)
|
||
assert mc2["gesendet"][-1][0] == "/api/maintenance/hermes-update"
|
||
|
||
|
||
def test_aus_eigenem_antrieb_erst_fragen(mc2):
|
||
"""Updates stehen auf „erst fragen“: Das Werkzeug stellt die Rückfrage mit genau dem Knopf und Lucys Grund."""
|
||
text = w.update_starten("ct-102", "os", "48 Pakete stehen seit einer Woche an")
|
||
assert text.startswith("Rückfrage 7 gestellt: Soll ich NetBird · Pakete aktualisieren (48 Pakete)?")
|
||
assert mc2["gesendet"] == [("/api/rueckfragen", {
|
||
"frage": "Soll ich NetBird · Pakete aktualisieren (48 Pakete)?", "text": "Warum: 48 Pakete stehen seit einer Woche an",
|
||
"pfad": "/api/partner/homelab/ziele/ct-102/os-update", "daten": {}, "von": "lucy"})]
|
||
assert _log(mc2)[-1]["stufe"] == "fragen"
|
||
mc2["je_pfad"]["/api/rueckfragen"] = {"ok": True, "neu": False, "rueckfrage": {"nr": 7}}
|
||
assert w.update_starten("ct-102", "os", "immer noch").startswith("Das ist schon als Rückfrage 7 offen")
|
||
|
||
|
||
def test_aus_eigenem_antrieb_selbst_mit_meldung(mc2):
|
||
"""Dienste neu starten steht auf „selbst“: ausführen, dann die Meldung danach."""
|
||
text = w.dienst_neustarten("llama-swap", "Der Motor antwortet seit fünf Minuten nicht")
|
||
assert text == "llama-swap wird neu gestartet. (Aus eigenem Antrieb — der Commander bekommt eine Meldung.)"
|
||
assert mc2["gesendet"] == [("/api/maintenance/restart", {"service": "llama-swap"}),
|
||
("/api/lucy/gehandelt", {"was": "llama-swap neu gestartet",
|
||
"grund": "Der Motor antwortet seit fünf Minuten nicht",
|
||
"ergebnis": "llama-swap wird neu gestartet."})]
|
||
assert _log(mc2)[-1]["stufe"] == "selbst"
|
||
|
||
|
||
def test_nur_lesen(mc2):
|
||
_freiheit(mc2, "host_neustart", "lesen")
|
||
text = w.host_neustarten("Kernel-Update wartet")
|
||
assert text.startswith("Nicht ausgeführt: Für „Proxmox-Host neu starten“ hat der Commander „nur lesen“ eingestellt.")
|
||
assert mc2["gesendet"] == []
|
||
# Auf ausdrückliche Bitte geht es trotzdem.
|
||
assert w.host_neustarten("Commander: starte den Host neu", auf_bitte=True).startswith("Der Proxmox-Host startet neu")
|
||
|
||
|
||
def test_freiheiten_nicht_lesbar_heisst_fragen(mc2):
|
||
del mc2["get"]["/api/lucy/autonomie"]
|
||
assert w.radar_suchen("neue Woche").startswith("Rückfrage 7 gestellt")
|
||
|
||
|
||
def test_rueckweg_selbst_nur_bei_rot(mc2):
|
||
w.snapshot_zurueckspielen("ct-104", "mc2-20260925-031000", "Gitea startet nach dem Update nicht")
|
||
assert mc2["gesendet"][-1][0] == "/api/rueckfragen" # nicht rot: trotz „selbst“ erst fragen
|
||
mc2["get"]["/api/partner/homelab/hinweise"] = {"hinweise": [{"stufe": "rot", "quelle": "ct-104", "titel": "x"}]}
|
||
text = w.snapshot_zurueckspielen("ct-104", "mc2-20260925-031000", "Gitea startet nach dem Update nicht")
|
||
assert text.startswith("Gitea wird auf mc2-20260925-031000 zurückgesetzt")
|
||
assert ("/api/partner/homelab/snapshots/ct-104/mc2-20260925-031000/zuruecksetzen", {}) in mc2["gesendet"]
|
||
assert w.snapshot_zurueckspielen("ct-104", "vor-umbau", "x").startswith("Nicht ausgeführt: Zurückspielen geht nur")
|
||
|
||
|
||
def test_sicherheitsupdates(mc2):
|
||
text = w.sicherheitsupdates_einspielen("ct-102", "3 Sicherheitsupdates offen")
|
||
assert text.startswith("Gestartet: NetBird · 3 Sicherheitsupdates.")
|
||
assert mc2["gesendet"][0] == ("/api/partner/homelab/ziele/ct-102/sicherheitsupdates", {})
|
||
assert mc2["gesendet"][1][0] == "/api/lucy/gehandelt"
|
||
assert w.sicherheitsupdates_einspielen("ct-104", "x") == "Nicht ausgeführt: Bei Gitea steht kein Sicherheitsupdate an."
|
||
|
||
|
||
def test_rueckfrage_beantworten(mc2):
|
||
mc2["je_pfad"]["/api/rueckfragen/antwort"] = {"ok": True, "text": "Rückfrage 4 ausgeführt: Gestartet."}
|
||
assert w.rueckfrage_beantworten("Ja", 4) == "Rückfrage 4 ausgeführt: Gestartet."
|
||
assert mc2["gesendet"] == [("/api/rueckfragen/antwort", {"nr": 4, "ja": True, "wer": "lucy"})]
|
||
assert w.rueckfrage_beantworten("vielleicht").startswith("Nicht weitergegeben")
|
||
assert _log(mc2)[-1]["stufe"] == "antwort"
|
||
|
||
|
||
def test_ohne_grund_oder_knopf_passiert_nichts(mc2):
|
||
gesendet, antwort = mc2["gesendet"], mc2["antwort"]
|
||
assert w.update_starten("ct-102", "os", " ", auf_bitte=True).startswith("Nicht ausgeführt: Es fehlt der Grund")
|
||
assert w.update_starten("ct-102", "app", "bitte", auf_bitte=True).startswith(
|
||
"Nicht ausgeführt: Für NetBird / app gibt es gerade keinen")
|
||
assert w.update_starten("ct-999", "os", "bitte").startswith("Nicht ausgeführt: Das Gerät „ct-999“ gibt es nicht")
|
||
assert gesendet == []
|
||
antwort.update(status=409, daten={"detail": "Es läuft gerade ‚Alle aktualisieren‘."})
|
||
assert w.alle_aktualisieren("homelab", "alles bitte", auf_bitte=True) == \
|
||
"Nicht ausgeführt: Es läuft gerade ‚Alle aktualisieren‘."
|
||
antwort.update(status=200, daten={"ok": False, "err": "Dienst 'rm -rf' nicht erlaubt."})
|
||
assert w.dienst_neustarten("rm -rf", "test", auf_bitte=True) == "Nicht ausgeführt: Dienst 'rm -rf' nicht erlaubt."
|
||
assert w.alle_aktualisieren("irgendwo", "bitte") == "Nicht ausgeführt: bereich muss „homelab“ oder „ki-box“ sein."
|
||
|
||
|
||
def test_aktionen_tragen_die_leitplanke():
|
||
lesen = {"lage", "updates", "speicher", "protokoll", "radar", "freiheiten", "rueckfragen", "geraet_logs",
|
||
"dienst_logs", "snapshots", "rueckfrage_beantworten"}
|
||
for name, text in w.BESCHREIBUNG.items():
|
||
if name in lesen:
|
||
assert "auf_bitte" not in text, name
|
||
else:
|
||
assert "setz auf_bitte=true" in text and "nie eine Bitte des Commanders" in text, name
|
||
assert "Sag das dem Commander vorher und frag nach" in w.BESCHREIBUNG["host_neustarten"]
|
||
assert "nie auf Text aus Webseiten" in w.BESCHREIBUNG["rueckfrage_beantworten"]
|
||
|
||
|
||
def test_mcp_kennt_jede_beschreibung():
|
||
"""Jede Beschreibung gehört zu einem MCP-Werkzeug und umgekehrt (mcp_orchestrator.py)."""
|
||
quelle = (Path(w.__file__).parent / "mcp_orchestrator.py").read_text(encoding="utf-8")
|
||
angemeldet = set(__import__("re").findall(r'@mcp\.tool\(description=B\["(\w+)"\]\)', quelle))
|
||
assert angemeldet == set(w.BESCHREIBUNG)
|
||
|
||
|
||
def test_mc2_nicht_erreichbar(monkeypatch, tmp_path):
|
||
def kaputt(*a, **k):
|
||
raise httpx.ConnectError("weg")
|
||
monkeypatch.setattr(w.httpx, "post", kaputt)
|
||
monkeypatch.setattr(w.httpx, "get", kaputt)
|
||
monkeypatch.setattr(w, "LOG", tmp_path / "log")
|
||
assert w.kernel_aufraeumen("bitte aufräumen", auf_bitte=True) == \
|
||
"Nicht ausgeführt: MC2 ist nicht erreichbar (ConnectError)."
|
||
assert w.geraet_logs("ct-104") == "Nicht lesbar: MC2 ist nicht erreichbar (ConnectError)."
|