fix(kanban): cap corrupt-backup retention at 10 files per board DB

Content-addressed quarantine backups dedupe identical corrupt bytes,
but corruption that keeps mutating between failures (partial repairs,
further damage across dispatcher retries, multi-profile fleets) mints a
new sha-named backup every round — a user accumulated 124
.corrupt.*.bak files with no bound.

After each NEW backup is created, prune oldest-by-mtime backups beyond
_CORRUPT_BACKUP_RETENTION (module constant, default 10), including the
copied -wal/-shm sidecars. The just-created backup is always exempt
(copy2 preserves the source mtime, which can be older than existing
backups). Pruning is best-effort and never masks the corruption error
about to be raised; dedupe of identical corrupt bytes is unchanged.

hermes_cli/kanban_db.py

@@ -1286,6 +1286,14 @@ _INIT_LOCK = threading.RLock()
 _SQLITE_HEADER = b"SQLite format 3\x00"
 DEFAULT_BUSY_TIMEOUT_MS = 120_000
 
+# Maximum number of ``<db>.corrupt.<hash>.bak`` quarantine files retained per
+# board DB. Content-addressing already dedupes identical corrupt bytes, but
+# repeatedly-mutating corruption (partial repairs, further damage between
+# dispatcher retries) mints a new fingerprint each time; without a cap a user
+# accumulated 124 backups. Oldest-by-mtime files beyond the cap are pruned
+# right after each new backup is created.
+_CORRUPT_BACKUP_RETENTION = 10
+
 # Bounded acquire for the cross-process init lock (#36644). The original bare
 # blocking flock had no timeout, so a wedged holder blocked the dispatcher's
 # next-tick connect forever. We retry a non-blocking acquire up to this
@@ -1567,6 +1575,54 @@ class KanbanDbCorruptError(RuntimeError):
         )
 
 
