Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -1927,10 +1927,15 @@ def _run_checkpoint_auto_maintenance() -> None:
if not cfg.get("auto_prune", False):
return
from tools.checkpoint_manager import maybe_auto_prune_checkpoints
# delete_orphans is intentionally never honoured here: a missing
# workdir at startup is ambiguous (deleted project vs. an unmounted
# external volume / network share / VPN not yet up) and this sweep
# runs unattended. Orphan cleanup is only ever done via the explicit
# `hermes checkpoints prune` command, which the user has to invoke.
maybe_auto_prune_checkpoints(
retention_days=int(cfg.get("retention_days", 7)),
min_interval_hours=int(cfg.get("min_interval_hours", 24)),
delete_orphans=bool(cfg.get("delete_orphans", True)),
delete_orphans=False,
max_total_size_mb=int(cfg.get("max_total_size_mb", 500)),
)
except Exception as exc:
Expand Down
14 changes: 10 additions & 4 deletions gateway/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -3415,18 +3415,24 @@ def __init__(self, config: Optional[GatewayConfig] = None):
except Exception as exc:
logger.debug("state.db auto-maintenance skipped: %s", exc)

# Opportunistic shadow-repo cleanup — deletes orphan/stale
# checkpoint repos under ~/.hermes/checkpoints/. Opt-in via
# checkpoints.auto_prune, idempotent via .last_prune marker.
# Opportunistic shadow-repo cleanup — deletes stale checkpoint repos
# under ~/.hermes/checkpoints/. Opt-in via checkpoints.auto_prune,
# idempotent via .last_prune marker.
try:
from hermes_cli.config import load_config as _load_full_config
_ckpt_cfg = (_load_full_config().get("checkpoints") or {})
if _ckpt_cfg.get("auto_prune", False):
from tools.checkpoint_manager import maybe_auto_prune_checkpoints
# delete_orphans is intentionally never honoured here: a
# missing workdir at startup is ambiguous (deleted project
# vs. an unmounted external volume / network share / VPN
# not yet up) and this sweep runs unattended. Orphan cleanup
# is only ever done via the explicit `hermes checkpoints
# prune` command, which the user has to invoke.
maybe_auto_prune_checkpoints(
retention_days=int(_ckpt_cfg.get("retention_days", 7)),
min_interval_hours=int(_ckpt_cfg.get("min_interval_hours", 24)),
delete_orphans=bool(_ckpt_cfg.get("delete_orphans", True)),
delete_orphans=False,
max_total_size_mb=int(_ckpt_cfg.get("max_total_size_mb", 500)),
)
except Exception as exc:
Expand Down
37 changes: 34 additions & 3 deletions hermes_cli/checkpoints.py
Original file line number Diff line number Diff line change
Expand Up @@ -109,20 +109,49 @@ def cmd_list(args: argparse.Namespace) -> int:


def cmd_prune(args: argparse.Namespace) -> int:
from tools.checkpoint_manager import prune_checkpoints
from tools.checkpoint_manager import prune_checkpoints, store_status

retention_days = args.retention_days
max_size_mb = args.max_size_mb
delete_orphans = not args.keep_orphans

if delete_orphans and not args.force:
info = store_status()
orphans = [
p for p in info.get("projects", [])
if not p.get("exists")
]
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.")
print("Pass --keep-orphans to prune stale entries only.")
if not _confirm("Delete these orphan projects?"):
print("Aborted.")
return 1

print("Pruning checkpoint store…")
print(f" retention_days: {retention_days}")
print(f" delete_orphans: {not args.keep_orphans}")
print(f" delete_orphans: {delete_orphans}")
print(f" max_total_size_mb: {max_size_mb}")
print()

result = prune_checkpoints(
retention_days=retention_days,
delete_orphans=not args.keep_orphans,
delete_orphans=delete_orphans,
max_total_size_mb=max_size_mb,
)
print(f"Scanned: {result['scanned']}")
Expand Down Expand Up @@ -225,6 +254,8 @@ def register_cli(parser: argparse.ArgumentParser) -> None:
"per project until total size <= this (default 500)")
p_prune.add_argument("--keep-orphans", action="store_true",
help="Skip deleting projects whose workdir no longer exists")
p_prune.add_argument("-f", "--force", action="store_true",
help="Skip the orphan-deletion confirmation prompt")
p_prune.set_defaults(func=cmd_prune)

