mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-29 18:46:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

fix(checkpoints): include pre-v2 shadow repos in orphan preview

Browse source 
store_status()["projects"] only ever covered v2 metadata, so the
`hermes checkpoints prune` confirmation prompt was blind to pre-v2
base/<hash>/HEAD shadow repos that prune_checkpoints() deletes
separately via shutil.rmtree — a pre-v2-only or mixed store could
lose checkpoint history without ever hitting the confirmation.

Extract the pre-v2 scan into _pre_v2_shadow_repos() and have both
store_status() (preview, new pre_v2_projects key) and
prune_checkpoints() (deletion) read from it, so the CLI prompt can
no longer diverge from what actually gets removed.

Addresses review from egilewski on #69141.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
joaomarcos 2026-07-22 15:54:20 -03:00 • committed by Teknium
parent cd3f653b4e
commit dd8c0c7be8
2 changed files with 71 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

16
hermes_cli/checkpoints.py
View file

@ -116,17 +116,25 @@ def cmd_prune(args: argparse.Namespace) -> int:
delete_orphans = not args.keep_orphans
if delete_orphans and not args.force:
info = store_status()
orphans = [
p for p in store_status().get("projects", [])
p for p in info.get("projects", [])
if not p.get("exists")
]
if orphans:
print(f"This will permanently delete {len(orphans)} orphan checkpoint "
"project(s) whose workdir is not currently reachable:")
pre_v2_orphans = [
p for p in info.get("pre_v2_projects", [])
if not p.get("exists")
]
if orphans or pre_v2_orphans:
print(f"This will permanently delete {len(orphans) + len(pre_v2_orphans)} "
"orphan checkpoint project(s) whose workdir is not currently reachable:")
print()
for p in orphans:
wd = p.get("workdir") or "(unknown)"
print(f" {wd} ({p.get('commits', 0)} commit(s))")
for p in pre_v2_orphans:
wd = p.get("workdir") or "(unknown)"
print(f" {wd} (pre-v2 shadow repo)")
print()
print("A workdir can be unreachable because the project was deleted,")
print("or because an external volume / network share / VPN is down.")

74
tools/checkpoint_manager.py
View file

@ -545,6 +545,39 @@ def _list_projects(store: Path) -> List[Dict]:
return out
def _pre_v2_shadow_repos(base: Path) -> List[Dict]:
"""Return pre-v2 per-project shadow repos still directly under ``base``.
Pre-v2 layout kept one shadow git repo per working directory directly
under ``CHECKPOINT_BASE`` (identified by a ``HEAD`` file). This is the
single source of truth for that scan so a preview built from it (e.g.
``store_status``) always matches what ``prune_checkpoints`` deletes.
"""
out: List[Dict] = []
if not base.exists():
return out
for child in base.iterdir():
if not child.is_dir():
continue
if child.name == _STORE_DIRNAME or child.name.startswith(_LEGACY_PREFIX):
continue
if not (child / "HEAD").exists():
continue
workdir: Optional[str] = None
wd_marker = child / "HERMES_WORKDIR"
if wd_marker.exists():
try:
workdir = wd_marker.read_text(encoding="utf-8").strip()
except (OSError, UnicodeDecodeError):
workdir = None
out.append({
"path": child,
"workdir": workdir,
"exists": bool(workdir) and Path(workdir).exists(),
})
return out
def _dir_file_count(path: str) -> int:
"""Quick file count estimate (stops early if over _MAX_FILES)."""
count = 0
@ -1325,22 +1358,16 @@ def prune_checkpoints(
except OSError as exc:
result["errors"] += 1
logger.warning("Failed to delete legacy archive %s: %s", child, exc)
continue
# Only count as a pre-v2 shadow repo if it has a HEAD.
if not (child / "HEAD").exists():
continue
# Pre-v2 per-project shadow repos. Scanned via the same helper
# `store_status()` uses for its orphan preview, so a confirmation prompt
# built from that preview always matches what gets deleted here.
for repo in _pre_v2_shadow_repos(base):
child = repo["path"]
result["scanned"] += 1
reason: Optional[str] = None
if delete_orphans:
workdir: Optional[str] = None
wd_marker = child / "HERMES_WORKDIR"
if wd_marker.exists():
try:
workdir = wd_marker.read_text(encoding="utf-8").strip()
except (OSError, UnicodeDecodeError):
workdir = None
if workdir is None or not Path(workdir).exists():
reason = "orphan"
if delete_orphans and not repo["exists"]:
reason = "orphan"
if reason is None and retention_days > 0:
newest = 0.0
try:
@ -1572,7 +1599,14 @@ def store_status(checkpoint_base: Optional[Path] = None) -> Dict:
``{"base": path, "store_size_bytes": N, "legacy_size_bytes": N,
"total_size_bytes": N, "project_count": N, "projects": [...],
"legacy_archives": [...]}``
"pre_v2_projects": [...], "legacy_archives": [...]}``
``pre_v2_projects`` covers shadow repos still on the pre-v2 per-project
layout (``base/<hash>/HEAD``) — distinct from ``legacy_archives``, which
are already-migrated ``legacy-<ts>/`` dirs. Callers that preview an
orphan-deletion sweep must include both ``projects`` and
``pre_v2_projects``, since ``prune_checkpoints`` deletes orphans from
both layouts.
"""
base = checkpoint_base or CHECKPOINT_BASE
out: Dict = {
@ -1582,6 +1616,7 @@ def store_status(checkpoint_base: Optional[Path] = None) -> Dict:
"total_size_bytes": 0,
"project_count": 0,
"projects": [],
"pre_v2_projects": [],
"legacy_archives": [],
}
if not base.exists():
@ -1613,6 +1648,15 @@ def store_status(checkpoint_base: Optional[Path] = None) -> Dict:
})
out["project_count"] = len(out["projects"])
out["pre_v2_projects"] = [
{
"path": str(r["path"]),
"workdir": r["workdir"],
"exists": r["exists"],
}
for r in _pre_v2_shadow_repos(base)
]
for child in base.iterdir():
if child.is_dir() and child.name.startswith(_LEGACY_PREFIX):
try: