Playlists: mirror-update trigger prunes removed tracks (the other half)

Symmetric to the post-download reconcile (which handles ADDITIONS): when a
playlist's membership is re-synced (the mirror step — scheduled refresh or the
manual mirror endpoint), rebuild its folder from current membership WITH prune
IF it's organize-by-playlist. So a track that just LEFT the playlist has its
symlink cleaned up the instant membership changes, not only on the next download.

Factored a shared _rebuild_one_from_db (used by the manual 'Rebuild' button and
the mirror hook) + rebuild_mirrored_playlist_if_organized. Gated to organized
playlists, non-fatal at both mirror call sites.

Now the invariant 'folder = the playlist's current owned members' holds on every
change: additions caught at download, removals caught at mirror. 2 new tests
(removed track pruned; non-organized skipped). 985 + 277 tests pass.

core/automation/handlers/refresh_mirrored.py

@@ -292,6 +292,18 @@ def _commit_refresh(
         image_url=pl.get('image_url'),
     )
 
+    # Membership just changed — if this playlist is organize-by-playlist, rebuild
+    # its folder (with prune) so a track that LEFT the playlist has its symlink
+    # cleaned up now. Gated to organized playlists, non-fatal — never disturbs
+    # the refresh. (Additions are handled by the post-download reconcile.)
+    try:
+        from core.playlists.materialize_service import rebuild_mirrored_playlist_if_organized
+        rebuild_mirrored_playlist_if_organized(
+            db, deps.config_manager, pl.get('id'), profile_id=pl.get('profile_id', 1)
+        )
+    except Exception as _mat_err:
+        deps.logger.debug(f"[Playlist Folder] mirror-refresh cleanup skipped: {_mat_err}")
+
     if old_ids != new_ids:
         added = len(new_ids - old_ids)
         removed = len(old_ids - new_ids)

core/playlists/materialize_service.py

@@ -126,42 +126,61 @@ def reconcile_batch_playlists(batch: dict, download_tasks: dict, config_manager)
     return results
 
 
-def rebuild_organized_playlists_from_db(db, config_manager, *, profile_id: int = 1):
-    """Rebuild every "organize by playlist" folder from CURRENT library ownership —
-    for the manual "Rebuild" button. Unlike the post-download path (which uses the
-    batch's payload), there's no batch here, so each playlist's owned files are
-    re-derived via the app's own matcher — ``check_track_exists`` (by name+artist),
-    NOT source IDs — then resolved to disk and handed to the materializer. This
-    self-heals the folders after a library reorganize moves files or membership
-    changes. Returns a list of ``(playlist_name, RebuildSummary)``."""
+def _rebuild_one_from_db(db, config_manager, playlist: dict):
+    """Rebuild ONE playlist's folder from its CURRENT membership × ownership —
+    re-matching each member via the app's own ``check_track_exists`` (by name, not
+    source IDs), resolving to disk, and rebuilding WITH prune. Because it's driven
+    by current membership, a track that has LEFT the playlist drops out of the set
+    and its symlink is pruned. Returns ``(playlist_name, RebuildSummary)``."""
     from core.library.path_resolver import resolve_library_file_path
 
     root = docker_resolve_path(config_manager.get("playlists.materialize_path", "./Playlists"))
     mode = normalize_mode(config_manager.get("playlists.materialize_mode", "symlink"))
-    results = []
-    for pl in (db.get_mirrored_playlists(profile_id) or []):
-        if not pl.get("organize_by_playlist"):
+    real_paths: List[str] = []
+    seen = set()
+    for t in (db.get_mirrored_playlist_tracks(playlist["id"]) or []):
+        title = (t.get("track_name") or "").strip()
+        artist = (t.get("artist_name") or "").strip()
+        if not title:
             continue
-        real_paths: List[str] = []
-        seen = set()
-        for t in (db.get_mirrored_playlist_tracks(pl["id"]) or []):
-            title = (t.get("track_name") or "").strip()
-            artist = (t.get("artist_name") or "").strip()
-            if not title:
-                continue
-            try:
-                db_track, conf = db.check_track_exists(title, artist, confidence_threshold=0.7)
-            except Exception:
-                continue
-            if db_track is None or conf < 0.7:
-                continue
-            real = resolve_library_file_path(getattr(db_track, "file_path", None), config_manager=config_manager)
-            if real and real not in seen:
-                seen.add(real)
-                real_paths.append(real)
-        name = pl.get("name") or "Unnamed Playlist"
-        results.append((name, rebuild_playlist_folder(root, name, real_paths, mode)))
-    return results
+        try:
+            db_track, conf = db.check_track_exists(title, artist, confidence_threshold=0.7)
+        except Exception:
+            continue
+        if db_track is None or conf < 0.7:
+            continue
+        real = resolve_library_file_path(getattr(db_track, "file_path", None), config_manager=config_manager)
+        if real and real not in seen:
+            seen.add(real)
+            real_paths.append(real)
+    name = playlist.get("name") or "Unnamed Playlist"
+    return name, rebuild_playlist_folder(root, name, real_paths, mode)   # prune_stale=True
+
+
+def rebuild_organized_playlists_from_db(db, config_manager, *, profile_id: int = 1):
+    """Rebuild EVERY "organize by playlist" folder from current library ownership —
+    for the manual "Rebuild" button. Self-heals after a library reorganize moves
+    files or membership changes. Returns a list of ``(playlist_name, summary)``."""
+    return [
+        _rebuild_one_from_db(db, config_manager, pl)
+        for pl in (db.get_mirrored_playlists(profile_id) or [])
+        if pl.get("organize_by_playlist")
+    ]
+
+
+def rebuild_mirrored_playlist_if_organized(db, config_manager, playlist_id, *, profile_id: int = 1):
+    """Mirror-update hook: after a playlist's membership is re-synced, rebuild its
+    folder (with prune) IF it's organize-by-playlist — so a track that just LEFT
+    the playlist has its symlink cleaned up the instant membership changes (the
+    mirror image of the post-download reconcile that handles additions). Returns
+    the summary, or ``None`` when the playlist isn't organized / can't be found."""
+    if playlist_id is None:
+        return None
+    pl = db.get_mirrored_playlist(playlist_id)
+    if not pl or not pl.get("organize_by_playlist"):
+        return None
+    _name, summary = _rebuild_one_from_db(db, config_manager, pl)
+    return summary
 
 
 __all__ = [
@@ -169,4 +188,5 @@ __all__ = [
     "materialize_playlist_from_batch",
     "reconcile_batch_playlists",
     "rebuild_organized_playlists_from_db",
+    "rebuild_mirrored_playlist_if_organized",
 ]

tests/test_playlist_materialize_service.py

@@ -10,6 +10,7 @@ from pathlib import Path
 from core.playlists.materialize_service import (
     collect_batch_real_paths,
     materialize_playlist_from_batch,
+    rebuild_mirrored_playlist_if_organized,
     rebuild_organized_playlists_from_db,
     reconcile_batch_playlists,
 )
@@ -39,6 +40,12 @@ class _RebuildDB:
                     {"track_name": "Gone", "artist_name": "y"}]   # not owned
         return [{"track_name": "B", "artist_name": "z"}]
 
