feat(config): add background symlink rescan and simplify cache validation

- Added threading import and optional `_rescan_thread` for background operations
- Simplified `_load_symlink_cache` to only validate path mappings, removing fingerprint checks
- Updated `_initialize_symlink_mappings` to rebuild preview roots and schedule rescan when cache is loaded
- Added `_schedule_symlink_rescan` method to perform background validation of symlinks
- Cleared `_path_mappings` at start of `_scan_symbolic_links` to prevent stale entries
- Background rescan improves performance by deferring symlink validation after cache load

py/config.py

@@ -1,8 +1,9 @@
 import os
 import platform
+import threading
 from pathlib import Path
 import folder_paths # type: ignore
-from typing import Any, Dict, Iterable, List, Mapping, Optional, Set
+from typing import Any, Dict, Iterable, List, Mapping, Optional, Set, Tuple
 import logging
 import json
 import urllib.parse
@@ -81,6 +82,8 @@ class Config:
         self._path_mappings: Dict[str, str] = {}
         # Normalized preview root directories used to validate preview access
         self._preview_root_paths: Set[Path] = set()
+        # Optional background rescan thread
+        self._rescan_thread: Optional[threading.Thread] = None
         self.loras_roots = self._init_lora_paths()
         self.checkpoints_roots = None
         self.unet_roots = None
@@ -297,51 +300,11 @@ class Config:
             logger.info("Symlink cache payload is not a dict: %s", type(payload))
             return False
 
-        cached_fingerprint = payload.get("fingerprint")
         cached_mappings = payload.get("path_mappings")
-        if not isinstance(cached_fingerprint, dict) or not isinstance(cached_mappings, Mapping):
+        if not isinstance(cached_mappings, Mapping):
-            logger.info("Symlink cache missing fingerprint or path mappings")
+            logger.info("Symlink cache missing path mappings")
             return False
 
-        current_fingerprint = self._build_symlink_fingerprint()
-        cached_roots = cached_fingerprint.get("roots")
-        cached_stats = cached_fingerprint.get("stats")
-        if (
-            not isinstance(cached_roots, list)
-            or not isinstance(cached_stats, Mapping)
-            or sorted(cached_roots) != sorted(current_fingerprint["roots"])  # type: ignore[index]
-        ):
-            logger.info("Symlink cache invalidated: roots changed")
-            return False
-
-        for root in current_fingerprint["roots"]:  # type: ignore[assignment]
-            cached_stat = cached_stats.get(root) if isinstance(cached_stats, Mapping) else None
-            current_stat = current_fingerprint["stats"].get(root)  # type: ignore[index]
-            if not isinstance(cached_stat, Mapping) or not current_stat:
-                logger.info("Symlink cache invalidated: missing stats for %s", root)
-                return False
-
-            cached_mtime = cached_stat.get("mtime_ns")
-            cached_inode = cached_stat.get("inode")
-            current_mtime = current_stat.get("mtime_ns")
-            current_inode = current_stat.get("inode")
-
-            if cached_inode != current_inode:
-                logger.info("Symlink cache invalidated: inode changed for %s", root)
-                return False
-
-            if cached_mtime != current_mtime:
-                cached_noise = cached_stat.get("noise_mtime_ns")
-                current_noise = current_stat.get("noise_mtime_ns")
-                if not (
-                    cached_noise
-                    and current_noise
-                    and cached_mtime == cached_noise
-                    and current_mtime == current_noise
-                ):
-                    logger.info("Symlink cache invalidated: mtime changed for %s", root)
-                    return False
-
         normalized_mappings: Dict[str, str] = {}
         for target, link in cached_mappings.items():
             if not isinstance(target, str) or not isinstance(link, str):
@@ -368,23 +331,30 @@ class Config:
 
     def _initialize_symlink_mappings(self) -> None:
         start = time.perf_counter()
-        if not self._load_symlink_cache():
+        cache_loaded = self._load_symlink_cache()
-            self._scan_symbolic_links()
+
-            self._save_symlink_cache()
+        if cache_loaded:
-            logger.info(
-                "Symlink mappings rebuilt and cached in %.2f ms",
-                (time.perf_counter() - start) * 1000,
-            )
-        else:
             logger.info(
                 "Symlink mappings restored from cache in %.2f ms",
                 (time.perf_counter() - start) * 1000,
             )
+            self._rebuild_preview_roots()
+            self._schedule_symlink_rescan()
+            return
+
+        self._scan_symbolic_links()
+        self._save_symlink_cache()
         self._rebuild_preview_roots()