p_clear = subs.add_parser(
Expand Down
11 changes: 9 additions & 2 deletions hermes_cli/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -1369,15 +1369,22 @@ def _ensure_hermes_home_managed(home: Path):
"max_file_size_mb": 10,
# Auto-maintenance: hermes sweeps the checkpoint base at startup
# (at most once per ``min_interval_hours``) and:
# * deletes project entries whose workdir no longer exists (orphan)
# * deletes project entries whose last_touch is older than
# ``retention_days``
# * GCs the single shared store to reclaim unreachable objects
# * enforces ``max_total_size_mb`` across remaining projects
# * deletes ``legacy-*`` archives older than ``retention_days``
#
# NOTE: this automatic sweep never deletes "orphan" entries (workdir
# no longer found on disk). A missing workdir at startup is
# ambiguous — it can mean the project was deleted, or that an
# external volume / network share / VPN is simply not mounted yet —
# and this sweep runs unattended, so it must never guess. Orphan
# cleanup is only available via the explicit
# ``hermes checkpoints prune`` command (add ``--keep-orphans`` to
# skip it), where a human is looking at the output.
"auto_prune": True,
"retention_days": 7,
"delete_orphans": True,
"min_interval_hours": 24,
},

Expand Down
208 changes: 208 additions & 0 deletions tests/hermes_cli/test_checkpoints_prune.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,208 @@
"""Tests for `hermes checkpoints prune`'s orphan confirmation flow.

Covers the P1 raised on PR #69141: the confirmation preview must cover
BOTH v2 projects (`store_status()["projects"]`) and pre-v2 shadow repos
(`store_status()["pre_v2_projects"]`), since `prune_checkpoints()` deletes
orphans from both layouts. Exercises decline / accept / --force across
pre-v2-only and mixed (v2 + pre-v2) stores.
"""

from __future__ import annotations

import argparse

import pytest


def _ns(**kwargs) -> argparse.Namespace:
defaults = {"retention_days": 7, "max_size_mb": 500, "keep_orphans": False, "force": False}
defaults.update(kwargs)
return argparse.Namespace(**defaults)


def _prune_result(**kwargs) -> dict:
result = {"scanned": 0, "deleted_orphan": 0, "deleted_stale": 0, "errors": 0, "bytes_freed": 0}
result.update(kwargs)
return result


_V2_ORPHAN_ONLY_STATUS = {
"projects": [],
"pre_v2_projects": [],
}

_PRE_V2_ONLY_STATUS = {
"projects": [],
"pre_v2_projects": [
{"path": "/home/user/.hermes/checkpoints/deadbeefcafebabe", "workdir": None, "exists": False},
],
}

_MIXED_STATUS = {
"projects": [
{"hash": "abc123", "workdir": "/gone/v2-project", "exists": False, "commits": 4},
],
"pre_v2_projects": [
{"path": "/home/user/.hermes/checkpoints/deadbeefcafebabe", "workdir": "/gone/pre-v2-project", "exists": False},
],
}


def _patch_checkpoint_manager(monkeypatch, status: dict, prune_calls: list):
import tools.checkpoint_manager as ckpt_mgr

monkeypatch.setattr(ckpt_mgr, "store_status", lambda *a, **k: status)

def _fake_prune(**kwargs):
prune_calls.append(kwargs)
return _prune_result(
deleted_orphan=len(status["projects"]) + len(status["pre_v2_projects"]),
)

monkeypatch.setattr(ckpt_mgr, "prune_checkpoints", _fake_prune)


# ─── pre-v2-only store ──────────────────────────────────────────────────────


def test_pre_v2_only_decline_aborts_without_deleting(monkeypatch, capsys):
import hermes_cli.checkpoints as checkpoints_cli

prune_calls: list = []
_patch_checkpoint_manager(monkeypatch, _PRE_V2_ONLY_STATUS, prune_calls)
monkeypatch.setattr("builtins.input", lambda _prompt: "n")

rc = checkpoints_cli.cmd_prune(_ns())

assert rc == 1
assert prune_calls == []
out = capsys.readouterr().out
assert "pre-v2 shadow repo" in out
assert "Aborted" in out


def test_pre_v2_only_accept_deletes(monkeypatch, capsys):
import hermes_cli.checkpoints as checkpoints_cli

