Merge branch 'worktree-agent-abdb83e97b099a62d' into wartung/restarbeiten

2026-09-24 20:25:22 +02:00
parent 6afec70640 8e3fecbe42
commit 37179b8864
16 changed files with 1461 additions and 71 deletions
@@ -3,5 +3,7 @@
  apps.py      was in welchem Container steckt, wo seine Oberfläche liegt, woher die neueste Version kommt
  kanal.py     Aufträge und Berichte des Ausführers auf dem Proxmox-Host (gemeinsames Geheimnis)
  inventar.py  Bericht + neueste Versionen + Erreichbarkeit → Ziele im gemeinsamen Modell (kern/ziele.py)
-  updates.py   „Jetzt updaten“: Snapshot → Update → Prüfung → bei Rot zurück, mit Meldung
+  karenz.py    Wartezeit nach einer Änderung am Update-Skript (community-scripts, ungepinnt von GitHub)
  updates.py   „Jetzt updaten“: Snapshot bzw. Sicherung → Update → Prüfung → bei Rot zurück, mit Meldung
  pflege.py    wöchentliches Suchen (Paketlisten der Gäste), im Wächter-Takt des Stewards
 """
@@ -7,7 +7,9 @@ Dazu prüft dieser Teil selbst, ob die Weboberfläche jedes Gasts antwortet.
 Wie ehrlich der Stand ist, steht dabei: Sind die Paketlisten im Gast älter als zwei Wochen, heißt es
 „unbekannt“ (mit Knopf zum Suchen) statt „aktuell“; ist der Bericht älter als eine halbe Stunde, sagt
-die Übersicht, dass der Ausführer schweigt.
+die Übersicht, dass der Ausführer schweigt. Ist das Update-Skript einer App frisch geändert, bleibt sie
 „neu“, aber ohne Knopf (karenz.py). Rückweg und Rückfrage sagen, was vor dem Update geschieht: Snapshot,
 wo keiner geht eine Sicherung, sonst keiner.
 """
 import re
@@ -20,7 +22,7 @@ from kern.github import neueste_version
 from kern.zeit import LOCAL_TZ
 from kern.ziele import Aktion, BausteinStand, ZielStand
-from services.homelab import apps, arcane, kanal
+from services.homelab import apps, arcane, kanal, karenz
 BERICHT_ALT_S = 30 * 60
 LISTEN_ALT_S = 14 * 24 * 3600
@@ -70,6 +72,30 @@ def _datum(ts: float | None) -> str:
    return datetime.fromtimestamp(ts, LOCAL_TZ).strftime("%d.%m.%Y") if ts else "?"
 def _rueckweg(gast: dict) -> str:
    """Die Zeile „Rückweg“ der Karte: was vor jedem Update geschieht."""
    if gast.get("snapshot_moeglich"):
        return "Snapshot vor jedem Update"
    grund = gast.get("snapshot_grund") or "kein Snapshot möglich"
    if gast.get("sicherung_moeglich"):
        return f"Sicherung auf „{gast.get('sicherung_speicher') or 'dem Host'}“ vor jedem Update ({grund})"
    if gast.get("sicherung_grund"):
        return f"Kein automatischer Rückweg: {grund}. Auch keine Sicherung: {gast['sicherung_grund']}"
    return grund
 def _rueckweg_frage(gast: dict) -> str:
    """Der Satz zum Rückweg in der Rückfrage vor „Jetzt updaten“."""
    if gast.get("snapshot_moeglich"):
        return "Vorher wird ein Snapshot angelegt; ist die Prüfung danach rot, geht es von selbst zurück."
    if gast.get("sicherung_moeglich"):
        return "Vorher wird eine Sicherung angelegt; ist die Prüfung danach rot, wird sie zurückgespielt."
    satz = "Ein Snapshot ist hier nicht möglich — scheitert das Update, gibt es keinen automatischen Rückweg."
    if gast.get("sicherung_grund"):
        satz += f" Eine Sicherung geht auch nicht: {gast['sicherung_grund']}"
    return satz
 def _app_baustein(gast: dict, app: apps.App, zid: str) -> BausteinStand:
    installiert = (gast.get("app") or {}).get("version")
    stand = BausteinStand(id="app", name=app.name, zustand="aktuell", installiert=installiert)
@@ -103,10 +129,15 @@ def _app_baustein(gast: dict, app: apps.App, zid: str) -> BausteinStand:
        stand.grund = f"{app.name} aktualisiert sich über die eigene Oberfläche; das Community-Script tut es nicht."
    elif ist_neuer:
        stand.zustand, stand.kurz = "neu", f"{installiert} → {neu['version']}"
-        rueckweg = ("Vorher wird ein Snapshot angelegt; ist die Prüfung danach rot, geht es von selbst zurück."
+        # Frisch geänderte Update-Skripte erst nach der Wartezeit (karenz.py): „neu“, aber ohne Knopf.
-                    if gast.get("snapshot_moeglich") else
+        wartezeit = karenz.pruefen(app.kennung)
-                    "Ein Snapshot ist hier nicht möglich — scheitert das Update, gibt es keinen automatischen Rückweg.")
+        if wartezeit["gesperrt"]:
-        stand.aktion = Aktion("POST", f"/api/homelab/ziele/{zid}/update", f"{app.name} jetzt aktualisieren? {rueckweg}")
+            stand.grund = wartezeit["gesperrt"]
            return stand
        frage = f"{app.name} jetzt aktualisieren? {_rueckweg_frage(gast)}"
        if not wartezeit["geprueft"]:
            frage += f" {karenz.NICHT_GEPRUEFT}"
        stand.aktion = Aktion("POST", f"/api/homelab/ziele/{zid}/update", frage)
    return stand
@@ -124,7 +155,7 @@ def _os_baustein(gast: dict, zid: str, jetzt: float) -> BausteinStand:
    elif anzahl > 0:
        stand.zustand, stand.kurz = "neu", f"{anzahl} Pakete"
        stand.aktion = Aktion("POST", f"/api/homelab/ziele/{zid}/os-update",
-                              f"{anzahl} Pakete im Gast jetzt einspielen? Vorher wird, wo möglich, ein Snapshot angelegt.")
+                              f"{anzahl} Pakete im Gast jetzt einspielen? {_rueckweg_frage(gast)}")
    return stand
@@ -172,8 +203,7 @@ def _gast_ziel(gast: dict, jetzt: float) -> ZielStand:
    laeuft = gast.get("status") == "running"
    ziel = ZielStand(id=zid, name=app.name if app else str(gast.get("name") or zid), art=art, instanz="homelab",
                     erreichbar=(erreichbar(url) if laeuft else False) if url else (laeuft or None),
-                     rueckweg=("Snapshot vor jedem Update" if gast.get("snapshot_moeglich")
+                     rueckweg=_rueckweg(gast))
                               else gast.get("snapshot_grund") or "kein Snapshot möglich"))
    if not laeuft:
        ziel.bausteine.append(BausteinStand(id="status", name="Status", zustand="unbekannt",
                                            grund=f"Der Gast ist {gast.get('status') or 'unbekannt'}."))
@@ -238,14 +268,18 @@ def ziele() -> dict:
 def erreichbarkeit() -> list[dict]:
-    """Für den Wächter: jede Weboberfläche laufender, freigegebener Gäste — ohne GitHub-Abfragen."""
+    """Für den Wächter: jede Weboberfläche laufender, freigegebener Gäste — ohne GitHub-Abfragen. Gäste mitten in
    einem Update-Lauf bleiben außen vor: Sicherung, Update und Zurückspielen legen sie kurz still, und was dabei
    herauskommt, meldet der Lauf selbst (sonst käme ein zweiter Alarm samt „wieder ok“)."""
    from services.homelab import updates  # updates importiert dieses Modul
    eingang = kanal.bericht()
    im_update = updates.laufende_ziele()
    ergebnis = []
    for g in (eingang or {}).get("bericht", {}).get("gaeste") or []:
        app = apps.app_fuer((g.get("app") or {}).get("kennung"), g.get("name"))
        url = apps.adresse(app, g.get("ip"))
        if url and g.get("erlaubt"):
        zid = f"{'ct' if g['art'] == 'lxc' else 'vm'}-{g['vmid']}"
        if url and g.get("erlaubt") and zid not in im_update:
            ergebnis.append({"id": zid, "name": app.name if app else g.get("name"), "url": url,
                             "laeuft": g.get("status") == "running",
                             "ok": g.get("status") == "running" and erreichbar(url)})
