From dd8c0c7be89d4b2e899cae91b5ea6d81b6d52385 Mon Sep 17 00:00:00 2001 From: joaomarcos Date: Wed, 22 Jul 2026 15:54:20 -0300 Subject: [PATCH] fix(checkpoints): include pre-v2 shadow repos in orphan preview MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit store_status()["projects"] only ever covered v2 metadata, so the `hermes checkpoints prune` confirmation prompt was blind to pre-v2 base//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 --- hermes_cli/checkpoints.py | 16 ++++++-- tools/checkpoint_manager.py | 74 +++++++++++++++++++++++++++++-------- 2 files changed, 71 insertions(+), 19 deletions(-) diff --git a/hermes_cli/checkpoints.py b/hermes_cli/checkpoints.py index 758af6cb22bf..a5cdbe6daed3 100644 --- a/hermes_cli/checkpoints.py +++ b/hermes_cli/checkpoints.py @@ -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.") diff --git a/tools/checkpoint_manager.py b/tools/checkpoint_manager.py index cf5c1d726e17..6ba7c9ce439f 100644 --- a/tools/checkpoint_manager.py +++ b/tools/checkpoint_manager.py @@ -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//HEAD``) — distinct from ``legacy_archives``, which + are already-migrated ``legacy-/`` 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: