Add SoulSync standalone library — no media server required

New 'soulsync' media server option manages the library directly from
the filesystem, bypassing Plex/Jellyfin/Navidrome entirely.

Two paths populate the library:
1. Downloads/imports write artist/album/track to DB immediately at
   post-processing completion, with pre-populated enrichment IDs
   (Spotify, Deezer, MusicBrainz) so workers skip re-discovery
2. soulsync_client.py scans Transfer folder for incremental/deep scan
   via DatabaseUpdateWorker (same interface as server clients)

New files:
- core/soulsync_client.py: filesystem scanner implementing the same
  interface as Plex/Jellyfin/Navidrome clients. Recursive folder scan,
  Mutagen tag reading, artist/album/track grouping, hash-based stable
  IDs, incremental scan by modification time.

Modified:
- web_server.py: _record_soulsync_library_entry() at post-processing
  completion, client init, scan endpoint integration, status endpoint,
  web_scan_manager media_clients dict, test-connection cache updates
- config/settings.py: accept 'soulsync' in set_active_media_server,
  get_active_media_server_config, is_configured, validate_config
- core/web_scan_manager.py: add soulsync to server_client_map

Dedup: checks existing artist/album by name across ALL server sources
before inserting to avoid duplicates. Enrichment IDs only written when
the column is empty (won't overwrite existing data).

config/settings.py

@@ -622,8 +622,8 @@ class ConfigManager:
         return self.get('active_media_server', 'plex')
 
     def set_active_media_server(self, server: str):
-        """Set the active media server (plex, jellyfin, or navidrome)"""
-        if server not in ['plex', 'jellyfin', 'navidrome']:
+        """Set the active media server (plex, jellyfin, navidrome, or soulsync)"""
+        if server not in ['plex', 'jellyfin', 'navidrome', 'soulsync']:
             raise ValueError(f"Invalid media server: {server}")
         self.set('active_media_server', server)
 
@@ -636,6 +636,8 @@ class ConfigManager:
             return self.get_jellyfin_config()
         elif active_server == 'navidrome':
             return self.get_navidrome_config()
+        elif active_server == 'soulsync':
+            return {'transfer_path': self.get('soulseek.transfer_path', './Transfer')}
         else:
             return {}
 
@@ -655,6 +657,8 @@ class ConfigManager:
         elif active_server == 'navidrome':
             navidrome = self.get_navidrome_config()
             media_server_configured = bool(navidrome.get('base_url')) and bool(navidrome.get('username')) and bool(navidrome.get('password'))
+        elif active_server == 'soulsync':
+            media_server_configured = True  # SoulSync standalone is always configured
 
         return (
             bool(spotify.get('client_id')) and
@@ -675,6 +679,7 @@ class ConfigManager:
         validation['plex'] = bool(self.get('plex.base_url')) and bool(self.get('plex.token'))
         validation['jellyfin'] = bool(self.get('jellyfin.base_url')) and bool(self.get('jellyfin.api_key'))
         validation['navidrome'] = bool(self.get('navidrome.base_url')) and bool(self.get('navidrome.username')) and bool(self.get('navidrome.password'))
+        validation['soulsync'] = True  # Standalone mode is always valid
         validation['active_media_server'] = active_server
 
         return validation

core/soulsync_client.py

@@ -0,0 +1,442 @@
+"""SoulSync Standalone Library Client — filesystem-based media server replacement.
+
+Implements the same interface as Plex/Jellyfin/Navidrome clients so the
+DatabaseUpdateWorker can scan the Transfer folder directly without an
+external media server. Reads audio file tags via Mutagen, groups by
+artist/album folder structure, and returns compatible data objects.
+"""
+
+import hashlib
+import os
+import re
+from datetime import datetime
+from typing import Any, Callable, Dict, List, Optional, Set
+
+from utils.logging_config import get_logger
+
+logger = get_logger("soulsync_client")
+
+AUDIO_EXTENSIONS = {'.mp3', '.flac', '.ogg', '.opus', '.m4a', '.aac', '.wav', '.wma', '.aiff', '.aif', '.ape'}
+
+
+def _stable_id(text: str) -> str:
+    """Generate a stable integer-like ID from a string (for DB compatibility)."""
+    return str(abs(int(hashlib.md5(text.encode('utf-8', errors='replace')).hexdigest(), 16)) % (10 ** 9))
+
+
+def _read_tags(file_path: str) -> Dict[str, Any]:
+    """Read audio tags from a file. Returns dict with title, artist, album, etc."""
+    result = {
+        'title': '', 'artist': '', 'album_artist': '', 'album': '',
+        'track_number': 0, 'disc_number': 1, 'year': '',
+        'genre': '', 'duration_ms': 0, 'bitrate': 0,
+    }
+    try:
+        from mutagen import File as MutagenFile
+        audio = MutagenFile(file_path, easy=True)
+        if audio:
+            if audio.tags:
+                tags = audio.tags
+                result['title'] = (tags.get('title', [''])[0] or '').strip()
+                result['artist'] = (tags.get('artist', [''])[0] or '').strip()
+                result['album_artist'] = (tags.get('albumartist', [''])[0] or '').strip()
+                result['album'] = (tags.get('album', [''])[0] or '').strip()
+                result['genre'] = (tags.get('genre', [''])[0] or '').strip()
+
+                date_str = (tags.get('date', [''])[0] or tags.get('year', [''])[0] or '').strip()
+                if date_str and len(date_str) >= 4:
+                    result['year'] = date_str[:4]
+
+                tn = tags.get('tracknumber', ['0'])[0]
+                try:
+                    result['track_number'] = int(str(tn).split('/')[0])
+                except (ValueError, TypeError):
+                    pass
+
+                dn = tags.get('discnumber', ['1'])[0]
+                try:
+                    result['disc_number'] = int(str(dn).split('/')[0])
+                except (ValueError, TypeError):
+                    pass
+
+            # Duration and bitrate from audio info
+            if hasattr(audio, 'info') and audio.info:
+                if hasattr(audio.info, 'length'):
+                    result['duration_ms'] = int(audio.info.length * 1000)
+                if hasattr(audio.info, 'bitrate'):
+                    result['bitrate'] = int(audio.info.bitrate / 1000) if audio.info.bitrate else 0
+    except Exception as e:
+        logger.debug(f"Could not read tags from {os.path.basename(file_path)}: {e}")
+
+    # Fallback: parse filename if no title
+    if not result['title']:
+        basename = os.path.splitext(os.path.basename(file_path))[0]
+        # Strip leading track numbers like "01 - Title" or "01. Title"
+        cleaned = re.sub(r'^\d+[\s.\-_]+', '', basename).strip()
+        result['title'] = cleaned or basename
+
+    return result
+
+
+class SoulSyncTrack:
+    """Track object compatible with DatabaseUpdateWorker expectations."""
+
+    def __init__(self, file_path: str, tags: Dict[str, Any], artist_ref=None, album_ref=None):
+        self.file_path = file_path
+        self._tags = tags
+        self._artist_ref = artist_ref
+        self._album_ref = album_ref
+
+        self.ratingKey = _stable_id(file_path)
+        self.title = tags['title']
+        self.duration = tags['duration_ms']
+        self.trackNumber = tags['track_number'] or None
+        self.discNumber = tags['disc_number'] or 1
+        self.year = int(tags['year']) if tags['year'] else None
+        self.userRating = None
+        self.addedAt = datetime.fromtimestamp(os.path.getmtime(file_path)) if os.path.exists(file_path) else datetime.now()
+        self.path = file_path
+        self.bitRate = tags['bitrate']
+        self.suffix = os.path.splitext(file_path)[1].lstrip('.').lower()
+
+    def artist(self):
+        return self._artist_ref
+
+    def album(self):
+        return self._album_ref
+
+
+class SoulSyncAlbum:
+    """Album object compatible with DatabaseUpdateWorker expectations."""
+
+    def __init__(self, album_key: str, title: str, year: Optional[int],
+                 artist_ref=None, track_list: List[SoulSyncTrack] = None):
+        self.ratingKey = _stable_id(album_key)
+        self.title = title
+        self.year = year
+        self._artist_ref = artist_ref
+        self._tracks = track_list or []
+        self.thumb = None
+        self.addedAt = datetime.now()
+        self.leafCount = len(self._tracks)  # Plex compat: track count
+        self.duration = sum(t.duration for t in self._tracks)  # Total duration in ms
+
+        # Collect genres from track tags
+        genre_set = set()
+        for t in self._tracks:
+            if t._tags.get('genre'):
+                genre_set.add(t._tags['genre'])
+        self.genres = list(genre_set)
+
+        # Set addedAt from earliest track
+        if self._tracks:
+            self.addedAt = min(t.addedAt for t in self._tracks)
+
+        # Check for cover art in the album folder
+        if self._tracks:
+            album_dir = os.path.dirname(self._tracks[0].file_path)
+            for cover_name in ['cover.jpg', 'cover.png', 'folder.jpg', 'folder.png']:
+                cover_path = os.path.join(album_dir, cover_name)
+                if os.path.isfile(cover_path):
+                    self.thumb = cover_path
+                    break
+
+    def artist(self):
+        return self._artist_ref
+
+    def tracks(self):
+        return self._tracks
+
+
+class SoulSyncArtist:
+    """Artist object compatible with DatabaseUpdateWorker expectations."""
+
+    def __init__(self, artist_key: str, title: str, album_list: List[SoulSyncAlbum] = None):
+        self.ratingKey = _stable_id(artist_key)
+        self.title = title
+        self._albums = album_list or []
+        self.genres = []
+        self.summary = ''
+        self.thumb = None
+        self.addedAt = datetime.now()
+
+        # Collect genres from tracks
+        genre_set = set()
+        for album in self._albums:
+            for track in album.tracks():
+                if track._tags.get('genre'):
+                    genre_set.add(track._tags['genre'])
+        self.genres = list(genre_set)
+
+        # Set addedAt from earliest album
+        if self._albums:
+            self.addedAt = min(a.addedAt for a in self._albums)
+
+        # Use first album's thumb as artist thumb
+        for album in self._albums:
+            if album.thumb:
+                self.thumb = album.thumb
+                break
+
+    def albums(self):
+        return self._albums
+
+
+class SoulSyncClient:
+    """Filesystem-based media server client for standalone SoulSync operation.
+
+    Scans the Transfer folder recursively, reads audio file tags, and
+    returns artist/album/track objects in the same format as the
+    Plex/Jellyfin/Navidrome clients. Designed as a drop-in replacement
+    for the DatabaseUpdateWorker.
+    """
+
+    def __init__(self):
+        from config.settings import config_manager
+        self._config_manager = config_manager
+        self._transfer_path = ''
+        self._progress_callback = None
+        self._cache = None  # Cached scan result
+        self._cache_time = 0
+        self._cache_ttl = 300  # 5 minute cache
+        self._last_scan_time = None
+        self._reload_config()
+
+    def _reload_config(self):
+        transfer = self._config_manager.get('soulseek.transfer_path', './Transfer')
+        # Docker path resolution
+        if os.path.exists('/.dockerenv') and len(transfer) >= 3 and transfer[1] == ':':
+            drive = transfer[0].lower()
+            rest = transfer[2:].replace('\\', '/')
+            transfer = f"/host/mnt/{drive}{rest}"
+        self._transfer_path = transfer
+
+    def reload_config(self):
+        self._reload_config()
+        self._cache = None
+
+    def ensure_connection(self) -> bool:
+        self._reload_config()
+        return os.path.isdir(self._transfer_path)
+
+    def is_connected(self) -> bool:
+        return os.path.isdir(self._transfer_path)
+
+    def set_progress_callback(self, callback: Callable):
+        self._progress_callback = callback
+
+    def clear_cache(self):
+        self._cache = None
+        self._cache_time = 0
+
+    def get_cache_stats(self) -> Dict[str, int]:
+        if not self._cache:
+            return {'artists': 0, 'albums': 0, 'tracks': 0}
+        return {
+            'artists': len(self._cache),
+            'albums': sum(len(a.albums()) for a in self._cache),
+            'tracks': sum(sum(len(alb.tracks()) for alb in a.albums()) for a in self._cache),
+        }
+
+    def _emit_progress(self, msg: str):
+        if self._progress_callback:
+            try:
+                self._progress_callback(msg)
+            except Exception:
+                pass
+
+    # ── Core Scanning ──
+
+    def _scan_transfer(self, since_mtime: float = 0) -> List[SoulSyncArtist]:
+        """Scan the Transfer folder and build artist/album/track hierarchy."""
+        if not os.path.isdir(self._transfer_path):
+            logger.warning(f"Transfer path not found: {self._transfer_path}")
+            return []
+
+        self._emit_progress(f"Scanning {self._transfer_path}...")
+        logger.info(f"[SoulSync] Scanning Transfer folder: {self._transfer_path}")
+
+        # Walk filesystem and collect all audio files with tags
+        file_entries = []  # (file_path, tags)
+        scanned = 0
+
+        for root, dirs, files in os.walk(self._transfer_path):
+            for filename in files:
+                ext = os.path.splitext(filename)[1].lower()
+                if ext not in AUDIO_EXTENSIONS:
+                    continue
+
+                file_path = os.path.join(root, filename)
+
+                # Incremental: skip files older than since_mtime
+                if since_mtime > 0:
+                    try:
+                        if os.path.getmtime(file_path) < since_mtime:
+                            continue
+                    except OSError:
+                        continue
+
+                tags = _read_tags(file_path)
+                file_entries.append((file_path, tags))
+                scanned += 1
+
+                if scanned % 100 == 0:
+                    self._emit_progress(f"Reading tags: {scanned} files...")
+
+        logger.info(f"[SoulSync] Found {len(file_entries)} audio files")
+        self._emit_progress(f"Found {len(file_entries)} audio files, building library...")
+
+        # Group by artist → album
+        # Key: (artist_name_lower) → { album_name_lower → [(file_path, tags)] }
+        artist_map: Dict[str, Dict[str, List]] = {}
+        artist_names: Dict[str, str] = {}  # lower → canonical name
+
+        for file_path, tags in file_entries:
+            # Prefer album artist, fall back to track artist, then folder name
+            artist_name = tags['album_artist'] or tags['artist']
+            if not artist_name:
+                # Try to extract from folder structure (Transfer/Artist/Album/track)
+                rel = os.path.relpath(file_path, self._transfer_path).replace('\\', '/')
+                parts = rel.split('/')
+                if len(parts) >= 3:
+                    artist_name = parts[0]
+                elif len(parts) >= 2:
+                    artist_name = parts[0]
+                else:
+                    artist_name = 'Unknown Artist'
+
+            album_name = tags['album']
+            if not album_name:
+                # Try folder name
+                album_dir = os.path.basename(os.path.dirname(file_path))
+                if album_dir and album_dir != os.path.basename(self._transfer_path):
+                    album_name = album_dir
+                else:
+                    album_name = tags['title'] or 'Unknown Album'
+
+            a_key = artist_name.lower().strip()
+            al_key = album_name.lower().strip()
+
+            if a_key not in artist_map:
+                artist_map[a_key] = {}
+                artist_names[a_key] = artist_name
+            if al_key not in artist_map[a_key]:
+                artist_map[a_key][al_key] = []
+
+            artist_map[a_key][al_key].append((file_path, tags))
+
+        # Build object hierarchy
+        artists = []
+        for a_key, albums_dict in artist_map.items():
+            canonical_artist = artist_names[a_key]
+            album_objects = []
+
+            for al_key, track_entries in albums_dict.items():
+                # Get canonical album name from first track
+                canonical_album = track_entries[0][1]['album'] or al_key
+                year = None
+                for _, t in track_entries:
+                    if t['year']:
+                        try:
+                            year = int(t['year'])
+                        except ValueError:
+                            pass
+                        break
+
+                # Build tracks
+                track_objects = []
+                for fp, tg in sorted(track_entries, key=lambda x: (x[1]['disc_number'], x[1]['track_number'])):
+                    track_objects.append(SoulSyncTrack(fp, tg))
+
+                album_key = f"{canonical_artist}::{canonical_album}"
+                album_obj = SoulSyncAlbum(album_key, canonical_album, year, track_list=track_objects)
+
+                # Link tracks back to album
+                for t in track_objects:
+                    t._album_ref = album_obj
+
+                album_objects.append(album_obj)
+
+            artist_obj = SoulSyncArtist(canonical_artist, canonical_artist, album_objects)
+
+            # Link albums and tracks back to artist
+            for album in album_objects:
+                album._artist_ref = artist_obj
+                for track in album.tracks():
+                    track._artist_ref = artist_obj
+
+            artists.append(artist_obj)
+
+        logger.info(f"[SoulSync] Built library: {len(artists)} artists, "
+                     f"{sum(len(a.albums()) for a in artists)} albums, "
+                     f"{sum(sum(len(al.tracks()) for al in a.albums()) for a in artists)} tracks")
+
+        return artists
+
+    def _get_cached_scan(self) -> List[SoulSyncArtist]:
+        """Return cached scan or perform a new one."""
+        import time
+        now = time.time()
+        if self._cache and (now - self._cache_time) < self._cache_ttl:
+            return self._cache
+        self._cache = self._scan_transfer()
+        self._cache_time = now
+        self._last_scan_time = datetime.now().isoformat()
+        return self._cache
+
+    # ── Public Interface (matches Plex/Jellyfin/Navidrome) ──
+
+    def get_all_artists(self) -> List[SoulSyncArtist]:
+        """Get all artists from the Transfer folder."""
+        return self._get_cached_scan()
+
+    def get_all_artist_ids(self) -> Set[str]:
+        """Get all artist IDs for removal detection."""
+        return {a.ratingKey for a in self._get_cached_scan()}
+
+    def get_all_album_ids(self) -> Set[str]:
+        """Get all album IDs for removal detection."""
+        ids = set()
+        for artist in self._get_cached_scan():
+            for album in artist.albums():
+                ids.add(album.ratingKey)
+        return ids
+
+    def get_recently_added_albums(self, max_results: int = 400) -> List[SoulSyncAlbum]:
+        """Get recently added/modified albums (for incremental scan)."""
+        import time
+        # Use last scan time or default to 7 days ago
+        since = 0
+        if self._last_scan_time:
+            try:
+                since = datetime.fromisoformat(self._last_scan_time).timestamp()
+            except (ValueError, TypeError):
+                pass
+        if since == 0:
+            since = time.time() - (7 * 86400)  # 7 days ago
+
+        # Scan only recent files
+        artists = self._scan_transfer(since_mtime=since)
+        all_albums = []
+        for artist in artists:
+            all_albums.extend(artist.albums())
+
+        # Sort by most recent first
+        all_albums.sort(key=lambda a: a.addedAt, reverse=True)
+        return all_albums[:max_results]
+
+    def get_recently_updated_albums(self, max_results: int = 400) -> List[SoulSyncAlbum]:
+        """Alias for get_recently_added_albums (filesystem has no update concept)."""
+        return self.get_recently_added_albums(max_results)
+
+    def get_recently_added_tracks(self, max_results: int = 400) -> List[SoulSyncTrack]:
+        """Get recently added tracks."""
+        albums = self.get_recently_added_albums(max_results * 2)
+        all_tracks = []
+        for album in albums:
+            all_tracks.extend(album.tracks())
+        all_tracks.sort(key=lambda t: t.addedAt, reverse=True)
+        return all_tracks[:max_results]
+
+    def get_recently_updated_tracks(self, max_results: int = 400) -> List[SoulSyncTrack]:
+        return self.get_recently_added_tracks(max_results)

core/web_scan_manager.py

@@ -52,7 +52,8 @@ class WebScanManager:
             server_client_map = {
                 'jellyfin': 'jellyfin_client',
                 'navidrome': 'navidrome_client',
-                'plex': 'plex_client'
+                'plex': 'plex_client',
+                'soulsync': 'soulsync_library_client',
             }
 
             # Try to get the configured active server

web_server.py

@@ -332,7 +332,7 @@ def _make_context_key(username, filename):
 # Each client is initialized independently so one failure doesn't take down everything.
 # Previously, a single exception set ALL clients to None, breaking the entire app.
 print("Initializing SoulSync services for Web UI...")
-spotify_client = plex_client = jellyfin_client = navidrome_client = soulseek_client = tidal_client = matching_engine = sync_service = web_scan_manager = None
+spotify_client = plex_client = jellyfin_client = navidrome_client = soulsync_library_client = soulseek_client = tidal_client = matching_engine = sync_service = web_scan_manager = None
 
 try:
     spotify_client = SpotifyClient()
@@ -358,6 +358,13 @@ try:
 except Exception as e:
     print(f"  Navidrome client failed to initialize: {e}")
 
+try:
+    from core.soulsync_client import SoulSyncClient
+    soulsync_library_client = SoulSyncClient()
+    print("  SoulSync library client initialized")
+except Exception as e:
+    print(f"  SoulSync library client failed to initialize: {e}")
+
 try:
     soulseek_client = DownloadOrchestrator()
     print("  Download orchestrator initialized")
@@ -402,7 +409,8 @@ try:
     media_clients = {
         'plex_client': plex_client,
         'jellyfin_client': jellyfin_client,
-        'navidrome_client': navidrome_client
+        'navidrome_client': navidrome_client,
+        'soulsync_library_client': soulsync_library_client,
     }
     web_scan_manager = WebScanManager(media_clients, delay_seconds=60)
     print("  Web scan manager initialized")
@@ -2132,6 +2140,166 @@ def _record_download_provenance(context):
         pass  # Non-critical, never block download flow
 
 
+def _record_soulsync_library_entry(context, spotify_artist, album_info):
+    """Write artist/album/track to library DB after successful download/import.
+
+    Only runs when active server is 'soulsync' (standalone mode). Creates
+    DB records with server_source='soulsync' and pre-populates enrichment
+    IDs so enrichment workers don't need to re-discover them.
+    """
+    try:
+        active_server = config_manager.get_active_media_server()
+        if active_server != 'soulsync':
+            return
+
+        final_path = context.get('_final_processed_path')
+        if not final_path:
+            return
+
+        spotify_album = context.get('spotify_album', {}) or {}
+        track_info = context.get('track_info', {}) or {}
+        original_search = context.get('original_search_result', {}) or {}
+
+        artist_name = (spotify_artist or {}).get('name', '')
+        if not artist_name:
+            artist_name = original_search.get('spotify_clean_artist', '') or original_search.get('artist', '')
+        if not artist_name or artist_name in ('Unknown', 'Unknown Artist'):
+            return
+
+        album_name = ''
+        if album_info and isinstance(album_info, dict):
+            album_name = album_info.get('album_name', '')
+        if not album_name:
+            album_name = spotify_album.get('name', '') or original_search.get('album', '')
+        if not album_name:
+            album_name = track_info.get('name', 'Unknown')
+
+        track_name = original_search.get('spotify_clean_title', '') or track_info.get('name', '') or original_search.get('title', '')
+        track_number = (track_info.get('track_number') or (album_info.get('track_number') if isinstance(album_info, dict) else None)) or 1
+        duration_ms = track_info.get('duration_ms', 0) or 0
+
+        year = None
+        release_date = spotify_album.get('release_date', '')
+        if release_date and len(release_date) >= 4:
+            try:
+                year = int(release_date[:4])
+            except ValueError:
+                pass
+
+        image_url = spotify_album.get('image_url', '')
+        if not image_url:
+            images = spotify_album.get('images', [])
+            if images and isinstance(images, list) and len(images) > 0:
+                img = images[0]
+                image_url = img.get('url', '') if isinstance(img, dict) else str(img)
+
+        # Enrichment IDs from context — saves enrichment workers from re-discovering
+        spotify_artist_id = (spotify_artist or {}).get('id', '')
+        if spotify_artist_id in ('auto_import', 'from_sync_modal', 'explicit_artist', ''):
+            spotify_artist_id = ''
+        spotify_album_id = spotify_album.get('id', '')
+        if spotify_album_id in ('from_sync_modal', 'explicit_album', ''):
+            spotify_album_id = ''
+        spotify_track_id = track_info.get('id', '') or original_search.get('id', '')
+
+        genres = (spotify_artist or {}).get('genres', [])
+        genres_json = json.dumps(genres) if genres else ''
+
+        bitrate = 0
+        try:
+            from mutagen import File as MutagenFile
+            audio = MutagenFile(final_path)
+            if audio and hasattr(audio, 'info') and audio.info and hasattr(audio.info, 'bitrate'):
+                bitrate = int(audio.info.bitrate / 1000) if audio.info.bitrate else 0
+        except Exception:
+            pass
+
+        import hashlib
+        def _sid(text):
+            return str(abs(int(hashlib.md5(text.encode('utf-8', errors='replace')).hexdigest(), 16)) % (10 ** 9))
+
+        artist_id = _sid(artist_name.lower().strip())
+        album_id = _sid(f"{artist_name}::{album_name}".lower().strip())
+        track_id = _sid(final_path)
+        total_tracks = spotify_album.get('total_tracks', 0) or 0
+
+        db = get_database()
+        with db._get_connection() as conn:
+            cursor = conn.cursor()
+
+            # ── Artist: find existing or create ──
+            # Check any source (not just soulsync) to avoid ID collisions
+            cursor.execute("SELECT id, server_source FROM artists WHERE id = ?", (artist_id,))
+            existing_by_id = cursor.fetchone()
+            if existing_by_id:
+                # ID exists — reuse it (may be from another server source)
+                artist_id = existing_by_id[0]
+            else:
+                # Check by name across all sources
+                cursor.execute("SELECT id FROM artists WHERE name COLLATE NOCASE = ? LIMIT 1", (artist_name,))
+                existing_by_name = cursor.fetchone()
+                if existing_by_name:
+                    artist_id = existing_by_name[0]
+                else:
+                    cursor.execute("""
+                        INSERT INTO artists (id, name, genres, thumb_url, server_source, created_at, updated_at)
+                        VALUES (?, ?, ?, ?, 'soulsync', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
+                    """, (artist_id, artist_name, genres_json, image_url))
+                    if spotify_artist_id:
+                        try:
+                            cursor.execute("UPDATE artists SET spotify_artist_id = ? WHERE id = ? AND (spotify_artist_id IS NULL OR spotify_artist_id = '')",
+                                           (spotify_artist_id, artist_id))
+                        except Exception:
+                            pass
+
+            # ── Album: find existing or create ──
+            cursor.execute("SELECT id FROM albums WHERE id = ?", (album_id,))
+            existing_album_by_id = cursor.fetchone()
+            if existing_album_by_id:
+                album_id = existing_album_by_id[0]
+            else:
+                # Check by title + artist across all sources
+                cursor.execute("SELECT id FROM albums WHERE title COLLATE NOCASE = ? AND artist_id = ? LIMIT 1",
+                               (album_name, artist_id))
+                existing_album_by_name = cursor.fetchone()
+                if existing_album_by_name:
+                    album_id = existing_album_by_name[0]
+                else:
+                    cursor.execute("""
+                        INSERT INTO albums (id, artist_id, title, year, thumb_url, genres, track_count,
+                                            server_source, created_at, updated_at)
+                        VALUES (?, ?, ?, ?, ?, ?, ?, 'soulsync', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
+                    """, (album_id, artist_id, album_name, year, image_url, genres_json, total_tracks))
+                    if spotify_album_id:
+                        try:
+                            cursor.execute("UPDATE albums SET spotify_album_id = ? WHERE id = ? AND (spotify_album_id IS NULL OR spotify_album_id = '')",
+                                           (spotify_album_id, album_id))
+                        except Exception:
+                            pass
+
+            # ── Track ──
+            cursor.execute("SELECT id FROM tracks WHERE file_path = ?", (final_path,))
+            if not cursor.fetchone():
+                cursor.execute("""
+                    INSERT INTO tracks (id, album_id, artist_id, title, track_number,
+                                        duration, file_path, bitrate, server_source,
+                                        created_at, updated_at)
+                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'soulsync', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
+                """, (track_id, album_id, artist_id, track_name, track_number,
+                      duration_ms, final_path, bitrate))
+                if spotify_track_id:
+                    try:
+                        cursor.execute("UPDATE tracks SET spotify_track_id = ? WHERE id = ?", (spotify_track_id, track_id))
+                    except Exception:
+                        pass
+
+            conn.commit()
+            print(f"[SoulSync Library] Added: {artist_name} / {album_name} / {track_name}")
+
+    except Exception as e:
+        logger.debug(f"[SoulSync Library] Non-critical error: {e}")
+
+
 # --- Register Public REST API Blueprint (v1) ---
 try:
     from api import create_api_blueprint, limiter
@@ -4786,6 +4954,9 @@ def get_status():
             elif active_server == "navidrome" and navidrome_client:
                 # Use existing instance
                 media_server_status = navidrome_client.is_connected()
+            elif active_server == "soulsync":
+                # Standalone mode — always connected if Transfer folder exists
+                media_server_status = soulsync_library_client.is_connected() if soulsync_library_client else False
             media_server_response_time = (time.time() - media_server_start) * 1000
             _status_cache['media_server'] = {
                 'connected': media_server_status,
@@ -6621,7 +6792,7 @@ def test_connection_endpoint():
             _status_cache['spotify']['source'] = _get_metadata_fallback_source()
             _status_cache_timestamps['spotify'] = current_time
             print("Updated Spotify status cache after successful test")
-        elif service in ['plex', 'jellyfin', 'navidrome']:
+        elif service in ['plex', 'jellyfin', 'navidrome', 'soulsync']:
             _status_cache['media_server']['connected'] = True
             _status_cache['media_server']['type'] = service
             _status_cache_timestamps['media_server'] = current_time
@@ -6671,7 +6842,7 @@ def test_dashboard_connection_endpoint():
             _status_cache['spotify']['source'] = _get_metadata_fallback_source()
             _status_cache_timestamps['spotify'] = current_time
             print("Updated Spotify status cache after successful dashboard test")
-        elif service in ['plex', 'jellyfin', 'navidrome']:
+        elif service in ['plex', 'jellyfin', 'navidrome', 'soulsync']:
             _status_cache['media_server']['connected'] = True
             _status_cache['media_server']['type'] = service
             _status_cache_timestamps['media_server'] = current_time
@@ -21522,6 +21693,7 @@ def _post_process_matched_download(context_key, context, file_path):
         _emit_track_downloaded(context)
         _record_library_history_download(context)
         _record_download_provenance(context)
+        _record_soulsync_library_entry(context, spotify_artist, album_info)
 
         # RETAG DATA CAPTURE: Record completed album/single downloads for retag tool
         try:
@@ -24272,6 +24444,8 @@ def _run_db_update_task(full_refresh, server_type):
         media_client = jellyfin_client
     elif server_type == "navidrome":
         media_client = navidrome_client
+    elif server_type == "soulsync":
+        media_client = soulsync_library_client
 
     if not media_client:
         _db_update_error_callback(f"Media client for '{server_type}' not available.")
@@ -24318,6 +24492,10 @@ def _run_deep_scan_task(server_type):
         media_client = jellyfin_client
     elif server_type == "navidrome":
         media_client = navidrome_client
+    elif server_type == "soulsync":
+        media_client = soulsync_library_client
+        if media_client:
+            media_client.clear_cache()  # Force fresh scan for deep scan
 
     if not media_client:
         _db_update_error_callback(f"Media client for '{server_type}' not available.")