@@ -6,6 +6,8 @@ X-MC2-Ausfuehrer). Das erzeugt dieser Teil beim ersten Bedarf in <Datenordner>/a
 für den Dienst lesbar); die Einrichtung auf dem Proxmox-Host kopiert es in /etc/mc2-ausfuehrer.json.
 Aufträge liegen in <Datenordner>/ausfuehrer-auftraege.json und überleben so einen Neustart dieses Teils.
 Zwei Prozesse schreiben hinein: die Oberfläche (Knöpfe, Abholen durch den Ausführer) und seit 24.09.2026 der
 Steward (wöchentliches Suchen, pflege.py). Lesen, ändern, schreiben geschieht deshalb unter einer Dateisperre.
 """
 import hmac
@@ -15,12 +17,19 @@ import secrets
 import threading
 import time
 import uuid
 from collections.abc import Iterator
 from contextlib import contextmanager
 from pathlib import Path
 from kern.einstellungen import einstellungen
 try:
    import fcntl  # Linux: Sperre über Prozessgrenzen
 except ImportError:  # Windows (Entwicklung): nur die Thread-Sperre
    fcntl = None
 AKTIONEN = {"bericht", "snapshot", "update", "os_update", "suchen", "zurueck", "snapshot_loeschen",
-            "host_update", "host_neustart"}
+            "host_update", "host_neustart", "sichern", "sicherung_zurueck", "sicherung_loeschen"}
 VERLOREN_S = 3 * 3600         # abgeholt, aber nie beantwortet (Ausführer abgestürzt, Host neu gestartet)
 BEHALTEN = 60                 # so viele erledigte Aufträge bleiben sichtbar
@@ -32,6 +41,24 @@ def _dir() -> Path:
    return einstellungen().daten_dir
@contextmanager
 def _sperre() -> Iterator[None]:
    """Für Lesen-ändern-schreiben der Auftragsliste: Thread-Sperre, unter Linux zusätzlich flock. Ohne sie
    überschrieben sich Oberfläche und Steward gegenseitig (ein abgeholter Auftrag stünde wieder auf „wartet“)."""
    with _lock:
        if fcntl is None:
            yield
            return
        pfad = _dir() / "ausfuehrer-auftraege.lock"
        pfad.parent.mkdir(parents=True, exist_ok=True)
        with open(pfad, "a", encoding="utf-8") as f:
            fcntl.flock(f, fcntl.LOCK_EX)
            try:
                yield
            finally:
                fcntl.flock(f, fcntl.LOCK_UN)
 def _json_schreiben(pfad: Path, daten: object) -> None:
    pfad.parent.mkdir(parents=True, exist_ok=True)
    tmp = pfad.with_suffix(pfad.suffix + ".tmp")
@@ -118,7 +145,7 @@ def anlegen(aktion: str, parameter: dict | None = None) -> str:
    auftrag = {"id": uuid.uuid4().hex[:12], "aktion": aktion, "parameter": parameter or {},
               "status": "wartet", "erstellt": time.time(), "abgeholt": None, "fertig": None,
               "code": None, "text": None}
-    with _lock:
+    with _sperre():
        auftraege = _alle()
        auftraege.append(auftrag)
        _merken(auftraege)
@@ -129,7 +156,7 @@ def naechster() -> dict | None:
    """Für den Ausführer: der älteste wartende Auftrag, ab jetzt „läuft“. Hängengebliebene werden
    vorher als verloren abgeschlossen."""
    jetzt = time.time()
-    with _lock:
+    with _sperre():
        auftraege = _alle()
        for a in auftraege:
            if a["status"] == "laeuft" and jetzt - (a["abgeholt"] or jetzt) > VERLOREN_S:
@@ -142,7 +169,7 @@ def naechster() -> dict | None:
 def ergebnis(auftrag_id: str, code: int, text: str) -> bool:
-    with _lock:
+    with _sperre():
        auftraege = _alle()
        auftrag = next((a for a in auftraege if a["id"] == auftrag_id), None)
        if auftrag is None or auftrag["status"] != "laeuft":
@@ -0,0 +1,61 @@
 """Wartezeit nach einer Änderung am Update-Skript (24.09.2026).
 „Jetzt updaten“ lässt im Gast `update` laufen. Das lädt ct/<kennung>.sh ungepinnt von GitHub
 (community-scripts/ProxmoxVE, Zweig main) und führt es als root aus. Eine Änderung, die erst ein paar Stunden
 alt ist, hat noch kaum jemand im Betrieb gesehen. Deshalb gilt für Apps mit dem Weg „skript“: Ist die letzte
 Änderung an ct/<kennung>.sh jünger als MC_HOMELAB_KARENZ_H Stunden (Standard 48), bleibt der Baustein „neu“,
 aber ohne Knopf, und updates.starten() lehnt ab — beide mit demselben Satz.
 Wann das Skript zuletzt geändert wurde, sagt die GitHub-API (kern/github.py, 15 Minuten gemerkt). Lässt sich das
 nicht abfragen, wird nicht blockiert; die Rückfrage vor dem Update sagt es dann (NICHT_GEPRUEFT).
 """
 import os
 import re
 import time
 from datetime import datetime
 from kern.github import github_json
 from kern.zeit import LOCAL_TZ
 REPO = "community-scripts/ProxmoxVE"
 STANDARD_H = 48.0
 NICHT_GEPRUEFT = "Ob das Skript kürzlich geändert wurde, ließ sich nicht prüfen."
 _KENNUNG = re.compile(r"^[a-z0-9-]+$")
 def karenz_s() -> float:
    """Die Wartezeit in Sekunden (MC_HOMELAB_KARENZ_H; 0 schaltet sie ab)."""
    try:
        stunden = float(os.environ.get("MC_HOMELAB_KARENZ_H", "") or STANDARD_H)
    except ValueError:
        stunden = STANDARD_H
    return max(0.0, stunden) * 3600
 def skript_geaendert(kennung: str) -> float:
    """Wann ct/<kennung>.sh auf main zuletzt geändert wurde (Unix-Sekunden). Wirft, wenn es sich nicht sagen lässt."""
    if not _KENNUNG.match(kennung or ""):
        raise ValueError(f"ungültige Kennung {kennung!r}")
    daten = github_json(f"repos/{REPO}/commits?path=ct/{kennung}.sh&sha=main&per_page=1")
    # Das Datum, an dem die Änderung auf main landete (committer, nicht author: ein Pull-Request kann älter sein).
    datum = str(daten[0]["commit"]["committer"]["date"])
    return datetime.fromisoformat(datum.replace("Z", "+00:00")).timestamp()
 def sperrtext(geaendert: float, frei_ab: float) -> str:
    a, b = datetime.fromtimestamp(geaendert, LOCAL_TZ), datetime.fromtimestamp(frei_ab, LOCAL_TZ)
    return f"Das Update-Skript wurde am {a:%d.%m.} geändert; zur Sicherheit erst ab {b:%d.%m. %H:%M}."
 def pruefen(kennung: str, jetzt: float | None = None) -> dict:
    """{"gesperrt": Satz oder None, "geprueft": bool}. Gesperrt, solange die letzte Änderung am Skript jünger ist
    als die Wartezeit. Scheitert die Abfrage: nicht gesperrt, aber geprueft=False."""
    try:
        geaendert = skript_geaendert(kennung)
    except Exception:
        return {"gesperrt": None, "geprueft": False}
    frei_ab = geaendert + karenz_s()
    if (time.time() if jetzt is None else jetzt) >= frei_ab:
        return {"gesperrt": None, "geprueft": True}
    return {"gesperrt": sperrtext(geaendert, frei_ab), "geprueft": True}
@@ -0,0 +1,180 @@
 """Pflege im Hintergrund: wöchentlich nach Updates suchen (24.09.2026).
 Die Übersicht ist nur so ehrlich wie die Paketlisten in den Gästen, und die erneuert dort niemand von selbst
 (AdGuard stand im September 2026 auf Listen vom Oktober 2025). Deshalb stößt der Homelab-Teil den Auftrag
 „suchen“ (apt-get update bzw. apk update im Gast, ändert keine Pakete) selbst an, sobald die Listen eines Gasts
 älter als sieben Tage sind:
  • nur freigegebene, laufende Container — der Ausführer prüft das Etikett ohnehin selbst noch einmal
  • höchstens ein Auftrag je Gast und Tag; nie während eines Update-Laufs für diesen Gast und nie, solange für
    ihn ein anderer Auftrag offen ist; nur, wenn der Ausführer gerade berichtet
  • bevorzugt nachts 02:00–05:00 (Berlin); war die Instanz oder der Ausführer in der letzten Nacht nicht da,
    eben gleich
  • keine Meldungen. Scheitert die Suche für einen Gast FEHLSCHLAEGE_HINWEIS-mal hintereinander, wird daraus
    ein gelber Hinweis des Wächters (gelb geht nicht an Telegram).
 Der Steward der Rolle homelab ruft pruefe_paketlisten() in jedem Wächter-Takt auf (services/waechter.py,
 PRUEFUNGEN). Zustand in <Datenordner>/homelab-pflege.json; den schreibt nur der Steward. Die Aufträge selbst
 landen im Kanal (kanal.py, unter Dateisperre, weil auch die Oberfläche dort schreibt).
 """
 import json
 import logging
 import threading
 import time
 from datetime import datetime, timedelta
 from pathlib import Path
 from kern.einstellungen import einstellungen
 from kern.zeit import LOCAL_TZ
 from services.homelab import apps, inventar, kanal, updates
 log = logging.getLogger(__name__)
 LISTEN_ALT_S = 7 * 24 * 3600
 NACHT_VON, NACHT_BIS = 2, 5          # volle Stunden, Europe/Berlin
 FEHLSCHLAEGE_HINWEIS = 2
 _lock = threading.Lock()
 def _pfad() -> Path:
    return einstellungen().daten_dir / "homelab-pflege.json"
 def zustand() -> dict:
    """{"nacht": Datum der letzten Nacht, in der gesucht werden konnte, "gaeste": {vmid: {…}}}"""
    try:
        daten = json.loads(_pfad().read_text(encoding="utf-8"))
    except (OSError, ValueError):
        daten = {}
    if not isinstance(daten, dict):
        daten = {}
    if not isinstance(daten.get("gaeste"), dict):
        daten["gaeste"] = {}
    return daten
 def _schreiben(daten: dict) -> None:
    _pfad().parent.mkdir(parents=True, exist_ok=True)
    tmp = _pfad().with_suffix(".tmp")
    tmp.write_text(json.dumps(daten, ensure_ascii=False, indent=1), encoding="utf-8")
    tmp.replace(_pfad())
 def faellig(gaeste: list[dict], jetzt: float) -> list[dict]:
    """Die Container, deren Paketlisten eine neue Suche brauchen: freigegeben, laufend, Listen älter als 7 Tage."""
    ergebnis = []
    for g in gaeste:
        listen = (g.get("os_updates") or {}).get("listen_stand")
        if (g.get("art") == "lxc" and g.get("status") == "running" and g.get("erlaubt")
                and inventar.EIGENES_ETIKETT not in (g.get("etiketten") or [])
                and listen and jetzt - listen > LISTEN_ALT_S):
            ergebnis.append(g)
    return ergebnis
 def _fehlerzeile(text: str | None) -> str:
    """Die aussagekräftigste Zeile einer gescheiterten Suche (apt meldet Fehler mit „E:“)."""
    zeilen = [z.strip() for z in str(text or "").splitlines() if z.strip()]
    wichtig = [z for z in zeilen if z.startswith(("E:", "ERROR", "Abgelehnt", "Zeitlimit", "Fehler"))]
    return (wichtig or zeilen or ["ohne Angabe"])[-1][:200]
 def _einsammeln(daten: dict) -> bool:
    """Ergebnisse der eigenen Aufträge übernehmen: fertig → Zähler auf null, gescheitert → eins mehr."""
    geaendert = False
    for eintrag in daten["gaeste"].values():
        aid = eintrag.get("auftrag")
        if not aid:
            continue
        a = kanal.auftrag(aid)
        if a is not None and a.get("status") in ("wartet", "laeuft"):
            continue
        if a is not None and a.get("status") == "fertig":
            eintrag["fehlschlaege"], eintrag["fehler"] = 0, None
        elif a is not None:           # fehler, verloren
            eintrag["fehlschlaege"] = int(eintrag.get("fehlschlaege") or 0) + 1
            eintrag["fehler"] = _fehlerzeile(a.get("text"))
        eintrag["auftrag"] = None     # nicht mehr in der Liste: vergessen, nicht zählen
        geaendert = True
    return geaendert
 def _offene_gaeste() -> set:
    """Gäste mit einem offenen Auftrag gleich welcher Art (auch von Hand oder aus einem Update-Lauf)."""
    return {(a.get("parameter") or {}).get("vmid") for a in kanal.liste() if a.get("status") in ("wartet", "laeuft")}
 def _befunde(daten: dict, faellige: list[dict]) -> list:
    """Gelbe Hinweise für Gäste, deren Listen weiter alt sind und deren Suche wiederholt scheiterte."""
    from services.waechter import Befund  # der Wächter registriert diese Prüfung, also erst hier
    befunde = []
    for g in faellige:
        eintrag = daten["gaeste"].get(str(g["vmid"])) or {}
        anzahl = int(eintrag.get("fehlschlaege") or 0)
        if anzahl < FEHLSCHLAEGE_HINWEIS:
            continue
        name = eintrag.get("name") or str(g.get("name") or g["vmid"])
        befunde.append(Befund(
            id=f"pflege:ct-{g['vmid']}", stufe="gelb", titel=f"{name}: Die Suche nach Updates scheitert",
            text=(f"Die Paketlisten ließen sich {anzahl}-mal hintereinander nicht erneuern (wöchentliche Suche). "
                  f"Zuletzt: {eintrag.get('fehler') or 'ohne Angabe'}"),
            quelle=f"ct-{g['vmid']}", sofort=True))
    return befunde
 def pruefe_paketlisten(jetzt: float | None = None) -> list:
    """Im Wächter-Takt: Ergebnisse einsammeln, fällige Suchen anstoßen. Rückgabe: Befunde für den Wächter —
    nur, wenn die Suche für einen Gast wiederholt scheitert (gelb)."""
    from services import waechter
    jetzt = time.time() if jetzt is None else jetzt
    lokal = datetime.fromtimestamp(jetzt, LOCAL_TZ)
    heute = lokal.date().isoformat()
    nachts = NACHT_VON <= lokal.hour < NACHT_BIS
    # Die zuletzt begonnene Nacht: vor 02:00 die von gestern.
    letzte_nacht = (lokal.date() - timedelta(days=0 if lokal.hour >= NACHT_VON else 1)).isoformat()
    eingang = kanal.bericht()
    bericht = (eingang or {}).get("bericht") or {}
    gaeste = bericht.get("gaeste") or []
    zuletzt = kanal.zuletzt()
    # Aufträge nur, wenn der Ausführer gerade berichtet und nicht im Nur-Lesen-Modus läuft (dann lehnte er ab).
    verbunden = bool(eingang and zuletzt and jetzt - zuletzt < inventar.BERICHT_ALT_S
                     and not (bericht.get("host") or {}).get("nur_lesen"))
    faellige = faellig(gaeste, jetzt)
    with _lock:
        daten = zustand()
        geaendert = _einsammeln(daten)
        faellig_ids = {str(g["vmid"]) for g in faellige}
        for vmid, eintrag in daten["gaeste"].items():
            # Listen wieder frisch (auch von Hand gesucht) oder Gast nicht mehr dabei: Zähler vergessen.
            if gaeste and eintrag.get("fehlschlaege") and vmid not in faellig_ids:
                eintrag["fehlschlaege"], eintrag["fehler"] = 0, None
                geaendert = True
        if verbunden and nachts and daten.get("nacht") != heute:
            daten["nacht"] = heute
            geaendert = True
        # Nachts sowieso; tagsüber nur, wenn die letzte Nacht verpasst wurde (Instanz oder Ausführer war weg).
        if verbunden and (nachts or daten.get("nacht") != letzte_nacht):
            offen = _offene_gaeste()
            im_update = updates.laufende_ziele()
            for g in faellige:
                vmid = g["vmid"]
                eintrag = daten["gaeste"].setdefault(str(vmid), {})
                if eintrag.get("tag") == heute or eintrag.get("auftrag") or vmid in offen \
                        or f"ct-{vmid}" in im_update:
                    continue
                app = apps.app_fuer((g.get("app") or {}).get("kennung"), g.get("name"))
                eintrag["name"] = app.name if app else str(g.get("name") or vmid)
                if waechter.TROCKEN:
                    log.info("pflege (trocken): würde die Paketlisten von %s erneuern", eintrag["name"])
                    continue
                eintrag.update(tag=heute, auftrag=kanal.anlegen("suchen", {"vmid": vmid}), angestossen=jetzt)
                geaendert = True
                log.info("pflege: Paketlisten von %s sind alt → suchen", eintrag["name"])
        if geaendert:
            _schreiben(daten)
    return _befunde(daten, faellige)
@@ -4,12 +4,15 @@ Nur per Knopf, mit Erfolgsmeldung und Erreichbarkeits-Check; ist er rot, geht es
 und es wird gemeldet. Host-Updates gibt es mit Warnung, der Neustart ist ein eigener Knopf.
 Ablauf für einen Gast:
-  1. Snapshot (wo möglich; beim PBS verhindert der Bind-Mount ihn — dann ohne Rückweg, die Rückfrage sagt es)
+  1. Rückweg anlegen: ein Snapshot; wo keiner geht (PBS: Bind-Mount), eine Sicherung (vzdump auf einen lokalen
-  2. Update: die App per Community-Script (`update`, still) oder die Pakete des Gasts
+     Speicher des Hosts, nie auf den PBS); geht beides nicht, ohne Rückweg — die Rückfrage sagt es. Scheitert
     dieser Schritt, beginnt das Update gar nicht.
  2. Update: die App per Community-Script (`update`, still) oder die Pakete des Gasts. Ein frisch geändertes
     Update-Skript wartet erst MC_HOMELAB_KARENZ_H Stunden (karenz.py).
  3. 20 s warten, frischen Bericht holen
  4. Prüfen: läuft der Gast, antwortet die Weboberfläche, ist die App-Version jetzt neuer?
-  5. Rot und Snapshot da → zurück auf den Snapshot → dringende Meldung.
+  5. Rot → zurück auf den Snapshot bzw. die Sicherung zurückspielen → dringende Meldung.
-     Grün → ältere Snapshots des Orchestrators für diesen Gast weg → Meldung „eingespielt“.
+     Grün → ältere Snapshots bzw. Sicherungen des Orchestrators für diesen Gast weg → Meldung „eingespielt“.
 Jeder Lauf steht in <Datenordner>/homelab-laeufe.json und im strukturierten Update-Verlauf. Pro Ziel
 läuft höchstens ein Lauf; startet dieser Teil neu, gilt ein offener Lauf als unterbrochen.
 """
@@ -27,13 +30,14 @@ from kern.einstellungen import einstellungen
 from kern.zeit import LOCAL_TZ
 from services import announce, update_verlauf
-from services.homelab import apps, arcane, inventar, kanal
+from services.homelab import apps, arcane, inventar, kanal, karenz
 log = logging.getLogger(__name__)
 WARTEN_NACH_UPDATE_S = 20
 ZEITLIMIT_UPDATE_S = 45 * 60
 ZEITLIMIT_KURZ_S = 15 * 60
 ZEITLIMIT_SICHERUNG_S = 40 * 60     # Sicherung und Zurückspielen (der Ausführer gibt nach 30 min auf)
 BETREFF = "[Homelab-Update]"
 _lock = threading.RLock()     # _merken liest und schreibt unter derselben Sperre
@@ -65,6 +69,15 @@ def laeufe(anzahl: int = 20) -> list[dict]:
    return list(reversed(_laeufe()))[:anzahl]
 def laufende_ziele() -> set[str]:
    """Ziele mit einem laufenden Update. Liest nur die Datei — geht also auch im Steward (Wächter, Pflege)."""
    return {str(x.get("ziel")) for x in _laeufe() if x.get("status") == "laeuft"}
 def laeuft(ziel_id: str) -> bool:
    return ziel_id in laufende_ziele()
 def unterbrochene_abschliessen() -> None:
    """Beim Start: Läufe, die noch „läuft“ heißen, gehörten zum vorigen Prozess."""
    for lauf in _laeufe():
@@ -123,39 +136,72 @@ def pruefen(vorher: dict, nachher: dict | None, baustein: str) -> list[str]:
    return fehler
-def _gast_lauf(lauf: dict, gast: dict) -> None:
+def _rueckweg_anlegen(lauf: dict, gast: dict) -> tuple[str, str] | None:
-    vmid, baustein = gast["vmid"], lauf["baustein"]
+    """Vor dem Update: ("snapshot", Name), wo keiner geht ("sicherung", Archiv), sonst None. Wirft RuntimeError,
-    snapshot = None
+    wenn der Schritt scheitert (etwa zu wenig Platz) — dann beginnt das Update gar nicht."""
-    try:
+    vmid = gast["vmid"]
    if gast.get("snapshot_moeglich"):
        a = _auftrag(lauf, "snapshot", {"vmid": vmid}, ZEITLIMIT_KURZ_S)
        m = re.search(r"snapshot=(mc2-\d{8}-\d{6})", a.get("text") or "")
-            snapshot = m.group(1) if m else None
+        return ("snapshot", m.group(1)) if m else None
    if gast.get("sicherung_moeglich"):
        a = _auftrag(lauf, "sichern", {"vmid": vmid}, ZEITLIMIT_SICHERUNG_S)
        m = re.search(r"^sicherung=(\S+)$", a.get("text") or "", re.MULTILINE)
        if not m:
            raise RuntimeError("sichern: Der Ausführer nannte keine Sicherung")
        return ("sicherung", m.group(1))
    return None
 def _zurueck(lauf: dict, vmid: int, rueckweg: tuple[str, str]) -> str:
    """Den Rückweg gehen; Rückgabe: was geschah (für die Meldung). Wirft RuntimeError, wenn er scheitert."""
    art, name = rueckweg
    if art == "snapshot":
        _auftrag(lauf, "zurueck", {"vmid": vmid, "snapshot": name}, ZEITLIMIT_KURZ_S)
        return f"automatisch zurück auf den Snapshot {name}"
    _auftrag(lauf, "sicherung_zurueck", {"vmid": vmid, "sicherung": name}, ZEITLIMIT_SICHERUNG_S)
    return f"automatisch die Sicherung von vorher zurückgespielt ({name.rsplit('/', 1)[-1]})"
 def _gast_lauf(lauf: dict, gast: dict) -> None:
    vmid, baustein = gast["vmid"], lauf["baustein"]
    try:
        rueckweg = _rueckweg_anlegen(lauf, gast)
    except RuntimeError as exc:
        _ende(lauf, "fehler", f"{lauf['name']}: Update nicht begonnen — {str(exc).rstrip('.')}. "
                              "Am Gerät wurde nichts geändert.")
        return
    try:
        _auftrag(lauf, "update" if baustein == "app" else "os_update", {"vmid": vmid}, ZEITLIMIT_UPDATE_S)
        time.sleep(WARTEN_NACH_UPDATE_S)
        fehler = pruefen(gast, _frischer_gast(lauf, vmid), baustein)
    except RuntimeError as exc:
        fehler = [str(exc)]
    if not fehler:
