# NousResearch/hermes-agent · Filesystem checkpointing and rollback before destructive operations ## Open now - Nothing open on GitHub. These members all concern a proposed feature for automatically creating checkpoints or backups of the filesystem before destructive file modifications, with support for rolling back to a prior state. Several reference Git-based or native checkpoint mechanisms and a rollback command. ## TL;DR - 5 member(s): 0 open issue(s), 3 closed issue(s), 0 open PR(s), 1 merged PR(s), 1 PR(s) closed without merging. - A fix was merged to main in PR #824. Verify it addresses your case. - **Recommended:** read PR #824 (and any competing PRs below) before filing. Confirm the merged change covers your symptom. ## Maintainer resolution verdict - Internal-tracker verdict over this cluster's issue members: `partially_resolved` · deciding PR(s): `pr-824`. This is distinct from the fix status above. Weigh it yourself. ## How this cluster was grouped - Link evidence (`STRUCTURAL_PARTIAL`): some members are connected by structural evidence (shared changed files, or a closing/superseded reference); the rest are joined by an automated relevance judgement over their text. - Members outside that structural core may belong to a different underlying problem. - Pairwise link density `0.27`: the share of member pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. `1.00` means every pair is linked. ## Closed / duplicate issues - #452: \[Feature\]: feat: filesystem checkpointing and rollback before destructive operations - #4489: \[Feature\]: Auto-Checkpoint with Git Before File Modifications - #29062: Auto-backup before every operation with instant rollback support ## Prior & competing PRs - **#824: feat: filesystem checkpoints and /rollback command**. merged to main · fix tier `verified` - maintainer direction: category `merged`; implication `superseded` - files: `gateway/run.py`, `hermes_cli/main.py`, `run_agent.py`, `tests/tools/test_checkpoint_manager.py`, `tools/checkpoint_manager.py`, `website/docs/reference/cli-commands.md`, `website/docs/user-guide/configuration.md`, `website/docs/user-guide/features/checkpoints.md` - fix hunks: - `gateway/run.py` ```diff @@ -771,7 +771,7 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: _known_commands = {"new", "reset", "help", "status", "stop", "model", "personality", "retry", "undo", "sethome", "set-home", "compress", "usage", "insights", "reload-mcp", "reload_mcp", - "update", "title", "resume", "provider"} + "update", "title", "resume", "provider", "rollback"} if command and command in _known_commands: await self.hooks.emit(f"command:{command}", { "platform": source.platform.value if source.platform else "", ``` ```diff @@ -830,6 +830,9 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: if command == "resume": return await self._handle_resume_command(event) + + if command == "rollback": + return await self._handle_rollback_command(event) # Skill slash commands: /skill-name loads the skill and sends to agent if command: ``` ```diff @@ -1400,6 +1403,7 @@ async def _handle_help_command(self, event: MessageEvent) -> str: "`/resume [name]` — Resume a previously-named session", "`/usage` — Show token usage for this session", "`/insights [days]` — Show usage insights and analytics", + "`/rollback [number]` — List or restore filesystem checkpoints", "`/reload-mcp` — Reload MCP servers from config", "`/update` — Update Hermes Agent to the latest version", "`/help` — Show this message", ``` ```diff @@ -1746,6 +1750,65 @@ async def _handle_set_home_command(self, event: MessageEvent) -> str: f"Cron jobs and cross-platform messages will be delivered here." ) + async def _handle_rollback_command(self, event: MessageEvent) -> str: + """Handle /rollback command — list or restore filesystem checkpoints.""" + from tools.checkpoint_manager import CheckpointManager, format_checkpoint_list + + # Read checkpoint config from config.yaml + cp_cfg = {} + try: + import yaml as _y + _cfg_path = _hermes_home / "config.yaml" + if _cfg_path.exists(): + with open(_cfg_path, encoding="utf-8") as _f: + _data = _y.safe_load(_f) or {} + cp_cfg = _data.get("checkpoints", {}) + if isinstance(cp_cfg, bool): + cp_cfg = {"enabled": cp_cfg} + except Exception: + pass + + if not cp_cfg.get("enabled", False): + return ( + "Checkpoints are not enabled.\n" + "Enable in config.yaml:\n```\ncheckpoints:\n enabled: true\n```" + ) + + mgr = CheckpointManager( + enabled=True, + max_snapshots=cp_cfg.get("max_snapshots", 50), + ) + + cwd = os.getenv("MESSAGING_CWD", str(Path.home())) + arg = event.get_command_args().strip() + + if not arg: + checkpoints = mgr.list_checkpoints(cwd) + return format_checkpoint_list(checkpoints, cwd) + + # Restore by number or hash + checkpoints = mgr.list_checkpoints(cwd) + if not checkpoints: + return f"No checkpoints found for {cwd}" + + target_hash = None + try: + idx = int(arg) - 1 + if 0 <= idx < len(checkpoints): + target_hash = checkpoints[idx]["hash"] + else: + return f"Invalid checkpoint number. Use 1-{len(checkpoints)}." + except ValueError: + target_hash = arg + + result = mgr.restore(cwd, target_hash) + if result["success"]: + return ( + f"✅ Restored to checkpoint {result['restored_to']}: {result['reason']}\n" + f"A pre-rollback snapshot was saved automatically." + ) + return f"❌ {result['error']}" + async def _handle_compress_command(self, event: MessageEvent) -> str: """Handle /compress command -- manually compress conversation context.""" source = event.source ``` - `hermes_cli/main.py` ```diff @@ -489,6 +489,7 @@ def cmd_chat(args): "query": args.query, "resume": getattr(args, "resume", None), "worktree": getattr(args, "worktree", False), + "checkpoints": getattr(args, "checkpoints", False), } # Filter out None values kwargs = {k: v for k, v in kwargs.items() if v is not None} ``` ```diff @@ -1927,6 +1928,12 @@ def main(): default=False, help="Run in an isolated git worktree (for parallel agents on the same repo)" ) + chat_parser.add_argument( + "--checkpoints", + action="store_true", + default=False, + help="Enable filesystem checkpoints before destructive file operations (use /rollback to restore)" + ) chat_parser.set_defaults(func=cmd_chat) # ========================================================================= ``` - `run_agent.py` ```diff @@ -185,6 +185,8 @@ def __init__( honcho_session_key: str = None, iteration_budget: "IterationBudget" = None, fallback_model: Dict[str, Any] = None, + checkpoints_enabled: bool = False, + checkpoint_max_snapshots: int = 50, ): """ Initialize the AI Agent. ``` ```diff @@ -486,6 +488,13 @@ def __init__( # Cached system prompt -- built once per session, only rebuilt on compression self._cached_system_prompt: Optional[str] = None + # Filesystem checkpoint manager (transparent — not a tool) + from tools.checkpoint_manager import CheckpointManager + self._checkpoint_mgr = CheckpointManager( + enabled=checkpoints_enabled, + max_snapshots=checkpoint_max_snapshots, + ) + # SQLite session store (optional -- provided by CLI or gateway) self._session_db = session_db if self._session_db: ``` ```diff @@ -2706,6 +2715,18 @@ def _execute_tool_calls(self, assistant_message, messages: list, effective_task_ except Exception as cb_err: logging.debug(f"Tool progress callback error: {cb_err}") + # Checkpoint: snapshot working dir before file-mutating tools + if function_name in ("write_file", "patch") and self._checkpoint_mgr.enabled: + try: + file_path = function_args.get("path", "") + if file_path: + work_dir = self._checkpoint_mgr.get_working_dir_for_path(file_path) + self._checkpoint_mgr.ensure_checkpoint( + work_dir, f"before {function_name}" + ) + except Exception: + pass # never block tool execution + tool_start_time = time.time() if function_name == "todo": ``` ```diff @@ -3215,6 +3236,9 @@ def run_conversation( self.clear_interrupt() while api_call_count < self.max_iterations and self.iteration_budget.remaining > 0: + # Reset per-turn checkpoint dedup so each iteration can take one snapshot + self._checkpoint_mgr.new_turn() + # Check for interrupt request (e.g., user sent new message) if self._interrupt_requested: interrupted = True ``` - `tests/tools/test_checkpoint_manager.py` ```diff @@ -0,0 +1,385 @@ +"""Tests for tools/checkpoint_manager.py — CheckpointManager.""" + +import os +import json +import shutil +import pytest +from pathlib import Path +from unittest.mock import patch + +from tools.checkpoint_manager import ( + CheckpointManager, + _shadow_repo_path, + _init_shadow_repo, + _run_git, + _git_env, + _dir_file_count, + format_checkpoint_list, + DEFAULT_EXCLUDES, + CHECKPOINT_BASE, +) + + +# ========================================================================= +# Fixtures +# ========================================================================= + [redacted:email]() +def work_dir(tmp_path): + """Temporary working directory.""" + d = tmp_path / "project" + d.mkdir() + (d / "main.py").write_text("print('hello')\\n") + (d / "README.md").write_text("# Project\\n") + return d + + [redacted:email]() +def checkpoint_base(tmp_path): + """Isolated checkpoint base — never writes to ~/.hermes/.""" + return tmp_path / "checkpoints" + + [redacted:email]() +def mgr(work_dir, checkpoint_base, monkeypatch): + """CheckpointManager with redirected checkpoint base.""" + monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base) + return CheckpointManager(enabled=True, max_snapshots=50) + + [redacted:email]() +def disabled_mgr(checkpoint_base, monkeypatch): + """Disabled CheckpointManager.""" + monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base) + return CheckpointManager(enabled=False) + + +# ========================================================================= +# Shadow repo path +# ========================================================================= + +class TestShadowRepoPath: + def test_deterministic(self, work_dir, checkpoint_base, monkeypatch): + monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base) + p1 = _shadow_repo_path(str(work_dir)) + p2 = _shadow_repo_path(str(work_dir)) + assert p1 == p2 + + def test_different_dirs_different_paths(self, tmp_path, checkpoint_base, monkeypatch): + monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base) + p1 = _shadow_repo_path(str(tmp_path / "a")) + p2 = _shadow_repo_path(str(tmp_path / "b")) + assert p1 != p2 + + def test_under_checkpoint_base(self, work_dir, checkpoint_base, monkeypatch): + monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base) + p = _shadow_repo_path(str(work_dir)) + assert str(p).startswith(str(checkpoint_base)) + + +# ========================================================================= +# Shadow repo init +# ========================================================================= + +class TestShadowRepoInit: + def test_creates_git_repo(self, work_dir, checkpoint_base, monkeypatch): + monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base) + shadow = _shadow_repo_path(str(work_dir)) + err = _init_shadow_repo(shadow, str(work_dir)) + assert err is None + assert (shadow / "HEAD").exists() + + def test_no_git_in_project_dir(self, work_dir, checkpoint_base, monkeypatch): + monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base) + shadow = _shadow_repo_path(str(work_dir)) + _init_shadow_repo(shadow, str(work_dir)) + assert not (work_dir / ".git").exists() + + def test_has_exclude_file(self, work_dir, checkpoint_base, monkeypatch): + monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base) + shadow = _shadow_repo_path(str(work_dir)) + _init_shadow_repo(shadow, str(work_dir)) + exclude = shadow / "info" / "exclude" + assert exclude.exists() + content = exclude.read_text() + assert "node_modules/" in content + assert ".env" in content + + def test_has_workdir_file(self, work_dir, checkpoint_base, monkeypatch): + monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base) + shadow = _shadow_repo_path(str(work_dir)) + _init_shadow_repo(shadow, str(work_dir)) + workdir_file = shadow / "HERMES_WORKDIR" + assert workdir_file.exists() + assert str(work_dir.resolve()) in workdir_file.read_text() + + def test_idempotent(self, work_dir, checkpoint_base, monkeypatch): + monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base) + shadow = _shadow_repo_path(str(work_dir)) + err1 = _init_shadow_repo(shadow, str(work_dir)) + err2 = _init_shadow_repo(shadow, str(work_dir)) + assert err1 is None + assert err2 is None + + +# ========================================================================= +# CheckpointManager — disabled +# ========================================================================= + +class TestDisabledManager: + def test_ensure_checkpoint_returns_false(self, disabled_mgr, work_dir): + assert disabled_mgr.ensure_checkpoint(str(work_dir)) is False + + def test_new_turn_works(self, disabled_mgr): + disabled_mgr.new_turn() # should not raise + + +# ========================================================================= +# CheckpointManager — taking checkpoints +# ========================================================================= + +class TestTakeCheckpoint: + def test_first_checkpoint(self, mgr, work_dir): + result = mgr.ensure_checkpoint(str(work_dir), "initial") + assert result is True + + def test_dedup_same_turn(self, mgr, work_dir): + r1 = mgr.ensure_checkpoint(str(work_dir), "first") + r2 = mgr.ensure_checkpoint(str(work_dir), "second") + assert r1 is True + assert r2 is False # dedup'd + + def test_new_turn_resets_dedup(self, mgr, work_dir): + r1 = mgr.ensure_checkpoint(str(work_dir), "turn 1") + assert r1 is True + + mgr.new_turn() + + # Modify a file so there's something to commit + (work_dir / "main.py").write_text("print('modified')\\n") + r2 = mgr.ensure_checkpoint(str(work_dir), "turn 2") + assert r2 is True + + def test_no_changes_skips_commit(self, mgr, work_dir): + # First checkpoint + mgr.ensure_checkpoint(str(work_dir), "initial") + mgr.new_turn() + + # No file changes — should return False (nothing to commit) + r = mgr.ensure_checkpoint(str(work_dir), "no changes") + assert r is False + + def test_skip_root_dir(self, mgr): + r = mgr.ensure_checkpoint("/", "root") + assert r is False + + def test_skip_home_dir(self, mgr): + r = mgr.ensure_checkpoint(str(Path.home()), "home") + assert r is False + + +# ========================================================================= +# CheckpointManager — listing checkpoints +# ========================================================================= + +class TestListCheckpoints: + def test_empty_when_no_checkpoints(self, mgr, work_dir): + result = mgr.list_checkpoints(str(work_dir)) + assert result == [] + + def test_list_after_take(self, mgr, work_dir): + mgr.ensure_checkpoint(str(work_dir), "test checkpoint") + result = mgr.list_checkpoints(str(work_dir)) + assert len(result) == 1 + assert result[0]["reason"] == "test checkpoint" + assert "hash" in result[0] + assert "short_hash" in result[0] + assert "timestamp" in result[0] + + def test_multiple_checkpoints_ordered(self, mgr, work_dir): + mgr.ensure_checkpoint(str(work_dir), "first") + mgr.new_turn() + + (work_dir / "main.py").write_text("v2\\n") + mgr.ensure_checkpoint(str(work_dir), "second") + mgr.new_turn() + + (work_dir / "main.py").write_text("v3\\n") + mgr.ensure_checkpoint(str(work_dir), "third") + + result = mgr.list_checkpoints(str(work_dir)) + assert len(result) == 3 + # Most recent first + assert result[0]["reason"] == "third" + assert result[2]["reason"] == "first" + + +# ========================================================================= +# CheckpointManager — restoring +# ========================================================================= + +class TestRestore: + def test_restore_to_previous(self, mgr, work_dir): + # Write original content + (work_dir / "main.py").write_text("original\\n") + mgr.ensure_checkpoint(str(work_dir), "original state") + mgr.new_turn() + + # Modify the file + (work_dir / "main.py").write_text("modified\\n") + + # Get the checkpoint hash + checkpoints = mgr.list_checkpoints(str(work_dir)) + assert len(checkpoints) == 1 + + # Restore + result = mgr.restore(str(work_dir), checkpoints[0]["hash"]) + assert result["success"] is True + + # File should be back to original + assert (work_dir / "main.py").read_text() == "original\\n" + + def test_restore_invalid_hash(self, mgr, work_dir): + mgr.ensure_checkpoint(str(work_dir), "initial") + result = mgr.restore(str(work_dir), "deadbeef1234") + assert result["success"] is False + + def test_restore_no_checkpoints(self, mgr, work_dir): + result = mgr.restore(str(work_dir), "abc123") + assert result["success"] is False + + def test_restore_creates_pre_rollback_snapshot(self, mgr, work_dir): + (work_dir / "main.py").write_text("v1\\n") + mgr.ensure_checkpoint(str(work_dir), "v1") + mgr.new_turn() + + (work_dir / "main.py").write_text("v2\\n") + + checkpoints = mgr.list_checkpoints(str(work_dir)) + mgr.restore(str(work_dir), checkpoints[0]["hash"]) + + # Should now have 2 checkpoints: original + pre-rollback + all_cps = mgr.list_checkpoints(str(work_dir)) + assert len(all_cps) >= 2 + assert "pre-rollback" in all_cps[0]["reason"] + + +# ========================================================================= +# CheckpointManager — working dir resolution +# ========================================================================= + +class TestWorkingDirResolution: + def test_resolves_git_project_root(self, tmp_path): + mgr = CheckpointManager(enabled=True) + project = tmp_path / "myproject" + project.mkdir() + (project / ".git").mkdir() + subdir = project / "src" + subdir.mkdir() + filepath = subdir / "main.py" + filepath.write_text("x\\n") + + result = mgr.get_working_dir_for_path(str(filepath)) + assert result == str(project) + + def test_resolves_pyproject_root(self, tmp_path): + mgr = CheckpointManager(enabled=True) + project = tmp_path / "pyproj" + project.mkdir() + (project / "pyproject.toml").write_text("[project]\\n") + subdir = project / "src" + subdir.mkdir() + + result = mgr.get_working_dir_for_path(str(subdir / "file.py")) + assert result == str(project) + + def test_falls_back_to_parent(self, tmp_path): + mgr = CheckpointManager(enabled=True) + filepath = tmp_path / "random" / "file.py" + filepath.parent.mkdir(parents=True) + filepath.write_text("x\\n") + + result = mgr.get_working_dir_for_path(str(filepath)) + assert result == str(filepath.parent) + + +# ========================================================================= +# Git env isolation +# ========================================================================= + +class TestGitEnvIsolation: + def test_sets_git_dir(self, tmp_path): + shadow = tmp_path / "shadow" + env = _git_env(shadow, str(tmp_path / "work")) + assert env["GIT_DIR"] == str(shadow) + + def test_sets_work_tree(self, tmp_path): + shadow = tmp_path / "shadow" + work = tmp_path / "work" + env = _git_env(shadow, str(work)) + assert env["GIT_WORK_TREE"] == str(work.resolve()) + + def test_clears_index_file(self, tmp_path, monkeypatch): + monkeypatch.setenv("GIT_INDEX_FILE", "/some/index") + shadow = tmp_path / "shadow" + env = _git_env(shadow, str(tmp_path)) + assert "GIT_INDEX_FILE" not in env + + +# ========================================================================= +# format_checkpoint_list +# ========================================================================= + +class TestFormatCheckpointList: + def test_empty_list(self): + result = format_checkpoint_list([], "/some/dir") + assert "No checkpoints" in result + + def test_formats_entries(self): + cps = [ + {"hash": "abc123", "short_hash": "abc1", "timestamp": "2026-03-09T21:15:00-07:00", "reason": "before write_file"}, + {"hash": "def456", "short_hash": "def4", "timestamp": "2026-03-09T21:10:00-07:00", "reason": "before patch"}, + ] + result = format_checkpoint_list(cps, "/home/user/project") + assert "abc1" in result + assert "def4" in result + assert "before write_file" in result + assert "/rollback" in result + + +# ========================================================================= +# File count guard +# ========================================================================= + +class TestDirFileCount: + def test_counts_files(self, work_dir): + count = _dir_file_count(str(work_dir)) + assert count >= 2 # main.py + README.md + + def test_nonexistent_dir(self, tmp_path): + count = _dir_file_count(str(tmp_path / "nonexistent")) + assert count == 0 + + +# ========================================================================= +# Error resilience +# ========================================================================= + +class TestErrorResilience: + def test_no_git_installed(self, work_dir, checkpoint_base, monkeypatch): + monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base) + mgr = CheckpointManager(enabled=True) + # Mock git not found + monkeypatch.setattr("shutil.which", lambda x: None) + mgr._git_available = None # reset lazy probe + result = mgr.ensure_checkpoint(str(work_dir), "test") + assert result is False + + def test_checkpoint_failure_does_not_raise(self, mgr, work_dir, monkeypatch): + """Checkpoint failures should never raise — they're silently logged.""" + def broken_run_git(*args, **kwargs): + raise OSError("git exploded") + monkeypatch.setattr("tools.checkpoint_manager._run_git", broken_run_git) + # Should not raise + result = mgr.ensure_checkpoint(str(work_dir), "test") + assert result is False ``` - `tools/checkpoint_manager.py` ```diff @@ -0,0 +1,441 @@ +""" +Checkpoint Manager — Transparent filesystem snapshots via shadow git repos. + +Creates automatic snapshots of working directories before file-mutating +operations (write_file, patch), triggered once per conversation turn. +Provides rollback to any previous checkpoint. + +This is NOT a tool — the LLM never sees it. It's transparent infrastructure +controlled by the ``checkpoints`` config flag or ``--checkpoints`` CLI flag. + +Architecture: + ~/.hermes/checkpoints/{sha256(abs_dir)[:16]}/ — shadow git repo + HEAD, refs/, objects/ — standard git internals + HERMES_WORKDIR — original dir path + info/exclude — default excludes + +The shadow repo uses GIT_DIR + GIT_WORK_TREE so no git state leaks +into the user's project directory. +""" + +import hashlib +import logging +import os +import shutil +import subprocess +import time +from pathlib import Path +from typing import Dict, List, Optional, Set + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +CHECKPOINT_BASE = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes")) / "checkpoints" + +DEFAULT_EXCLUDES = [ + "node_modules/", + "dist/", + "build/", + ".env", + ".env.*", + ".env.local", + ".env.*.local", + "__pycache__/", + "*.pyc", + "*.pyo", + ".DS_Store", + "*.log", + ".cache/", + ".next/", + ".nuxt/", + "coverage/", + ".pytest_cache/", + ".venv/", + "venv/", + ".git/", +] + +# Git subprocess timeout (seconds). +_GIT_TIMEOUT: int = max(10, min(60, int(os.getenv("HERMES_CHECKPOINT_TIMEOUT", "30")))) + +# Max files to snapshot — skip huge directories to avoid slowdowns. +_MAX_FILES = 50_000 + + +# --------------------------------------------------------------------------- +# Shadow repo helpers +# --------------------------------------------------------------------------- + +def _shadow_repo_path(working_dir: str) -> Path: + """Deterministic shadow repo path: sha256(abs_path)[:16].""" + abs_path = str(Path(working_dir).resolve()) + dir_hash = hashlib.sha256(abs_path.encode()).hexdigest()[:16] + return CHECKPOINT_BASE / dir_hash + + +def _git_env(shadow_repo: Path, working_dir: str) -> dict: + """Build env dict that redirects git to the shadow repo.""" + env = os.environ.copy() + env["GIT_DIR"] = str(shadow_repo) + env["GIT_WORK_TREE"] = str(Path(working_dir).resolve()) + env.pop("GIT_INDEX_FILE", None) + env.pop("GIT_NAMESPACE", None) + env.pop("GIT_ALTERNATE_OBJECT_DIRECTORIES", None) + return env + + +def _run_git( + args: List[str], + shadow_repo: Path, + working_dir: str, + timeout: int = _GIT_TIMEOUT, +) -> tuple: + """Run a git command against the shadow repo. Returns (ok, stdout, stderr).""" + env = _git_env(shadow_repo, working_dir) + try: + result = subprocess.run( + ["git"] + args, + capture_output=True, + text=True, + timeout=timeout, + env=env, + cwd=str(Path(working_dir).resolve()), + ) + return result.returncode == 0, result.stdout.strip(), result.stderr.strip() + except subprocess.TimeoutExpired: + return False, "", f"git timed out after {timeout}s: git {' '.join(args)}" + except FileNotFoundError: + return False, "", "git not found" + except Exception as exc: + return False, "", str(exc) + + +def _init_shadow_repo(shadow_repo: Path, working_dir: str) -> Optional[str]: + """Initialise shadow repo if needed. Returns error string or None.""" + if (shadow_repo / "HEAD").exists(): + return None + + shadow_repo.mkdir(parents=True, exist_ok=True) + + ok, _, err = _run_git(["init"], shadow_repo, working_dir) + if not ok: + return f"Shadow repo init failed: {err}" + + _run_git(["config", "user.email", "hermes@local"], shadow_repo, working_dir) + _run_git(["config", "user.name", "Hermes Checkpoint"], shadow_repo, working_dir) + + info_dir = shadow_repo / "info" + info_dir.mkdir(exist_ok=True) + (info_dir / "exclude").write_text( + "\n".join(DEFAULT_EXCLUDES) + "\n", encoding="utf-8" + ) + + (shadow_repo / "HERMES_WORKDIR").write_text( + str(Path(working_dir).resolve()) + "\n", encoding="utf-8" + ) + + logger.debug("Initialised checkpoint repo at %s for %s", shadow_repo, working_dir) + return None + + +def _dir_file_count(path: str) -> int: + """Quick file count estimate (stops early if over _MAX_FILES).""" + count = 0 + try: + for _ in Path(path).rglob("*"): + count += 1 + if count > _MAX_FILES: + return count + except (PermissionError, OSError): + pass + return count + + +# --------------------------------------------------------------------------- +# CheckpointManager +# --------------------------------------------------------------------------- + +class CheckpointManager: + """Manages automatic filesystem checkpoints. + + Designed to be owned by AIAgent. Call ``new_turn()`` at the start of + each conversation turn and ``ensure_checkpoint(dir, reason)`` before + any file-mutating tool call. The manager deduplicates so at most one + snapshot is taken per directory per turn. + + Parameters + ---------- + enabled : bool + Master switch (from config / CLI flag). + max_snapshots : int + Keep at most this many checkpoints per directory. + """ + + def __init__(self, enabled: bool = False, max_snapshots: int = 50): + self.enabled = enabled + self.max_snapshots = max_snapshots + self._checkpointed_dirs: Set[str] = set() + self._git_available: Optional[bool] = None # lazy probe + + # ------------------------------------------------------------------ + # Turn lifecycle + # ------------------------------------------------------------------ + + def new_turn(self) -> None: + """Reset per-turn dedup. Call at the start of each agent iteration.""" + self._checkpointed_dirs.clear() + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def ensure_checkpoint(self, working_dir: str, reason: str = "auto") -> bool: + """Take a checkpoint if enabled and not already done this turn. + + Returns True if a checkpoint was taken, False otherwise. + Never raises — all errors are silently logged. + """ + if not self.enabled: + return False + + # Lazy git probe + if self._git_available is None: + self._git_available = shutil.which("git") is not None + if not self._git_available: + logger.debug("Checkpoints disabled: git not found") + if not self._git_available: + return False + + abs_dir = str(Path(working_dir).resolve()) + + # Skip root, home, and other overly broad directories + if abs_dir in ("/", str(Path.home())): + logger.debug("Checkpoint skipped: directory too broad (%s)", abs_dir) + return False + + # Already checkpointed this turn? + if abs_dir in self._checkpointed_dirs: + return False + + self._checkpointed_dirs.add(abs_dir) + + try: + return self._take(abs_dir, reason) + except Exception as e: + logger.debug("Checkpoint failed (non-fatal): %s", e) + return False + + def list_checkpoints(self, working_dir: str) -> List[Dict]: + """List available checkpoints for a directory. + + Returns a list of dicts with keys: hash, short_hash, timestamp, reason. + Most recent first. + """ + abs_dir = str(Path(working_dir).resolve()) + shadow = _shadow_repo_path(abs_dir) + + if not (shadow / "HEAD").exists(): + return [] + + ok, stdout, _ = _run_git( + ["log", "--format=%H|%h|%aI|%s", "--no-walk=unsorted", + "--all" if False else "HEAD", # just HEAD lineage + "-n", str(self.max_snapshots)], + shadow, abs_dir, + ) + + # Simpler: just use regular log + ok, stdout, _ = _run_git( + ["log", "--format=%H|%h|%aI|%s", "-n", str(self.max_snapshots)], + shadow, abs_dir, + ) + + if not ok or not stdout: + return [] + + results = [] + for line in stdout.splitlines(): + parts = line.split("|", 3) + if len(parts) == 4: + results.append({ + "hash": parts[0], + "short_hash": parts[1], + "timestamp": parts[2], + "reason": parts[3], + }) + return results + + def restore(self, working_dir: str, commit_hash: str) -> Dict: + """Restore files to a checkpoint state. + + Uses ``git checkout -- .`` which restores tracked files + without moving HEAD — safe and reversible. + + Returns dict with success/error info. + """ + abs_dir = str(Path(working_dir).resolve()) + shadow = _shadow_repo_path(abs_dir) + + if not (shadow / "HEAD").exists(): + return {"success": False, "error": "No checkpoints exist for this directory"} + + # Verify the commit exists + ok, _, err = _run_git( + ["cat-file", "-t", commit_hash], shadow, abs_dir, + ) + if not ok: + return {"success": False, "error": f"Checkpoint '{commit_hash}' not found"} + + # Take a checkpoint of current state before restoring (so you can undo the undo) + self._take(abs_dir, f"pre-rollback snapshot (restoring to {commit_hash[:8]})") + + # Restore + ok, stdout, err = _run_git( + ["checkout", commit_hash, "--", "."], + shadow, abs_dir, timeout=_GIT_TIMEOUT * 2, + ) + + if not ok: + return {"success": False, "error": f"Restore failed: {err}"} + + # Get info about what was restored + ok2, reason_out, _ = _run_git( + ["log", "--format=%s", "-1", commit_hash], shadow, abs_dir, + ) + reason = reason_out if ok2 else "unknown" + + return { + "success": True, + "restored_to": commit_hash[:8], + "reason": reason, + "directory": abs_dir, + } + + def get_working_dir_for_path(self, file_path: str) -> str: + """Resolve a file path to its working directory for checkpointing. + + Walks up from the file's parent to find a reasonable project root + (directory containing .git, pyproject.toml, package.json, etc.). + Falls back to the file's parent directory. + """ + path = Path(file_path).resolve() + if path.is_dir(): + candidate = path + else: + candidate = path.parent + + # Walk up looking for project root markers + markers = {".git", "pyproject.toml", "package.json", "Cargo.toml", + "go.mod", "Makefile", "pom.xml", ".hg", "Gemfile"} + check = candidate + while check != check.parent: + if any((check / m).exists() for m in markers): + return str(check) + check = check.parent + + # No project root found — use the file's parent + return str(candidate) + + # ------------------------------------------------------------------ + # Internal + # ------------------------------------------------------------------ + + def _take(self, working_dir: str, reason: str) -> bool: + """Take a snapshot. Returns True on success.""" + shadow = _shadow_repo_path(working_dir) + + # Init if needed + err = _init_shadow_repo(shadow, working_dir) + if err: + logger.debug("Checkpoint init failed: %s", err) + return False + + # Quick size guard — don't try to snapshot enormous directories + if _dir_file_count(working_dir) > _MAX_FILES: + logger.debug("Checkpoint skipped: >%d files in %s", _MAX_FILES, working_dir) + return False + + # Stage everything + ok, _, err = _run_git( + ["add", "-A"], shadow, working_dir, timeout=_GIT_TIMEOUT * 2, + ) + if not ok: + logger.debug("Checkpoint git-add failed: %s", err) + return False + + # Check if there's anything to commit + ok_diff, diff_out, _ = _run_git( + ["diff", "--cached", "--quiet"], shadow, working_dir, + ) + if ok_diff: + # No changes to commit + logger.debug("Checkpoint skipped: no changes in %s", working_dir) + return False + + # Commit + ok, _, err = _run_git( + ["commit", "-m", reason, "--allow-empty-message"], + shadow, working_dir, timeout=_GIT_TIMEOUT * 2, + ) + if not ok: + logger.debug("Checkpoint commit failed: %s", err) + return False + + logger.debug("Checkpoint taken in %s: %s", working_dir, reason) + + # Prune old snapshots + self._prune(shadow, working_dir) + + return True + + def _prune(self, shadow_repo: Path, working_dir: str) -> None: + """Keep only the last max_snapshots commits via orphan reset.""" + ok, stdout, _ = _run_git( + ["rev-list", "--count", "HEAD"], shadow_repo, working_dir, + ) + if not ok: + return + + try: + count = int(stdout) + except ValueError: + return + + if count <= self.max_snapshots: + return + + # Get the hash of the commit at the cutoff point + ok, cutoff_hash, _ = _run_git( + ["rev-list", "--reverse", "HEAD", "--skip=0", + f"--max-count=1"], + shadow_repo, working_dir, + ) + + # For simplicity, we don't actually prune — git's pack mechanism + # handles this efficiently, and the objects are small. The log + # listing is already limited by max_snapshots. + # Full pruning would require rebase --onto or filter-branch which + # is fragile for a background feature. We just limit the log view. + logger.debug("Checkpoint repo has %d commits (limit %d)", count, self.max_snapshots) + + +def format_checkpoint_list(checkpoints: List[Dict], directory: str) -> str: + """Format checkpoint list for display to user.""" + if not checkpoints: + return f"No checkpoints found for {directory}" + + lines = [f"📸 Checkpoints for {directory}:\n"] + for i, cp in enumerate(checkpoints, 1): + # Parse ISO timestamp to something readable + ts = cp["timestamp"] + if "T" in ts: + ts = ts.split("T")[1].split("+")[0].split("-")[0][:5] # HH:MM + date = cp["timestamp"].split("T")[0] + ts = f"{date} {ts}" + lines.append(f" {i}. {cp['short_hash']} {ts} {cp['reason']}") + + lines.append(f"\nUse /rollback to restore, e.g. /rollback 1") + return "\n".join(lines) ``` - `website/docs/reference/cli-commands.md` ```diff @@ -24,6 +24,7 @@ These are commands you run from your shell. | `hermes chat --toolsets "web,terminal"` / `-t` | Use specific toolsets | | `hermes chat --verbose` | Enable verbose/debug output | | `hermes --worktree` / `-w` | Start in an isolated git worktree (for parallel agents) | +| `hermes --checkpoints` | Enable filesystem checkpoints before destructive file operations | ### Provider & Model Management ``` ```diff @@ -202,6 +203,8 @@ These work in messaging platforms (Telegram, Discord, Slack, WhatsApp) but not t | `/sethome` | Set this chat as the home channel | | `/status` | Show session info | | `/reload-mcp` | Reload MCP servers from config | +| `/rollback` | List filesystem checkpoints for the current directory | +| `/rollback ` | Restore files to checkpoint #N | | `/update` | Update Hermes Agent to the latest version | --- ``` - `website/docs/user-guide/configuration.md` ```diff @@ -663,6 +663,16 @@ browser: record_sessions: false # Auto-record browser sessions as WebM videos to ~/.hermes/browser_recordings/ ``` +## Checkpoints + +Automatic filesystem snapshots before destructive file operations. See the [Checkpoints feature page](/docs/user-guide/features/checkpoints) for details. + +```yaml +checkpoints: + enabled: false # Enable automatic checkpoints (also: hermes --checkpoints) + max_snapshots: 50 # Max checkpoints to keep per directory +``` + ## Delegation Configure subagent behavior for the delegate tool: ``` - `website/docs/user-guide/features/checkpoints.md` ```diff @@ -0,0 +1,97 @@ +# Filesystem Checkpoints + +Hermes can automatically snapshot your working directory before making file changes, giving you a safety net to roll back if something goes wrong. + +## How It Works + +When enabled, Hermes takes a **one-time snapshot** at the start of each conversation turn before the first file-modifying operation (`write_file` or `patch`). This creates a point-in-time backup you can restore to at any time. + +Under the hood, checkpoints use a **shadow git repository** stored at `~/.hermes/checkpoints/`. This is completely separate from your project's git — no `.git` directory is created in your project, and your own git history is never touched. + +## Enabling Checkpoints + +### Per-session (CLI flag) + +```bash +hermes --checkpoints +``` + +### Permanently (config.yaml) + +```yaml +# ~/.hermes/config.yaml +checkpoints: + enabled: true + max_snapshots: 50 # max checkpoints per directory (default: 50) +``` + +## Rolling Back + +Use the `/rollback` slash command: + +``` +/rollback # List all available checkpoints +/rollback 1 # Restore to checkpoint #1 (most recent) +/rollback 3 # Restore to checkpoint #3 (further back) +/rollback abc1234 # Restore by git commit hash +``` + +Example output: + +``` +📸 Checkpoints for /home/user/project: + + 1. abc1234 2026-03-10 14:22 before write_file + 2. def5678 2026-03-10 14:15 before patch + 3. ghi9012 2026-03-10 14:08 before write_file + +Use /rollback to restore, e.g. /rollback 1 +``` + +When you restore, Hermes automatically takes a **pre-rollback snapshot** first — so you can always undo your undo. + +## What Gets Checkpointed + +Checkpoints capture the entire working directory (the project root), excluding common large/sensitive patterns: + +- `node_modules/`, `dist/`, `build/` +- `.env`, `.env.*` +- `__pycache__/`, `*.pyc` +- `.venv/`, `venv/` +- `.git/` +- `.DS_Store`, `*.log` + +## Performance + +Checkpoints are designed to be lightweight: + +- **Once per turn** — only the first file operation triggers a snapshot, not every write +- **Skips large directories** — directories with >50,000 files are skipped automatically +- **Skips when nothing changed** — if no files were modified since the last checkpoint, no commit is created +- **Non-blocking** — if a checkpoint fails for any reason, the file operation proceeds normally + +## How It Determines the Project Root + +When you write to a file like `src/components/Button.tsx`, Hermes walks up the directory tree looking for project markers (`.git`, `pyproject.toml`, `package.json`, `Cargo.toml`, etc.) to find the project root. This ensures the entire project is checkpointed, not just the file's parent directory. + +## Platforms + +Checkpoints work on both: +- **CLI** — uses your current working directory +- **Gateway** (Telegram, Discord, etc.) — uses `MESSAGING_CWD` + +The `/rollback` command is available on all platforms. + +## FAQ + +**Does this conflict with my project's git?** +No. Checkpoints use a completely separate shadow git repository via `GIT_DIR` environment variables. Your project's `.git/` is never touched. + +**How much disk space do checkpoints use?** +Git is very efficient at storing diffs. For most projects, checkpoint data is negligible. Old checkpoints are pruned when `max_snapshots` is exceeded. + +**Can I checkpoint without git installed?** +No — git must be available on your PATH. If it's not installed, checkpoints silently disable. + +**Can I roll back across sessions?** +Yes! Checkpoints persist in `~/.hermes/checkpoints/` and survive across sessions. You can roll back to a checkpoint from yesterday. ``` - **#559: feat(checkpoint): filesystem checkpointing and rollback before destructive operations**. closed without merge - maintainer direction: category `uncategorized`; implication `needs_llm` ## Fix status - Merged to main in PR #824 (cited above). No release data is tracked. Verify downstream. - Fix last verified (merged) `2026-03-10`. This confirmation is old and may be out of date (a revert/reopen would not have invalidated it here); re-verify. - Subsystem: `agent`