+    def get_mirrored_playlist(self, playlist_id):
+        for pl in self.get_mirrored_playlists():
+            if pl["id"] == playlist_id:
+                return pl
+        return None
+
     def check_track_exists(self, title, artist, confidence_threshold=0.8,
                            server_source=None, album=None, candidate_tracks=None):
         fp = self.owned.get(title)
@@ -191,6 +198,45 @@ def test_reconcile_noop_for_plain_batch(tmp_path: Path):
     assert not (tmp_path / "Playlists").exists()
 
 
+def test_mirror_cleanup_prunes_removed_track(tmp_path: Path):
+    """Mirror-update hook: after a track LEAVES the playlist, its symlink is pruned."""
+    a = tmp_path / "Music" / "A.mp3"
+    gone = tmp_path / "Music" / "Gone.mp3"
+    for f in (a, gone):
+        f.parent.mkdir(parents=True, exist_ok=True)
+        f.write_bytes(b"audio")
+
+    class _DB:
+        def get_mirrored_playlist(self, pid):
+            return {"id": 1, "name": "Mix", "organize_by_playlist": True} if pid == 1 else None
+
+        def get_mirrored_playlist_tracks(self, pid):
+            return [{"track_name": "A", "artist_name": "x"}]   # 'Gone' was removed upstream
+
+        def check_track_exists(self, title, artist, confidence_threshold=0.8,
+                               server_source=None, album=None, candidate_tracks=None):
+            fp = {"A": str(a), "Gone": str(gone)}.get(title)
+            return (_Track(fp), 1.0) if fp else (None, 0.0)
+
+    from core.playlists.materialize import rebuild_playlist_folder
+    rebuild_playlist_folder(str(tmp_path / "Playlists"), "Mix", [str(a), str(gone)], "symlink")
+    assert (tmp_path / "Playlists" / "Mix" / "Gone.mp3").exists()   # both present before
+
+    summary = rebuild_mirrored_playlist_if_organized(_DB(), _Cfg(str(tmp_path / "Playlists")), 1, profile_id=1)
+    assert summary is not None
+    pdir = tmp_path / "Playlists" / "Mix"
+    assert (pdir / "A.mp3").exists()
+    assert not (pdir / "Gone.mp3").exists()                        # pruned on mirror update
+
+
+def test_mirror_cleanup_skips_non_organized(tmp_path: Path):
+    db = _RebuildDB({})
+    cfg = _Cfg(str(tmp_path / "Playlists"))
+    assert rebuild_mirrored_playlist_if_organized(db, cfg, 2, profile_id=1) is None     # Off (organize=0)
+    assert rebuild_mirrored_playlist_if_organized(db, cfg, 999, profile_id=1) is None   # unknown
+    assert rebuild_mirrored_playlist_if_organized(db, cfg, None, profile_id=1) is None
+
+
 def test_rebuild_from_db_only_organized_and_owned(tmp_path: Path):
     """The manual button: rebuild every organized playlist by re-matching with
     check_track_exists (name), linking only owned tracks."""

web_server.py

@@ -33682,6 +33682,15 @@ def mirror_playlist_endpoint():
         if playlist_id is None:
             return jsonify({"error": "Failed to mirror playlist"}), 500
 
+        # Membership just changed — if organize-by-playlist, rebuild its folder
+        # (with prune) so a removed track's symlink is cleaned up now. Gated +
+        # non-fatal, so it can never break the mirror response.
+        try:
+            from core.playlists.materialize_service import rebuild_mirrored_playlist_if_organized
+            rebuild_mirrored_playlist_if_organized(database, config_manager, playlist_id, profile_id=profile_id)
+        except Exception as _mat_err:
+            logger.debug("[Playlist Folder] mirror-time cleanup skipped: %s", _mat_err)
+
         try:
             if automation_engine:
                 automation_engine.emit('mirrored_playlist_created', {