fix(checkpoints): bind orphan confirmation to previewed identities

Address P1 from PR review: cmd_prune()'s y/N preview reads
store_status() but the confirmed deletion re-scans both the v2 and
pre-v2 layouts from scratch. A workdir that goes missing while the
human is answering the prompt gets swept in as if it had been shown
and approved.

prune_checkpoints() now accepts orphan_allowlist — a set of v2 project
hashes and/or pre-v2 shadow repo paths. When set, only orphans whose
identity is in the set are deleted; anything newly orphaned since the
scan survives the run. cmd_prune() builds this set from the exact
projects it just displayed and passed confirmation for. --force still
passes None (no preview shown, so nothing to bind to).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
joaomarcos 2026-07-23 13:47:59 -03:00 committed by Teknium
parent 3ad8552f78
commit a373fab1ce
3 changed files with 143 additions and 3 deletions

View file

@ -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']}")

View file

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

View file

@ -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)