Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
136 changes: 136 additions & 0 deletions hermes_cli/backup.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
"__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
Expand Down Expand Up @@ -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
13 changes: 13 additions & 0 deletions hermes_cli/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -1037,6 +1037,19 @@ def _ensure_hermes_home_managed(home: Path):
"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,
}
Expand Down
94 changes: 94 additions & 0 deletions hermes_cli/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -9577,6 +9665,12 @@ def cmd_claw(args):
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)

# =========================================================================
Expand Down
Loading
Loading