-        _aufraeumen(lauf, vmid, behalten=snapshot)
+        _aufraeumen(lauf, vmid, rueckweg)
        _ende(lauf, "eingespielt", f"{lauf['name']}: eingespielt, Prüfung grün.")
        return
-    if snapshot:
+    if rueckweg:
        try:
-            _auftrag(lauf, "zurueck", {"vmid": vmid, "snapshot": snapshot}, ZEITLIMIT_KURZ_S)
+            wie = _zurueck(lauf, vmid, rueckweg)
            _ende(lauf, "zurueckgerollt",
-                  f"{lauf['name']}: Update gescheitert ({' '.join(fehler)}) — automatisch zurück auf den Snapshot "
+                  f"{lauf['name']}: Update gescheitert ({' '.join(fehler)}) — {wie}. Läuft wieder wie vorher.",
-                  f"{snapshot}. Läuft wieder wie vorher.", dringend=True)
+                  dringend=True)
            return
        except RuntimeError as exc:
            fehler.append(f"Auch der Rückweg scheiterte: {exc}")
            if rueckweg[0] == "sicherung":
                fehler.append(f"Die Sicherung {rueckweg[1]} liegt weiter auf dem Proxmox-Host.")
    _ende(lauf, "fehler", f"{lauf['name']}: Update gescheitert — {' '.join(fehler)}"
-                          + ("" if snapshot else " Es gab keinen Snapshot, also keinen automatischen Rückweg."),
+                          + ("" if rueckweg else " Es gab weder Snapshot noch Sicherung, also keinen automatischen "
                                                 "Rückweg."),
          dringend=True)
-def _aufraeumen(lauf: dict, vmid: int, behalten: str | None) -> None:
+def _aufraeumen(lauf: dict, vmid: int, rueckweg: tuple[str, str] | None) -> None:
-    """Ältere Snapshots des Orchestrators für diesen Gast löschen; den neuesten behalten."""
+    """Ältere Snapshots des Orchestrators für diesen Gast löschen, nach einer Sicherung auch ältere Sicherungen;
    der Rückweg dieses Laufs (der neueste) bleibt."""
    art, behalten = rueckweg or (None, None)
    gast = inventar.gast(lauf["ziel"]) or {}
    for name in gast.get("snapshots") or []:
        if name.startswith("mc2-") and name != behalten:
@@ -163,6 +209,14 @@ def _aufraeumen(lauf: dict, vmid: int, behalten: str | None) -> None:
                _auftrag(lauf, "snapshot_loeschen", {"vmid": vmid, "snapshot": name}, ZEITLIMIT_KURZ_S)
            except RuntimeError:
                lauf["schritte"].append(f"Alter Snapshot {name} blieb stehen.")
    if art != "sicherung":
        return
    for volid in gast.get("sicherungen") or []:
        if volid != behalten:
            try:
                _auftrag(lauf, "sicherung_loeschen", {"vmid": vmid, "sicherung": volid}, ZEITLIMIT_KURZ_S)
            except RuntimeError:
                lauf["schritte"].append(f"Alte Sicherung {volid} blieb stehen.")
 def _docker_lauf(lauf: dict, basis: str) -> None:
@@ -224,7 +278,7 @@ def starten(ziel_id: str, baustein: str) -> dict:
    """Knopf „Jetzt updaten“. Rückgabe {"ok": True, "lauf": id} oder {"ok": False, "detail": …}."""
    if baustein not in ("app", "os", "pakete", "neustart", "docker"):
        return {"ok": False, "detail": "Unbekannter Baustein."}
-    if any(x.get("ziel") == ziel_id and x.get("status") == "laeuft" for x in _laeufe()):
+    if laeuft(ziel_id):
        return {"ok": False, "detail": "Für dieses Gerät läuft schon ein Update."}
    if ziel_id == "pve":
        name, gast = "Proxmox-Host", None
@@ -247,6 +301,8 @@ def starten(ziel_id: str, baustein: str) -> dict:
        if baustein == "app" and (not app or app.weg != "skript"):
            return {"ok": False, "detail": f"{name} aktualisiert sich nicht über das Community-Script "
                                          "(eigene Oberfläche oder Pakete)."}
        if baustein == "app" and (sperre := karenz.pruefen(app.kennung)["gesperrt"]):
            return {"ok": False, "detail": sperre}
    lauf = _neuer_lauf(ziel_id, baustein, name)
    ziel = _host_lauf if gast is None else (lambda lf: _gast_lauf(lf, gast))
    threading.Thread(target=ziel, args=(lauf,), name=f"homelab-{ziel_id}", daemon=True).start()
@@ -487,6 +487,7 @@ PRUEFUNGEN = {
            pruefe_festgehalten, pruefe_partner),
    "homelab": (pruefe_platte, pruefe_partner, pruefe_ausfuehrer, pruefe_gaeste),
 }[ROLLE]
 PRUEFUNGEN += (__import__("services.homelab.pflege").homelab.pflege.pruefe_paketlisten,) if ROLLE == "homelab" else ()
 BETREFF_PROBLEM, BETREFF_OK = {"box": ("[Box-Problem]", "[Box wieder ok]"),
                               "homelab": ("[Homelab-Problem]", "[Homelab wieder ok]")}[ROLLE]
@@ -32,6 +32,7 @@ def daten(tmp_path, monkeypatch):
        "tag": neueste.get(repo, "2026-07-24-r1"), "tag_roh": roh.get(repo, neueste.get(repo, "2026-07-24-r1")),
        "version": neueste.get(repo, "2026-07-24-r1"), "datum": "2026-07-24"})
    monkeypatch.setattr(inventar, "erreichbar", lambda url: True)
    monkeypatch.setattr(inventar.karenz, "skript_geaendert", lambda kennung: 0.0)   # Update-Skripte alt, kein Netz
    monkeypatch.setattr(inventar.arcane, "app_version",
                        lambda url: {"installiert": "2.12.0", "neu": "2.13.1", "update": True})
    monkeypatch.delenv("MC_ARCANE_KEY", raising=False)
@@ -0,0 +1,110 @@
 """Wartezeit nach einer Änderung am Update-Skript (24.09.2026): ct/<kennung>.sh kommt ungepinnt von GitHub und
 läuft als root im Gast — frisch Geändertes erst nach MC_HOMELAB_KARENZ_H Stunden."""
 import json
 from datetime import datetime
 from pathlib import Path
 import pytest
 from kern import einstellungen as einstellungen_mod
 from kern.zeit import LOCAL_TZ
 from services.homelab import inventar, kanal, karenz, updates
 BERICHT = json.loads((Path(__file__).parent / "fixtures" / "pve-bericht.json").read_text(encoding="utf-8"))
 # 23.09.2026 14:05 in Berlin = 12:05 UTC, so wie GitHub es schreibt.
 GEAENDERT = "2026-09-23T12:05:11Z"
 GEAENDERT_TS = datetime(2026, 9, 23, 14, 5, 11, tzinfo=LOCAL_TZ).timestamp()
 def _github(datum: str = GEAENDERT, pfade: list | None = None):
    def antwort(pfad, timeout=8):
        if pfade is not None:
            pfade.append(pfad)
        return [{"sha": "abc", "commit": {"author": {"date": "2026-09-01T08:00:00Z"},
                                          "committer": {"date": datum}}}]
    return antwort
 def test_frische_aenderung_sperrt_mit_berliner_zeiten(monkeypatch):
    pfade: list[str] = []
    monkeypatch.setattr(karenz, "github_json", _github(pfade=pfade))
    monkeypatch.delenv("MC_HOMELAB_KARENZ_H", raising=False)
    stand = karenz.pruefen("gitea", jetzt=GEAENDERT_TS + 10 * 3600)
    assert stand == {"gesperrt": "Das Update-Skript wurde am 23.09. geändert; zur Sicherheit erst ab 25.09. 14:05.",
                     "geprueft": True}
    # Das Datum, an dem die Änderung auf main landete (committer), nicht das des Autors.
    assert pfade == ["repos/community-scripts/ProxmoxVE/commits?path=ct/gitea.sh&sha=main&per_page=1"]
    assert karenz.pruefen("gitea", jetzt=GEAENDERT_TS + 48 * 3600) == {"gesperrt": None, "geprueft": True}
 def test_wartezeit_einstellbar(monkeypatch):
    monkeypatch.setattr(karenz, "github_json", _github())
    monkeypatch.setenv("MC_HOMELAB_KARENZ_H", "0")
    assert karenz.pruefen("gitea", jetzt=GEAENDERT_TS + 60)["gesperrt"] is None
    monkeypatch.setenv("MC_HOMELAB_KARENZ_H", "72")
    assert "erst ab 26.09. 14:05" in karenz.pruefen("gitea", jetzt=GEAENDERT_TS + 50 * 3600)["gesperrt"]
    monkeypatch.setenv("MC_HOMELAB_KARENZ_H", "viel")
    assert karenz.karenz_s() == 48 * 3600
 def test_ohne_antwort_von_github_nicht_blockieren(monkeypatch):
    def kaputt(pfad, timeout=8):
        raise RuntimeError("403 rate limit")
    monkeypatch.setattr(karenz, "github_json", kaputt)
    assert karenz.pruefen("gitea") == {"gesperrt": None, "geprueft": False}
    monkeypatch.setattr(karenz, "github_json", lambda pfad, timeout=8: [])      # Skript gibt es nicht
    assert karenz.pruefen("gitea")["geprueft"] is False
    assert karenz.pruefen("../../etc")["geprueft"] is False                      # fremde Kennung: keine Abfrage
 # --- Übersicht und Knopf -------------------------------------------------------------------------
@pytest.fixture
 def gitea_neu(tmp_path, monkeypatch):
    """Gitea 1.27.2 installiert, 1.27.3 verfügbar (der Stand vor dem ersten echten Lauf am 24.09.)."""
    monkeypatch.setenv("MC_DATEN_DIR", str(tmp_path))
    einstellungen_mod.einstellungen.cache_clear()
    monkeypatch.setattr(inventar, "neueste_version", lambda repo: {
        "tag": "v1.27.3", "tag_roh": "v1.27.3", "version": "1.27.3" if repo == "go-gitea/gitea" else "0.0.1",
        "datum": "2026-09-20"})
    monkeypatch.setattr(inventar, "erreichbar", lambda url: True)
    monkeypatch.setattr(inventar.arcane, "app_version", lambda url: None)
    monkeypatch.delenv("MC_HOMELAB_KARENZ_H", raising=False)
    kanal.bericht_speichern(BERICHT)
    yield
    einstellungen_mod.einstellungen.cache_clear()
 def _gitea_app() -> dict:
    ziele = {z["id"]: z for z in inventar.ziele()["ziele"]}
    return {b["id"]: b for b in ziele["ct-104"]["bausteine"]}["app"]
 def test_frisch_geaendertes_skript_neu_aber_ohne_knopf(gitea_neu, monkeypatch):
    jetzt = datetime.now(LOCAL_TZ).timestamp()
    monkeypatch.setattr(karenz, "skript_geaendert", lambda kennung: jetzt - 3600)
    app = _gitea_app()
    assert (app["zustand"], app["kurz"], app["aktion"]) == ("neu", "1.27.2 → 1.27.3", None)
    assert app["grund"].startswith("Das Update-Skript wurde am ") and "zur Sicherheit erst ab" in app["grund"]
    # Der Knopf fehlt nicht nur: auch die Schnittstelle lehnt mit demselben Satz ab, und kein Lauf beginnt.
    antwort = updates.starten("ct-104", "app")
    assert antwort == {"ok": False, "detail": app["grund"]}
    assert updates.laufende_ziele() == set() and kanal.liste() == []
 def test_altes_skript_knopf_wie_bisher(gitea_neu, monkeypatch):
    monkeypatch.setattr(karenz, "skript_geaendert", lambda kennung: 0.0)
    app = _gitea_app()
    assert app["aktion"]["frage"] == ("Gitea jetzt aktualisieren? Vorher wird ein Snapshot angelegt; ist die Prüfung "
                                      "danach rot, geht es von selbst zurück.")
 def test_ungeprueft_sagt_es_die_rueckfrage(gitea_neu, monkeypatch):
    def kaputt(kennung):
        raise RuntimeError("GitHub antwortet nicht")
    monkeypatch.setattr(karenz, "skript_geaendert", kaputt)
    app = _gitea_app()
    assert app["aktion"] is not None and app["aktion"]["frage"].endswith(karenz.NICHT_GEPRUEFT)
    assert app["aktion"]["frage"] == ("Gitea jetzt aktualisieren? Vorher wird ein Snapshot angelegt; ist die Prüfung "
                                      "danach rot, geht es von selbst zurück. Ob das Skript kürzlich geändert wurde, "
                                      "ließ sich nicht prüfen.")
@@ -0,0 +1,167 @@
 """Wöchentliches Suchen (services/homelab/pflege.py, 24.09.2026): Der Homelab-Teil erneuert alte Paketlisten in den
 Gästen selbst — nachts, einmal je Gast und Tag, nie neben einem Update. Mit dem echten Bericht vom 24.09.
 (tests/fixtures/pve-bericht.json): AdGuard, NPMplus, NetBird, PVE Scripts Local und Gitea haben alte Listen, die des
 PBS sind frisch."""
 import json
 import os
 import subprocess
 import sys
 import time
 from collections import Counter
 from datetime import datetime
 from pathlib import Path
 import pytest
 from kern import einstellungen as einstellungen_mod
 from kern.zeit import LOCAL_TZ
 from services import waechter
 from services.homelab import kanal, pflege, updates
 BERICHT = json.loads((Path(__file__).parent / "fixtures" / "pve-bericht.json").read_text(encoding="utf-8"))
 BACKEND = Path(__file__).resolve().parents[1]
 NACHT = datetime(2026, 9, 25, 3, 0, tzinfo=LOCAL_TZ).timestamp()
 TAG = datetime(2026, 9, 25, 14, 0, tzinfo=LOCAL_TZ).timestamp()
 NAECHSTE_NACHT = datetime(2026, 9, 26, 3, 0, tzinfo=LOCAL_TZ).timestamp()
 ALTE_LISTEN = {100, 101, 102, 103, 104}
