mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-04-28 06:51:16 +08:00
478444c262
Every working dir hermes ever touches gets its own shadow git repo under
~/.hermes/checkpoints/{sha256(abs_dir)[:16]}/. The per-repo _prune is a
no-op (comment in CheckpointManager._prune says so), so abandoned repos
from deleted/moved projects or one-off tmp dirs pile up forever. Field
reports put the typical offender at 1000+ repos / ~12 GB on active
contributor machines.
Adds an opt-in startup sweep that mirrors the sessions.auto_prune
pattern from #13861 / #16286:
- tools/checkpoint_manager.py: new prune_checkpoints() and
maybe_auto_prune_checkpoints() helpers. Deletes shadow repos that
are orphan (HERMES_WORKDIR marker points to a path that no longer
exists) or stale (newest in-repo mtime older than retention_days).
Idempotent via a CHECKPOINT_BASE/.last_prune marker file so it only
runs once per min_interval_hours regardless of how many hermes
processes start up.
- hermes_cli/config.py: new checkpoints.auto_prune /
retention_days / delete_orphans / min_interval_hours knobs.
Default auto_prune: false so users who rely on /rollback against
long-ago sessions never lose data silently.
- cli.py / gateway/run.py: startup hooks gated on checkpoints.auto_prune,
called right next to the existing state.db maintenance block.
- Docs updated with the new config knobs.
- 11 regression tests: orphan/stale deletion, precedence, byte-freed
tracking, non-shadow dir skip, interval gating, corrupt marker
recovery.
Refs #3015 (session-file disk growth was fixed in #16286; this covers
the checkpoint side noted out-of-scope there).
910 lines
36 KiB
Python
910 lines
36 KiB
Python
"""Tests for tools/checkpoint_manager.py — CheckpointManager."""
|
|
|
|
import logging
|
|
import subprocess
|
|
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
|
|
# =========================================================================
|
|
|
|
@pytest.fixture()
|
|
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
|
|
|
|
|
|
@pytest.fixture()
|
|
def checkpoint_base(tmp_path):
|
|
"""Isolated checkpoint base — never writes to ~/.hermes/."""
|
|
return tmp_path / "checkpoints"
|
|
|
|
|
|
@pytest.fixture()
|
|
def fake_home(tmp_path, monkeypatch):
|
|
"""Set a deterministic fake home for expanduser/path-home behavior."""
|
|
home = tmp_path / "home"
|
|
home.mkdir()
|
|
monkeypatch.setenv("HOME", str(home))
|
|
monkeypatch.setenv("USERPROFILE", str(home))
|
|
monkeypatch.delenv("HOMEDRIVE", raising=False)
|
|
monkeypatch.delenv("HOMEPATH", raising=False)
|
|
monkeypatch.setattr(Path, "home", classmethod(lambda cls: home))
|
|
return home
|
|
|
|
|
|
@pytest.fixture()
|
|
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)
|
|
|
|
|
|
@pytest.fixture()
|
|
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))
|
|
|
|
def test_tilde_and_expanded_home_share_shadow_repo(self, fake_home, checkpoint_base, monkeypatch):
|
|
monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base)
|
|
project = fake_home / "project"
|
|
project.mkdir()
|
|
|
|
tilde_path = f"~/{project.name}"
|
|
expanded_path = str(project)
|
|
|
|
assert _shadow_repo_path(tilde_path) == _shadow_repo_path(expanded_path)
|
|
|
|
|
|
# =========================================================================
|
|
# 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_successful_checkpoint_does_not_log_expected_diff_exit(self, mgr, work_dir, caplog):
|
|
with caplog.at_level(logging.ERROR, logger="tools.checkpoint_manager"):
|
|
result = mgr.ensure_checkpoint(str(work_dir), "initial")
|
|
assert result is True
|
|
assert not any("diff --cached --quiet" in r.getMessage() for r in caplog.records)
|
|
|
|
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"
|
|
|
|
def test_tilde_path_lists_same_checkpoints_as_expanded_path(self, checkpoint_base, fake_home, monkeypatch):
|
|
monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base)
|
|
mgr = CheckpointManager(enabled=True, max_snapshots=50)
|
|
project = fake_home / "project"
|
|
project.mkdir()
|
|
(project / "main.py").write_text("v1\n")
|
|
|
|
tilde_path = f"~/{project.name}"
|
|
assert mgr.ensure_checkpoint(tilde_path, "initial") is True
|
|
|
|
listed = mgr.list_checkpoints(str(project))
|
|
assert len(listed) == 1
|
|
assert listed[0]["reason"] == "initial"
|
|
|
|
|
|
# =========================================================================
|
|
# 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"]
|
|
|
|
def test_tilde_path_supports_diff_and_restore_flow(self, checkpoint_base, fake_home, monkeypatch):
|
|
monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base)
|
|
mgr = CheckpointManager(enabled=True, max_snapshots=50)
|
|
project = fake_home / "project"
|
|
project.mkdir()
|
|
file_path = project / "main.py"
|
|
file_path.write_text("original\n")
|
|
|
|
tilde_path = f"~/{project.name}"
|
|
assert mgr.ensure_checkpoint(tilde_path, "initial") is True
|
|
mgr.new_turn()
|
|
|
|
file_path.write_text("changed\n")
|
|
checkpoints = mgr.list_checkpoints(str(project))
|
|
diff_result = mgr.diff(tilde_path, checkpoints[0]["hash"])
|
|
assert diff_result["success"] is True
|
|
assert "main.py" in diff_result["diff"]
|
|
|
|
restore_result = mgr.restore(tilde_path, checkpoints[0]["hash"])
|
|
assert restore_result["success"] is True
|
|
assert file_path.read_text() == "original\n"
|
|
|
|
|
|
# =========================================================================
|
|
# 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, monkeypatch):
|
|
mgr = CheckpointManager(enabled=True)
|
|
filepath = tmp_path / "random" / "file.py"
|
|
filepath.parent.mkdir(parents=True)
|
|
filepath.write_text("x\\n")
|
|
|
|
# The walk-up scan for project markers (.git, pyproject.toml, etc.)
|
|
# stops at tmp_path — otherwise stray markers in ``/tmp`` (e.g.
|
|
# ``/tmp/pyproject.toml`` left by other tools on the host) get
|
|
# picked up as the project root and this test flakes on shared CI.
|
|
import pathlib as _pl
|
|
_real_exists = _pl.Path.exists
|
|
|
|
def _guarded_exists(self):
|
|
s = str(self)
|
|
stop = str(tmp_path)
|
|
if not s.startswith(stop) and any(
|
|
s.endswith("/" + m) or s == "/" + m
|
|
for m in (".git", "pyproject.toml", "package.json",
|
|
"Cargo.toml", "go.mod", "Makefile", "pom.xml",
|
|
".hg", "Gemfile")
|
|
):
|
|
return False
|
|
return _real_exists(self)
|
|
|
|
monkeypatch.setattr(_pl.Path, "exists", _guarded_exists)
|
|
|
|
result = mgr.get_working_dir_for_path(str(filepath))
|
|
assert result == str(filepath.parent)
|
|
|
|
def test_resolves_tilde_path_to_project_root(self, fake_home):
|
|
mgr = CheckpointManager(enabled=True)
|
|
project = fake_home / "myproject"
|
|
project.mkdir()
|
|
(project / "pyproject.toml").write_text("[project]\n")
|
|
subdir = project / "src"
|
|
subdir.mkdir()
|
|
filepath = subdir / "main.py"
|
|
filepath.write_text("x\n")
|
|
|
|
result = mgr.get_working_dir_for_path(f"~/{project.name}/src/main.py")
|
|
assert result == str(project)
|
|
|
|
|
|
# =========================================================================
|
|
# 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
|
|
|
|
def test_expands_tilde_in_work_tree(self, fake_home, tmp_path):
|
|
shadow = tmp_path / "shadow"
|
|
work = fake_home / "work"
|
|
work.mkdir()
|
|
|
|
env = _git_env(shadow, f"~/{work.name}")
|
|
assert env["GIT_WORK_TREE"] == str(work.resolve())
|
|
|
|
|
|
# =========================================================================
|
|
# 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_run_git_allows_expected_nonzero_without_error_log(self, tmp_path, caplog):
|
|
work = tmp_path / "work"
|
|
work.mkdir()
|
|
completed = subprocess.CompletedProcess(
|
|
args=["git", "diff", "--cached", "--quiet"],
|
|
returncode=1,
|
|
stdout="",
|
|
stderr="",
|
|
)
|
|
with patch("tools.checkpoint_manager.subprocess.run", return_value=completed):
|
|
with caplog.at_level(logging.ERROR, logger="tools.checkpoint_manager"):
|
|
ok, stdout, stderr = _run_git(
|
|
["diff", "--cached", "--quiet"],
|
|
tmp_path / "shadow",
|
|
str(work),
|
|
allowed_returncodes={1},
|
|
)
|
|
assert ok is False
|
|
assert stdout == ""
|
|
assert stderr == ""
|
|
assert not caplog.records
|
|
|
|
def test_run_git_invalid_working_dir_reports_path_error(self, tmp_path, caplog):
|
|
missing = tmp_path / "missing"
|
|
with caplog.at_level(logging.ERROR, logger="tools.checkpoint_manager"):
|
|
ok, stdout, stderr = _run_git(
|
|
["status"],
|
|
tmp_path / "shadow",
|
|
str(missing),
|
|
)
|
|
assert ok is False
|
|
assert stdout == ""
|
|
assert "working directory not found" in stderr
|
|
assert not any("Git executable not found" in r.getMessage() for r in caplog.records)
|
|
|
|
def test_run_git_missing_git_reports_git_not_found(self, tmp_path, monkeypatch, caplog):
|
|
work = tmp_path / "work"
|
|
work.mkdir()
|
|
|
|
def raise_missing_git(*args, **kwargs):
|
|
raise FileNotFoundError(2, "No such file or directory", "git")
|
|
|
|
monkeypatch.setattr("tools.checkpoint_manager.subprocess.run", raise_missing_git)
|
|
with caplog.at_level(logging.ERROR, logger="tools.checkpoint_manager"):
|
|
ok, stdout, stderr = _run_git(
|
|
["status"],
|
|
tmp_path / "shadow",
|
|
str(work),
|
|
)
|
|
assert ok is False
|
|
assert stdout == ""
|
|
assert stderr == "git not found"
|
|
assert any("Git executable not found" in r.getMessage() for r in caplog.records)
|
|
|
|
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
|
|
|
|
|
|
# =========================================================================
|
|
# Security / Input validation
|
|
# =========================================================================
|
|
|
|
class TestSecurity:
|
|
def test_restore_rejects_argument_injection(self, mgr, work_dir):
|
|
mgr.ensure_checkpoint(str(work_dir), "initial")
|
|
# Try to pass a git flag as a commit hash
|
|
result = mgr.restore(str(work_dir), "--patch")
|
|
assert result["success"] is False
|
|
assert "Invalid commit hash" in result["error"]
|
|
assert "must not start with '-'" in result["error"]
|
|
|
|
result = mgr.restore(str(work_dir), "-p")
|
|
assert result["success"] is False
|
|
assert "Invalid commit hash" in result["error"]
|
|
|
|
def test_restore_rejects_invalid_hex_chars(self, mgr, work_dir):
|
|
mgr.ensure_checkpoint(str(work_dir), "initial")
|
|
# Git hashes should not contain characters like ;, &, |
|
|
result = mgr.restore(str(work_dir), "abc; rm -rf /")
|
|
assert result["success"] is False
|
|
assert "expected 4-64 hex characters" in result["error"]
|
|
|
|
result = mgr.diff(str(work_dir), "abc&def")
|
|
assert result["success"] is False
|
|
assert "expected 4-64 hex characters" in result["error"]
|
|
|
|
def test_restore_rejects_path_traversal(self, mgr, work_dir):
|
|
mgr.ensure_checkpoint(str(work_dir), "initial")
|
|
# Real commit hash but malicious path
|
|
checkpoints = mgr.list_checkpoints(str(work_dir))
|
|
target_hash = checkpoints[0]["hash"]
|
|
|
|
# Absolute path outside
|
|
result = mgr.restore(str(work_dir), target_hash, file_path="/etc/passwd")
|
|
assert result["success"] is False
|
|
assert "got absolute path" in result["error"]
|
|
|
|
# Relative traversal outside path
|
|
result = mgr.restore(str(work_dir), target_hash, file_path="../outside_file.txt")
|
|
assert result["success"] is False
|
|
assert "escapes the working directory" in result["error"]
|
|
|
|
def test_restore_accepts_valid_file_path(self, mgr, work_dir):
|
|
mgr.ensure_checkpoint(str(work_dir), "initial")
|
|
checkpoints = mgr.list_checkpoints(str(work_dir))
|
|
target_hash = checkpoints[0]["hash"]
|
|
|
|
# Valid path inside directory
|
|
result = mgr.restore(str(work_dir), target_hash, file_path="main.py")
|
|
assert result["success"] is True
|
|
|
|
# Another valid path with subdirectories
|
|
(work_dir / "subdir").mkdir()
|
|
(work_dir / "subdir" / "test.txt").write_text("hello")
|
|
mgr.new_turn()
|
|
mgr.ensure_checkpoint(str(work_dir), "second")
|
|
checkpoints = mgr.list_checkpoints(str(work_dir))
|
|
target_hash = checkpoints[0]["hash"]
|
|
|
|
result = mgr.restore(str(work_dir), target_hash, file_path="subdir/test.txt")
|
|
assert result["success"] is True
|
|
|
|
|
|
# =========================================================================
|
|
# GPG / global git config isolation
|
|
# =========================================================================
|
|
# Regression tests for the bug where users with ``commit.gpgsign = true``
|
|
# in their global git config got a pinentry popup (or a failed commit)
|
|
# every time the agent took a background snapshot.
|
|
|
|
import os as _os
|
|
|
|
|
|
class TestGpgAndGlobalConfigIsolation:
|
|
def test_git_env_isolates_global_and_system_config(self, tmp_path):
|
|
"""_git_env must null out GIT_CONFIG_GLOBAL / GIT_CONFIG_SYSTEM so the
|
|
shadow repo does not inherit user-level gpgsign, hooks, aliases, etc."""
|
|
env = _git_env(tmp_path / "shadow", str(tmp_path))
|
|
assert env["GIT_CONFIG_GLOBAL"] == _os.devnull
|
|
assert env["GIT_CONFIG_SYSTEM"] == _os.devnull
|
|
assert env["GIT_CONFIG_NOSYSTEM"] == "1"
|
|
|
|
def test_init_sets_commit_gpgsign_false(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))
|
|
# Inspect the shadow's own config directly — the settings must be
|
|
# written into the repo, not just inherited via env vars.
|
|
result = subprocess.run(
|
|
["git", "config", "--file", str(shadow / "config"), "--get", "commit.gpgsign"],
|
|
capture_output=True, text=True,
|
|
)
|
|
assert result.stdout.strip() == "false"
|
|
|
|
def test_init_sets_tag_gpgsign_false(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))
|
|
result = subprocess.run(
|
|
["git", "config", "--file", str(shadow / "config"), "--get", "tag.gpgSign"],
|
|
capture_output=True, text=True,
|
|
)
|
|
assert result.stdout.strip() == "false"
|
|
|
|
def test_checkpoint_works_with_global_gpgsign_and_broken_gpg(
|
|
self, work_dir, checkpoint_base, monkeypatch, tmp_path
|
|
):
|
|
"""The real bug scenario: user has global commit.gpgsign=true but GPG
|
|
is broken or pinentry is unavailable. Before the fix, every snapshot
|
|
either failed or spawned a pinentry window. After the fix, snapshots
|
|
succeed without ever invoking GPG."""
|
|
monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base)
|
|
|
|
# Fake HOME with global gpgsign=true and a deliberately broken GPG
|
|
# binary. If isolation fails, the commit will try to exec this
|
|
# nonexistent path and the checkpoint will fail.
|
|
fake_home = tmp_path / "fake_home"
|
|
fake_home.mkdir()
|
|
(fake_home / ".gitconfig").write_text(
|
|
"[user]\n email = real@user.com\n name = Real User\n"
|
|
"[commit]\n gpgsign = true\n"
|
|
"[tag]\n gpgSign = true\n"
|
|
"[gpg]\n program = /nonexistent/fake-gpg-binary\n"
|
|
)
|
|
monkeypatch.setenv("HOME", str(fake_home))
|
|
monkeypatch.delenv("GPG_TTY", raising=False)
|
|
monkeypatch.delenv("DISPLAY", raising=False) # block GUI pinentry
|
|
|
|
mgr = CheckpointManager(enabled=True)
|
|
assert mgr.ensure_checkpoint(str(work_dir), reason="with-global-gpgsign") is True
|
|
assert len(mgr.list_checkpoints(str(work_dir))) == 1
|
|
|
|
def test_checkpoint_works_on_prefix_shadow_without_local_gpgsign(
|
|
self, work_dir, checkpoint_base, monkeypatch, tmp_path
|
|
):
|
|
"""Users with shadow repos created before the fix will not have
|
|
commit.gpgsign=false in their shadow's own config. The inline
|
|
``--no-gpg-sign`` flag on the commit call must cover them."""
|
|
monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base)
|
|
|
|
# Simulate a pre-fix shadow repo: init without commit.gpgsign=false
|
|
# in its own config. _init_shadow_repo now writes it, so we must
|
|
# manually remove it to mimic the pre-fix state.
|
|
shadow = _shadow_repo_path(str(work_dir))
|
|
_init_shadow_repo(shadow, str(work_dir))
|
|
subprocess.run(
|
|
["git", "config", "--file", str(shadow / "config"),
|
|
"--unset", "commit.gpgsign"],
|
|
capture_output=True, text=True, check=False,
|
|
)
|
|
subprocess.run(
|
|
["git", "config", "--file", str(shadow / "config"),
|
|
"--unset", "tag.gpgSign"],
|
|
capture_output=True, text=True, check=False,
|
|
)
|
|
|
|
# And simulate hostile global config
|
|
fake_home = tmp_path / "fake_home"
|
|
fake_home.mkdir()
|
|
(fake_home / ".gitconfig").write_text(
|
|
"[commit]\n gpgsign = true\n"
|
|
"[gpg]\n program = /nonexistent/fake-gpg-binary\n"
|
|
)
|
|
monkeypatch.setenv("HOME", str(fake_home))
|
|
monkeypatch.delenv("GPG_TTY", raising=False)
|
|
monkeypatch.delenv("DISPLAY", raising=False)
|
|
|
|
mgr = CheckpointManager(enabled=True)
|
|
assert mgr.ensure_checkpoint(str(work_dir), reason="prefix-shadow") is True
|
|
assert len(mgr.list_checkpoints(str(work_dir))) == 1
|
|
|
|
|
|
# =========================================================================
|
|
# Auto-maintenance: prune_checkpoints + maybe_auto_prune_checkpoints
|
|
# =========================================================================
|
|
|
|
class TestPruneCheckpoints:
|
|
"""Sweep orphan/stale shadow repos under CHECKPOINT_BASE (issue #3015 follow-up)."""
|
|
|
|
def _seed_shadow_repo(
|
|
self, base: Path, dir_hash: str, workdir: Path, mtime: float = None
|
|
) -> Path:
|
|
"""Create a minimal shadow repo on disk without invoking real git."""
|
|
import time as _time
|
|
shadow = base / dir_hash
|
|
shadow.mkdir(parents=True)
|
|
(shadow / "HEAD").write_text("ref: refs/heads/main\n")
|
|
(shadow / "HERMES_WORKDIR").write_text(str(workdir) + "\n")
|
|
(shadow / "info").mkdir()
|
|
(shadow / "info" / "exclude").write_text("node_modules/\n")
|
|
if mtime is not None:
|
|
for p in shadow.rglob("*"):
|
|
import os
|
|
os.utime(p, (mtime, mtime))
|
|
import os
|
|
os.utime(shadow, (mtime, mtime))
|
|
return shadow
|
|
|
|
def test_deletes_orphan_when_workdir_missing(self, tmp_path):
|
|
from tools.checkpoint_manager import prune_checkpoints
|
|
|
|
base = tmp_path / "checkpoints"
|
|
alive_work = tmp_path / "alive"
|
|
alive_work.mkdir()
|
|
alive_repo = self._seed_shadow_repo(base, "aaaa" * 4, alive_work)
|
|
orphan_repo = self._seed_shadow_repo(
|
|
base, "bbbb" * 4, tmp_path / "was-deleted"
|
|
)
|
|
|
|
result = prune_checkpoints(retention_days=0, checkpoint_base=base)
|
|
|
|
assert result["scanned"] == 2
|
|
assert result["deleted_orphan"] == 1
|
|
assert result["deleted_stale"] == 0
|
|
assert alive_repo.exists()
|
|
assert not orphan_repo.exists()
|
|
|
|
def test_deletes_stale_by_mtime_when_workdir_alive(self, tmp_path):
|
|
from tools.checkpoint_manager import prune_checkpoints
|
|
import time as _time
|
|
|
|
base = tmp_path / "checkpoints"
|
|
work = tmp_path / "work"
|
|
work.mkdir()
|
|
|
|
fresh_repo = self._seed_shadow_repo(base, "cccc" * 4, work)
|
|
stale_work = tmp_path / "stale_work"
|
|
stale_work.mkdir()
|
|
old = _time.time() - 60 * 86400 # 60 days ago
|
|
stale_repo = self._seed_shadow_repo(base, "dddd" * 4, stale_work, mtime=old)
|
|
|
|
result = prune_checkpoints(
|
|
retention_days=30, delete_orphans=False, checkpoint_base=base
|
|
)
|
|
|
|
assert result["deleted_orphan"] == 0
|
|
assert result["deleted_stale"] == 1
|
|
assert fresh_repo.exists()
|
|
assert not stale_repo.exists()
|
|
|
|
def test_orphan_takes_priority_over_stale(self, tmp_path):
|
|
"""Orphan detection counts first — reason="orphan" even if also stale."""
|
|
from tools.checkpoint_manager import prune_checkpoints
|
|
import time as _time
|
|
|
|
base = tmp_path / "checkpoints"
|
|
old = _time.time() - 60 * 86400
|
|
self._seed_shadow_repo(base, "eeee" * 4, tmp_path / "gone", mtime=old)
|
|
|
|
result = prune_checkpoints(retention_days=30, checkpoint_base=base)
|
|
assert result["deleted_orphan"] == 1
|
|
assert result["deleted_stale"] == 0
|
|
|
|
def test_delete_orphans_disabled_keeps_orphans(self, tmp_path):
|
|
from tools.checkpoint_manager import prune_checkpoints
|
|
|
|
base = tmp_path / "checkpoints"
|
|
orphan = self._seed_shadow_repo(base, "ffff" * 4, tmp_path / "gone")
|
|
|
|
result = prune_checkpoints(
|
|
retention_days=0, delete_orphans=False, checkpoint_base=base
|
|
)
|
|
assert result["deleted_orphan"] == 0
|
|
assert orphan.exists()
|
|
|
|
def test_skips_non_shadow_dirs(self, tmp_path):
|
|
"""Dirs without HEAD (non-initialised) are left alone."""
|
|
from tools.checkpoint_manager import prune_checkpoints
|
|
|
|
base = tmp_path / "checkpoints"
|
|
base.mkdir()
|
|
(base / "garbage-dir").mkdir()
|
|
(base / "garbage-dir" / "random.txt").write_text("hi")
|
|
|
|
result = prune_checkpoints(retention_days=0, checkpoint_base=base)
|
|
assert result["scanned"] == 0
|
|
assert (base / "garbage-dir").exists()
|
|
|
|
def test_tracks_bytes_freed(self, tmp_path):
|
|
from tools.checkpoint_manager import prune_checkpoints
|
|
|
|
base = tmp_path / "checkpoints"
|
|
orphan = self._seed_shadow_repo(base, "1234" * 4, tmp_path / "gone")
|
|
(orphan / "objects").mkdir()
|
|
(orphan / "objects" / "pack.bin").write_bytes(b"x" * 5000)
|
|
|
|
result = prune_checkpoints(retention_days=0, checkpoint_base=base)
|
|
assert result["deleted_orphan"] == 1
|
|
assert result["bytes_freed"] >= 5000
|
|
|
|
def test_base_missing_returns_empty_counts(self, tmp_path):
|
|
from tools.checkpoint_manager import prune_checkpoints
|
|
|
|
result = prune_checkpoints(checkpoint_base=tmp_path / "does-not-exist")
|
|
assert result == {
|
|
"scanned": 0, "deleted_orphan": 0, "deleted_stale": 0,
|
|
"errors": 0, "bytes_freed": 0,
|
|
}
|
|
|
|
|
|
class TestMaybeAutoPruneCheckpoints:
|
|
def _seed(self, base, dir_hash, workdir):
|
|
base.mkdir(parents=True, exist_ok=True)
|
|
shadow = base / dir_hash
|
|
shadow.mkdir()
|
|
(shadow / "HEAD").write_text("ref: refs/heads/main\n")
|
|
(shadow / "HERMES_WORKDIR").write_text(str(workdir) + "\n")
|
|
return shadow
|
|
|
|
def test_first_call_prunes_and_writes_marker(self, tmp_path):
|
|
from tools.checkpoint_manager import maybe_auto_prune_checkpoints
|
|
|
|
base = tmp_path / "checkpoints"
|
|
self._seed(base, "0000" * 4, tmp_path / "gone")
|
|
|
|
out = maybe_auto_prune_checkpoints(checkpoint_base=base)
|
|
assert out["skipped"] is False
|
|
assert out["result"]["deleted_orphan"] == 1
|
|
assert (base / ".last_prune").exists()
|
|
|
|
def test_second_call_within_interval_skips(self, tmp_path):
|
|
from tools.checkpoint_manager import maybe_auto_prune_checkpoints
|
|
|
|
base = tmp_path / "checkpoints"
|
|
self._seed(base, "1111" * 4, tmp_path / "gone")
|
|
|
|
first = maybe_auto_prune_checkpoints(
|
|
checkpoint_base=base, min_interval_hours=24
|
|
)
|
|
assert first["skipped"] is False
|
|
|
|
self._seed(base, "2222" * 4, tmp_path / "also-gone")
|
|
second = maybe_auto_prune_checkpoints(
|
|
checkpoint_base=base, min_interval_hours=24
|
|
)
|
|
assert second["skipped"] is True
|
|
# The second orphan must still exist — skip was honoured.
|
|
assert (base / ("2222" * 4)).exists()
|
|
|
|
def test_corrupt_marker_treated_as_no_prior_run(self, tmp_path):
|
|
from tools.checkpoint_manager import maybe_auto_prune_checkpoints
|
|
|
|
base = tmp_path / "checkpoints"
|
|
base.mkdir()
|
|
(base / ".last_prune").write_text("not-a-timestamp")
|
|
self._seed(base, "3333" * 4, tmp_path / "gone")
|
|
|
|
out = maybe_auto_prune_checkpoints(checkpoint_base=base)
|
|
assert out["skipped"] is False
|
|
assert out["result"]["deleted_orphan"] == 1
|
|
|
|
def test_missing_base_no_raise(self, tmp_path):
|
|
from tools.checkpoint_manager import maybe_auto_prune_checkpoints
|
|
|
|
out = maybe_auto_prune_checkpoints(
|
|
checkpoint_base=tmp_path / "does-not-exist"
|
|
)
|
|
assert out["skipped"] is False
|
|
assert out["result"]["scanned"] == 0
|
|
|