prune_calls: list = []
_patch_checkpoint_manager(monkeypatch, _PRE_V2_ONLY_STATUS, prune_calls)
monkeypatch.setattr("builtins.input", lambda _prompt: "y")

rc = checkpoints_cli.cmd_prune(_ns())

assert rc == 0
assert len(prune_calls) == 1
assert prune_calls[0]["delete_orphans"] is True


def test_pre_v2_only_force_skips_prompt(monkeypatch, capsys):
import hermes_cli.checkpoints as checkpoints_cli

prune_calls: list = []
_patch_checkpoint_manager(monkeypatch, _PRE_V2_ONLY_STATUS, prune_calls)

def _unexpected_input(_prompt):
raise AssertionError("input() must not be called when --force is passed")

monkeypatch.setattr("builtins.input", _unexpected_input)

rc = checkpoints_cli.cmd_prune(_ns(force=True))

assert rc == 0
assert len(prune_calls) == 1


# ─── mixed store (v2 + pre-v2) ──────────────────────────────────────────────


def test_mixed_store_decline_aborts_without_deleting(monkeypatch, capsys):
import hermes_cli.checkpoints as checkpoints_cli

prune_calls: list = []
_patch_checkpoint_manager(monkeypatch, _MIXED_STATUS, prune_calls)
monkeypatch.setattr("builtins.input", lambda _prompt: "n")

rc = checkpoints_cli.cmd_prune(_ns())

assert rc == 1
assert prune_calls == []
out = capsys.readouterr().out
# Both layouts must appear in the preview, not just the v2 one.
assert "/gone/v2-project" in out
assert "/gone/pre-v2-project" in out
assert "This will permanently delete 2 orphan checkpoint project(s)" in out


def test_mixed_store_accept_deletes_both_layouts(monkeypatch, capsys):
import hermes_cli.checkpoints as checkpoints_cli

prune_calls: list = []
_patch_checkpoint_manager(monkeypatch, _MIXED_STATUS, prune_calls)
monkeypatch.setattr("builtins.input", lambda _prompt: "y")

rc = checkpoints_cli.cmd_prune(_ns())

assert rc == 0
assert len(prune_calls) == 1
out = capsys.readouterr().out
assert "Deleted orphan: 2" in out


def test_mixed_store_force_skips_prompt_deletes_both(monkeypatch, capsys):
import hermes_cli.checkpoints as checkpoints_cli

prune_calls: list = []
_patch_checkpoint_manager(monkeypatch, _MIXED_STATUS, prune_calls)

def _unexpected_input(_prompt):
raise AssertionError("input() must not be called when --force is passed")

monkeypatch.setattr("builtins.input", _unexpected_input)

rc = checkpoints_cli.cmd_prune(_ns(force=True))

assert rc == 0
assert len(prune_calls) == 1
assert prune_calls[0]["delete_orphans"] is True


# ─── --keep-orphans skips the prompt entirely, on either layout ───────────


@pytest.mark.parametrize("status", [_PRE_V2_ONLY_STATUS, _MIXED_STATUS], ids=["pre_v2_only", "mixed"])
def test_keep_orphans_skips_prompt(monkeypatch, capsys, status):
import hermes_cli.checkpoints as checkpoints_cli

prune_calls: list = []
_patch_checkpoint_manager(monkeypatch, status, prune_calls)

def _unexpected_input(_prompt):
raise AssertionError("input() must not be called when --keep-orphans is passed")

monkeypatch.setattr("builtins.input", _unexpected_input)

rc = checkpoints_cli.cmd_prune(_ns(keep_orphans=True))

assert rc == 0
assert len(prune_calls) == 1
assert prune_calls[0]["delete_orphans"] is False


# ─── no orphans present: never prompts even without --force ───────────────


def test_no_orphans_skips_prompt(monkeypatch, capsys):
import hermes_cli.checkpoints as checkpoints_cli

prune_calls: list = []
_patch_checkpoint_manager(monkeypatch, _V2_ORPHAN_ONLY_STATUS, prune_calls)

def _unexpected_input(_prompt):
raise AssertionError("input() must not be called when there are no orphans")

monkeypatch.setattr("builtins.input", _unexpected_input)

rc = checkpoints_cli.cmd_prune(_ns())

assert rc == 0
assert len(prune_calls) == 1
Loading