fix(scan): mass-prune guard v2 — partial degraded listing refused on auto scan

The v1 zero-listing guard (f6e9727) was bypassed by a partial degraded
mount: if even one song was visible, current_files was non-empty and
delete_missing ran freely, pruning every invisible DB row (Creed r1 HIGH).

Guard contract (god ruling):
- Auto scans (startup, periodic, /api/rescan): refuse when
  would_remove >= max(_PRUNE_MAX_ABS=1, _PRUNE_MAX_FRAC=0.5 * existing).
  Zero-listing also refused (would_remove == existing).
  Scan sets stage='error', leaves DB intact.
- /api/rescan/full: allow_mass_prune=True → guard logs a warning and
  proceeds; delete_missing runs (user-authorised explicit intent).

Changes:
- lib/scan.py: _PRUNE_MAX_ABS/FRAC constants; combined guard (zero +
  partial) before delete_missing; allow_mass_prune param on
  background_scan(); _scan_mass_prune_next global threaded through
  kick_scan() and _scan_runner()
- server.py: trigger_full_rescan calls kick_scan(allow_mass_prune=True)
- tests/test_scan_prune_guard.py: Case B (Creed partial, RED f6e9727→
  GREEN here) + Case C (full-rescan bypass)

Gates: pytest 2816/2816, JS 1155/1155

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01H2bM5jSbMskpdxm2CmuQVj

lib/scan.py

