diff --git a/hermes_cli/checkpoints.py b/hermes_cli/checkpoints.py index a5cdbe6daed3..4c0c448478ec 100644 --- a/hermes_cli/checkpoints.py +++ b/hermes_cli/checkpoints.py @@ -25,7 +25,7 @@ import argparse import time from datetime import datetime from pathlib import Path -from typing import Any +from typing import Any, Optional def _fmt_bytes(n: int) -> str: @@ -115,6 +115,12 @@ def cmd_prune(args: argparse.Namespace) -> int: max_size_mb = args.max_size_mb delete_orphans = not args.keep_orphans + # When set, restricts orphan deletion to exactly the identities shown in + # the confirmation preview below (v2 project hashes / pre-v2 shadow repo + # paths). `None` means "no restriction" — used for --force, where there + # is no preview to bind to. + orphan_allowlist: Optional[set] = None + if delete_orphans and not args.force: info = store_status() orphans = [ @@ -142,6 +148,12 @@ def cmd_prune(args: argparse.Namespace) -> int: if not _confirm("Delete these orphan projects?"): print("Aborted.") return 1 + # Bind the deletion to exactly what was just displayed and + # confirmed — a project that becomes orphaned only *after* this + # preview (e.g. its workdir disappears while waiting on input()) + # must not be swept up under this same confirmation. + orphan_allowlist = {p["hash"] for p in orphans} + orphan_allowlist.update(p["path"] for p in pre_v2_orphans) print("Pruning checkpoint store…") print(f" retention_days: {retention_days}") @@ -153,6 +165,7 @@ def cmd_prune(args: argparse.Namespace) -> int: retention_days=retention_days, delete_orphans=delete_orphans, max_total_size_mb=max_size_mb, + orphan_allowlist=orphan_allowlist, ) print(f"Scanned: {result['scanned']}") print(f"Deleted orphan: {result['deleted_orphan']}") diff --git a/tests/tools/test_checkpoint_manager.py b/tests/tools/test_checkpoint_manager.py index 5c6db10c0119..903f28908c88 100644 --- a/tests/tools/test_checkpoint_manager.py +++ b/tests/tools/test_checkpoint_manager.py @@ -1,5 +1,6 @@ """Tests for tools/checkpoint_manager.py — CheckpointManager (v2 single-store).""" +import argparse import json import logging import os @@ -934,6 +935,112 @@ class TestPruneCheckpointsV2: assert not old_legacy.exists() +class TestPruneCheckpointsOrphanAllowlist: + """P1 fix on PR #69141: the confirmation preview must bind to exactly + what gets deleted. A project that only becomes orphaned *after* the + preview was built (e.g. its workdir vanishes while a human is answering + the y/N prompt) must survive a rescan-based prune unless its identity + was in the previewed/approved set. + """ + + def test_v2_allowlist_restricts_deletion_to_approved_hash(self, tmp_path, monkeypatch): + base = tmp_path / "checkpoints" + monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", base) + + previewed = tmp_path / "previewed-gone" + previewed.mkdir() + (previewed / "f").write_text("a") + newly_gone = tmp_path / "newly-gone" + newly_gone.mkdir() + (newly_gone / "f").write_text("b") + + m = CheckpointManager(enabled=True) + assert m.ensure_checkpoint(str(previewed), "previewed") is True + m.new_turn() + assert m.ensure_checkpoint(str(newly_gone), "newly-gone") is True + + import shutil as _shutil + _shutil.rmtree(previewed) + _shutil.rmtree(newly_gone) + + previewed_hash = _project_hash(str(previewed)) + newly_gone_hash = _project_hash(str(newly_gone)) + + result = prune_checkpoints( + retention_days=0, checkpoint_base=base, + orphan_allowlist={previewed_hash}, + ) + + assert result["deleted_orphan"] == 1 + assert not (base / "store" / "projects" / f"{previewed_hash}.json").exists() + # Not in the allowlist -> survives this prune even though it is orphaned. + assert (base / "store" / "projects" / f"{newly_gone_hash}.json").exists() + + def test_pre_v2_allowlist_restricts_deletion_to_approved_path(self, tmp_path): + base = tmp_path / "checkpoints" + previewed_repo = _seed_legacy_repo(base, "aaaa" * 4, tmp_path / "previewed-gone") + newly_gone_repo = _seed_legacy_repo(base, "bbbb" * 4, tmp_path / "newly-gone") + + result = prune_checkpoints( + retention_days=0, checkpoint_base=base, + orphan_allowlist={str(previewed_repo)}, + ) + + assert result["deleted_orphan"] == 1 + assert not previewed_repo.exists() + assert newly_gone_repo.exists() + + def test_allowlist_none_deletes_all_current_orphans(self, tmp_path, monkeypatch): + """Default (no allowlist) keeps prior behaviour, e.g. for --force.""" + base = tmp_path / "checkpoints" + orphan_a = _seed_legacy_repo(base, "cccc" * 4, tmp_path / "gone-a") + orphan_b = _seed_legacy_repo(base, "dddd" * 4, tmp_path / "gone-b") + + result = prune_checkpoints(retention_days=0, checkpoint_base=base) + + assert result["deleted_orphan"] == 2 + assert not orphan_a.exists() + assert not orphan_b.exists() + + def test_end_to_end_timing_change_during_confirmation_prompt(self, tmp_path, monkeypatch): + """Reproduces the exact PR #69141 review scenario end-to-end through + `hermes checkpoints prune`: the preview shows one pre-v2 orphan; a + second project's workdir is removed by the input() callback while + the human is "answering" the prompt. Only the previewed orphan may + be deleted. + """ + import hermes_cli.checkpoints as checkpoints_cli + + base = tmp_path / "checkpoints" + monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", base) + + previewed_work = tmp_path / "was-deleted-before-preview" + still_alive_work = tmp_path / "still-alive-during-preview" + still_alive_work.mkdir() + _seed_legacy_repo(base, "eeee" * 4, previewed_work) + second_repo = _seed_legacy_repo(base, "ffff" * 4, still_alive_work) + + def _confirm_and_go_stale(_prompt): + # Simulate the workdir disappearing after the preview was shown + # but before the human's answer is processed. + import shutil as _shutil + _shutil.rmtree(still_alive_work) + return "y" + + monkeypatch.setattr("builtins.input", _confirm_and_go_stale) + + args = argparse.Namespace( + retention_days=0, max_size_mb=0, keep_orphans=False, force=False, + ) + rc = checkpoints_cli.cmd_prune(args) + + assert rc == 0 + # The orphan shown in the preview is gone. + assert not (base / ("eeee" * 4)).exists() + # The one that only went orphan mid-confirmation must survive. + assert second_repo.exists() + + class TestMaybeAutoPruneCheckpoints: def test_first_call_prunes_and_writes_marker(self, tmp_path): base = tmp_path / "checkpoints" diff --git a/tools/checkpoint_manager.py b/tools/checkpoint_manager.py index 6ba7c9ce439f..b4791fca6e63 100644 --- a/tools/checkpoint_manager.py +++ b/tools/checkpoint_manager.py @@ -1293,6 +1293,7 @@ def prune_checkpoints( delete_orphans: bool = True, checkpoint_base: Optional[Path] = None, max_total_size_mb: int = 0, + orphan_allowlist: Optional[set] = None, ) -> Dict[str, int]: """Delete stale/orphan checkpoints and reclaim store space. @@ -1302,6 +1303,17 @@ def prune_checkpoints( (the original project was deleted / moved); OR * its ``last_touch`` is older than ``retention_days`` days. + ``orphan_allowlist``, when not ``None``, restricts orphan deletion to + the given identities (v2 project ``_hash`` strings and/or pre-v2 shadow + repo paths as ``str``). This lets a caller that showed the user a + confirmation preview (built from ``store_status()``) bind the resulting + deletion to exactly what was displayed — a project that only becomes + orphaned *after* the preview (e.g. its workdir vanishes while the human + is answering the prompt) is skipped rather than swept up under the + earlier confirmation. Pass ``None`` (the default) to delete every + currently-orphaned project, e.g. for ``--force`` or unattended callers + that never show a preview. + Additionally, if ``max_total_size_mb > 0`` and the store exceeds that after orphan/stale pruning, the oldest commit per remaining project is dropped until the store is under the cap. @@ -1366,7 +1378,11 @@ def prune_checkpoints( child = repo["path"] result["scanned"] += 1 reason: Optional[str] = None - if delete_orphans and not repo["exists"]: + if ( + delete_orphans + and not repo["exists"] + and (orphan_allowlist is None or str(child) in orphan_allowlist) + ): reason = "orphan" if reason is None and retention_days > 0: newest = 0.0 @@ -1405,7 +1421,11 @@ def prune_checkpoints( continue result["scanned"] += 1 reason = None - if delete_orphans and (not workdir or not Path(workdir).exists()): + if ( + delete_orphans + and (not workdir or not Path(workdir).exists()) + and (orphan_allowlist is None or dir_hash in orphan_allowlist) + ): reason = "orphan" elif retention_days > 0: last_touch = float(meta.get("last_touch", 0) or 0)