floor pre-migration prune keep to at least 1

hermes_cli/backup.py

@@ -1840,8 +1840,18 @@ def _prune_pre_migration_backups(backup_dir: Path, keep: int) -> int:
 
     Only touches files matching ``pre-migration-*.zip`` so other backups in
     the same directory are never touched.
+
+    ``keep`` is floored to 1 because this helper is only called immediately
+    after a fresh backup is written: ``keep=0`` would delete every
+    ``pre-migration-*.zip`` including the one just created (``backups[0:]``),
+    leaving the migrate path with nothing to restore. Same floor as
+    ``_prune_pre_update_backups``.
     """
-    keep = max(keep, 0)
+    try:
+        keep_n = int(keep)
+    except (TypeError, ValueError):
+        keep_n = _PRE_MIGRATION_DEFAULT_KEEP
+    keep_n = max(keep_n, 1)
     if not backup_dir.exists():
         return 0
 
@@ -1853,7 +1863,7 @@ def _prune_pre_migration_backups(backup_dir: Path, keep: int) -> int:
     )
 
     deleted = 0
-    for p in backups[keep:]:
+    for p in backups[keep_n:]:
         try:
             p.unlink()
             deleted += 1

plugins/platforms/telegram/adapter.py

@@ -2161,7 +2161,9 @@ def _record_polling_progress(self, generation: int) -> None:
             return
         self._polling_progress_event.set()
         self._polling_network_error_count = 0
-        if generation == self._polling_conflict_recovery_generation:
+        # Bare/test adapters may not have run ``__init__``; treat missing as
+        # "no conflict recovery in flight" (same defensive shape as teardown).
+        if generation == getattr(self, "_polling_conflict_recovery_generation", None):
             self._polling_conflict_recovery_generation = None
         else:
             self._polling_conflict_count = 0

tests/hermes_cli/test_pre_migration_prune_keep_floor.py

@@ -0,0 +1,62 @@
+"""keep<=0 must not wipe every pre-migration backup during prune."""
+
+from __future__ import annotations
+
+from hermes_cli.backup import _PRE_MIGRATION_PREFIX, _prune_pre_migration_backups
+
+
+def _seed(backup_dir, names):
+    backup_dir.mkdir(parents=True, exist_ok=True)
+    for name in names:
+        (backup_dir / name).write_bytes(b"zip")
+
+
+def test_prune_pre_migration_keep_zero_preserves_newest(tmp_path):
+    root = tmp_path / "backups"
+    _seed(
+        root,
+        [
+            f"{_PRE_MIGRATION_PREFIX}20260101T000003.zip",
+            f"{_PRE_MIGRATION_PREFIX}20260101T000002.zip",
+            f"{_PRE_MIGRATION_PREFIX}20260101T000001.zip",
+            "hand-made.zip",
+        ],
+    )
+
+    deleted = _prune_pre_migration_backups(root, keep=0)
+
+    remaining = sorted(p.name for p in root.iterdir())
+    assert deleted == 2
+    assert remaining == [
+        "hand-made.zip",
+        f"{_PRE_MIGRATION_PREFIX}20260101T000003.zip",
+    ]
+
+
+def test_prune_pre_migration_negative_keep_floors_to_one(tmp_path):
+    root = tmp_path / "backups"
+    _seed(
+        root,
+        [
+            f"{_PRE_MIGRATION_PREFIX}20260101T000002.zip",
+            f"{_PRE_MIGRATION_PREFIX}20260101T000001.zip",
+        ],
+    )
+
+    deleted = _prune_pre_migration_backups(root, keep=-3)
+
+    remaining = sorted(p.name for p in root.iterdir())
+    assert deleted == 1
+    assert remaining == [f"{_PRE_MIGRATION_PREFIX}20260101T000002.zip"]
+
+
+def test_prune_pre_migration_invalid_keep_uses_default(tmp_path):
+    root = tmp_path / "backups"
+    names = [f"{_PRE_MIGRATION_PREFIX}20260101T{i:06d}.zip" for i in range(8)]
+    _seed(root, names)
+
+    deleted = _prune_pre_migration_backups(root, keep="nope")
+
+    remaining = [p for p in root.iterdir() if p.name.startswith(_PRE_MIGRATION_PREFIX)]
+    assert len(remaining) == 5  # _PRE_MIGRATION_DEFAULT_KEEP
+    assert deleted == 3