@@ -171,6 +171,13 @@ _SCAN_STATUS_INIT = {"running": False, "stage": "idle", "total": 0, "done": 0, "
 
 _scan_status = dict(_SCAN_STATUS_INIT)
 
+# Mass-prune guard thresholds for automatic scans (not full rescan).
+# Refuse when would_remove >= max(_PRUNE_MAX_ABS, _PRUNE_MAX_FRAC * existing).
+# At 50 % of the library (or even a single row on tiny libraries) a sudden
+# disappearance almost certainly means a degraded mount, not a real deletion.
+_PRUNE_MAX_ABS = 1
+_PRUNE_MAX_FRAC = 0.5
+
 
 def _make_scan_executor():
     """Build the executor for the background metadata scan.
@@ -213,12 +220,17 @@ def _make_scan_executor():
     )
 
 
-def background_scan(force: bool = False):
+def background_scan(force: bool = False, allow_mass_prune: bool = False):
     """Scan the library and cache song metadata on startup. Uses a process pool to bypass the GIL for CPU-bound metadata parsing.
 
     `force` skips the directory-signature fast path and always does the full
     listing/stat pass — the manual Refresh sets it (see _dir_signature_file).
 
+    `allow_mass_prune` permits the scan to prune more than the catastrophic
+    threshold (_PRUNE_MAX_FRAC of the library). Only set by /api/rescan/full
+    (explicit user intent); automatic and plain /api/rescan scans leave it
+    False so a degraded-mount partial listing can't silently wipe the library.
+
     Never sets `_scan_status["running"] = False` — ownership of that flag
     lives in `_scan_runner` so a `kick_scan()` racing this function's
     terminal write cannot observe a stale False and start a second runner.
@@ -313,25 +325,42 @@ def background_scan(force: bool = False):
 
     current_files = {_relpath(f, dlc) for f in all_songs}
 
-    # Guard: refuse to prune when the listing returned zero songs but the DB is
-    # non-empty. An empty listing on a non-empty library almost certainly means
-    # the DLC mount was temporarily inaccessible (FUSE remount, dirty-flag RO
-    # fallback, brief unmount) rather than every song being genuinely deleted.
-    # delete_missing({}) would remove ALL rows — a catastrophic silent purge.
-    # Treat it as a listing failure instead and leave the DB intact.
-    if not current_files:
-        with appstate.meta_db._lock:
-            _existing = appstate.meta_db.conn.execute(
-                "SELECT COUNT(*) FROM songs").fetchone()[0]
-        if _existing > 0:
+    # Guard: refuse (or warn) when the listing suggests a degraded mount.
+    # Two cases share the same logic:
+    #   1. Zero listing  — current_files empty, DB non-empty: would erase everything.
+    #   2. Partial listing — current_files non-empty but so many DB rows are absent
+    #      that it looks like a mount glitch rather than deliberate deletions.
+    #      Threshold: would_remove >= max(_PRUNE_MAX_ABS, _PRUNE_MAX_FRAC * existing).
+    # allow_mass_prune (only True for /api/rescan/full) lets the prune proceed with
+    # a warning so the user's explicit intent is honoured even in the degraded case.
+    with appstate.meta_db._lock:
+        _existing = appstate.meta_db.conn.execute(
+            "SELECT COUNT(*) FROM songs").fetchone()[0]
+
+    if _existing > 0:
+        if not current_files:
+            _would_remove = _existing
+        else:
+            with appstate.meta_db._lock:
+                _db_files = {r[0] for r in appstate.meta_db.conn.execute(
+                    "SELECT filename FROM songs").fetchall()}
+            _would_remove = len(_db_files - current_files)
+
+        _threshold = max(_PRUNE_MAX_ABS, _PRUNE_MAX_FRAC * _existing)
+        if _would_remove >= _threshold:
             _msg = (
-                f"Scan: listing returned 0 songs but the DB has {_existing} rows "
-                "— possible mount/permission issue. Skipping prune to avoid data loss."
+                f"Scan: would remove {_would_remove} of {_existing} DB rows "
+                f"(threshold {int(_threshold)}) with only {len(current_files)} song(s) visible "
+                "— possible mount/permission issue. Check the DLC mount; use Settings → "
+                "Rescan Library (full) to authorise a large prune."
             )
-            log.error("%s", _msg)
-            _scan_status = {**_SCAN_STATUS_INIT, "running": True,
-                            "stage": "error", "error": _msg}
-            return
+            if allow_mass_prune:
+                log.warning("%s — proceeding (user-authorised full rescan)", _msg)
+            else:
+                log.error("%s", _msg)
+                _scan_status = {**_SCAN_STATUS_INIT, "running": True,
+                                "stage": "error", "error": _msg}
+                return
 
     # Clean up stale DB entries. delete_missing reports both deltas (rows pruned
     # + genuinely-new files) so the scan can surface an added/removed summary.
@@ -426,6 +455,10 @@ _scan_rescan_pending = False
 # Set by kick_scan(force=True); consumed by _scan_runner for the next pass so a
 # manual Refresh bypasses the directory-signature fast path.
 _scan_force_next = False
+# Set by kick_scan(allow_mass_prune=True); allows the next pass to prune past the
+# catastrophic threshold. Sticky like _scan_force_next: if any queued request asks
+# for it, the follow-up pass honours it.
+_scan_mass_prune_next = False
 
 
 # Handles to the running scan / enrichment worker threads. Both use the shared
@@ -436,7 +469,7 @@ _scan_force_next = False
 _scan_thread: threading.Thread | None = None
 
 
-def kick_scan(force: bool = False) -> bool:
+def kick_scan(force: bool = False, allow_mass_prune: bool = False) -> bool:
     """Request a library rescan, single-flight + coalescing.
 
     `force` skips the directory-signature fast path for the resulting pass (the
@@ -445,6 +478,10 @@ def kick_scan(force: bool = False) -> bool:
     onto a running or queued scan keeps the force intent: the pass is forced if
     ANY pending request asked for it.
 
+    `allow_mass_prune` permits the resulting pass to prune past the catastrophic
+    threshold. Sticky: if any pending request set it, the follow-up pass honours it.
+    Only /api/rescan/full passes True — plain rescans and startup scans never do.
+
     Returns True if a new scan thread was started, False if one was already
     running. In the latter case a follow-up pass is queued and runs as soon
     as the current scan finishes so files landing mid-scan (e.g. an upload
@@ -452,10 +489,12 @@ def kick_scan(force: bool = False) -> bool:
     until the next periodic pass. Multiple late-arriving requests coalesce
     into a single follow-up.
     """
-    global _scan_rescan_pending, _scan_thread, _scan_force_next
+    global _scan_rescan_pending, _scan_thread, _scan_force_next, _scan_mass_prune_next
     with _scan_kick_lock:
         if force:
             _scan_force_next = True
+        if allow_mass_prune:
+            _scan_mass_prune_next = True
         if _scan_status["running"]:
             _scan_rescan_pending = True
             return False
@@ -469,15 +508,17 @@ def kick_scan(force: bool = False) -> bool:
 
 def _scan_runner():
     """Run _background_scan, then re-run if requests arrived mid-scan."""
-    global _scan_rescan_pending, _scan_force_next
+    global _scan_rescan_pending, _scan_force_next, _scan_mass_prune_next
     while True:
-        # Consume the force flag for THIS pass; a forced request queued mid-scan
-        # sets it again for the follow-up.
+        # Consume both flags for THIS pass; requests queued mid-scan set them
+        # again for the follow-up (sticky: any requester who asked for it wins).
         with _scan_kick_lock:
             forced = _scan_force_next
             _scan_force_next = False
+            mass_prune = _scan_mass_prune_next
+            _scan_mass_prune_next = False
         try:
-            background_scan(force=forced)
+            background_scan(force=forced, allow_mass_prune=mass_prune)
         except Exception:
             log.exception("background scan failed unexpectedly")
 

server.py

@@ -1136,7 +1136,7 @@ def trigger_full_rescan():
         # delete_missing() prunes anything genuinely gone at the end.
         meta_db.conn.execute("UPDATE songs SET mtime = -1")
         meta_db.conn.commit()
-    if not scan.kick_scan(force=True):
+    if not scan.kick_scan(force=True, allow_mass_prune=True):
         return {"message": "Scan already in progress"}
     return {"message": "Full rescan started"}
 

tests/test_scan_prune_guard.py

@@ -1,13 +1,21 @@
-"""Scan prune guard — background_scan() must refuse to prune when listing returns 0
-songs but the songs table is non-empty (feedBack#P1-libpurge).
+"""Scan prune guard — background_scan() must refuse to prune when the listing looks
+degraded (feedBack#P1-libpurge).
 
-Failing input:
-    - builtin_content seeding mocked (simulates RO mount where writes fail)
-    - dlc dir contains no feedpak/sloppak/wem files
-    - songs table has 1 row
+Three cases:
 
-Without the guard: delete_missing({}) fires, all rows deleted.
-With the guard:    scan aborts with stage='error', row survives.
+Case A — zero listing (original guard):
+    Failing input: dlc dir completely empty, songs table has 1 row.
+    Without guard: delete_missing({}) fires → all rows deleted.
+    With guard:    scan aborts with stage='error', row survives.
+
+Case B — partial listing (Creed r1 HIGH):
+    Failing input: DB has 2 rows, dlc dir shows only 1 file (neither DB row visible).
+    Without guard: delete_missing prunes both invisible rows → catastrophic loss.
+    With guard:    would_remove(2) >= threshold(1) → stage='error', both rows survive.
+
+Case C — full rescan bypass:
+    Same partial-degraded setup, but scan.kick_scan(allow_mass_prune=True).
+    Guard logs a warning and proceeds; delete_missing runs normally.
 """
 import importlib
 import sys
@@ -87,3 +95,129 @@ def test_empty_listing_refuses_prune_when_db_nonempty(prune_guard_env):
         f"got {scan_mod._scan_status['stage']!r}"
     )
     assert scan_mod._scan_status["error"] is not None, "error message must be set"
+
+
+# ── Case B: partial listing (Creed r1 HIGH) ───────────────────────────────────
+
+@pytest.fixture()
+def partial_prune_env(tmp_path, monkeypatch, reset_scan_state):
+    """DB has 2 rows (neither on disk), dlc dir shows 1 unrelated visible file."""
+    import concurrent.futures
+
+    monkeypatch.setenv("CONFIG_DIR", str(tmp_path))
+    monkeypatch.delenv("DLC_DIR", raising=False)
+
+    dlc = tmp_path / "dlc"
+    dlc.mkdir()
+    (tmp_path / "config.json").write_text('{"dlc_dir": "%s"}' % dlc)
+
+    # One visible .feedpak file on disk — makes current_files non-empty so the
+    # old zero-listing guard would not fire, but both DB rows are absent.
+    import zipfile
+    visible = dlc / "only-visible.feedpak"
+    with zipfile.ZipFile(visible, "w") as zf:
+        zf.writestr("manifest.yaml", "title: Visible\nartist: Test\n")
+
+    monkeypatch.setattr("builtin_content.seed_builtin_diagnostic_sloppaks",
+                        lambda *a, **kw: None)
+    monkeypatch.setattr("builtin_content.seed_builtin_starter_content",
+                        lambda *a, **kw: None)
+
+    sys.modules.pop("server", None)
+    mod = importlib.import_module("server")
+
+    import scan as scan_mod
+    monkeypatch.setattr(
+        scan_mod, "_make_scan_executor",
+        lambda: concurrent.futures.ThreadPoolExecutor(max_workers=1),
+    )
+
+    yield mod, scan_mod
+
+    conn = getattr(getattr(mod, "meta_db", None), "conn", None)
+    if conn is not None:
+        getattr(sys.modules.get("server"), "_join_background_db_threads", lambda: None)()
+        conn.close()
+
+
+def test_partial_listing_refuses_prune_when_mass_threshold_exceeded(partial_prune_env):
+    """Creed r1 HIGH: 2 DB rows absent from listing, 1 visible file → auto scan refused.
+
+    Failing input:
+        - DB: lost-one.feedpak, lost-two.feedpak (neither on disk)
+        - dlc dir: only-visible.feedpak (not in DB)
+        - Auto scan (allow_mass_prune=False)
+    Expected:
+        - Both DB rows survive (count unchanged)
+        - stage='error', error message set
+    Fails on f6e9727 (old zero-only guard): would_remove=2, current_files non-empty
+    → old guard skips → delete_missing prunes both rows.
+    """
+    mod, scan_mod = partial_prune_env
+    import appstate
+
+    _song_meta = {"title": "T", "artist": "A", "album": "", "duration": 1.0,
+                  "tuning": "", "arrangements": [], "format": "archive"}
+    appstate.meta_db.put("lost-one.feedpak", 11111.0, 500, _song_meta)
+    appstate.meta_db.put("lost-two.feedpak", 22222.0, 500, _song_meta)
+
+    count_before = appstate.meta_db.conn.execute(
+        "SELECT COUNT(*) FROM songs").fetchone()[0]
+    assert count_before == 2, f"pre-condition: 2 rows in DB, got {count_before}"
+
+    # Auto scan — allow_mass_prune stays False (default)
+    scan_mod.background_scan()
+
+    count_after = appstate.meta_db.conn.execute(
+        "SELECT COUNT(*) FROM songs").fetchone()[0]
+    assert count_after == count_before, (
+        f"mass-prune guard must refuse when would_remove={count_before - count_after} "
+        f"exceeds threshold; DB had {count_before} row(s), now has {count_after}"
+    )
+    assert scan_mod._scan_status["stage"] == "error", (
+        f"scan must set stage='error' when the guard fires, "
+        f"got {scan_mod._scan_status['stage']!r}"
+    )
+    assert scan_mod._scan_status["error"] is not None, "error message must be set"
+
+
+# ── Case C: full rescan bypasses the guard ────────────────────────────────────
+
+def test_full_rescan_allows_prune_past_threshold(partial_prune_env):
+    """Full rescan (allow_mass_prune=True) proceeds even when would_remove >= threshold.
+
+    Same partial-degraded setup as Case B, but the user explicitly invoked
+    /api/rescan/full which sets allow_mass_prune=True. The guard logs a warning
+    and does not abort; delete_missing runs and prunes the absent rows.
+
+    Failing input: same as Case B.
+    Expected:      both absent rows pruned, stage='complete' (or 'scanning').
+    """
+    mod, scan_mod = partial_prune_env
+    import appstate
+
+    _song_meta = {"title": "T", "artist": "A", "album": "", "duration": 1.0,
+                  "tuning": "", "arrangements": [], "format": "archive"}
+    appstate.meta_db.put("lost-one.feedpak", 11111.0, 500, _song_meta)
+    appstate.meta_db.put("lost-two.feedpak", 22222.0, 500, _song_meta)
+
+    count_before = appstate.meta_db.conn.execute(
+        "SELECT COUNT(*) FROM songs").fetchone()[0]
+    assert count_before == 2
+
+    # Full rescan — allow_mass_prune=True (user-authorised)
+    scan_mod.background_scan(allow_mass_prune=True)
+
+    count_after = appstate.meta_db.conn.execute(
+        "SELECT COUNT(*) FROM songs").fetchone()[0]
+    # The two absent rows are pruned; only-visible may or may not have been indexed
+    # (it has a minimal manifest so sloppak detection may skip it — that's fine,
+    # the key invariant is that the guard did NOT abort).
+    assert scan_mod._scan_status["stage"] != "error", (
+        f"full rescan must not abort on mass-prune threshold; "
+        f"got stage={scan_mod._scan_status['stage']!r}"
+    )
+    assert count_after < count_before, (
+        f"full rescan must have pruned the absent rows; "
+        f"DB had {count_before} row(s), now has {count_after}"
+    )