Feat: Voll-Zustands-Backup + getesteter Restore + taeglicher Timer

Behebt 3 von 4 Backup-Luecken (Schicht 1, lokal):
- backup.sh sichert jetzt den ECHTEN Zustand als ein Tarball mc2-state-<ts>.tar.gz:
  mem0 (Chroma+history.db), ~/.hermes (config.yaml, .env, plugins/), llama-swap config.
  Vorher wurde nur die alte/leere mc2-memory.db gesichert. chmod 600 (enthaelt .env).
- restore.sh: --list / --dry-run / [--yes] <datei|latest>; macht VOR dem Zurueckspielen
  ein Sicherheits-Backup, stoppt/startet Dienste, Health-Check. Live round-trip verifiziert.
- mc2-backup.timer/.service: taegliches Backup ~03:30 (vorher gab es KEINE Automatik).
- backend/services/backup.py delegiert an backup.sh (eine Quelle der Wahrheit); UI-Button
  + /api/system/backups zeigen die Tarballs.
- docs/BACKUP.md: Backup/Restore-Anleitung.

Offen (Schicht 2): Off-Box-Spiegel (Box ist Bare Metal -> PBS-Client oder rsync in LXC).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Hitonabi
2026-06-27 21:25:39 +02:00
parent 9fb321e45e
commit 8554e7b29c
13 changed files with 251 additions and 48 deletions
+52 -30
View File
@@ -1,51 +1,73 @@
"""
Backup der „Verfassung" (Shared-Memory-SQLite) + aller Configs.
Snapshot in einen Zeitstempel-Ordner; die letzten N bleiben erhalten.
Voll-Zustands-Backup (mem0 + Hermes-Configs/Secrets + llama-swap config).
Delegiert an deploy/backup.sh (eine Quelle der Wahrheit, identisch zum systemd-Timer);
Restore läuft bewusst nur per CLI (deploy/restore.sh) — siehe docs/BACKUP.md.
"""
import shutil
import time
import subprocess
import tarfile
from pathlib import Path
from config import CONFIG_PATH, MEMORY_DB, MODELS_DIR
from config import MODELS_DIR
BACKUP_DIR = Path(MODELS_DIR) / "mc2-backups"
RETAIN = 7
SRC_ROOT = Path(__file__).resolve().parents[2]
BACKUP_SH = SRC_ROOT / "deploy" / "backup.sh"
def _safe_copy(src: Path, dst_dir: Path) -> str | None:
def _ts(p: Path) -> str:
"""Zeitstempel aus 'mc2-state-<ts>.tar.gz' (Path.stem ließe '.tar' stehen)."""
return p.name[len("mc2-state-"):-len(".tar.gz")]
def _latest() -> Path | None:
if not BACKUP_DIR.exists():
return None
snaps = sorted(BACKUP_DIR.glob("mc2-state-*.tar.gz"), reverse=True)
return snaps[0] if snaps else None
def _components(tarball: Path) -> list[str]:
"""Top-Level-Einträge im Tarball (zur Anzeige im UI)."""
try:
if src and src.exists():
shutil.copy2(src, dst_dir / src.name)
return src.name
with tarfile.open(tarball, "r:gz") as t:
top = {m.name.split("/")[1] for m in t.getmembers()
if m.name.startswith("./") and "/" in m.name[2:]}
top |= {m.name[2:] for m in t.getmembers()
if m.name.startswith("./") and "/" not in m.name[2:] and m.isfile()}
return sorted(x for x in top if x)
except Exception:
pass
return None
return []
def backup_now() -> dict:
"""Erstellt einen Snapshot (Memory-DB inkl. WAL/SHM + Configs). Alte Snapshots
über RETAIN hinaus werden entfernt."""
ts = time.strftime("%Y%m%d-%H%M%S")
dst = BACKUP_DIR / ts
dst.mkdir(parents=True, exist_ok=True)
saved = []
for src in (MEMORY_DB, Path(str(MEMORY_DB) + "-wal"), Path(str(MEMORY_DB) + "-shm"),
CONFIG_PATH):
if (name := _safe_copy(src, dst)):
saved.append(name)
# Aufräumen: nur die letzten RETAIN Snapshots behalten.
snaps = sorted([p for p in BACKUP_DIR.iterdir() if p.is_dir()], reverse=True)
for old in snaps[RETAIN:]:
shutil.rmtree(old, ignore_errors=True)
return {"ok": bool(saved), "snapshot": ts, "files": saved, "dir": str(dst)}
"""Erstellt einen Voll-Zustands-Snapshot via deploy/backup.sh."""
try:
r = subprocess.run(["/bin/bash", str(BACKUP_SH)], capture_output=True, text=True, timeout=180)
if r.returncode != 0:
return {"ok": False, "snapshot": "", "files": [], "error": (r.stderr or r.stdout).strip()[-300:]}
except Exception as exc: # noqa: BLE001
return {"ok": False, "snapshot": "", "files": [], "error": str(exc)}
latest = _latest()
if not latest:
return {"ok": False, "snapshot": "", "files": [], "error": "Kein Backup erzeugt"}
return {
"ok": True,
"snapshot": _ts(latest),
"files": _components(latest),
"size_mb": round(latest.stat().st_size / 1_000_000, 2),
}
def list_backups() -> list[dict]:
if not BACKUP_DIR.exists():
return []
out = []
for p in sorted([d for d in BACKUP_DIR.iterdir() if d.is_dir()], reverse=True):
files = [f.name for f in p.iterdir() if f.is_file()]
out.append({"snapshot": p.name, "files": files})
for p in sorted(BACKUP_DIR.glob("mc2-state-*.tar.gz"), reverse=True):
out.append({
"snapshot": _ts(p),
"file": p.name,
"size_mb": round(p.stat().st_size / 1_000_000, 2),
})
return out