@pytest.fixture
 def uhr(tmp_path, monkeypatch):
    """Daten in tmp; der Ausführer hat zuletzt um uhr["zuletzt"] berichtet."""
    monkeypatch.setenv("MC_DATEN_DIR", str(tmp_path))
    einstellungen_mod.einstellungen.cache_clear()
    stand = {"zuletzt": NACHT}
    monkeypatch.setattr(kanal, "zuletzt", lambda: stand["zuletzt"])
    monkeypatch.setattr(waechter, "TROCKEN", False)
    kanal.bericht_speichern(BERICHT)
    yield stand
    einstellungen_mod.einstellungen.cache_clear()
 def _suchen() -> Counter:
    """Wie viele Aufträge „suchen“ es je Gast gibt."""
    return Counter(a["parameter"]["vmid"] for a in kanal.liste() if a["aktion"] == "suchen")
 def _ausfuehrer_antwortet(scheitert: set[int] = frozenset()) -> None:
    while (a := kanal.naechster()):
        vmid = a["parameter"].get("vmid")
        if vmid in scheitert:
            kanal.ergebnis(a["id"], 100, "Hit:1 http://deb.debian.org/debian trixie InRelease\n"
                                         "E: Failed to fetch http://security.debian.org  Temporary failure resolving")
        else:
            kanal.ergebnis(a["id"], 0, "Hit:1 http://deb.debian.org/debian trixie InRelease")
 def test_nachts_einmal_je_gast_und_tag(uhr):
    assert pflege.pruefe_paketlisten(jetzt=NACHT) == []
    assert set(_suchen()) == ALTE_LISTEN                   # PBS: frische Listen; die Arcane-VM ist kein Container
    pflege.pruefe_paketlisten(jetzt=NACHT + 60)            # nächster Takt: offen, nichts doppelt
    _ausfuehrer_antwortet()
    pflege.pruefe_paketlisten(jetzt=NACHT + 120)           # erledigt, Listen im Bericht noch alt: heute nicht mehr
    assert _suchen() == Counter({v: 1 for v in ALTE_LISTEN})
    zustand = pflege.zustand()
    assert zustand["nacht"] == "2026-09-25"
    assert zustand["gaeste"]["100"]["name"] == "AdGuard Home" and zustand["gaeste"]["100"]["fehlschlaege"] == 0
    uhr["zuletzt"] = NAECHSTE_NACHT                          # am nächsten Tag wieder, falls die Listen alt bleiben
    pflege.pruefe_paketlisten(jetzt=NAECHSTE_NACHT)
    assert _suchen() == Counter({v: 2 for v in ALTE_LISTEN})
 def test_tagsueber_nur_wenn_die_nacht_verpasst_wurde(uhr):
    uhr["zuletzt"] = TAG
    pflege._schreiben({"nacht": "2026-09-25", "gaeste": {}})       # die Instanz war letzte Nacht da
    pflege.pruefe_paketlisten(jetzt=TAG)
    assert _suchen() == Counter()                                 # also bis zur nächsten Nacht warten
    pflege._schreiben({"nacht": "2026-09-24", "gaeste": {}})       # letzte Nacht war sie aus
    pflege.pruefe_paketlisten(jetzt=TAG)
    assert set(_suchen()) == ALTE_LISTEN                          # dann eben jetzt
    assert pflege.zustand()["nacht"] == "2026-09-24"              # tagsüber gilt keine Nacht als gesehen
 def test_vor_zwei_uhr_zaehlt_die_nacht_von_gestern(uhr):
    kurz_vor_zwei = datetime(2026, 9, 26, 1, 30, tzinfo=LOCAL_TZ).timestamp()
    uhr["zuletzt"] = kurz_vor_zwei
    pflege._schreiben({"nacht": "2026-09-25", "gaeste": {}})
    pflege.pruefe_paketlisten(jetzt=kurz_vor_zwei)
    assert _suchen() == Counter()
 def test_nie_neben_einem_update_oder_einem_offenen_auftrag(uhr):
    updates._merken({"id": "l1", "ziel": "ct-104", "baustein": "app", "name": "Gitea", "status": "laeuft",
                     "start": time.time(), "ende": None, "ergebnis": None, "text": None, "schritte": []})
    kanal.anlegen("os_update", {"vmid": 100})
    pflege.pruefe_paketlisten(jetzt=NACHT)
    assert set(_suchen()) == ALTE_LISTEN - {100, 104}
 def test_ohne_ausfuehrer_keine_auftraege(uhr):
    uhr["zuletzt"] = NACHT - 3600                                  # letzter Bericht vor einer Stunde
    pflege.pruefe_paketlisten(jetzt=NACHT)
    assert _suchen() == Counter() and pflege.zustand().get("nacht") is None
 def test_ausfuehrer_im_nur_lesen_modus_bekommt_keine_auftraege(uhr):
    nur_lesen = json.loads(json.dumps(BERICHT))
    nur_lesen["host"]["nur_lesen"] = True
    kanal.bericht_speichern(nur_lesen)
    pflege.pruefe_paketlisten(jetzt=NACHT)
    assert _suchen() == Counter() and pflege.zustand().get("nacht") is None
 def test_trockenlauf_legt_nichts_an(uhr, monkeypatch):
    monkeypatch.setattr(waechter, "TROCKEN", True)
    pflege.pruefe_paketlisten(jetzt=NACHT)
    assert _suchen() == Counter()
 def test_wiederholtes_scheitern_wird_ein_gelber_hinweis(uhr):
    pflege.pruefe_paketlisten(jetzt=NACHT)
    _ausfuehrer_antwortet(scheitert={100})
    assert pflege.pruefe_paketlisten(jetzt=NACHT + 60) == []       # einmal ist noch kein Muster
    uhr["zuletzt"] = NAECHSTE_NACHT
    pflege.pruefe_paketlisten(jetzt=NAECHSTE_NACHT)                # neuer Tag, neuer Versuch
    _ausfuehrer_antwortet(scheitert={100})
    befunde = pflege.pruefe_paketlisten(jetzt=NAECHSTE_NACHT + 60)
    assert [(b.id, b.stufe, b.titel, b.sofort) for b in befunde] == [
        ("pflege:ct-100", "gelb", "AdGuard Home: Die Suche nach Updates scheitert", True)]
    assert "2-mal hintereinander" in befunde[0].text and "E: Failed to fetch" in befunde[0].text
    # Listen wieder frisch (etwa von Hand gesucht): Hinweis weg, Zähler vergessen.
    frisch = json.loads(json.dumps(BERICHT))
    next(g for g in frisch["gaeste"] if g["vmid"] == 100)["os_updates"]["listen_stand"] = int(NAECHSTE_NACHT)
    kanal.bericht_speichern(frisch)
    assert pflege.pruefe_paketlisten(jetzt=NAECHSTE_NACHT + 120) == []
    assert pflege.zustand()["gaeste"]["100"]["fehlschlaege"] == 0
 def test_der_waechter_zeigt_den_hinweis_ohne_telegram(uhr, tmp_path, monkeypatch):
    pflege._schreiben({"nacht": "2026-09-25", "gaeste": {"100": {
        "name": "AdGuard Home", "tag": "2026-09-25", "auftrag": None, "fehlschlaege": 3, "fehler": "E: kaputt"}}})
    uhr["zuletzt"] = TAG
    monkeypatch.setattr(waechter, "STORE_PATH", tmp_path / "mc2-waechter.json")
    monkeypatch.setattr(waechter, "_stand", {"hinweise": {}, "kandidaten": {}, "verlauf": [], "auto": {}, "stand": 0.0})
    monkeypatch.setattr(waechter, "_update_laeuft", lambda: False)
    meldungen: list[str] = []
    monkeypatch.setattr(waechter, "_telegram", lambda betreff, text: meldungen.append(text))
    def pruefe_paketlisten():
        return pflege.pruefe_paketlisten(jetzt=TAG)
    monkeypatch.setattr(waechter, "PRUEFUNGEN", (pruefe_paketlisten,))
    waechter.takt()
    hinweis = waechter._stand["hinweise"]["pflege:ct-100"]
    assert hinweis["stufe"] == "gelb" and "3-mal" in hinweis["text"] and meldungen == []
@pytest.mark.skipif(os.name != "posix", reason="Die Dateisperre gibt es nur unter Linux (am PC: nur Thread-Sperre).")
 def test_auftraege_aus_zwei_prozessen_gehen_nicht_verloren(tmp_path):
    """Oberfläche und Steward schreiben beide in die Auftragsliste; ohne Dateisperre überschrieben sie sich."""
    code = ("from services.homelab import kanal\n"
            "for i in range(40):\n"
            "    kanal.anlegen('suchen', {'vmid': 100 + i})\n")
    env = {**os.environ, "MC_DATEN_DIR": str(tmp_path), "MC_ROLLE": "homelab"}
    prozesse = [subprocess.Popen([sys.executable, "-c", code], cwd=BACKEND, env=env) for _ in range(3)]
    assert all(p.wait(timeout=120) == 0 for p in prozesse)
    auftraege = json.loads((tmp_path / "ausfuehrer-auftraege.json").read_text(encoding="utf-8"))
    assert len(auftraege) == 120
