From 065bc9184687ad41dcd54ca5060b20bdadf67b8e Mon Sep 17 00:00:00 2001 From: KoNit-K <124019182+KoNit-K@users.noreply.github.com> Date: Tue, 15 Sep 2026 17:36:32 +0800 Subject: [PATCH] fix(checkpoints): report legacy archive deletion failures --- hermes_cli/checkpoints.py | 3 ++ tests/hermes_cli/test_checkpoints_prune.py | 22 ++++++++++- tests/tools/test_checkpoint_manager.py | 44 ++++++++++++++++++++++ tools/checkpoint_manager.py | 4 +- 4 files changed, 70 insertions(+), 3 deletions(-) diff --git a/hermes_cli/checkpoints.py b/hermes_cli/checkpoints.py index 48decf0a86..3c86daafff 100644 --- a/hermes_cli/checkpoints.py +++ b/hermes_cli/checkpoints.py @@ -180,6 +180,9 @@ def cmd_clear_legacy(args: argparse.Namespace) -> int: result = clear_legacy() print(f"Deleted {result['deleted']} archive(s), reclaimed {_fmt_bytes(result['bytes_freed'])}.") + if result["errors"]: + print(f"Failed to delete {result['errors']} archive(s). See logs for details.") + return 2 return 0 diff --git a/tests/hermes_cli/test_checkpoints_prune.py b/tests/hermes_cli/test_checkpoints_prune.py index 909c1746a2..6c4b61acaf 100644 --- a/tests/hermes_cli/test_checkpoints_prune.py +++ b/tests/hermes_cli/test_checkpoints_prune.py @@ -26,6 +26,27 @@ def _prune_result(**kwargs) -> dict: return result +def test_clear_legacy_returns_error_when_an_archive_cannot_be_deleted(monkeypatch, capsys): + import hermes_cli.checkpoints as checkpoints_cli + import tools.checkpoint_manager as ckpt_mgr + + monkeypatch.setattr( + ckpt_mgr, + "store_status", + lambda: {"legacy_archives": [{"name": "legacy-read-only", "size_bytes": 1}]}, + ) + monkeypatch.setattr( + ckpt_mgr, + "clear_legacy", + lambda: {"deleted": 0, "errors": 1, "bytes_freed": 0}, + ) + + rc = checkpoints_cli.cmd_clear_legacy(_ns(force=True)) + + assert rc == 2 + assert "Failed to delete 1 archive(s)." in capsys.readouterr().out + + _V2_ORPHAN_ONLY_STATUS = { "projects": [], "pre_v2_projects": [], @@ -122,4 +143,3 @@ def test_empty_preview_binds_empty_allowlist(monkeypatch, capsys): - diff --git a/tests/tools/test_checkpoint_manager.py b/tests/tools/test_checkpoint_manager.py index f36f97e112..7756597b48 100644 --- a/tests/tools/test_checkpoint_manager.py +++ b/tests/tools/test_checkpoint_manager.py @@ -1187,6 +1187,50 @@ class TestClearFunctions: # Store preserved assert (base / "store" / "HEAD").exists() + def test_clear_legacy_reports_partial_deletion_failures(self, tmp_path, monkeypatch): + base = tmp_path / "checkpoints" + failed = base / "legacy-failed" + deleted = base / "legacy-deleted" + for archive in (failed, deleted): + archive.mkdir(parents=True) + (archive / "data").write_bytes(b"x") + + real_rmtree = shutil.rmtree + + def fail_one(path, *args, **kwargs): + if Path(path) == failed: + raise OSError("read-only file") + return real_rmtree(path, *args, **kwargs) + + monkeypatch.setattr("tools.checkpoint_manager.shutil.rmtree", fail_one) + + result = clear_legacy(base) + + assert result["deleted"] == 1 + assert result["errors"] == 1 + assert not deleted.exists() + assert failed.exists() + + def test_clear_legacy_reports_all_deletion_failures(self, tmp_path, monkeypatch): + base = tmp_path / "checkpoints" + archives = [base / "legacy-first", base / "legacy-second"] + for archive in archives: + archive.mkdir(parents=True) + + def always_fail(*args, **kwargs): + raise OSError("read-only file") + + monkeypatch.setattr( + "tools.checkpoint_manager.shutil.rmtree", + always_fail, + ) + + result = clear_legacy(base) + + assert result["deleted"] == 0 + assert result["errors"] == 2 + assert all(archive.exists() for archive in archives) + # ========================================================================= # Orphan pruning must not act on an unreachable volume diff --git a/tools/checkpoint_manager.py b/tools/checkpoint_manager.py index 5e192c018c..790e5124b8 100644 --- a/tools/checkpoint_manager.py +++ b/tools/checkpoint_manager.py @@ -1254,9 +1254,9 @@ def clear_all(checkpoint_base: Optional[Path] = None) -> Dict[str, int]: def clear_legacy(checkpoint_base: Optional[Path] = None) -> Dict[str, int]: - """Delete all ``legacy-*`` archive directories. Returns ``{"bytes_freed": N, "deleted": count}``.""" + """Delete all ``legacy-*`` archive directories and report any failures.""" base = checkpoint_base or _resolve_checkpoint_base() - out = {"bytes_freed": 0, "deleted": 0} + out = {"bytes_freed": 0, "deleted": 0, "errors": 0} if not base.exists(): return out for child in _legacy_archives(base):