+def _prune_corrupt_backups(
+    parent: Path, base_name: str, keep: Optional[Path] = None,
+) -> None:
+    """Cap the number of retained ``<db>.corrupt.<hash>.bak`` files.
+
+    Content-addressed backups dedupe identical corrupt bytes, but a board
+    whose file keeps changing between corruption events (partial repairs,
+    ongoing damage, fleets of retrying dispatchers) can still accumulate
+    backups without bound — a user reported 124 of them. After creating a
+    new backup we keep only the ``_CORRUPT_BACKUP_RETENTION`` most recent
+    (by mtime) and delete the rest, including their copied ``-wal``/``-shm``
+    sidecars. ``keep`` (the just-created backup) is never pruned regardless
+    of its mtime — ``shutil.copy2`` preserves the source file's timestamp,
+    which may be older than existing backups. Best-effort: prune failures
+    never mask the corruption error the caller is about to raise.
+    """
+    try:
+        backups = [
+            candidate
+            for candidate in parent.glob(f"{base_name}.corrupt.*.bak")
+            if candidate.is_file() and candidate != keep
+        ]
+    except OSError:
+        return
+    budget = _CORRUPT_BACKUP_RETENTION - (1 if keep is not None else 0)
+    budget = max(budget, 0)
+    if len(backups) <= budget:
+        return
+
+    def _mtime(item: Path) -> float:
+        try:
+            return item.stat().st_mtime
+        except OSError:
+            return 0.0
+
+    backups.sort(key=_mtime, reverse=True)
+    for stale in backups[budget:]:
+        for victim in (
+            stale,
+            stale.with_name(stale.name + "-wal"),
+            stale.with_name(stale.name + "-shm"),
+        ):
+            try:
+                victim.unlink(missing_ok=True)
+            except OSError:
+                pass
+
+
 def _backup_corrupt_db(path: Path) -> Optional[Path]:
     """Copy a corrupt DB (and its WAL/SHM sidecars) to a content-addressed backup.
 
@@ -1607,6 +1663,9 @@ def _backup_corrupt_db(path: Path) -> Optional[Path]:
             shutil.copy2(resolved, candidate)
         except OSError:
             return None
+        # A NEW backup landed on disk — enforce the retention cap so
+        # mutating-corruption loops can't accumulate quarantines forever.
+        _prune_corrupt_backups(parent, base_name, keep=candidate)
     for suffix in ("-wal", "-shm"):
         sidecar = parent / (base_name + suffix)
         if sidecar.parent != parent or not sidecar.exists():

tests/hermes_cli/test_kanban_db_repair.py

@@ -194,3 +194,83 @@ def test_repaired_db_connects_normally_afterwards(tmp_path):
     finally:
         conn.close()
     assert set(tmp_path.glob("kanban.db.corrupt.*.bak")) == before
+
+
+# ---------------------------------------------------------------------------
+# Corrupt-backup retention cap
+# ---------------------------------------------------------------------------
+
+def test_corrupt_backup_retention_cap_prunes_oldest(tmp_path, monkeypatch):
+    """Mutating corruption can't accumulate quarantine files forever.
+
+    Regression for the field report of 124 ``.corrupt.*.bak`` files: each
+    distinct corrupt byte-state mints a new content-addressed backup, so a
+    board whose file keeps changing between failures grows one backup per
+    mutation. The cap keeps only the newest ``_CORRUPT_BACKUP_RETENTION``.
+    """
+    monkeypatch.setattr(kb, "_CORRUPT_BACKUP_RETENTION", 3)
+    db_path = tmp_path / "kanban.db"
+    _write_page_corrupt_db(db_path)
+
+    minted: list[Path] = []
+    for i in range(8):
+        # Mutate the corrupt bytes → new sha → new backup each round.
+        with db_path.open("r+b") as fh:
+            fh.seek(200)
+            fh.write(bytes([i]) * 16)
+        kb._INITIALIZED_PATHS.discard(str(db_path.resolve()))
+        with pytest.raises(kb.KanbanDbCorruptError) as excinfo:
+            kb.connect(db_path=db_path)
+        assert excinfo.value.backup_path is not None
+        minted.append(excinfo.value.backup_path)
+        # The just-created backup always survives its own prune pass.
+        assert excinfo.value.backup_path.exists()
+
+    remaining = sorted(tmp_path.glob("kanban.db.corrupt.*.bak"))
+    assert len(remaining) == 3, (
+        f"expected retention cap of 3, found {len(remaining)}: {remaining}"
+    )
+    # The newest backup (this round's) is among the survivors.
+    assert minted[-1] in remaining
+
+
+def test_corrupt_backup_retention_prunes_sidecar_copies(tmp_path, monkeypatch):
+    """Pruned backups take their copied -wal/-shm sidecars with them."""
+    monkeypatch.setattr(kb, "_CORRUPT_BACKUP_RETENTION", 1)
+    db_path = tmp_path / "kanban.db"
+    _write_page_corrupt_db(db_path)
+
+    # Fabricate an old backup + sidecars that the next prune should remove.
+    import os
+    stale = tmp_path / "kanban.db.corrupt.deadbeef00000000.bak"
+    stale.write_bytes(b"old corrupt bytes")
+    (tmp_path / (stale.name + "-wal")).write_bytes(b"wal")
+    (tmp_path / (stale.name + "-shm")).write_bytes(b"shm")
+    past = 1_000_000_000
+    os.utime(stale, (past, past))
+
+    kb._INITIALIZED_PATHS.discard(str(db_path.resolve()))
+    with pytest.raises(kb.KanbanDbCorruptError) as excinfo:
+        kb.connect(db_path=db_path)
+    fresh = excinfo.value.backup_path
+    assert fresh is not None and fresh.exists()
+
+    assert not stale.exists()
+    assert not (tmp_path / (stale.name + "-wal")).exists()
+    assert not (tmp_path / (stale.name + "-shm")).exists()
+
+
+def test_identical_corrupt_bytes_still_reuse_one_backup(tmp_path):
+    """The retention cap must not break content-addressed dedupe."""
+    db_path = tmp_path / "kanban.db"
+    _write_page_corrupt_db(db_path)
+
+    backups: set[Path] = set()
+    for _ in range(5):
+        kb._INITIALIZED_PATHS.discard(str(db_path.resolve()))
+        with pytest.raises(kb.KanbanDbCorruptError) as excinfo:
+            kb.connect(db_path=db_path)
+        assert excinfo.value.backup_path is not None
+        backups.add(excinfo.value.backup_path)
+    assert len(backups) == 1
+    assert len(list(tmp_path.glob("kanban.db.corrupt.*.bak"))) == 1