@@ -0,0 +1,439 @@
 """PBS: Sicherung statt Snapshot (24.09.2026). Der Ausführer über Fakes (die Speicher und CT 105 so, wie sie am
 24.09. auf dem Proxmox-Host standen: pvesm status, storage.cfg, pct config 105), der Ablauf im Homelab-Teil mit
 einem nachgespielten Ausführer."""
 import importlib.util
 import json
 import threading
 import time
 from pathlib import Path
 import pytest
 from kern import einstellungen as einstellungen_mod
 from services.homelab import inventar, kanal, updates
 BERICHT = json.loads((Path(__file__).parent / "fixtures" / "pve-bericht.json").read_text(encoding="utf-8"))
 AUSFUEHRER = Path(__file__).resolve().parents[2] / "deploy" / "homelab" / "ausfuehrer.py"
 GIB = 1024 ** 3
 # /nodes/pve/storage am 24.09.2026 (Zahlen aus pvesm status, KiB → Bytes).
 SPEICHER = [
    {"storage": "local", "type": "dir", "content": "vztmpl,backup,iso", "active": 1, "enabled": 1, "shared": 0,
     "avail": 46835092 * 1024, "used": 46613140 * 1024, "total": 98497780 * 1024},
    {"storage": "local-lvm", "type": "lvmthin", "content": "rootdir,images", "active": 1, "enabled": 1, "shared": 0,
     "avail": 186793893 * 1024},
    {"storage": "pbs-qnap", "type": "pbs", "content": "backup", "active": 1, "enabled": 1, "shared": 1,
     "avail": 2282130304 * 1024},
 ]
 KONFIG_105 = {"rootfs": "local-lvm:vm-105-disk-0,size=10G", "mp0": "/mnt/qnap-backup/pbs-datastore,mp=/mnt/datastore",
              "unprivileged": 1, "onboot": 1, "ostype": "debian", "tags": "backup;community-script"}
 ALT = "local:backup/vzdump-lxc-105-2026_09_20-03_00_00.tar.zst"
 NOTIZ = "mc2-sicherung: vor einem Update durch den Homelab Orchestrator (proxmox-backup-server)"
 def _modul():
    spec = importlib.util.spec_from_file_location("ausfuehrer", AUSFUEHRER)
    modul = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(modul)
    return modul
 # --- Reine Auswertungen im Ausführer ----------------------------------------------------------
 def test_speicher_wahl_nimmt_local_und_nie_den_pbs():
    a = _modul()
    s, fehlt = a.sicherungsspeicher(SPEICHER)
    assert s["storage"] == "local" and fehlt is None
    # Ohne lokalen Speicher mit Sicherungen: klare Ablehnung, der PBS ist keine Ausweichlösung.
    s, fehlt = a.sicherungsspeicher([x for x in SPEICHER if x["storage"] != "local"])
    assert s is None and "Kein lokaler Speicher" in fehlt and "VZDump-Sicherung" in fehlt
    assert a.sicherungsspeicher([])[1].startswith("Die Speicher des Proxmox-Hosts")
    # Ein lokaler Speicher, der nur als „geteilt“ eingetragen ist (Netzfreigabe), zählt nicht als lokal.
    geteilt = [{**SPEICHER[0], "shared": 1}] + SPEICHER[1:]
    assert a.sicherungsspeicher(geteilt)[0] is None
 def test_speicher_wunsch_aus_der_konfiguration():
    a = _modul()
    assert a.sicherungsspeicher(SPEICHER, "local")[0]["storage"] == "local"
    s, fehlt = a.sicherungsspeicher(SPEICHER, "pbs-qnap")
    assert s is None and "nie" in fehlt and "selbst sichern" in fehlt
    assert "keine Sicherungen" in a.sicherungsspeicher(SPEICHER, "local-lvm")[1]
    assert "gibt es nicht" in a.sicherungsspeicher(SPEICHER, "usb")[1]
    aus = [{**SPEICHER[0], "active": 0}]
    assert "nicht verfügbar" in a.sicherungsspeicher(aus, "local")[1]
 def test_platzbedarf_ohne_bind_mount():
    a = _modul()
    # Läuft der Gast: die belegte rootfs. Der Bind-Mount (Datenspeicher des PBS) zählt nicht.
    assert a.sicherung_bedarf(KONFIG_105, 3 * GIB) == 3 * GIB
    # Läuft er nicht: die volle Größe der rootfs.
    assert a.sicherung_bedarf(KONFIG_105, None) == 10 * GIB
    # Weitere Volumes zählen nur mit backup=1.
    mit = {**KONFIG_105, "mp1": "local-lvm:vm-105-disk-1,mp=/daten,backup=1,size=8G",
           "mp2": "local-lvm:vm-105-disk-2,mp=/tmp2,size=4G"}
    assert a.sicherung_bedarf(mit, 3 * GIB) == 11 * GIB
    assert a.sicherung_bedarf({}, None) is None
 def test_platzpruefung_frei_groesser_als_belegt_mal_1_2():
    a = _modul()
    stand = a.sicherung_stand(SPEICHER)
    assert stand == {"speicher": "local", "art": "dir", "frei": 46835092 * 1024}
    assert a.sicherung_pruefen(stand, 3 * GIB) == (True, None)
    eng = {"speicher": "local", "frei": int(3.5 * GIB)}
    ok, grund = a.sicherung_pruefen(eng, 3 * GIB)          # 3,5 GB frei < 3 GB × 1,2
    assert not ok and grund == "Zu wenig Platz auf „local“: frei 3,5 GB, nötig mehr als 3,6 GB (belegt 3,0 GB × 1,2)."
    assert a.sicherung_pruefen({"speicher": "local", "frei": int(3.7 * GIB)}, 3 * GIB)[0] is True
    assert a.sicherung_pruefen({"speicher": None, "fehler": "Kein lokaler Speicher …"}, 3 * GIB) == (
        False, "Kein lokaler Speicher …")
 def test_modus_snapshot_wo_die_rootfs_es_kann():
    a = _modul()
    arten = {s["storage"]: s["type"] for s in SPEICHER}
    assert a.sicherung_modus(KONFIG_105, arten) == "snapshot"
    assert a.sicherung_modus({"rootfs": "local:105/vm-105-disk-0.raw,size=8G"}, arten) == "stop"
 def test_nur_eigene_sicherungen_zaehlen():
    a = _modul()
    eintraege = [
        {"volid": "local:backup/vzdump-lxc-105-2026_09_24-22_00_00.tar.zst", "ctime": 20, "notes": NOTIZ},
        {"volid": ALT, "ctime": 10, "notes": NOTIZ},
        {"volid": "local:backup/vzdump-lxc-105-2026_09_01-03_00_00.tar.zst", "ctime": 5, "notes": "von Hand"},
        {"volid": "local:backup/vzdump-lxc-104-2026_09_24-22_00_00.tar.zst", "ctime": 30, "notes": NOTIZ},
        {"volid": "pbs-qnap:backup/ct/105/2026-09-24T20:00:00Z", "ctime": 40, "notes": NOTIZ},
    ]
    assert [e["volid"] for e in a.eigene_sicherungen(eintraege, 105)] == [
        ALT, "local:backup/vzdump-lxc-105-2026_09_24-22_00_00.tar.zst"]
    assert len(a.eigene_sicherungen(eintraege)) == 3
    assert a.eigene_sicherungen(None) == []
 # --- Aktionen des Ausführers über Fakes -------------------------------------------------------
 class Host:
    """Der Proxmox-Host als Fake: beantwortet pvesh und führt „Befehle“ nur ins Protokoll."""
    def __init__(self, a, speicher=SPEICHER, belegt=3 * GIB):
        self.befehle: list[list[str]] = []
        self.inhalt = [{"volid": ALT, "vmid": 105, "ctime": 1790000000, "notes": NOTIZ},
                       {"volid": "local:backup/vzdump-lxc-105-2026_09_01-03_00_00.tar.zst", "vmid": 105,
                        "ctime": 1788000000, "notes": "von Hand"}]
        self.status = "running"
        self.antworten = {
            "/cluster/resources": [{"vmid": 105, "type": "lxc", "tags": "backup;community-script"},
                                   {"vmid": 106, "type": "qemu", "tags": ""}],
            "/nodes/pve/storage": speicher,
            "/nodes/pve/lxc/105/config": KONFIG_105,
            "/nodes/pve/lxc/105/status/current": lambda: {"status": self.status, "disk": belegt},
            "/nodes/pve/storage/local/content": lambda: self.inhalt,
        }
        a.NODE = "pve"
        a._pvesh = self.pvesh
        a._laufen = self.laufen
    def pvesh(self, pfad, *parameter):
        if pfad not in self.antworten:
            raise RuntimeError(f"pvesh {pfad}: gibt es im Fake nicht")
        wert = self.antworten[pfad]
        return wert() if callable(wert) else wert
    def laufen(self, befehl, zeitlimit=60, sanft=False):
        self.befehle.append(list(befehl))
        if befehl[0] == "vzdump":
            self.sanft_vzdump = sanft
            self.inhalt.append({"volid": "local:backup/vzdump-lxc-105-2026_09_24-22_10_00.tar.zst", "vmid": 105,
                                "ctime": int(time.time()), "notes": NOTIZ})
            return 0, "INFO: Backup job finished successfully"
        if befehl[:2] == ["pct", "shutdown"]:
            self.status = "stopped"
        return 0, "ok"
 def test_sichern_vzdump_auf_local_mit_notiz(monkeypatch):
    a = _modul()
    host = Host(a)
    antwort = a.ausfuehren({"aktion": "sichern", "parameter": {"vmid": 105}})
    assert antwort["code"] == 0
    assert antwort["text"].endswith("\nsicherung=local:backup/vzdump-lxc-105-2026_09_24-22_10_00.tar.zst")
    befehl = host.befehle[-1]
    assert befehl[:2] == ["vzdump", "105"] and host.sanft_vzdump is True
    optionen = dict(zip(befehl[2::2], befehl[3::2], strict=True))
    assert optionen["--storage"] == "local" and optionen["--mode"] == "snapshot"
    assert optionen["--remove"] == "0" and optionen["--compress"] == "zstd"
    assert optionen["--notes-template"].startswith("mc2-sicherung")
 def test_sichern_lehnt_ab_ohne_speicher_und_ohne_platz(monkeypatch):
    a = _modul()
    host = Host(a, speicher=[s for s in SPEICHER if s["storage"] != "local"])
    antwort = a.ausfuehren({"aktion": "sichern", "parameter": {"vmid": 105}})
    assert antwort["code"] == 2 and "Kein lokaler Speicher" in antwort["text"] and host.befehle == []
    host = Host(a, speicher=[{**SPEICHER[0], "avail": 2 * GIB}] + SPEICHER[1:], belegt=3 * GIB)
    antwort = a.ausfuehren({"aktion": "sichern", "parameter": {"vmid": 105}})
    assert antwort["code"] == 2 and "Zu wenig Platz" in antwort["text"] and host.befehle == []
    # Wunsch in der Konfiguration zeigt auf den PBS: nie.
    host = Host(a)
    monkeypatch.setattr(a, "SICHERUNG_SPEICHER", "pbs-qnap")
    antwort = a.ausfuehren({"aktion": "sichern", "parameter": {"vmid": 105}})
    assert antwort["code"] == 2 and "Proxmox-Backup-Server" in antwort["text"] and host.befehle == []
    # Ein Gast ohne erlaubtes Etikett bleibt unangetastet.
    assert "kein erlaubtes Etikett" in a.ausfuehren({"aktion": "sichern", "parameter": {"vmid": 106}})["text"]
 def test_sichern_ohne_kennzeichnung_ist_rot():
    a = _modul()
    host = Host(a)
    def ohne_notiz(befehl, zeitlimit=60, sanft=False):
        host.befehle.append(befehl)
        host.inhalt.append({"volid": "local:backup/vzdump-lxc-105-2026_09_24-22_10_00.tar.zst", "vmid": 105,
                            "ctime": int(time.time()), "notes": ""})
        return 0, "INFO: fertig"
    a._laufen = ohne_notiz
    antwort = a.ausfuehren({"aktion": "sichern", "parameter": {"vmid": 105}})
    assert antwort["code"] == 1 and "mc2-sicherung" in antwort["text"]
 def test_sicherung_zurueck_stoppt_spielt_zurueck_und_startet():
    a = _modul()
    host = Host(a)
    antwort = a.ausfuehren({"aktion": "sicherung_zurueck", "parameter": {"vmid": 105, "sicherung": ALT}})
    assert antwort["code"] == 0
    assert host.befehle == [
        ["pct", "shutdown", "105", "--timeout", "120", "--forceStop", "1"],
        ["pct", "restore", "105", ALT, "--force", "1", "--storage", "local-lvm"],   # der bisherige rootfs-Speicher
        ["pct", "start", "105"]]
 def test_zurueckgespielt_und_geloescht_wird_nur_eigenes():
    a = _modul()
    host = Host(a)
    fremd = "local:backup/vzdump-lxc-105-2026_09_01-03_00_00.tar.zst"        # von Hand, ohne mc2-Notiz
    for aktion in ("sicherung_zurueck", "sicherung_loeschen"):
        for volid in (fremd, "local:backup/vzdump-lxc-104-2026_09_20-03_00_00.tar.zst",
                      "pbs-qnap:backup/ct/105/2026-09-24T20:00:00Z", "/var/lib/vz/dump/x.tar.zst; reboot", None):
            antwort = a.ausfuehren({"aktion": aktion, "parameter": {"vmid": 105, "sicherung": volid}})
            assert antwort["code"] == 2, (aktion, volid)
    assert host.befehle == []
    assert a.ausfuehren({"aktion": "sicherung_loeschen", "parameter": {"vmid": 105, "sicherung": ALT}})["code"] == 0
    assert host.befehle == [["pvesm", "free", ALT]]
 def test_laufen_mit_zeitlimit_und_sanftem_abbruch():
    import sys
    a = _modul()
    assert a._laufen([sys.executable, "-c", "print('hallo')"]) == (0, "hallo")
    beginn = time.time()
    code, text = a._laufen([sys.executable, "-c", "import time; time.sleep(30)"], zeitlimit=1, sanft=True)
    assert (code, text) == (124, "Zeitlimit (1 s) überschritten") and time.time() - beginn < 20
    assert a._laufen(["gibt-es-nicht-mc2"])[0] == 127
 def test_unerwarteter_fehler_wird_beantwortet():
    """Sonst bliebe der Auftrag offen, und der Homelab-Teil wartete bis zu seinem Zeitlimit."""
    a = _modul()
    Host(a)
    a.AKTIONEN["sichern"] = lambda p: {}["kaputt"]
    antwort = a.ausfuehren({"aktion": "sichern", "parameter": {"vmid": 105}})
    assert antwort["code"] == 1 and "KeyError" in antwort["text"]
 def test_bericht_sagt_ob_eine_sicherung_geht():
    a = _modul()
    host = Host(a)
    host.antworten.update({
        "/cluster/resources": [{"vmid": 105, "type": "lxc", "node": "pve", "name": "proxmox-backup-server",
                                "status": "running", "tags": "backup;community-script", "disk": 3 * GIB}],
        "/nodes/pve/lxc/105/snapshot": [{"name": "current"}],
        "/version": {"version": "9.2.11"},
        "/nodes/pve/apt/update": [],
    })
    bericht = a.bericht()
    assert bericht["host"]["sicherung"] == {"speicher": "local", "art": "dir", "frei": 46835092 * 1024}
    assert bericht["host"]["nur_lesen"] is False           # die Pflege legt sonst keine Aufträge an
    pbs = bericht["gaeste"][0]
    assert pbs["snapshot_moeglich"] is False and "Bind-Mount" in pbs["snapshot_grund"]
    assert (pbs["sicherung_moeglich"], pbs["sicherung_grund"], pbs["sicherung_speicher"]) == (True, None, "local")
    assert pbs["sicherungen"] == [ALT]                       # die fremde Sicherung von Hand zählt nicht
    host.antworten["/nodes/pve/storage"] = SPEICHER[1:]      # kein lokaler Speicher mit Sicherungen
    pbs = a.bericht()["gaeste"][0]
    assert pbs["sicherung_moeglich"] is False and "Kein lokaler Speicher" in pbs["sicherung_grund"]
 # --- Der Ablauf im Homelab-Teil ---------------------------------------------------------------
@pytest.fixture
 def daten(tmp_path, monkeypatch):
    monkeypatch.setenv("MC_DATEN_DIR", str(tmp_path))
    monkeypatch.delenv("MC_AUSFUEHRER_TOKEN", raising=False)
    einstellungen_mod.einstellungen.cache_clear()
    monkeypatch.setattr(kanal, "_kontakt", {"zuletzt": None})
    monkeypatch.setattr(inventar, "neueste_version", lambda repo: None)
    monkeypatch.setattr(inventar, "erreichbar", lambda url: True)
    monkeypatch.setattr(inventar.karenz, "skript_geaendert", lambda kennung: 0.0)
    yield tmp_path
    einstellungen_mod.einstellungen.cache_clear()
 def _mit_sicherung(bericht: dict, **pbs) -> dict:
    neu = json.loads(json.dumps(bericht))
    gast = next(g for g in neu["gaeste"] if g["vmid"] == 105)
    gast.update(pbs)
    gast["os_updates"]["listen_stand"] = int(time.time())     # frische Listen: 58 Pakete bleiben „neu“
    return neu
 BERICHT_PBS = _mit_sicherung(BERICHT, sicherung_moeglich=True, sicherung_grund=None, sicherung_speicher="local",
                             sicherungen=[ALT])
 NEU = "local:backup/vzdump-lxc-105-2026_09_24-22_10_00.tar.zst"
 def _ausfuehrer(stopp: threading.Event, protokoll: list, fehlschlag: dict) -> None:
    """Nachgespielter Ausführer; fehlschlag: Aktion → (Code, Text)."""
    while not stopp.is_set():
        auftrag = kanal.naechster()
        if not auftrag:
            time.sleep(0.02)
            continue
        protokoll.append((auftrag["aktion"], auftrag["parameter"]))
        code, text = fehlschlag.get(auftrag["aktion"], (0, ""))
        if auftrag["aktion"] == "sichern" and code == 0:
            text = f"INFO: fertig\nsicherung={NEU}"
        elif auftrag["aktion"] == "bericht":
            text = json.dumps(_mit_sicherung(BERICHT_PBS, sicherungen=[ALT, NEU]))
        kanal.ergebnis(auftrag["id"], code, text)
@pytest.fixture
 def lauf(daten, monkeypatch):
    monkeypatch.setattr(updates, "WARTEN_NACH_UPDATE_S", 0)
    echt = kanal.warten
    monkeypatch.setattr(kanal, "warten", lambda aid, zeitlimit_s, takt_s=2.0: echt(aid, min(zeitlimit_s, 10), 0.02))
    meldungen: list[tuple[str, bool]] = []
    monkeypatch.setattr(updates, "_melden", lambda text, dringend=False: meldungen.append((text, dringend)))
    monkeypatch.setattr(updates.update_verlauf, "eintragen", lambda *a: None)
    kanal.bericht_speichern(BERICHT_PBS)
    stopp, protokoll, fehlschlag = threading.Event(), [], {}
    def starten(ziel: str, baustein: str) -> dict:
        threading.Thread(target=_ausfuehrer, args=(stopp, protokoll, fehlschlag), daemon=True).start()
        antwort = updates.starten(ziel, baustein)
        assert antwort["ok"], antwort
        ende = time.time() + 20
        while time.time() < ende:
            x = next((x for x in updates.laeufe() if x["id"] == antwort["lauf"]), {})
            if x.get("status") == "fertig":
                return x
            time.sleep(0.05)
        raise AssertionError("Lauf wurde nicht fertig")
    yield starten, protokoll, meldungen, fehlschlag
    stopp.set()
 def test_pbs_gruen_mit_sicherung_und_alte_sicherung_weg(lauf):
    starten, protokoll, meldungen, _ = lauf
    x = starten("ct-105", "os")
    assert x["ergebnis"] == "eingespielt"
    assert protokoll == [("sichern", {"vmid": 105}), ("os_update", {"vmid": 105}), ("bericht", {}),
                         ("sicherung_loeschen", {"vmid": 105, "sicherung": ALT})]   # die neue bleibt
    assert meldungen == [("Proxmox Backup Server: eingespielt, Prüfung grün.", False)]
 def test_pbs_rot_spielt_die_sicherung_zurueck(lauf, monkeypatch):
    starten, protokoll, meldungen, _ = lauf
    monkeypatch.setattr(inventar, "erreichbar", lambda url: False)
    x = starten("ct-105", "os")
    assert x["ergebnis"] == "zurueckgerollt"
    assert [p[0] for p in protokoll] == ["sichern", "os_update", "bericht", "sicherung_zurueck"]
    assert protokoll[-1][1] == {"vmid": 105, "sicherung": NEU}
    text, dringend = meldungen[0]
    assert dringend and "Sicherung von vorher zurückgespielt (vzdump-lxc-105-2026_09_24-22_10_00.tar.zst)" in text
 def test_pbs_rot_und_auch_das_zurueckspielen_scheitert(lauf, monkeypatch):
    starten, _, meldungen, fehlschlag = lauf
    monkeypatch.setattr(inventar, "erreichbar", lambda url: False)
    fehlschlag["sicherung_zurueck"] = (1, "restore failed")
    x = starten("ct-105", "os")
    assert x["ergebnis"] == "fehler"
    text, dringend = meldungen[0]
    assert dringend and "Auch der Rückweg scheiterte" in text and f"Die Sicherung {NEU} liegt weiter" in text
 def test_ohne_sicherung_beginnt_das_update_nicht(lauf):
    starten, protokoll, meldungen, fehlschlag = lauf
    fehlschlag["sichern"] = (2, "Abgelehnt: Zu wenig Platz auf „local“: frei 2,0 GB, nötig mehr als 3,6 GB.")
    x = starten("ct-105", "os")
    assert x["ergebnis"] == "fehler" and [p[0] for p in protokoll] == ["sichern"]
    text, dringend = meldungen[0]
    assert not dringend and "Update nicht begonnen" in text and "Zu wenig Platz" in text
    assert "nichts geändert" in text and ".." not in text
 def test_gescheiterter_snapshot_heisst_update_nicht_begonnen(lauf, monkeypatch):
    """Vorher hieß das „Update gescheitert … kein Rückweg“ (dringend), obwohl das Update nie lief."""
    starten, protokoll, meldungen, fehlschlag = lauf
    kanal.bericht_speichern(BERICHT)
    monkeypatch.setattr(inventar, "neueste_version", lambda repo: {
        "tag": "v1.27.3", "tag_roh": "v1.27.3", "version": "1.27.3", "datum": "2026-09-20"})
    fehlschlag["snapshot"] = (1, "snapshot feature is not available")
    x = starten("ct-104", "app")
    assert x["ergebnis"] == "fehler" and [p[0] for p in protokoll] == ["snapshot"]
    erwartet = ("Gitea: Update nicht begonnen — snapshot gescheitert: snapshot feature is not available. "
                "Am Gerät wurde nichts geändert.")
    assert meldungen == [(erwartet, False)]
 def test_weder_snapshot_noch_sicherung_wie_bisher(lauf, monkeypatch):
    starten, protokoll, meldungen, _ = lauf
    kanal.bericht_speichern(_mit_sicherung(BERICHT, sicherung_moeglich=False,
                                           sicherung_grund="Kein lokaler Speicher des Proxmox-Hosts …"))
    monkeypatch.setattr(inventar, "erreichbar", lambda url: False)
    x = starten("ct-105", "os")
    assert [p[0] for p in protokoll] == ["os_update", "bericht"] and x["ergebnis"] == "fehler"
    assert "weder Snapshot noch Sicherung" in meldungen[0][0]
 # --- Was die Übersicht dazu sagt ----------------------------------------------------------------
 def _pbs(bericht: dict) -> dict:
    kanal.bericht_speichern(bericht)
    return {z["id"]: z for z in inventar.ziele()["ziele"]}["ct-105"]
 def test_rueckweg_und_rueckfrage_mit_sicherung(daten):
    pbs = _pbs(BERICHT_PBS)
    assert pbs["rueckweg"].startswith("Sicherung auf „local“ vor jedem Update (Bind-Mount mp0")
    os_baustein = {b["id"]: b for b in pbs["bausteine"]}["os"]
    assert os_baustein["aktion"]["frage"] == ("58 Pakete im Gast jetzt einspielen? Vorher wird eine Sicherung "
                                              "angelegt; ist die Prüfung danach rot, wird sie zurückgespielt.")
 def test_ohne_sicherung_sagt_die_rueckfrage_warum(daten):
    pbs = _pbs(_mit_sicherung(BERICHT, sicherung_moeglich=False,
                              sicherung_grund="Zu wenig Platz auf „local“: frei 2,0 GB."))
    assert "Auch keine Sicherung: Zu wenig Platz" in pbs["rueckweg"]
    frage = {b["id"]: b for b in pbs["bausteine"]}["os"]["aktion"]["frage"]
    assert "keinen automatischen Rückweg" in frage and "Eine Sicherung geht auch nicht: Zu wenig Platz" in frage
    # Der alte Bericht (Ausführer noch ohne Sicherungen) bleibt wie bisher.
    assert _pbs(BERICHT)["rueckweg"] == "Bind-Mount mp0 (/mnt/qnap-backup/pbs-datastore) — nur Backup möglich"
 def test_waechter_laesst_gaeste_im_update_in_ruhe(daten):
    from services import waechter
    kanal.bericht_speichern(BERICHT_PBS)
    updates._merken({"id": "x1", "ziel": "ct-105", "baustein": "os", "name": "PBS", "status": "laeuft",
                     "start": time.time(), "ende": None, "ergebnis": None, "text": None, "schritte": []})
    ids = [e["id"] for e in inventar.erreichbarkeit()]
    assert "ct-105" not in ids and "ct-104" in ids
    assert all(b.id != "gast:ct-105" for b in waechter.pruefe_gaeste())
