diff --git a/hermes_cli/backup.py b/hermes_cli/backup.py index cae165b908..51437c03a6 100644 --- a/hermes_cli/backup.py +++ b/hermes_cli/backup.py @@ -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 ``/backups/pre-update-.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 diff --git a/hermes_cli/config.py b/hermes_cli/config.py index e061fff62c..f0777b80aa 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -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 ``/backups/`` and + # can be restored with ``hermes import ``. 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, } diff --git a/hermes_cli/main.py b/hermes_cli/main.py index ca0bef6106..b838fb9e0e 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -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) # ========================================================================= diff --git a/tests/hermes_cli/test_backup.py b/tests/hermes_cli/test_backup.py index 805862cd80..f12e4757b2 100644 --- a/tests/hermes_cli/test_backup.py +++ b/tests/hermes_cli/test_backup.py @@ -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