+        logger.info(
+            "Symlink mappings rebuilt and cached in %.2f ms",
+            (time.perf_counter() - start) * 1000,
+        )
 
     def _scan_symbolic_links(self):
         """Scan all symbolic links in LoRA, Checkpoint, and Embedding root directories"""
         start = time.perf_counter()
+        # Reset mappings before rescanning to avoid stale entries
+        self._path_mappings.clear()
         visited_dirs: Set[str] = set()
         for root in self._symlink_roots():
             self._scan_directory_links(root, visited_dirs)
@@ -394,6 +364,36 @@ class Config:
             len(self._path_mappings),
         )
 
+    def _schedule_symlink_rescan(self) -> None:
+        """Trigger a best-effort background rescan to refresh stale caches."""
+
+        if self._rescan_thread and self._rescan_thread.is_alive():
+            return
+
+        def worker():
+            try:
+                self._scan_symbolic_links()
+                self._save_symlink_cache()
+                self._rebuild_preview_roots()
+                logger.info("Background symlink rescan completed")
+            except Exception as exc:  # pragma: no cover - defensive logging
+                logger.info("Background symlink rescan failed: %s", exc)
+
+        thread = threading.Thread(
+            target=worker,
+            name="lora-manager-symlink-rescan",
+            daemon=True,
+        )
+        self._rescan_thread = thread
+        thread.start()
+
+    def _wait_for_rescan(self, timeout: Optional[float] = None) -> None:
+        """Block until the background rescan completes (testing convenience)."""
+
+        thread = self._rescan_thread
+        if thread:
+            thread.join(timeout=timeout)
+
     def _scan_directory_links(self, root: str, visited_dirs: Set[str]):
         """Iteratively scan directory symlinks to avoid deep recursion."""
         try:

tests/config/test_symlink_cache.py

@@ -62,6 +62,7 @@ def test_symlink_scan_skips_file_links(monkeypatch: pytest.MonkeyPatch, tmp_path
 
 def test_symlink_cache_reuses_previous_scan(monkeypatch: pytest.MonkeyPatch, tmp_path):
     loras_dir, settings_dir = _setup_paths(monkeypatch, tmp_path)
+    monkeypatch.setattr(config_module.Config, "_schedule_symlink_rescan", lambda self: None)
 
     target_dir = loras_dir / "target"
     target_dir.mkdir()
@@ -85,6 +86,7 @@ def test_symlink_cache_reuses_previous_scan(monkeypatch: pytest.MonkeyPatch, tmp
 
 def test_symlink_cache_survives_noise_mtime(monkeypatch: pytest.MonkeyPatch, tmp_path):
     loras_dir, settings_dir = _setup_paths(monkeypatch, tmp_path)
+    monkeypatch.setattr(config_module.Config, "_schedule_symlink_rescan", lambda self: None)
 
     target_dir = loras_dir / "target"
     target_dir.mkdir()
@@ -109,3 +111,35 @@ def test_symlink_cache_survives_noise_mtime(monkeypatch: pytest.MonkeyPatch, tmp
 
     second_cfg = config_module.Config()
     assert second_cfg.map_path_to_link(str(target_dir)) == _normalize(str(dir_link))
+
+
+def test_background_rescan_refreshes_cache(monkeypatch: pytest.MonkeyPatch, tmp_path):
+    loras_dir, _ = _setup_paths(monkeypatch, tmp_path)
+
+    target_dir = loras_dir / "target"
+    target_dir.mkdir()
+    dir_link = loras_dir / "dir_link"
+    dir_link.symlink_to(target_dir, target_is_directory=True)
+
+    # Build initial cache pointing at the first target
+    first_cfg = config_module.Config()
+    old_real = _normalize(os.path.realpath(target_dir))
+    assert first_cfg.map_path_to_link(str(target_dir)) == _normalize(str(dir_link))
+
+    # Retarget the symlink to a new directory without touching the cache file
+    new_target = loras_dir / "target_v2"
+    new_target.mkdir()
+    dir_link.unlink()
+    dir_link.symlink_to(new_target, target_is_directory=True)
+
+    second_cfg = config_module.Config()
+
+    # Cache may still point at the old real path immediately after load
+    initial_mapping = second_cfg.map_path_to_link(str(new_target))
+    assert initial_mapping in {str(new_target), _normalize(str(dir_link))}
+
+    # Background rescan should refresh the mapping to the new target and update the cache file
+    second_cfg._wait_for_rescan(timeout=2.0)
+    new_real = _normalize(os.path.realpath(new_target))
+    assert second_cfg._path_mappings.get(new_real) == _normalize(str(dir_link))
+    assert second_cfg.map_path_to_link(str(new_target)) == _normalize(str(dir_link))