@@ -217,7 +217,8 @@ import steward
 from services import waechter
 print(json.dumps({"pruefungen": [p.__name__ for p in waechter.PRUEFUNGEN], "daten": str(waechter.DATEN_DIR)}))
 """)
-    assert ergebnis["pruefungen"] == ["pruefe_platte", "pruefe_partner", "pruefe_ausfuehrer", "pruefe_gaeste"]
+    assert ergebnis["pruefungen"] == ["pruefe_platte", "pruefe_partner", "pruefe_ausfuehrer", "pruefe_gaeste",
                                      "pruefe_paketlisten"]
 def test_homelab_rolle_hat_die_homelab_schnittstellen(tmp_path):
@@ -6,15 +6,22 @@ beim Homelab-Teil ab („Pull“, User-Entscheid 24.09.): Der Proxmox-Host öffn
 Container des Homelab-Teils braucht keinen Proxmox-Schlüssel.
  • Bericht    alle 10 Minuten (oder auf Anfrage): Host-Updates, Gäste mit Status, App-Version,
-               OS-Updates, Snapshots und ob ein Snapshot überhaupt geht. Nur lesend.
+               OS-Updates, Snapshots und ob ein Snapshot überhaupt geht; wo nicht, ob eine Sicherung
               geht, und die Sicherungen des Orchestrators. Nur lesend.
  • Aufträge   eine feste Liste von Aktionen (AKTIONEN); alles andere wird abgelehnt. Welche Gäste
               angefasst werden dürfen, entscheidet dieser Host selbst anhand der Etiketten:
               community-script oder watcher, aber nicht watcher-aus. Dem Server wird dabei nicht
               vertraut — er kann nur aus der Liste wählen.
  • Sicherung  wo kein Snapshot geht (PBS: Bind-Mount), vzdump auf einen lokalen Speicher des Hosts —
               nie auf einen Speicher der Art pbs, sonst sicherte sich der PBS selbst. Nur eigene
               Sicherungen (Notiz „mc2-sicherung“) werden zurückgespielt oder gelöscht. An der
               Speicher-Konfiguration ändert der Ausführer nichts.
 Nur die Standardbibliothek (Debian-Python des Hosts). Konfiguration: /etc/mc2-ausfuehrer.json
  {"server": "http://192.168.178.x:9001", "token": "<gemeinsames Geheimnis>", "node": "pve", "nur_lesen": false}
-„nur_lesen": true liefert nur den Bericht und lehnt jeden anderen Auftrag ab.
+„nur_lesen": true liefert nur den Bericht und lehnt jeden anderen Auftrag ab. Optional
 „sicherung_speicher": "<Speicher>" legt fest, wohin die Sicherungen gehen (sonst der erste lokale Speicher,
 der Sicherungen annimmt). ausfuehrer-einrichten.sh schreibt die Datei neu; den Schlüssel danach wieder eintragen.
 Aufruf:  ausfuehrer.py               Dauerbetrieb (systemd)
         ausfuehrer.py --bericht     Bericht einmal auf die Konsole (nur lesend, zum Prüfen)
@@ -42,6 +49,17 @@ SNAPSHOT_NAME = re.compile(r"^mc2-\d{8}-\d{6}$")
 # Speicherarten, auf denen Proxmox Snapshots kann.
 SNAPSHOT_SPEICHER = {"lvmthin", "zfspool", "rbd", "btrfs", "cephfs"}
 # Sicherungen vor einem Update, wo kein Snapshot geht (24.09.2026). Jede trägt diese Notiz; nur solche fasst der
 # Ausführer später wieder an (zurückspielen, löschen).
 SICHERUNG_KENNUNG = "mc2-sicherung"
 SICHERUNG_NOTIZ = SICHERUNG_KENNUNG + ": vor einem Update durch den Homelab Orchestrator ({{guestname}})"
 SICHERUNG_ARCHIV = re.compile(
    r"^([A-Za-z][\w.-]*):backup/vzdump-lxc-(\d+)-\d{4}_\d{2}_\d{2}-\d{2}_\d{2}_\d{2}\.tar(?:\.(?:zst|gz|lzo))?$")
 # Dateispeicher auf dem Host selbst. Netzspeicher nur, wenn „sicherung_speicher“ es so will — pbs nie.
 LOKALE_SICHERUNGSARTEN = {"dir", "btrfs"}
 PLATZ_FAKTOR = 1.2
 ZEITLIMIT_SICHERUNG_S = 30 * 60
 # App-Version je Community-Script (Kennung aus /usr/bin/update im Gast). Neuere Skripte legen die Version
 # in /root/.<app> ab; für die älteren steht hier, wie man sie erfährt. Festes Wissen dieses Hosts.
 VERSION_PROBEN = {
@@ -59,14 +77,27 @@ log = logging.getLogger("ausfuehrer")
 # --- Hilfen -----------------------------------------------------------------------------
-def _laufen(befehl: list[str], zeitlimit: int = 60) -> tuple[int, str]:
+def _laufen(befehl: list[str], zeitlimit: int = 60, sanft: bool = False) -> tuple[int, str]:
    """Befehl mit Zeitlimit. sanft (vzdump, pct restore): beim Zeitlimit zuerst SIGTERM — dann räumen sie Sperre
    und Snapshot selbst weg; hart abgebrochen wird erst zwei Minuten später."""
    try:
-        r = subprocess.run(befehl, capture_output=True, text=True, timeout=zeitlimit)
+        prozess = subprocess.Popen(befehl, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
-    except subprocess.TimeoutExpired:
+                                   errors="replace")
        return 124, f"Zeitlimit ({zeitlimit} s) überschritten"
    except OSError as exc:
        return 127, str(exc)
-    return r.returncode, (r.stdout + r.stderr).strip()
+    try:
        aus, fehler = prozess.communicate(timeout=zeitlimit)
    except subprocess.TimeoutExpired:
        if sanft:
            prozess.terminate()
            try:
                prozess.communicate(timeout=120)
            except subprocess.TimeoutExpired:
                pass
        prozess.kill()
        prozess.communicate()
        return 124, f"Zeitlimit ({zeitlimit} s) überschritten"
    return prozess.returncode, (aus + fehler).strip()
 def _pvesh(pfad: str, *parameter: str) -> object:
@@ -91,11 +122,16 @@ def erlaubt(etiketten: list[str]) -> bool:
 # --- Bericht (nur lesend) -------------------------------------------------------------------
-def _speicherart() -> dict[str, str]:
+def _speicher_liste() -> list[dict]:
    """Die Speicher des Hosts (Name, Art, Inhalt, frei …) — leer, wenn sie sich nicht lesen lassen."""
    try:
-        return {s["storage"]: s["type"] for s in _pvesh(f"/nodes/{NODE}/storage")}
+        return [s for s in _pvesh(f"/nodes/{NODE}/storage") or [] if isinstance(s, dict) and s.get("storage")]
    except Exception:
-        return {}
+        return []
 def _speicherart() -> dict[str, str]:
    return {s["storage"]: s.get("type") for s in _speicher_liste()}
 def snapshot_moeglich(konfig: dict, arten: dict[str, str]) -> tuple[bool, str | None]:
@@ -112,6 +148,117 @@ def snapshot_moeglich(konfig: dict, arten: dict[str, str]) -> tuple[bool, str |
    return True, None
 # --- Sicherung statt Snapshot (reine Auswertungen, ohne Host testbar) ---------------------------
 def _inhalt(speicher: dict) -> set[str]:
    return {x.strip() for x in str(speicher.get("content") or "").split(",") if x.strip()}
 def sicherungsspeicher(speicher: list[dict], wunsch: str | None = None) -> tuple[dict | None, str | None]:
    """Wohin die Sicherung vor einem Update geht: (Speicher, None) oder (None, was fehlt). Nie auf einen Speicher
    der Art pbs (der PBS würde sich selbst sichern). Ohne Wunsch der erste lokale Speicher, der Sicherungen annimmt.
    Die Speicher-Konfiguration selbst ist Sache des Users; hier wird nur gewählt oder abgelehnt."""
    if not speicher:
        return None, "Die Speicher des Proxmox-Hosts ließen sich nicht lesen."
    if wunsch:
        s = next((x for x in speicher if x.get("storage") == wunsch), None)
        if s is None:
            return None, (f"Den eingestellten Sicherungsspeicher „{wunsch}“ (sicherung_speicher in {KONFIG}) "
                          "gibt es nicht.")
        if s.get("type") == "pbs":
            return None, (f"„{wunsch}“ ist ein Proxmox-Backup-Server-Speicher. Dorthin sichert der Orchestrator nie — "
                          "der PBS würde sich selbst sichern.")
        if "backup" not in _inhalt(s):
            return None, f"„{wunsch}“ nimmt keine Sicherungen an (Inhalt „VZDump-Sicherung“ fehlt)."
        if not s.get("active"):
            return None, f"Der Sicherungsspeicher „{wunsch}“ ist gerade nicht verfügbar."
        return s, None
    for s in speicher:
        if (s.get("type") in LOKALE_SICHERUNGSARTEN and not s.get("shared") and s.get("active")
                and "backup" in _inhalt(s)):
            return s, None
    return None, ("Kein lokaler Speicher des Proxmox-Hosts nimmt Sicherungen an. In Proxmox unter Rechenzentrum → "
                  "Speicher bei einem lokalen Verzeichnis (z. B. „local“) den Inhalt „VZDump-Sicherung“ zulassen oder "
                  f"„sicherung_speicher“ in {KONFIG} eintragen.")
 def sicherung_stand(speicher: list[dict], wunsch: str | None = None) -> dict:
    """Für den Bericht: {"speicher", "art", "frei"} oder {"speicher": None, "fehler": was fehlt}."""
    s, fehlt = sicherungsspeicher(speicher, wunsch)
    if s is None:
        return {"speicher": None, "fehler": fehlt}
    return {"speicher": s["storage"], "art": s.get("type"), "frei": s.get("avail")}
 _EINHEIT = {"": 1, "K": 1024, "M": 1024 ** 2, "G": 1024 ** 3, "T": 1024 ** 4}
 def _groesse(text: str) -> int | None:
    """„10G“ → Bytes (Größenangaben der Proxmox-Konfiguration)."""
    m = re.fullmatch(r"(\d+(?:\.\d+)?)([KMGT]?)", str(text or "").strip().upper())
    return int(float(m.group(1)) * _EINHEIT[m.group(2)]) if m else None
 def _volume(wert: object) -> tuple[str, dict[str, str]]:
    teile = str(wert or "").split(",")
    return teile[0], dict(t.split("=", 1) for t in teile[1:] if "=" in t)
 def sicherung_bedarf(konfig: dict, belegt_rootfs: int | None) -> int | None:
    """Wie viel die Sicherung höchstens umfasst: die belegte rootfs (läuft der Gast nicht: ihre volle Größe) plus
    weitere Volumes mit backup=1. Bind-Mounts sichert vzdump nie mit — der Datenspeicher des PBS bleibt draußen."""
    _, optionen = _volume(konfig.get("rootfs"))
    bedarf = belegt_rootfs or _groesse(optionen.get("size", ""))
    if not bedarf:
        return None
    for schluessel, wert in konfig.items():
        if re.fullmatch(r"mp\d+", schluessel):
            quelle, optionen = _volume(wert)
            if not quelle.startswith("/") and optionen.get("backup") in ("1", "yes", "on", "true"):
                bedarf += _groesse(optionen.get("size", "")) or 0
    return bedarf
 def platz_reicht(frei: int | None, bedarf: int | None) -> bool:
    return frei is not None and bedarf is not None and frei > bedarf * PLATZ_FAKTOR
 def _gb(anzahl: float | None) -> str:
    return f"{(anzahl or 0) / 1024 ** 3:.1f} GB".replace(".", ",")
 def sicherung_pruefen(stand: dict, bedarf: int | None) -> tuple[bool, str | None]:
    """Geht vor dem Update eine Sicherung? stand wie sicherung_stand(); vorher Platz prüfen: frei > belegt × 1,2."""
    if not stand.get("speicher"):
        return False, stand.get("fehler") or "Es gibt keinen Speicher für Sicherungen."
    if not bedarf:
        return False, "Wie viel Platz der Gast belegt, ließ sich nicht lesen."
    if not platz_reicht(stand.get("frei"), bedarf):
        return False, (f"Zu wenig Platz auf „{stand['speicher']}“: frei {_gb(stand.get('frei'))}, nötig mehr als "
                       f"{_gb(bedarf * PLATZ_FAKTOR)} (belegt {_gb(bedarf)} × 1,2).")
    return True, None
 def sicherung_modus(konfig: dict, arten: dict[str, str]) -> str:
    """snapshot, wenn die rootfs Snapshots kann: Der Gast läuft durch, Bind-Mounts lässt vzdump aus. Sonst stop —
    kurz aus, braucht aber keinen Zwischenspeicher wie „suspend“, auf den vzdump sonst ausweicht."""
    speicher = str(konfig.get("rootfs") or "").split(":")[0]
    return "snapshot" if arten.get(speicher) in SNAPSHOT_SPEICHER else "stop"
 def eigene_sicherungen(eintraege: object, vmid: int | None = None) -> list[dict]:
    """Nur Sicherungen des Orchestrators (Notiz beginnt mit mc2-sicherung, Archiv vzdump-lxc-<vmid>-…), älteste
    zuerst; mit vmid nur die dieses Gasts."""
    eigene = []
    for e in eintraege if isinstance(eintraege, list) else []:
        m = SICHERUNG_ARCHIV.match(str(e.get("volid") or "")) if isinstance(e, dict) else None
        if not m or not str(e.get("notes") or "").startswith(SICHERUNG_KENNUNG):
            continue
        if vmid is None or int(m.group(2)) == vmid:
            eigene.append(e)
    return sorted(eigene, key=lambda e: (e.get("ctime") or 0, e["volid"]))
 def app_kennung_aus(update_skript: str) -> str | None:
    """Kennung des Community-Scripts aus /usr/bin/update. Zwei Formate: das alte lädt ct/<kennung>.sh direkt,
    das neue (schreibt jedes erfolgreiche Update seit 09/2026) setzt `export SCRIPT_SLUG="<kennung>"`."""
@@ -191,12 +338,17 @@ def _snapshots(art: str, vmid: int) -> list[str]:
        return []
-def _gast(eintrag: dict, arten: dict[str, str]) -> dict:
+def _eigene_sicherungen(speicher: str, vmid: int | None = None) -> list[dict]:
    parameter = ["--content", "backup"] + (["--vmid", str(vmid)] if vmid is not None else [])
    return eigene_sicherungen(_pvesh(f"/nodes/{NODE}/storage/{speicher}/content", *parameter), vmid)
 def _gast(eintrag: dict, arten: dict[str, str], sicherung: dict, sicherungen: dict[int, list[str]]) -> dict:
    art, vmid = eintrag["type"], int(eintrag["vmid"])
    etiketten = _etiketten(eintrag.get("tags"))
    gast = {"vmid": vmid, "art": art, "name": eintrag.get("name"), "status": eintrag.get("status"),
            "etiketten": etiketten, "erlaubt": erlaubt(etiketten), "uptime": eintrag.get("uptime"),
-            "snapshots": _snapshots(art, vmid)}
+            "snapshots": _snapshots(art, vmid), "sicherungen": sicherungen.get(vmid, [])}
    try:
        konfig = _pvesh(f"/nodes/{NODE}/{art}/{vmid}/config")
    except Exception as exc:
@@ -204,6 +356,13 @@ def _gast(eintrag: dict, arten: dict[str, str]) -> dict:
        return gast
    gast["onboot"] = bool(konfig.get("onboot"))
    gast["snapshot_moeglich"], gast["snapshot_grund"] = snapshot_moeglich(konfig, arten)
    if art == "lxc" and not gast["snapshot_moeglich"]:
        # Kein Snapshot: Geht wenigstens eine Sicherung (Speicher da, genug Platz)?
        belegt = eintrag.get("disk") if gast["status"] == "running" else None
        gast["sicherung_moeglich"], gast["sicherung_grund"] = sicherung_pruefen(
            sicherung, sicherung_bedarf(konfig, belegt))
        if gast["sicherung_moeglich"]:
            gast["sicherung_speicher"] = sicherung["speicher"]
    if art == "lxc":
        gast["ostype"] = konfig.get("ostype")
        gast["ip"] = _ip_lxc(vmid) if gast["status"] == "running" else None
@@ -217,7 +376,7 @@ def _gast(eintrag: dict, arten: dict[str, str]) -> dict:
 def _host() -> dict:
-    host: dict = {"node": NODE, "neustart_noetig": os.path.exists("/var/run/reboot-required")}
+    host: dict = {"node": NODE, "neustart_noetig": os.path.exists("/var/run/reboot-required"), "nur_lesen": NUR_LESEN}
    try:
        host["version"] = _pvesh("/version").get("version")
        host["updates"] = [{"paket": p.get("Package"), "alt": p.get("OldVersion"), "neu": p.get("Version"),
@@ -230,12 +389,23 @@ def _host() -> dict:
 def bericht() -> dict:
-    arten = _speicherart()
+    speicher = _speicher_liste()
    arten = {s["storage"]: s.get("type") for s in speicher}
    sicherung = sicherung_stand(speicher, SICHERUNG_SPEICHER)
    sicherungen: dict[int, list[str]] = {}
    if sicherung.get("speicher"):
        try:
            for e in _eigene_sicherungen(sicherung["speicher"]):
                sicherungen.setdefault(int(SICHERUNG_ARCHIV.match(e["volid"]).group(2)), []).append(e["volid"])
        except Exception:
            pass
    gaeste = []
    for eintrag in _pvesh("/cluster/resources", "--type", "vm") or []:
        if eintrag.get("node") == NODE and not eintrag.get("template"):
-            gaeste.append(_gast(eintrag, arten))
+            gaeste.append(_gast(eintrag, arten, sicherung, sicherungen))
-    return {"zeit": datetime.now(timezone.utc).isoformat(timespec="seconds"), "host": _host(),
+    host = _host()
    host["sicherung"] = sicherung
    return {"zeit": datetime.now(timezone.utc).isoformat(timespec="seconds"), "host": host,
            "gaeste": sorted(gaeste, key=lambda g: g["vmid"])}
@@ -312,6 +482,82 @@ def a_snapshot_loeschen(p: dict) -> tuple[int, str]:
    return _laufen([_werkzeug(art), "delsnapshot", str(vmid), name], 600)
 def _eigene_sicherung(vmid: int, volid: object) -> str:
    """Die Sicherung, wenn sie eine des Orchestrators für genau diesen Gast ist — sonst ValueError. Nur solche
    werden zurückgespielt oder gelöscht: Archiv vzdump-lxc-<vmid>-…, Notiz mc2-sicherung, kein PBS-Speicher."""
    volid = str(volid or "")
    m = SICHERUNG_ARCHIV.match(volid)
    if not m or int(m.group(2)) != vmid:
        raise ValueError("nur Sicherungen des Orchestrators für diesen Gast (vzdump-lxc-<Gast>-…)")
    if _speicherart().get(m.group(1)) == "pbs":
        raise ValueError("Sicherungen auf einem PBS-Speicher fasst der Orchestrator nicht an")
    if not any(e["volid"] == volid for e in _eigene_sicherungen(m.group(1), vmid)):
        raise ValueError(f"{volid} fehlt oder ist keine Sicherung des Orchestrators (Notiz „{SICHERUNG_KENNUNG}“)")
    return volid
 def a_sichern(p: dict) -> tuple[int, str]:
    """vzdump vor einem Update, wo kein Snapshot geht. Nur auf einen Speicher, den sicherungsspeicher() wählt, und
    nur mit genug Platz. Die neue Sicherung steht am Ende des Texts: sicherung=<volid>."""
    art, vmid = _gast_pruefen(p.get("vmid"))
    if art != "lxc":
        raise ValueError("Sicherungen vor einem Update gibt es nur für Container")
    speicher = _speicher_liste()
    ziel, fehlt = sicherungsspeicher(speicher, SICHERUNG_SPEICHER)
    if ziel is None:
        raise ValueError(fehlt)
    konfig = _pvesh(f"/nodes/{NODE}/lxc/{vmid}/config")
    status = _pvesh(f"/nodes/{NODE}/lxc/{vmid}/status/current") or {}
    bedarf = sicherung_bedarf(konfig, status.get("disk") if status.get("status") == "running" else None)
    ok, grund = sicherung_pruefen({"speicher": ziel["storage"], "frei": ziel.get("avail")}, bedarf)
    if not ok:
        raise ValueError(grund)
    modus = sicherung_modus(konfig, {s["storage"]: s.get("type") for s in speicher})
    beginn = time.time()
    # --remove 0: keine Aufräumregeln des Speichers anwenden (die träfen auch fremde Sicherungen dieses Gasts).
    # --lockwait 10: läuft gerade eine andere Sicherung, nach zehn Minuten sauber aufgeben statt drei Stunden warten.
    code, text = _laufen(["vzdump", str(vmid), "--storage", ziel["storage"], "--mode", modus, "--compress", "zstd",
                          "--remove", "0", "--lockwait", "10", "--notes-template", SICHERUNG_NOTIZ],
                         ZEITLIMIT_SICHERUNG_S, sanft=True)
    if code != 0:
        return code, text
    neue = [e for e in _eigene_sicherungen(ziel["storage"], vmid) if (e.get("ctime") or 0) >= beginn - 120]
    if not neue:
        return 1, text + "\nDie neue Sicherung ist nicht zu finden oder trägt die Notiz mc2-sicherung nicht."
    return 0, f"{text}\nsicherung={neue[-1]['volid']}"
 def a_sicherung_zurueck(p: dict) -> tuple[int, str]:
    """Rückweg bei Rot, wo kein Snapshot ging: Gast stoppen, die Sicherung auf den bisherigen Speicher der rootfs
    zurückspielen (pct restore --force), starten. Bind-Mounts (der Datenspeicher des PBS) bleiben unberührt."""
    art, vmid = _gast_pruefen(p.get("vmid"))
    if art != "lxc":
        raise ValueError("nur für Container")
    volid = _eigene_sicherung(vmid, p.get("sicherung"))
    ziel = str(_pvesh(f"/nodes/{NODE}/lxc/{vmid}/config").get("rootfs") or "").split(":")[0]
    if not ziel or ziel.startswith("/"):
        raise ValueError("Der Speicher der rootfs ließ sich nicht lesen")
    protokoll = []
    if (_pvesh(f"/nodes/{NODE}/lxc/{vmid}/status/current") or {}).get("status") == "running":
        code, text = _laufen(["pct", "shutdown", str(vmid), "--timeout", "120", "--forceStop", "1"], 300)
        protokoll.append(text)
        if code != 0:
            return code, "\n".join(protokoll + ["Der Gast ließ sich nicht stoppen; zurückgespielt wurde nichts."])
    code, text = _laufen(["pct", "restore", str(vmid), volid, "--force", "1", "--storage", ziel],
                         ZEITLIMIT_SICHERUNG_S, sanft=True)
    protokoll.append(text)
    if code != 0:
        return code, "\n".join(protokoll + [f"Zurückspielen gescheitert; die Sicherung {volid} liegt weiter da."])
    code, text = _laufen(["pct", "start", str(vmid)], 120)
    protokoll.append(text)
    return code, "\n".join(t for t in protokoll if t)
 def a_sicherung_loeschen(p: dict) -> tuple[int, str]:
    _, vmid = _gast_pruefen(p.get("vmid"))
    return _laufen(["pvesm", "free", _eigene_sicherung(vmid, p.get("sicherung"))], 300)
 def a_host_update(p: dict) -> tuple[int, str]:
    # Kein Neustart: der ist ein eigener Knopf (User-Entscheid 24.09.2026).
    befehl = ("apt-get update -q && DEBIAN_FRONTEND=noninteractive apt-get -y "
@@ -330,7 +576,8 @@ def a_bericht(p: dict) -> tuple[int, str]:
 AKTIONEN = {"snapshot": a_snapshot, "update": a_update, "os_update": a_os_update, "suchen": a_suchen,
            "zurueck": a_zurueck,
            "snapshot_loeschen": a_snapshot_loeschen, "host_update": a_host_update,
-            "host_neustart": a_host_neustart, "bericht": a_bericht}
+            "host_neustart": a_host_neustart, "bericht": a_bericht,
            "sichern": a_sichern, "sicherung_zurueck": a_sicherung_zurueck, "sicherung_loeschen": a_sicherung_loeschen}
 def ausfuehren(auftrag: dict) -> dict:
@@ -344,6 +591,10 @@ def ausfuehren(auftrag: dict) -> dict:
        code, text = aktion(auftrag.get("parameter") or {})
    except (ValueError, RuntimeError) as exc:
        return {"code": 2, "text": f"Abgelehnt: {exc}"}
    except Exception as exc:
        # Sonst bliebe der Auftrag ohne Antwort, und der Homelab-Teil wartete bis zu seinem Zeitlimit.
        log.exception("Auftrag %s gescheitert", auftrag.get("aktion"))
        return {"code": 1, "text": f"Fehler im Ausführer: {exc.__class__.__name__}: {exc}"[:500]}
    return {"code": code, "text": text[-20000:]}
@@ -393,6 +644,7 @@ SERVER = str(_konfig.get("server") or "").rstrip("/")
 TOKEN = str(_konfig.get("token") or "")
 NODE = str(_konfig.get("node") or platform.node())
 NUR_LESEN = bool(_konfig.get("nur_lesen"))
 SICHERUNG_SPEICHER = str(_konfig.get("sicherung_speicher") or "").strip() or None
 def main() -> int:
@@ -225,28 +225,59 @@ flowchart LR
  gemeinsamen Geheimnis aus (Kopfzeile `X-MC2-Ausfuehrer`; erzeugt der Homelab-Teil in
  `/var/lib/mc2/ausfuehrer.token`, liegt auf dem Host in `/etc/mc2-ausfuehrer.json`, beides 0600). Er führt nur eine
  feste Liste von Aktionen aus (`bericht`, `snapshot`, `update`, `os_update`, `suchen`, `zurueck`,
-  `snapshot_loeschen`, `host_update`, `host_neustart`) und prüft selbst, ob ein Gast das Etikett
+  `snapshot_loeschen`, `sichern`, `sicherung_zurueck`, `sicherung_loeschen`, `host_update`, `host_neustart`) und
-  `community-script` oder `watcher` trägt und nicht `watcher-aus` — dem Server vertraut er dabei nicht.
+  prüft selbst, ob ein Gast das Etikett `community-script` oder `watcher` trägt und nicht `watcher-aus` — dem Server
  vertraut er dabei nicht.
 - **Bericht** (nur lesend, auch von Hand: `python3 ausfuehrer.py --bericht`): Host-Version und Paket-Updates, je
  Gast Status, IP, Etiketten, Snapshots, ob ein Snapshot geht (Bind-Mounts wie beim PBS verhindern ihn), die
  Community-Script-Kennung (aus `/usr/bin/update` im Gast), die App-Version (je App eigener Weg: `/root/.<app>`,
  `AdGuardHome --version`, `netbird version`, `dpkg-query`, Docker-Image-Datum) und die Paket-Updates im Gast samt
-  Alter der Paketlisten.
+  Alter der Paketlisten. Dazu der Sicherungsspeicher (`host.sicherung`: Name und frei, oder was fehlt), je Container
- **Homelab-Teil** `backend/services/homelab/`: `kanal.py` (Geheimnis, Auftragsliste, Bericht), `apps.py` (App-
+  ohne Snapshot, ob eine Sicherung geht (`sicherung_moeglich`, sonst `sicherung_grund`), und die eigenen
-  Katalog: Name, GitHub-Quelle, Weboberfläche), `inventar.py` (Bericht + neueste Versionen von GitHub +
+  Sicherungen je Gast.
-  eigene Webprüfung → Ziele im gemeinsamen Modell; Paketlisten älter als 14 Tage = „unklar“ mit Knopf „Nach Updates
+- **Sicherung statt Snapshot** (Ausführer, seit 24.09.): Wo kein Snapshot geht, sichert `sichern` den Container per
-  suchen“), `updates.py` („Jetzt updaten“). Schnittstellen `routers/homelab.py` unter `/api/homelab/…`.
+  `vzdump` — auf den ersten lokalen Speicher, der Sicherungen annimmt (Art `dir`/`btrfs`, nicht geteilt; heute
- **„Jetzt updaten“** (nur per Knopf): Snapshot (wo möglich) → `update` des Community-Scripts (`PHS_SILENT=1`) bzw.
+  `local` = `/var/lib/vz`), oder auf `sicherung_speicher` aus `/etc/mc2-ausfuehrer.json`. Nie auf einen Speicher der
-  Pakete → 20 s warten → frischer Bericht → Prüfung (Gast läuft, Weboberfläche antwortet, App-Version neu). Rot
+  Art `pbs`: Der PBS würde sich selbst sichern. Nimmt kein lokaler Speicher Sicherungen an, lehnt er ab und sagt im
-  und Snapshot da → automatisch zurück + dringende Meldung; grün → ältere `mc2-`-Snapshots weg + Meldung. Läufe in
+  Bericht, was fehlt; die Speicher-Konfiguration ändert er nicht. Vorher prüft er den Platz (frei > belegt × 1,2;
  belegt = rootfs, Bind-Mounts sichert vzdump nie mit). `vzdump` läuft mit `--mode snapshot` (rootfs auf lvmthin: der
  Gast läuft durch), sonst `stop`, dazu `--remove 0` (keine Aufräumregeln des Speichers) und der Notiz
  `mc2-sicherung: …`. Nur Sicherungen mit dieser Notiz spielt er zurück (`sicherung_zurueck`: Gast stoppen,
  `pct restore <vmid> <archiv> --force 1 --storage <bisheriger rootfs-Speicher>`, starten; Bind-Mount-Daten bleiben
  unberührt) oder löscht er (`sicherung_loeschen`, `pvesm free`). Zeitlimit 30 min; läuft es ab, bekommt `vzdump`
  bzw. `pct restore` erst SIGTERM, damit Sperre und Snapshot aufgeräumt werden.
 - **Homelab-Teil** `backend/services/homelab/`: `kanal.py` (Geheimnis, Auftragsliste unter Dateisperre, Bericht),
  `apps.py` (App-Katalog: Name, GitHub-Quelle, Weboberfläche, Update-Weg), `inventar.py` (Bericht + neueste Versionen
  von GitHub + eigene Webprüfung → Ziele im gemeinsamen Modell; Paketlisten älter als 14 Tage = „unklar“ mit Knopf
  „Nach Updates suchen“), `karenz.py` (Wartezeit nach einer Skriptänderung), `updates.py` („Jetzt updaten“),
  `pflege.py` (wöchentliches Suchen). Schnittstellen `routers/homelab.py` unter `/api/homelab/…`.
 - **„Jetzt updaten“** (nur per Knopf): Snapshot, wo keiner geht Sicherung (wo auch die nicht geht: ohne Rückweg, mit
  Warnung in der Rückfrage; scheitert der Schritt, beginnt das Update nicht) → `update` des Community-Scripts
  (`PHS_SILENT=1`) bzw. Pakete → 20 s warten → frischer Bericht → Prüfung (Gast läuft, Weboberfläche antwortet,
  App-Version neu). Rot → zurück auf den Snapshot bzw. die Sicherung zurückspielen + dringende Meldung; grün → ältere
  `mc2-`-Snapshots bzw. `mc2-sicherung`-Sicherungen dieses Gasts weg (die neueste bleibt) + Meldung. Läufe in
  `/var/lib/mc2/homelab-laeufe.json` und im strukturierten Update-Verlauf. Host: Pakete per Knopf mit Warnung,
  Neustart als eigener Knopf.
 - **Wartezeit nach Skriptänderung** (`karenz.py`): `update` lädt `ct/<kennung>.sh` ungepinnt von GitHub
  (community-scripts/ProxmoxVE, Zweig `main`) und führt es als root aus. Für Apps mit dem Weg „skript“ fragt der
  Homelab-Teil deshalb, wann das Skript zuletzt geändert wurde (`/repos/community-scripts/ProxmoxVE/commits?path=…`,
  über `kern/github.py`, 15 min gemerkt). Jünger als `MC_HOMELAB_KARENZ_H` (Standard 48 h): der Baustein bleibt
  „neu“, aber ohne Knopf, mit „Das Update-Skript wurde am TT.MM. geändert; zur Sicherheit erst ab TT.MM. HH:MM.“
  (Berliner Zeit); `updates.starten` lehnt mit demselben Satz ab. Antwortet GitHub nicht, blockiert nichts; die
  Rückfrage sagt dann „Ob das Skript kürzlich geändert wurde, ließ sich nicht prüfen.“
 - **Wöchentliches Suchen** (`pflege.py`, im Wächter-Takt des Stewards): Sind die Paketlisten eines freigegebenen,
  laufenden Containers älter als 7 Tage, legt der Homelab-Teil selbst den Auftrag `suchen` an — höchstens einmal je
  Gast und Tag, nie während eines Update-Laufs oder neben einem offenen Auftrag für diesen Gast, nur wenn der Ausführer
  gerade berichtet, bevorzugt nachts 02:00–05:00 (war die Instanz letzte Nacht aus, eben gleich). Keine Meldungen;
  scheitert die Suche für einen Gast zweimal hintereinander, wird es ein gelber Hinweis des Wächters. Zustand in
  `/var/lib/mc2/homelab-pflege.json`. Weil damit zwei Prozesse Aufträge anlegen, schreibt `kanal.py` unter `flock`.
 - **Arcane und Docker** (`services/homelab/arcane.py`): Arcanes eigene Version kommt öffentlich über
  `/api/app-version`; die Docker-Images mit neuerem Stand und der Updater brauchen einen Arcane-API-Schlüssel
  (`MC_ARCANE_KEY`, Kopfzeile `X-API-Key`). Docker-Updates laufen zuerst nur als Probelauf (`dryRun`); echt erst
  mit `MC_ARCANE_ECHT=1`. Die Arcane-VM braucht dafür kein Etikett, weil der Ausführer nicht beteiligt ist.
- **Wächter** in der Rolle `homelab`: Platte, Partner (die KI-Box), Ausführer (kein Bericht seit 30 min = rot) und
+- **Wächter** in der Rolle `homelab`: Platte, Partner (die KI-Box), Ausführer (kein Bericht seit 30 min = rot),
-  jede Weboberfläche der freigegebenen Gäste.
+  jede Weboberfläche der freigegebenen Gäste (außer mitten in ihrem Update-Lauf; das Ergebnis meldet der Lauf) und das
  wöchentliche Suchen (gelb, wenn es wiederholt scheitert).
 - **Oberfläche**: Die Seite „Homelab“ zeigt alle Geräte als Karten — die KI-Box (`/api/ziele`) und alles aus
  `/api/homelab/ziele` — mit Stand je Baustein, Rückweg und Knopf samt Rückfrage.
 - **Meldungen** ohne Hermes: `notify.sh` im Container nimmt den Zweitweg direkt an die Bot-API
@@ -72,7 +72,10 @@ soll kein Alarm sein).
 | Partner-Instanz (nur mit `MC_PARTNER_URL`) | „<Name> antwortet nicht" | – |
 | Abgestürzte Prüfung | – | „Eine Prüfung des Wächters lief nicht" |
-In der Rolle `homelab` prüft der Wächter bisher nur Platte und Partner und meldet mit „[Homelab-Problem]".
+In der Rolle `homelab` prüft der Wächter Platte, Partner, den Ausführer und die Weboberflächen der freigegebenen
 Gäste (einen Gast mitten in seinem Update-Lauf nicht: Das Ergebnis meldet der Lauf selbst) und meldet mit
 „[Homelab-Problem]". Im selben Takt stößt er das wöchentliche Suchen an (siehe „Homelab-Teil im Betrieb“); scheitert
 es für einen Gast zweimal hintereinander, gibt es einen gelben Hinweis „<App>: Die Suche nach Updates scheitert“.
 - **Selbstreparatur:** Nur ein abgestürzter Dienst (`ActiveState=failed`) wird neu gestartet, höchstens 2× je Stunde
  und Hinweis. Ein gestoppter Dienst (`inactive`) wird nur gemeldet. Ein schlafender Dienst (`inactive` und
@@ -169,6 +172,33 @@ den Container). Alles läuft am PC in Git-Bash, im Repo; SSH-Zugang zu `pve` (Sc
 Prüfen: `curl http://<ip>:9001/api/homelab/ziele` (nach etwa einer Minute stehen die Geräte darin),
 `systemctl status mc2-ausfuehrer` auf dem Host, Seite „Homelab“ der Oberfläche.
 Ein neuer Stand des Ausführers (`deploy/homelab/ausfuehrer.py`) kommt nicht mit dem Deploy; dafür Schritt 4 erneut
 laufen lassen. Achtung: `ausfuehrer-einrichten.sh` schreibt `/etc/mc2-ausfuehrer.json` neu — ein von Hand gesetztes
 `sicherung_speicher` danach wieder eintragen und `systemctl restart mc2-ausfuehrer`.
 ## Homelab-Teil im Betrieb
 - **Sicherung statt Snapshot** (Gäste, bei denen kein Snapshot geht; heute der PBS, CT 105, wegen seines
  Bind-Mounts): Vor dem Update sichert der Ausführer den Container per `vzdump` auf den ersten lokalen Speicher, der
  Sicherungen annimmt — auf dem Proxmox-PC `local` (`/var/lib/vz/dump`, liegt auf der Systemplatte des Hosts). Nie
  auf `pbs-qnap`. Anderer Speicher: `"sicherung_speicher": "<Name>"` in `/etc/mc2-ausfuehrer.json`, dann
  `systemctl restart mc2-ausfuehrer`. Nimmt kein lokaler Speicher Sicherungen an, sagt es die Karte („Rückweg“); die
  Lösung ist Sache des Users (Proxmox → Rechenzentrum → Speicher → `local` → Inhalt „VZDump-Sicherung“). Vorher prüft
  er den Platz (frei > belegt × 1,2), sonst beginnt das Update nicht.
 - **Eigene Sicherungen erkennen:** Notiz `mc2-sicherung: vor einem Update durch den Homelab Orchestrator (<Gast>)`.
  Nach einem grünen Lauf bleibt nur die neueste je Gast. Ansehen auf dem Host:
  `pvesh get /nodes/pve/storage/local/content --content backup --output-format json-pretty`. Eine Sicherung von Hand
  zurückspielen: `pct shutdown <vmid> && pct restore <vmid> <volid> --force 1 --storage local-lvm && pct start <vmid>`
  (der Bind-Mount und seine Daten bleiben, wie sie sind).
 - **Wartezeit nach Skriptänderung:** Ist `ct/<app>.sh` bei community-scripts jünger als `MC_HOMELAB_KARENZ_H` Stunden
  (Standard 48), zeigt die Karte „Update bereit“ ohne Knopf und nennt, ab wann es geht. Ändern oder abschalten (`0`):
  `MC_HOMELAB_KARENZ_H=…` in `/etc/mc2/homelab.env` im Container, dann `systemctl restart mc2-homelab`.
 - **Wöchentliches Suchen:** Der Steward legt für jeden freigegebenen, laufenden Container, dessen Paketlisten älter als
  7 Tage sind, den Auftrag `suchen` an (nur `apt-get update` bzw. `apk update` im Gast, es wird nichts installiert):
  nachts 02:00–05:00, höchstens einmal je Gast und Tag, nicht während eines Updates; war die Instanz nachts aus, gleich
  nach dem Start. Stand in `/var/lib/mc2/homelab-pflege.json` (Nacht, je Gast: letzter Tag, Fehlschläge, letzter
  Fehler). Schreibt nur der Steward; zum Zurücksetzen die Datei löschen und `systemctl restart mc2-homelab-steward`.
 ## Sicherung
 - **Wann:** `mc2-backup.timer` täglich 03:30 (+≤5 min); außerdem vor jedem Hermes-Update, vor jedem Zurückspielen und
@@ -18,7 +18,7 @@ zwei Minuten, Meldung auf Telegram, der Snapshot bleibt als Rückweg liegen. Dan
 | 1 · Box-Wart 1.0 | Ballast raus, Box-Diät, riskante Stellen, Deploy mit Prüftor und Rückweg, Sonntags-Timer, Oberfläche, Doku | erledigt 24.09. |
 | 2 · Kern und zweite Instanz | Rollen `box`/`homelab`, Partner-Aufsicht, Telegram-Zweitweg, Aufträge als systemd-Einheiten, Ziel-Modell mit Box-Adapter, strukturierter Update-Verlauf, Oberfläche „Homelab Orchestrator“ | erledigt 24.09. |
 | 3 · Homelab sehen | Container 107, Ausführer liefert den Bericht (kein Proxmox-Schlüssel), Inventar aller Gäste mit App-Version, Paketen, Webprüfung, Rückweg; Seite „Homelab“ = alle Geräte | erledigt 24.09. |
-| 4 · Jetzt updaten | Snapshot → Update → Prüfung → bei Rot zurück + Meldung; Host-Pakete mit Warnung, Neustart getrennt; Docker über Arcane als Probelauf | eingerichtet 24.09.; erster Lauf grün (Gitea), Rückweg bei Rot bisher nur im Test |
+| 4 · Jetzt updaten | Snapshot → Update → Prüfung → bei Rot zurück + Meldung; Host-Pakete mit Warnung, Neustart getrennt; Docker über Arcane als Probelauf; frisch geänderte Update-Skripte erst nach 48 h Wartezeit; PBS (kein Snapshot wegen Bind-Mount) mit Sicherung per vzdump auf `local`, bei Rot zurückgespielt | eingerichtet 24.09.; erster Lauf grün (Gitea), Rückweg bei Rot bisher nur im Test; Wartezeit und PBS-Sicherung gebaut 24.09., die Sicherung greift erst mit dem neu eingespielten Ausführer |
 | 5 · Android-App | Push, Cockpit, Updates freigeben, Lucy per Sprache mit Live-Modus | offen, eigenes Projekt |
 Technik: [ARCHITEKTUR.md](../ARCHITEKTUR.md), Abschnitt „Der Homelab-Teil“; Einrichtung und Betrieb:
@@ -31,9 +31,7 @@ Technik: [ARCHITEKTUR.md](../ARCHITEKTUR.md), Abschnitt „Der Homelab-Teil“;
   `systemctl restart mc2-homelab mc2-homelab-steward`. Docker-Updates bleiben Probelauf bis `MC_ARCANE_ECHT=1`.
 2. **Feste IP für Container 107.** Er hat seine Adresse per DHCP (192.168.178.31); Partner-Adresse der Box,
   Ausführer-Konfiguration und der Deploy-Schritt 8 hängen daran. In der Fritzbox die Adresse fest zuordnen.
-3. **Paketlisten in den Gästen sind alt** (AdGuard seit Oktober 2025). Die Übersicht sagt deshalb „unklar“ statt
+3. **Die Arcane-VM (106) trägt kein Etikett.** Für Arcane und Docker braucht es keins (eigene Schnittstelle); für
   „aktuell“; „Nach Updates suchen“ je Gast erneuert sie. Offen: ob der Ausführer das wöchentlich selbst tun soll.
 4. **Die Arcane-VM (106) trägt kein Etikett.** Für Arcane und Docker braucht es keins (eigene Schnittstelle); für
   Snapshots oder OS-Updates der VM über den Ausführer bräuchte sie `watcher` (heute nicht gebaut).
 ## Offene Einzelpunkte (Box-Wart)