hermes-agent/tests/tools/test_checkpoint_manager.py

"""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