mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-04-28 06:51:16 +08:00
feat(update): auto-backup HERMES_HOME before hermes update (#16539)
Every 'hermes update' now runs a full backup of ~/.hermes/ first, so users can always roll back to the exact state they had before the update if anything goes wrong (corrupted sessions.db, broken skills, config migrations that don't round-trip, etc.). Changes: - hermes_cli/backup.py: new create_pre_update_backup() helper. Writes to <HERMES_HOME>/backups/pre-update-<stamp>.zip using the same exclusion rules and SQLite safe-copy as 'hermes backup'. Auto-rotates (keep last N, pre-update-*.zip only — hand-dropped zips in backups/ are untouched). Adds 'backups' to _EXCLUDED_DIRS so subsequent backups don't nest prior ones. - hermes_cli/main.py: _run_pre_update_backup() wired into _cmd_update_impl before any git operation. Prints save path, restore command, and how to disable. Swallows failures so a broken backup never blocks the update itself. New --no-backup flag on 'hermes update' for one-off override. - hermes_cli/config.py: new 'updates' section in DEFAULT_CONFIG with pre_update_backup (default true) and backup_keep (default 5). Auto-surfaces in the dashboard config UI. - tests/hermes_cli/test_backup.py: +11 tests covering backup location, content parity with 'hermes backup', no-recursion, rotation, manual file preservation, config gate, --no-backup flag, flag-wins-over-config.
This commit is contained in:
@@ -36,6 +36,7 @@ _EXCLUDED_DIRS = {
|
||||
"__pycache__", # bytecode caches — regenerated on import
|
||||
".git", # nested git dirs (profiles shouldn't have these, but safety)
|
||||
"node_modules", # js deps if website/ somehow leaks in
|
||||
"backups", # prior auto-backups — don't nest backups exponentially
|
||||
}
|
||||
|
||||
# File-name suffixes to skip
|
||||
@@ -683,3 +684,138 @@ def run_quick_backup(args) -> None:
|
||||
print(f" Restore with: /snapshot restore {snap_id}")
|
||||
else:
|
||||
print("No state files found to snapshot.")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pre-update auto-backup
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_PRE_UPDATE_BACKUPS_DIR = "backups"
|
||||
_PRE_UPDATE_PREFIX = "pre-update-"
|
||||
_PRE_UPDATE_DEFAULT_KEEP = 5
|
||||
|
||||
|
||||
def _pre_update_backup_dir(hermes_home: Optional[Path] = None) -> Path:
|
||||
home = hermes_home or get_hermes_home()
|
||||
return home / _PRE_UPDATE_BACKUPS_DIR
|
||||
|
||||
|
||||
def _prune_pre_update_backups(backup_dir: Path, keep: int) -> int:
|
||||
"""Remove oldest pre-update backups beyond the keep limit.
|
||||
|
||||
Returns the number of files deleted. Only touches files matching
|
||||
``pre-update-*.zip`` so hand-made zips dropped in the same directory
|
||||
are never touched.
|
||||
"""
|
||||
if keep < 0:
|
||||
keep = 0
|
||||
if not backup_dir.exists():
|
||||
return 0
|
||||
|
||||
backups = sorted(
|
||||
(p for p in backup_dir.iterdir()
|
||||
if p.is_file() and p.name.startswith(_PRE_UPDATE_PREFIX) and p.suffix.lower() == ".zip"),
|
||||
key=lambda p: p.name,
|
||||
reverse=True,
|
||||
)
|
||||
|
||||
deleted = 0
|
||||
for p in backups[keep:]:
|
||||
try:
|
||||
p.unlink()
|
||||
deleted += 1
|
||||
except OSError as exc:
|
||||
logger.warning("Failed to prune backup %s: %s", p.name, exc)
|
||||
|
||||
return deleted
|
||||
|
||||
|
||||
def create_pre_update_backup(
|
||||
hermes_home: Optional[Path] = None,
|
||||
keep: int = _PRE_UPDATE_DEFAULT_KEEP,
|
||||
) -> Optional[Path]:
|
||||
"""Create a full zip backup of HERMES_HOME under ``backups/``.
|
||||
|
||||
Mirrors :func:`run_backup` (same exclusion rules, same SQLite safe-copy)
|
||||
but writes to ``<HERMES_HOME>/backups/pre-update-<timestamp>.zip`` and
|
||||
auto-prunes old pre-update backups.
|
||||
|
||||
Returns the path to the created zip, or ``None`` if no files were
|
||||
found or the backup could not be created. Never raises — the caller
|
||||
(``hermes update``) should continue even if the backup fails.
|
||||
"""
|
||||
hermes_root = hermes_home or get_default_hermes_root()
|
||||
if not hermes_root.is_dir():
|
||||
return None
|
||||
|
||||
backup_dir = _pre_update_backup_dir(hermes_root)
|
||||
try:
|
||||
backup_dir.mkdir(parents=True, exist_ok=True)
|
||||
except OSError as exc:
|
||||
logger.warning("Could not create pre-update backup dir %s: %s", backup_dir, exc)
|
||||
return None
|
||||
|
||||
stamp = datetime.now().strftime("%Y-%m-%d-%H%M%S")
|
||||
out_path = backup_dir / f"{_PRE_UPDATE_PREFIX}{stamp}.zip"
|
||||
|
||||
# Collect files (same logic as run_backup, minus the chatty progress prints)
|
||||
files_to_add: list[tuple[Path, Path]] = []
|
||||
try:
|
||||
for dirpath, dirnames, filenames in os.walk(hermes_root, followlinks=False):
|
||||
dp = Path(dirpath)
|
||||
# Prune excluded directories in-place so os.walk doesn't descend
|
||||
dirnames[:] = [d for d in dirnames if d not in _EXCLUDED_DIRS]
|
||||
|
||||
for fname in filenames:
|
||||
fpath = dp / fname
|
||||
try:
|
||||
rel = fpath.relative_to(hermes_root)
|
||||
except ValueError:
|
||||
continue
|
||||
|
||||
if _should_exclude(rel):
|
||||
continue
|
||||
|
||||
# Skip the output zip itself if it already exists
|
||||
try:
|
||||
if fpath.resolve() == out_path.resolve():
|
||||
continue
|
||||
except (OSError, ValueError):
|
||||
pass
|
||||
|
||||
files_to_add.append((fpath, rel))
|
||||
except OSError as exc:
|
||||
logger.warning("Pre-update backup: walk failed: %s", exc)
|
||||
return None
|
||||
|
||||
if not files_to_add:
|
||||
return None
|
||||
|
||||
try:
|
||||
with zipfile.ZipFile(out_path, "w", zipfile.ZIP_DEFLATED, compresslevel=6) as zf:
|
||||
for abs_path, rel_path in files_to_add:
|
||||
try:
|
||||
if abs_path.suffix == ".db":
|
||||
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as tmp:
|
||||
tmp_db = Path(tmp.name)
|
||||
try:
|
||||
if _safe_copy_db(abs_path, tmp_db):
|
||||
zf.write(tmp_db, arcname=str(rel_path))
|
||||
finally:
|
||||
tmp_db.unlink(missing_ok=True)
|
||||
else:
|
||||
zf.write(abs_path, arcname=str(rel_path))
|
||||
except (PermissionError, OSError, ValueError) as exc:
|
||||
logger.debug("Skipping %s in pre-update backup: %s", rel_path, exc)
|
||||
continue
|
||||
except OSError as exc:
|
||||
logger.warning("Pre-update backup: zip write failed: %s", exc)
|
||||
# Best-effort cleanup of partial file
|
||||
try:
|
||||
out_path.unlink(missing_ok=True)
|
||||
except OSError:
|
||||
pass
|
||||
return None
|
||||
|
||||
_prune_pre_update_backups(backup_dir, keep=keep)
|
||||
return out_path
|
||||
|
||||
@@ -1037,6 +1037,19 @@ DEFAULT_CONFIG = {
|
||||
"seen": {},
|
||||
},
|
||||
|
||||
# ``hermes update`` behaviour.
|
||||
"updates": {
|
||||
# Run a full ``hermes backup``-style zip of HERMES_HOME before every
|
||||
# ``hermes update``. Backups land in ``<HERMES_HOME>/backups/`` and
|
||||
# can be restored with ``hermes import <path>``. Set to false to
|
||||
# skip the backup entirely; use the ``--no-backup`` flag on a single
|
||||
# update invocation to override just that run.
|
||||
"pre_update_backup": True,
|
||||
# How many pre-update backup zips to retain. Older ones are pruned
|
||||
# automatically after each successful backup.
|
||||
"backup_keep": 5,
|
||||
},
|
||||
|
||||
# Config schema version - bump this when adding new required fields
|
||||
"_config_version": 22,
|
||||
}
|
||||
|
||||
@@ -6142,6 +6142,90 @@ def _ensure_fhs_path_guard() -> None:
|
||||
print(" (reload your shell or run 'source ~/.bashrc' to pick it up)")
|
||||
|
||||
|
||||
def _run_pre_update_backup(args) -> None:
|
||||
"""Create a full zip backup of HERMES_HOME before running the update.
|
||||
|
||||
Gated on ``updates.pre_update_backup`` in config (default true). The
|
||||
``--no-backup`` flag on ``hermes update`` overrides it for one run.
|
||||
Never raises — a backup failure should not block the update itself.
|
||||
"""
|
||||
# CLI flag wins over config
|
||||
if getattr(args, "no_backup", False):
|
||||
print("◆ Pre-update backup: skipped (--no-backup)")
|
||||
print()
|
||||
return
|
||||
|
||||
try:
|
||||
from hermes_cli.config import load_config
|
||||
cfg = load_config()
|
||||
except Exception as exc:
|
||||
logging.getLogger(__name__).debug("Could not load config for pre-update backup: %s", exc)
|
||||
cfg = {}
|
||||
|
||||
updates_cfg = cfg.get("updates", {}) if isinstance(cfg, dict) else {}
|
||||
enabled = updates_cfg.get("pre_update_backup", True)
|
||||
keep = updates_cfg.get("backup_keep", 5)
|
||||
|
||||
if not enabled:
|
||||
print("◆ Pre-update backup: disabled (updates.pre_update_backup=false in config.yaml)")
|
||||
print()
|
||||
return
|
||||
|
||||
try:
|
||||
from hermes_cli.backup import create_pre_update_backup
|
||||
except Exception as exc:
|
||||
print(f"⚠ Pre-update backup: could not load backup module ({exc}); continuing update.")
|
||||
print()
|
||||
return
|
||||
|
||||
print("◆ Creating pre-update backup...")
|
||||
t0 = _time.monotonic()
|
||||
try:
|
||||
out_path = create_pre_update_backup(keep=int(keep))
|
||||
except Exception as exc: # defensive — helper already swallows, but just in case
|
||||
print(f" ⚠ Backup failed: {exc}")
|
||||
print(" Continuing with update.")
|
||||
print()
|
||||
return
|
||||
|
||||
elapsed = _time.monotonic() - t0
|
||||
|
||||
if out_path is None:
|
||||
print(" ⚠ Backup skipped (no files found or write failed); continuing update.")
|
||||
print()
|
||||
return
|
||||
|
||||
try:
|
||||
size_bytes = out_path.stat().st_size
|
||||
except OSError:
|
||||
size_bytes = 0
|
||||
|
||||
# Human-readable size
|
||||
size_str = f"{size_bytes} B"
|
||||
for unit in ("KB", "MB", "GB"):
|
||||
if size_bytes < 1024:
|
||||
break
|
||||
size_bytes /= 1024
|
||||
size_str = f"{size_bytes:.1f} {unit}"
|
||||
|
||||
# Render path using display_hermes_home so the user sees ~/.hermes/...
|
||||
try:
|
||||
from hermes_constants import get_hermes_home, display_hermes_home
|
||||
home = get_hermes_home()
|
||||
try:
|
||||
display_path = f"{display_hermes_home()}/{out_path.relative_to(home)}"
|
||||
except ValueError:
|
||||
display_path = str(out_path)
|
||||
except Exception:
|
||||
display_path = str(out_path)
|
||||
|
||||
print(f" Saved: {display_path} ({size_str}, {elapsed:.1f}s)")
|
||||
print(f" Restore: hermes import {out_path}")
|
||||
print(f" Disable: set updates.pre_update_backup: false in config.yaml")
|
||||
print(f" (or pass --no-backup on a single update)")
|
||||
print()
|
||||
|
||||
|
||||
def cmd_update(args):
|
||||
"""Update Hermes Agent to the latest version.
|
||||
|
||||
@@ -6184,6 +6268,10 @@ def _cmd_update_impl(args, gateway_mode: bool):
|
||||
print("⚕ Updating Hermes Agent...")
|
||||
print()
|
||||
|
||||
# Pre-update backup — runs before any git/file mutation so users can
|
||||
# always roll back to the exact state they had before this update.
|
||||
_run_pre_update_backup(args)
|
||||
|
||||
# Try git-based update first, fall back to ZIP download on Windows
|
||||
# when git file I/O is broken (antivirus, NTFS filter drivers, etc.)
|
||||
use_zip_update = False
|
||||
@@ -9577,6 +9665,12 @@ Examples:
|
||||
default=False,
|
||||
help="Check whether an update is available without installing anything",
|
||||
)
|
||||
update_parser.add_argument(
|
||||
"--no-backup",
|
||||
action="store_true",
|
||||
default=False,
|
||||
help="Skip the pre-update backup for this run (overrides updates.pre_update_backup)",
|
||||
)
|
||||
update_parser.set_defaults(func=cmd_update)
|
||||
|
||||
# =========================================================================
|
||||
|
||||
@@ -1218,3 +1218,189 @@ class TestQuickSnapshot:
|
||||
snap_id = create_quick_snapshot(hermes_home=hermes_home)
|
||||
# Other state still present → snapshot succeeds.
|
||||
assert snap_id is not None
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pre-update backup (hermes update safety net)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestPreUpdateBackup:
|
||||
"""Tests for create_pre_update_backup — the auto-backup ``hermes update``
|
||||
runs before touching anything."""
|
||||
|
||||
@pytest.fixture
|
||||
def hermes_home(self, tmp_path):
|
||||
root = tmp_path / ".hermes"
|
||||
root.mkdir()
|
||||
_make_hermes_tree(root)
|
||||
return root
|
||||
|
||||
def test_creates_backup_under_backups_dir(self, hermes_home):
|
||||
from hermes_cli.backup import create_pre_update_backup
|
||||
out = create_pre_update_backup(hermes_home=hermes_home)
|
||||
assert out is not None
|
||||
assert out.exists()
|
||||
assert out.parent == hermes_home / "backups"
|
||||
assert out.name.startswith("pre-update-")
|
||||
assert out.suffix == ".zip"
|
||||
|
||||
def test_backup_contents_match_full_backup(self, hermes_home):
|
||||
"""Pre-update backup should include the same user data that
|
||||
``hermes backup`` would, and should exclude the same directories."""
|
||||
from hermes_cli.backup import create_pre_update_backup
|
||||
out = create_pre_update_backup(hermes_home=hermes_home)
|
||||
assert out is not None
|
||||
with zipfile.ZipFile(out) as zf:
|
||||
names = set(zf.namelist())
|
||||
# User data present
|
||||
assert "config.yaml" in names
|
||||
assert ".env" in names
|
||||
assert "sessions/abc123.json" in names
|
||||
assert "skills/my-skill/SKILL.md" in names
|
||||
assert "profiles/coder/config.yaml" in names
|
||||
# hermes-agent repo excluded
|
||||
assert not any(n.startswith("hermes-agent/") for n in names)
|
||||
# __pycache__ excluded
|
||||
assert not any("__pycache__" in n for n in names)
|
||||
# pid files excluded
|
||||
assert "gateway.pid" not in names
|
||||
|
||||
def test_does_not_recurse_into_prior_backups(self, hermes_home):
|
||||
"""The ``backups/`` directory must be excluded so that each backup
|
||||
doesn't grow exponentially by including all prior backups."""
|
||||
from hermes_cli.backup import create_pre_update_backup
|
||||
# First backup
|
||||
out1 = create_pre_update_backup(hermes_home=hermes_home)
|
||||
assert out1 is not None
|
||||
# Second backup — must not include the first
|
||||
out2 = create_pre_update_backup(hermes_home=hermes_home)
|
||||
assert out2 is not None
|
||||
with zipfile.ZipFile(out2) as zf:
|
||||
names = zf.namelist()
|
||||
assert not any(n.startswith("backups/") for n in names), (
|
||||
f"Pre-update backup recursed into backups/ — leaked: "
|
||||
f"{[n for n in names if n.startswith('backups/')]}"
|
||||
)
|
||||
|
||||
def test_rotation_keeps_only_n(self, hermes_home):
|
||||
"""After more than ``keep`` backups are created, older ones are
|
||||
pruned automatically."""
|
||||
import time as _t
|
||||
from hermes_cli.backup import create_pre_update_backup
|
||||
|
||||
created = []
|
||||
for _ in range(5):
|
||||
out = create_pre_update_backup(hermes_home=hermes_home, keep=3)
|
||||
created.append(out)
|
||||
_t.sleep(1.05) # ensure distinct seconds in timestamp
|
||||
|
||||
remaining = sorted(
|
||||
p.name for p in (hermes_home / "backups").iterdir()
|
||||
if p.name.startswith("pre-update-")
|
||||
)
|
||||
assert len(remaining) == 3
|
||||
# Oldest two should have been pruned
|
||||
assert created[0].name not in remaining
|
||||
assert created[1].name not in remaining
|
||||
# Newest three should remain
|
||||
assert created[4].name in remaining
|
||||
|
||||
def test_rotation_preserves_manual_files(self, hermes_home):
|
||||
"""Hand-dropped zips in ``backups/`` must not be touched by
|
||||
rotation — it only prunes files matching ``pre-update-*.zip``."""
|
||||
import time as _t
|
||||
from hermes_cli.backup import create_pre_update_backup
|
||||
|
||||
(hermes_home / "backups").mkdir(exist_ok=True)
|
||||
manual = hermes_home / "backups" / "my-manual.zip"
|
||||
manual.write_bytes(b"manual backup")
|
||||
|
||||
for _ in range(5):
|
||||
create_pre_update_backup(hermes_home=hermes_home, keep=2)
|
||||
_t.sleep(1.05)
|
||||
|
||||
assert manual.exists(), "Manual backup zip was incorrectly pruned"
|
||||
|
||||
def test_returns_none_if_root_missing(self, tmp_path):
|
||||
from hermes_cli.backup import create_pre_update_backup
|
||||
assert create_pre_update_backup(hermes_home=tmp_path / "does-not-exist") is None
|
||||
|
||||
|
||||
class TestRunPreUpdateBackup:
|
||||
"""Tests for the ``_run_pre_update_backup`` wrapper in main.py —
|
||||
covers config gate, ``--no-backup`` flag, and user-facing output."""
|
||||
|
||||
@pytest.fixture
|
||||
def hermes_home(self, tmp_path, monkeypatch):
|
||||
root = tmp_path / ".hermes"
|
||||
root.mkdir()
|
||||
_make_hermes_tree(root)
|
||||
# Point HERMES_HOME at the temp dir so config + backup paths resolve here
|
||||
monkeypatch.setenv("HERMES_HOME", str(root))
|
||||
# Make Path.home() point at tmp_path for anything that uses it
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
||||
# Bust caches for hermes_cli.config + hermes_constants so they pick up HERMES_HOME
|
||||
for mod in list(__import__("sys").modules.keys()):
|
||||
if mod.startswith("hermes_cli.config") or mod == "hermes_constants":
|
||||
del __import__("sys").modules[mod]
|
||||
return root
|
||||
|
||||
def test_default_enabled_creates_backup(self, hermes_home, capsys):
|
||||
from hermes_cli.main import _run_pre_update_backup
|
||||
_run_pre_update_backup(Namespace(no_backup=False))
|
||||
out = capsys.readouterr().out
|
||||
assert "Creating pre-update backup" in out
|
||||
assert "Saved:" in out
|
||||
assert "Restore:" in out
|
||||
assert "hermes import" in out
|
||||
assert "Disable:" in out
|
||||
# Actual backup was created
|
||||
backups = list((hermes_home / "backups").glob("pre-update-*.zip"))
|
||||
assert len(backups) == 1
|
||||
|
||||
def test_no_backup_flag_skips(self, hermes_home, capsys):
|
||||
from hermes_cli.main import _run_pre_update_backup
|
||||
_run_pre_update_backup(Namespace(no_backup=True))
|
||||
out = capsys.readouterr().out
|
||||
assert "skipped (--no-backup)" in out
|
||||
assert "Creating pre-update backup" not in out
|
||||
# No backup written
|
||||
assert not (hermes_home / "backups").exists() or not list(
|
||||
(hermes_home / "backups").glob("pre-update-*.zip")
|
||||
)
|
||||
|
||||
def test_config_disabled_skips(self, hermes_home, capsys):
|
||||
import yaml
|
||||
(hermes_home / "config.yaml").write_text(yaml.safe_dump({
|
||||
"_config_version": 22,
|
||||
"updates": {"pre_update_backup": False},
|
||||
}))
|
||||
# Ensure config module re-reads
|
||||
import sys as _sys
|
||||
for mod in list(_sys.modules.keys()):
|
||||
if mod.startswith("hermes_cli.config"):
|
||||
del _sys.modules[mod]
|
||||
|
||||
from hermes_cli.main import _run_pre_update_backup
|
||||
_run_pre_update_backup(Namespace(no_backup=False))
|
||||
out = capsys.readouterr().out
|
||||
assert "disabled" in out
|
||||
assert "updates.pre_update_backup=false" in out
|
||||
assert not list((hermes_home / "backups").glob("pre-update-*.zip")) \
|
||||
if (hermes_home / "backups").exists() else True
|
||||
|
||||
def test_cli_flag_overrides_enabled_config(self, hermes_home, capsys):
|
||||
"""--no-backup wins even when config says pre_update_backup: true."""
|
||||
import yaml
|
||||
(hermes_home / "config.yaml").write_text(yaml.safe_dump({
|
||||
"_config_version": 22,
|
||||
"updates": {"pre_update_backup": True},
|
||||
}))
|
||||
import sys as _sys
|
||||
for mod in list(_sys.modules.keys()):
|
||||
if mod.startswith("hermes_cli.config"):
|
||||
del _sys.modules[mod]
|
||||
|
||||
from hermes_cli.main import _run_pre_update_backup
|
||||
_run_pre_update_backup(Namespace(no_backup=True))
|
||||
out = capsys.readouterr().out
|
||||
assert "skipped (--no-backup)" in out
|
||||
|
||||
Reference in New Issue
Block a user