Auto-import: fall through to other metadata sources when primary returns no match

Discord report: 16 Bandcamp indie albums sat in staging because
auto-import couldn't identify them, but the manual search bar at the
bottom of the Import Music tab found the same albums fine. Trace:
`_search_metadata_source` only queried `get_primary_source()` — single
source, no fallback. Meanwhile `search_import_albums` (manual search bar)
already iterated `get_source_priority(get_primary_source())` and broke
on the first source with results. Asymmetric behavior, same album: manual
worked, auto-import didn't.

Fix: lift `_search_metadata_source` to use the same source-chain pattern.
Try primary first; if it returns nothing OR scores below the 0.4
threshold, fall through to the next source in priority order. First
source producing a strong-enough match wins. Result dict carries the
`source` that actually matched (not the primary name) so downstream
`_match_tracks` calls the right client. Defensive per-source try/except
so a rate-limited or auth-failed source doesn't abort the chain.
Unconfigured sources (client=None) silently skipped.

Cin-shape lift: scoring math extracted to pure `_score_album_search_result`
helper so the weight tweaks (album 50% / artist 20% / track-count 30%)
are pinned at the function boundary, independent of the orchestrator
(per-source iteration, exception containment, threshold check). Weight
constants exposed at module level (`_ALBUM_NAME_WEIGHT`,
`_ARTIST_NAME_WEIGHT`, `_TRACK_COUNT_WEIGHT`) — greppable, bumpable in
one place. Pre-extraction these were magic numbers inline.

27 new tests:
- 9 integration tests in `test_auto_import_multi_source_fallback.py`:
  primary-success path unchanged (no fallback fires, only primary client
  called), primary-empty falls through, primary-weak-score falls through,
  first fallback success stops the chain (no wasted API calls on
  remaining sources), all-sources-fail returns None, per-source
  exception contained, unconfigured-source skipped, result `source`
  field reflects winning source, `identification_confidence` from
  winning source.
- 18 helper tests in `test_album_search_scoring.py`: weights sum to
  1.0, album weight dominant (invariant pin), perfect-match returns 1.0,
  per-component contribution (album / artist / track-count), Bandcamp
  vs streaming track-count mismatch (7-files vs 4-tracks case still
  scores ~0.87 above threshold), zero-track-count and zero-file
  guards, huge-mismatch non-negative guard, list-of-strings artist
  shape, missing `.name` / `.artists` / `None` total_tracks edge cases.

Backwards compatible: single-source users see no change — chain just
has one entry. Existing test `test_search_metadata_source_extracts_artist_id_from_dict_artist`
needed one extra patch line for `get_source_priority`.

Full pytest sweep: 2754 passed.

core/auto_import_worker.py

@@ -199,6 +199,53 @@ def _quality_rank(ext: str) -> int:
     return ranks.get(ext.lower(), 1)
 
 
+# Weight constants for `_score_album_search_result` — exposed at module
+# level so they're greppable + bumpable in one place. Pre-fix these were
+# magic numbers inline.
+_ALBUM_NAME_WEIGHT = 0.5    # title fuzzy similarity
+_ARTIST_NAME_WEIGHT = 0.2   # primary artist fuzzy similarity (skipped when target is empty)
+_TRACK_COUNT_WEIGHT = 0.3   # how close the source's track count is to the file count
+
+
+def _score_album_search_result(album_result, target_album: str,
+                               target_artist: Optional[str],
+                               file_count: int) -> float:
+    """Pure scoring helper for `_search_metadata_source`.
+
+    Weights how well an `album_result` from a metadata source's
+    `search_albums` matches the search inputs. Returns float in [0.0, 1.0].
+    Pre-extraction this lived inline in the loop body; lifting it out
+    lets the weight math be pinned independently of the orchestrator
+    (per-source iteration, exception containment, threshold check).
+
+    `album_result` is expected to expose:
+      - `.name` (str)
+      - `.artists` (list of dict-like with 'name', optional 'id') or list[str]
+      - `.total_tracks` (int, optional)
+    """
+    score = 0.0
+
+    # Album name similarity (default 50%)
+    name = getattr(album_result, 'name', '') or ''
+    score += _similarity(target_album, name) * _ALBUM_NAME_WEIGHT
+
+    # Artist similarity (default 20%) — only when target_artist provided
+    if target_artist:
+        artists = getattr(album_result, 'artists', None) or []
+        r_artist = artists[0] if artists else ''
+        if isinstance(r_artist, dict):
+            r_artist = r_artist.get('name', '')
+        score += _similarity(target_artist, str(r_artist)) * _ARTIST_NAME_WEIGHT
+
+    # Track count match (default 30%) — only when both sides have a count
+    r_tracks = getattr(album_result, 'total_tracks', 0) or 0
+    if r_tracks > 0 and file_count > 0:
+        count_ratio = 1.0 - abs(r_tracks - file_count) / max(r_tracks, file_count)
+        score += max(0.0, count_ratio) * _TRACK_COUNT_WEIGHT
+
+    return score
+
+
 class AutoImportWorker:
     """Background worker that watches the staging folder and auto-imports music.
 
@@ -1260,87 +1307,120 @@ class AutoImportWorker:
     def _search_metadata_source(self, artist: Optional[str], album: str,
                                  method: str, candidate: FolderCandidate,
                                  query: str = None) -> Optional[Dict]:
-        """Search the active metadata source for an album match."""
+        """Search configured metadata sources for an album match.
+
+        Iterates `get_source_priority(get_primary_source())` so primary
+        is tried first and the rest are tried as fallback. Returns the
+        FIRST source whose best result clears the 0.4 score threshold.
+
+        Pre-fix this only queried the primary, which meant indie/niche
+        albums missing from the user's primary (e.g. Bandcamp releases
+        not on Spotify) failed auto-import even when manual search
+        could find them on Tidal/Deezer. The manual search bar at the
+        bottom of the Import tab already iterates the full source
+        chain via `search_import_albums` — this aligns auto-import
+        with that behavior.
+        """
         try:
-            from core.metadata_service import get_primary_source, get_client_for_source
-
-            source = get_primary_source()
-            client = get_client_for_source(source)
-            if not client or not hasattr(client, 'search_albums'):
-                return None
+            from core.metadata_service import (
+                get_primary_source,
+                get_source_priority,
+                get_client_for_source,
+            )
 
+            primary_source = get_primary_source()
+            source_chain = get_source_priority(primary_source)
             search_query = query or (f"{artist} {album}" if artist else album)
-            results = client.search_albums(search_query, limit=5)
-            if not results:
-                return None
 
-            # Score each result
-            best_result = None
-            best_score = 0
+            for source in source_chain:
+                client = get_client_for_source(source)
+                if not client or not hasattr(client, 'search_albums'):
+                    continue
 
-            for r in results:
-                score = 0
-                # Album name similarity (50%)
-                score += _similarity(album, r.name) * 0.5
-                # Artist similarity (20%)
-                if artist:
-                    r_artist = r.artists[0] if hasattr(r, 'artists') and r.artists else ''
-                    if isinstance(r_artist, dict):
-                        r_artist = r_artist.get('name', '')
-                    score += _similarity(artist, str(r_artist)) * 0.2
-                # Track count match (30%)
-                r_tracks = getattr(r, 'total_tracks', 0) or 0
+                try:
+                    results = client.search_albums(search_query, limit=5)
+                except Exception as e:
+                    # Per-source failures (rate limit, auth, transient HTTP)
+                    # shouldn't abort the fallback chain. Log + continue.
+                    logger.debug(
+                        f"Auto-import: search_albums failed on {source}: {e}"
+                    )
+                    continue
+
+                if not results:
+                    continue
+
+                # Score each result via the pure helper. Helper is
+                # tested independently in
+                # `tests/imports/test_album_search_scoring.py` so the
+                # weight math is pinned at the function boundary, not
+                # through the orchestrator path.
                 file_count = len(candidate.audio_files)
-                if r_tracks > 0 and file_count > 0:
-                    count_ratio = 1.0 - abs(r_tracks - file_count) / max(r_tracks, file_count)
-                    score += max(0, count_ratio) * 0.3
+                best_result = None
+                best_score = 0.0
+                for r in results:
+                    score = _score_album_search_result(r, album, artist, file_count)
+                    if score > best_score:
+                        best_score = score
+                        best_result = r
 
-                if score > best_score:
-                    best_score = score
-                    best_result = r
+                if not best_result or best_score < 0.4:
+                    # Primary returned weak/no match — fall through to next source
+                    if source != primary_source:
+                        logger.debug(
+                            f"Auto-import: {source} best score {best_score:.2f} "
+                            f"below threshold for '{album}', trying next source"
+                        )
+                    continue
 
-            if not best_result or best_score < 0.4:
-                return None
+                # Get image
+                image_url = ''
+                if hasattr(best_result, 'image_url'):
+                    image_url = best_result.image_url or ''
+                elif hasattr(best_result, 'images') and best_result.images:
+                    img = best_result.images[0]
+                    image_url = img.get('url', '') if isinstance(img, dict) else str(img)
 
-            # Get image
-            image_url = ''
-            if hasattr(best_result, 'image_url'):
-                image_url = best_result.image_url or ''
-            elif hasattr(best_result, 'images') and best_result.images:
-                img = best_result.images[0]
-                image_url = img.get('url', '') if isinstance(img, dict) else str(img)
+                r_artist = ''
+                r_artist_id = ''
+                if hasattr(best_result, 'artists') and best_result.artists:
+                    a = best_result.artists[0]
+                    if isinstance(a, dict):
+                        r_artist = a.get('name', str(a))
+                        # Surface the metadata-source artist ID so the
+                        # standalone-library write can land it on the right
+                        # `<source>_artist_id` column. Without this the
+                        # artists row gets created but with NULL on the
+                        # source-id, and watchlist scans can't recognise
+                        # the artist as already in library by stable ID.
+                        r_artist_id = str(a.get('id', '') or '')
+                    else:
+                        r_artist = str(a)
 
-            r_artist = ''
-            r_artist_id = ''
-            if hasattr(best_result, 'artists') and best_result.artists:
-                a = best_result.artists[0]
-                if isinstance(a, dict):
-                    r_artist = a.get('name', str(a))
-                    # Surface the metadata-source artist ID so the
-                    # standalone-library write can land it on the right
-                    # `<source>_artist_id` column. Without this the
-                    # artists row gets created but with NULL on the
-                    # source-id, and watchlist scans can't recognise
-                    # the artist as already in library by stable ID.
-                    r_artist_id = str(a.get('id', '') or '')
-                else:
-                    r_artist = str(a)
+                # Get release date
+                release_date = getattr(best_result, 'release_date', '') or ''
 
-            # Get release date
-            release_date = getattr(best_result, 'release_date', '') or ''
+                if source != primary_source:
+                    logger.info(
+                        f"Auto-import: identified '{album}' via fallback "
+                        f"source {source!r} (score {best_score:.2f}, primary "
+                        f"{primary_source!r} returned nothing usable)"
+                    )
 
-            return {
-                'album_id': best_result.id,
-                'album_name': best_result.name,
-                'artist_name': r_artist or artist or '',
-                'artist_id': r_artist_id,
-                'image_url': image_url,
-                'release_date': release_date,
-                'total_tracks': getattr(best_result, 'total_tracks', 0),
-                'source': source,
-                'method': method,
-                'identification_confidence': best_score,
-            }
+                return {
+                    'album_id': best_result.id,
+                    'album_name': best_result.name,
+                    'artist_name': r_artist or artist or '',
+                    'artist_id': r_artist_id,
+                    'image_url': image_url,
+                    'release_date': release_date,
+                    'total_tracks': getattr(best_result, 'total_tracks', 0),
+                    'source': source,
+                    'method': method,
+                    'identification_confidence': best_score,
+                }
+
+            return None
 
         except Exception as e:
             logger.debug(f"Metadata search failed for '{album}': {e}")

tests/imports/test_album_search_scoring.py

@@ -0,0 +1,217 @@
+"""Pin `_score_album_search_result` weight math.
+
+Helper extracted from the inline scoring block inside
+`_search_metadata_source` (auto-import album identification). Lifting
+it to a pure function lets each weight be tested in isolation
+without mocking the full source-chain orchestrator.
+
+Weights (constants in `core.auto_import_worker`):
+  - `_ALBUM_NAME_WEIGHT` = 0.5
+  - `_ARTIST_NAME_WEIGHT` = 0.2 (skipped when target_artist falsy)
+  - `_TRACK_COUNT_WEIGHT` = 0.3 (skipped when either side has 0 tracks)
+
+Maximum score is 1.0 when all three components match perfectly. The
+0.4 threshold in the orchestrator means a result needs at least one
+strong signal plus a partial second — pure track-count match alone
+(0.3) is below threshold.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+from core.auto_import_worker import (
+    _ALBUM_NAME_WEIGHT,
+    _ARTIST_NAME_WEIGHT,
+    _TRACK_COUNT_WEIGHT,
+    _score_album_search_result,
+)
+
+
+def _result(name: str, artist_name: str = "", total_tracks: int = 0):
+    """Minimal album-result stub matching the shape `search_albums`
+    returns. `artists` is the list-of-dicts shape every adapter uses."""
+    artists = [{"name": artist_name}] if artist_name else []
+    return SimpleNamespace(name=name, artists=artists, total_tracks=total_tracks)
+
+
+# ---------------------------------------------------------------------------
+# Component weights — pinned at the boundary
+# ---------------------------------------------------------------------------
+
+
+class TestWeightConstants:
+    def test_weights_sum_to_one(self):
+        """Total weight budget = 1.0. If a weight is bumped without
+        adjusting another, perfect-match score drifts above/below 1.0
+        and the 0.4 threshold semantics shift silently."""
+        total = _ALBUM_NAME_WEIGHT + _ARTIST_NAME_WEIGHT + _TRACK_COUNT_WEIGHT
+        assert total == pytest.approx(1.0, abs=1e-9), (
+            f"Weights must sum to 1.0; got {total}"
+        )
+
+    def test_album_weight_is_dominant(self):
+        """Album name has 50% — strongest signal. If artist or track
+        count weight ever exceeds album weight, the matching semantics
+        flip (e.g. a wrong-album-right-count result could outscore
+        a right-album-wrong-count one). Pin so a future weight tweak
+        doesn't break this invariant."""
+        assert _ALBUM_NAME_WEIGHT > _ARTIST_NAME_WEIGHT
+        assert _ALBUM_NAME_WEIGHT > _TRACK_COUNT_WEIGHT
+
+
+# ---------------------------------------------------------------------------
+# Perfect match → 1.0
+# ---------------------------------------------------------------------------
+
+
+class TestPerfectMatch:
+    def test_all_three_perfect_returns_one(self):
+        r = _result("Test Album", "Test Artist", total_tracks=10)
+        score = _score_album_search_result(r, "Test Album", "Test Artist", file_count=10)
+        assert score == pytest.approx(1.0, abs=1e-9)
+
+
+# ---------------------------------------------------------------------------
+# Album name component
+# ---------------------------------------------------------------------------
+
+
+class TestAlbumNameWeight:
+    def test_exact_album_match_no_other_signals(self):
+        """Album name matches perfectly but no artist provided and
+        no track count match — score is just 50%."""
+        r = _result("Test Album", total_tracks=0)
+        score = _score_album_search_result(r, "Test Album", target_artist=None, file_count=0)
+        assert score == pytest.approx(_ALBUM_NAME_WEIGHT, abs=1e-9)
+
+    def test_completely_different_album_zero_album_component(self):
+        r = _result("Totally Different", total_tracks=0)
+        score = _score_album_search_result(r, "Test Album", target_artist=None, file_count=0)
+        # SequenceMatcher would return some small non-zero similarity even
+        # for fully different strings, so just verify it's well below 0.5
+        assert score < _ALBUM_NAME_WEIGHT * 0.5
+
+
+# ---------------------------------------------------------------------------
+# Artist component
+# ---------------------------------------------------------------------------
+
+
+class TestArtistWeight:
+    def test_artist_match_adds_full_artist_weight(self):
+        r = _result("Album", "Artist", total_tracks=0)
+        with_artist = _score_album_search_result(r, "Album", "Artist", file_count=0)
+        without_artist = _score_album_search_result(r, "Album", None, file_count=0)
+        # Difference = artist weight
+        assert with_artist - without_artist == pytest.approx(_ARTIST_NAME_WEIGHT, abs=1e-9)
+
+    def test_target_artist_none_skips_artist_component(self):
+        """When target_artist is falsy (None / empty), artist weight
+        contributes zero — not a penalty, not a bonus. Lets album-
+        only searches (e.g. from a folder name with no artist info)
+        still hit the threshold via album + track count alone."""
+        r = _result("Album", "WrongArtist", total_tracks=10)
+        score = _score_album_search_result(r, "Album", None, file_count=10)
+        # Album + track count perfect = 0.5 + 0.3 = 0.8 (artist weight skipped)
+        assert score == pytest.approx(_ALBUM_NAME_WEIGHT + _TRACK_COUNT_WEIGHT, abs=1e-9)
+
+    def test_string_artist_not_dict_still_scored(self):
+        """Some adapters return `artists` as list-of-strings instead
+        of list-of-dicts. Helper must handle both shapes."""
+        r = SimpleNamespace(name="Album", artists=["Just A String"], total_tracks=0)
+        score_string = _score_album_search_result(r, "Album", "Just A String", 0)
+        assert score_string >= _ALBUM_NAME_WEIGHT + _ARTIST_NAME_WEIGHT - 0.05
+
+    def test_empty_artists_list_treats_as_no_artist(self):
+        r = _result("Album", artist_name="", total_tracks=0)  # no artist
+        score = _score_album_search_result(r, "Album", "Some Target", file_count=0)
+        # Artist sim against empty string is 0 → no artist weight contribution
+        assert score == pytest.approx(_ALBUM_NAME_WEIGHT, abs=1e-9)
+
+
+# ---------------------------------------------------------------------------
+# Track count component
+# ---------------------------------------------------------------------------
+
+
+class TestTrackCountWeight:
+    def test_exact_track_count_match_full_weight(self):
+        r = _result("Album", total_tracks=10)
+        score = _score_album_search_result(r, "Album", None, file_count=10)
+        # Album (perfect) + track count (perfect) — no artist component
+        assert score == pytest.approx(_ALBUM_NAME_WEIGHT + _TRACK_COUNT_WEIGHT, abs=1e-9)
+
+    def test_off_by_one_track_count_near_full(self):
+        """1 track off out of 10 → ratio = 1.0 - 1/10 = 0.9 → 0.9 * 0.3 = 0.27"""
+        r = _result("Album", total_tracks=10)
+        score = _score_album_search_result(r, "Album", None, file_count=9)
+        expected = _ALBUM_NAME_WEIGHT + (0.9 * _TRACK_COUNT_WEIGHT)
+        assert score == pytest.approx(expected, abs=1e-9)
+
+    def test_bandcamp_vs_streaming_track_count_mismatch(self):
+        """Reporter's exact case: Bandcamp 7-track album vs Spotify
+        4-track release. Track count ratio = 1.0 - 3/7 = ~0.571.
+        With perfect album + artist match, total = 0.5 + 0.2 + 0.171
+        = 0.871 → comfortably above the 0.4 threshold so the album
+        still identifies despite the count mismatch."""
+        r = _result("Work in Progress", "Godly the Ruler", total_tracks=4)
+        score = _score_album_search_result(r, "Work in Progress", "Godly the Ruler", file_count=7)
+        assert score > 0.4, (
+            f"Bandcamp-vs-streaming case must still pass threshold; got {score:.3f}"
+        )
+        # Sanity bound — score should land around 0.87
+        assert 0.85 < score < 0.90
+
+    def test_zero_track_count_from_source_skips_track_component(self):
+        """Some search responses don't include total_tracks. Helper
+        must not penalize — just skip the track-count component."""
+        r = _result("Album", total_tracks=0)
+        score = _score_album_search_result(r, "Album", None, file_count=10)
+        # Only album component contributes
+        assert score == pytest.approx(_ALBUM_NAME_WEIGHT, abs=1e-9)
+
+    def test_zero_file_count_skips_track_component(self):
+        """Defensive: candidate has 0 files (somehow). Don't divide
+        by zero or skew the score."""
+        r = _result("Album", total_tracks=10)
+        score = _score_album_search_result(r, "Album", None, file_count=0)
+        assert score == pytest.approx(_ALBUM_NAME_WEIGHT, abs=1e-9)
+
+    def test_huge_mismatch_track_count_no_negative_contribution(self):
+        """File count 1, source track count 100 → ratio = 1 - 99/100
+        = 0.01. Tiny but non-negative. `max(0, ...)` guards against
+        any future formula change introducing a negative."""
+        r = _result("Album", total_tracks=100)
+        score = _score_album_search_result(r, "Album", None, file_count=1)
+        assert score >= _ALBUM_NAME_WEIGHT  # at least the album component
+
+
+# ---------------------------------------------------------------------------
+# Edge cases — defensive
+# ---------------------------------------------------------------------------
+
+
+class TestEdgeCases:
+    def test_album_result_without_name_attribute(self):
+        """If the result somehow lacks `.name` (unusual adapter
+        return), helper falls back to '' and scores 0 album sim."""
+        r = SimpleNamespace(artists=[], total_tracks=0)  # no `.name`
+        score = _score_album_search_result(r, "Test Album", None, file_count=0)
+        assert score == 0.0
+
+    def test_album_result_without_artists_attribute(self):
+        """If `.artists` is missing, treat as empty list."""
+        r = SimpleNamespace(name="Album", total_tracks=0)  # no .artists
+        score = _score_album_search_result(r, "Album", "Target Artist", file_count=0)
+        # Album matches perfectly; artist sim against missing is 0
+        assert score == pytest.approx(_ALBUM_NAME_WEIGHT, abs=1e-9)
+
+    def test_album_result_with_none_total_tracks(self):
+        """Some adapters return None for missing total_tracks instead
+        of 0. `getattr(..., 'total_tracks', 0) or 0` should handle it."""
+        r = SimpleNamespace(name="Album", artists=[], total_tracks=None)
+        score = _score_album_search_result(r, "Album", None, file_count=10)
+        assert score == pytest.approx(_ALBUM_NAME_WEIGHT, abs=1e-9)

tests/imports/test_auto_import_context_shape.py

@@ -480,6 +480,7 @@ def test_search_metadata_source_extracts_artist_id_from_dict_artist():
 
     worker = AutoImportWorker(database=MagicMock(), process_callback=lambda *a, **k: None)
     with patch("core.metadata_service.get_primary_source", return_value="spotify"), \
+         patch("core.metadata_service.get_source_priority", return_value=["spotify"]), \
          patch("core.metadata_service.get_client_for_source", return_value=fake_client):
         result = worker._search_metadata_source(
             "Test Artist", "Test Album", "tags", candidate,

tests/imports/test_auto_import_multi_source_fallback.py

@@ -0,0 +1,365 @@
+"""Pin auto-import multi-source fallback for album identification.
+
+Discord report (mushy, paraphrased): 16 Bandcamp albums sat in staging
+because auto-import couldn't identify them. Manual search at the
+bottom of the Import Music tab found the same albums fine via Tidal
+or Deezer — the user's primary metadata source (Spotify) just didn't
+have them.
+
+Root cause: `_search_metadata_source` only queried the primary source.
+The manual `search_import_albums` path already iterates the full
+`get_source_priority(get_primary_source())` chain and breaks on first
+source that returns results. This brings auto-import to parity.
+
+Fix semantics (option C — "primary first, fall through on weak"):
+  - Try primary source first
+  - Score result; if best ≥ 0.4 → return with that source
+  - Otherwise fall through to next source in priority order
+  - First source that produces a result above threshold wins
+  - Returns None only if ALL sources fail / score below threshold
+
+Tests pin:
+  - Primary success path unchanged (returns primary result, no fallback fired)
+  - Primary returns nothing → fallback fires to next source
+  - Primary scores below threshold → fallback fires
+  - First fallback succeeds → no further sources queried
+  - All sources fail → None
+  - Per-source exception is contained (doesn't abort the chain)
+  - Result `source` field reflects WHICH source actually matched
+  - `identification_confidence` is the score from the winning source
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from core.auto_import_worker import AutoImportWorker, FolderCandidate
+
+
+def _make_album(name: str, artist_name: str, total_tracks: int,
+                album_id: str = "alb-id", artist_id: str = "art-id"):
+    """Build a fake album result matching `search_albums` return shape."""
+    return SimpleNamespace(
+        id=album_id,
+        name=name,
+        artists=[{"id": artist_id, "name": artist_name}],
+        total_tracks=total_tracks,
+        image_url="https://img.example/cover.jpg",
+        release_date="2024-01-01",
+    )
+
+
+def _make_worker():
+    """Bare AutoImportWorker bypassing __init__ side effects."""
+    return AutoImportWorker(database=MagicMock(), process_callback=lambda *a, **k: None)
+
+
+def _make_candidate(file_count: int = 7, name: str = "TestAlbum"):
+    """Folder candidate with N files (no actual disk reads)."""
+    return FolderCandidate(
+        path=f"/staging/{name}",
+        name=name,
+        audio_files=[f"/staging/{name}/{i:02d}.flac" for i in range(1, file_count + 1)],
+    )
+
+
+# ---------------------------------------------------------------------------
+# Primary success path — fallback never fires
+# ---------------------------------------------------------------------------
+
+
+class TestPrimarySuccess:
+    def test_primary_returns_strong_match_no_fallback(self):
+        """Pre-fix behavior preserved: if primary scores above 0.4,
+        return its result and don't touch other sources."""
+        worker = _make_worker()
+        candidate = _make_candidate(file_count=10)
+
+        spotify_client = MagicMock()
+        spotify_client.search_albums.return_value = [
+            _make_album("Test Album", "Test Artist", total_tracks=10),
+        ]
+        tidal_client = MagicMock()  # should NEVER be called
+
+        def client_dispatch(source, **kwargs):
+            return {"spotify": spotify_client, "tidal": tidal_client}.get(source)
+
+        with patch("core.metadata_service.get_primary_source", return_value="spotify"), \
+             patch("core.metadata_service.get_source_priority",
+                   return_value=["spotify", "tidal", "deezer"]), \
+             patch("core.metadata_service.get_client_for_source",
+                   side_effect=client_dispatch):
+            result = worker._search_metadata_source(
+                "Test Artist", "Test Album", "tags", candidate,
+            )
+
+        assert result is not None
+        assert result["source"] == "spotify"
+        assert result["album_name"] == "Test Album"
+        spotify_client.search_albums.assert_called_once()
+        tidal_client.search_albums.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# Primary fails — fallback fires
+# ---------------------------------------------------------------------------
+
+
+class TestFallbackOnNoResults:
+    def test_primary_empty_falls_through_to_next_source(self):
+        """Reporter's exact case: Spotify doesn't have the Bandcamp
+        indie album. Tidal does. Auto-import must find it via Tidal."""
+        worker = _make_worker()
+        candidate = _make_candidate(file_count=7)
+
+        spotify_client = MagicMock()
+        spotify_client.search_albums.return_value = []  # not on Spotify
+        tidal_client = MagicMock()
+        tidal_client.search_albums.return_value = [
+            _make_album("Work in Progress", "Godly the Ruler", total_tracks=7,
+                        album_id="tidal-alb-1", artist_id="tidal-art-1"),
+        ]
+
+        def client_dispatch(source, **kwargs):
+            return {"spotify": spotify_client, "tidal": tidal_client}.get(source)
+
+        with patch("core.metadata_service.get_primary_source", return_value="spotify"), \
+             patch("core.metadata_service.get_source_priority",
+                   return_value=["spotify", "tidal", "deezer"]), \
+             patch("core.metadata_service.get_client_for_source",
+                   side_effect=client_dispatch):
+            result = worker._search_metadata_source(
+                "Godly the Ruler", "Work in Progress", "tags", candidate,
+            )
+
+        assert result is not None
+        assert result["source"] == "tidal", "Result must carry the source that actually matched"
+        assert result["album_id"] == "tidal-alb-1"
+        assert result["artist_id"] == "tidal-art-1"
+        spotify_client.search_albums.assert_called_once()
+        tidal_client.search_albums.assert_called_once()
+
+
+class TestFallbackOnWeakScore:
+    def test_primary_below_threshold_falls_through(self):
+        """Primary returns results but none score above 0.4 (e.g.
+        wrong-album false-matches). Fall through to next source for
+        a stronger match."""
+        worker = _make_worker()
+        candidate = _make_candidate(file_count=7)
+
+        spotify_client = MagicMock()
+        # Wrong album — name barely matches, no artist match, wrong track count
+        spotify_client.search_albums.return_value = [
+            _make_album("Different", "Wrong Artist", total_tracks=2),
+        ]
+        deezer_client = MagicMock()
+        deezer_client.search_albums.return_value = [
+            _make_album("Work in Progress", "Godly the Ruler", total_tracks=7),
+        ]
+
+        def client_dispatch(source, **kwargs):
+            return {"spotify": spotify_client, "deezer": deezer_client}.get(source)
+
+        with patch("core.metadata_service.get_primary_source", return_value="spotify"), \
+             patch("core.metadata_service.get_source_priority",
+                   return_value=["spotify", "deezer"]), \
+             patch("core.metadata_service.get_client_for_source",
+                   side_effect=client_dispatch):
+            result = worker._search_metadata_source(
+                "Godly the Ruler", "Work in Progress", "tags", candidate,
+            )
+
+        assert result is not None
+        assert result["source"] == "deezer"
+        assert result["album_name"] == "Work in Progress"
+        # Both clients called — primary returned weak, fallback picked up
+        spotify_client.search_albums.assert_called_once()
+        deezer_client.search_albums.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# Chain semantics
+# ---------------------------------------------------------------------------
+
+
+class TestChainSemantics:
+    def test_first_fallback_success_stops_chain(self):
+        """When fallback succeeds, no further sources are queried.
+        Don't waste API budget on Deezer if Tidal already gave us a
+        strong result."""
+        worker = _make_worker()
+        candidate = _make_candidate(file_count=10)
+
+        spotify_client = MagicMock()
+        spotify_client.search_albums.return_value = []
+        tidal_client = MagicMock()
+        tidal_client.search_albums.return_value = [
+            _make_album("Test", "Artist", total_tracks=10),
+        ]
+        deezer_client = MagicMock()  # should NEVER be called
+
+        def client_dispatch(source, **kwargs):
+            return {"spotify": spotify_client,
+                    "tidal": tidal_client,
+                    "deezer": deezer_client}.get(source)
+
+        with patch("core.metadata_service.get_primary_source", return_value="spotify"), \
+             patch("core.metadata_service.get_source_priority",
+                   return_value=["spotify", "tidal", "deezer"]), \
+             patch("core.metadata_service.get_client_for_source",
+                   side_effect=client_dispatch):
+            result = worker._search_metadata_source("Artist", "Test", "tags", candidate)
+
+        assert result is not None
+        assert result["source"] == "tidal"
+        deezer_client.search_albums.assert_not_called()
+
+    def test_all_sources_fail_returns_none(self):
+        """If every source returns nothing or scores below threshold,
+        the whole search returns None (caller proceeds to next
+        identification strategy)."""
+        worker = _make_worker()
+        candidate = _make_candidate(file_count=7)
+
+        empty_client = MagicMock()
+        empty_client.search_albums.return_value = []
+
+        with patch("core.metadata_service.get_primary_source", return_value="spotify"), \
+             patch("core.metadata_service.get_source_priority",
+                   return_value=["spotify", "tidal", "deezer"]), \
+             patch("core.metadata_service.get_client_for_source",
+                   return_value=empty_client):
+            result = worker._search_metadata_source(
+                "Unknown Artist", "Nonexistent Album", "tags", candidate,
+            )
+
+        assert result is None
+        # All 3 sources got queried
+        assert empty_client.search_albums.call_count == 3
+
+    def test_per_source_exception_does_not_abort_chain(self):
+        """If one source raises (rate limit, auth, transient HTTP),
+        the chain continues to the next source instead of aborting
+        the whole identification attempt."""
+        worker = _make_worker()
+        candidate = _make_candidate(file_count=10)
+
+        spotify_client = MagicMock()
+        spotify_client.search_albums.side_effect = RuntimeError("rate limit")
+        tidal_client = MagicMock()
+        tidal_client.search_albums.return_value = [
+            _make_album("Test", "Artist", total_tracks=10),
+        ]
+
+        def client_dispatch(source, **kwargs):
+            return {"spotify": spotify_client, "tidal": tidal_client}.get(source)
+
+        with patch("core.metadata_service.get_primary_source", return_value="spotify"), \
+             patch("core.metadata_service.get_source_priority",
+                   return_value=["spotify", "tidal"]), \
+             patch("core.metadata_service.get_client_for_source",
+                   side_effect=client_dispatch):
+            result = worker._search_metadata_source("Artist", "Test", "tags", candidate)
+
+        assert result is not None
+        assert result["source"] == "tidal", (
+            "Primary exception must not block the fallback chain"
+        )
+
+    def test_unconfigured_source_skipped_gracefully(self):
+        """If `get_client_for_source` returns None for a source
+        (user hasn't configured it), skip and continue."""
+        worker = _make_worker()
+        candidate = _make_candidate(file_count=10)
+
+        tidal_client = MagicMock()
+        tidal_client.search_albums.return_value = [
+            _make_album("Test", "Artist", total_tracks=10),
+        ]
+
+        # Spotify returns None (no client configured); Tidal works
+        def client_dispatch(source, **kwargs):
+            if source == "spotify":
+                return None
+            return {"tidal": tidal_client}.get(source)
+
+        with patch("core.metadata_service.get_primary_source", return_value="spotify"), \
+             patch("core.metadata_service.get_source_priority",
+                   return_value=["spotify", "tidal"]), \
+             patch("core.metadata_service.get_client_for_source",
+                   side_effect=client_dispatch):
+            result = worker._search_metadata_source("Artist", "Test", "tags", candidate)
+
+        assert result is not None
+        assert result["source"] == "tidal"
+
+
+# ---------------------------------------------------------------------------
+# Result shape preservation
+# ---------------------------------------------------------------------------
+
+
+class TestResultShape:
+    def test_result_carries_correct_source_for_downstream_match(self):
+        """`_match_tracks` reads `identification['source']` to know
+        which client to ask for the album's tracklist. Result MUST
+        carry the source that actually matched, not the primary
+        source name."""
+        worker = _make_worker()
+        candidate = _make_candidate(file_count=8)
+
+        spotify_client = MagicMock()
+        spotify_client.search_albums.return_value = []
+        deezer_client = MagicMock()
+        deezer_client.search_albums.return_value = [
+            _make_album("Test", "Artist", total_tracks=8, album_id="dz-123"),
+        ]
+
+        def client_dispatch(source, **kwargs):
+            return {"spotify": spotify_client, "deezer": deezer_client}.get(source)
+
+        with patch("core.metadata_service.get_primary_source", return_value="spotify"), \
+             patch("core.metadata_service.get_source_priority",
+                   return_value=["spotify", "deezer"]), \
+             patch("core.metadata_service.get_client_for_source",
+                   side_effect=client_dispatch):
+            result = worker._search_metadata_source("Artist", "Test", "tags", candidate)
+
+        assert result["source"] == "deezer"
+        assert result["album_id"] == "dz-123", (
+            "Album ID must be the Deezer ID so _match_tracks queries "
+            "Deezer's get_album with the right ID format"
+        )
+
+    def test_identification_confidence_reflects_winning_source(self):
+        """`identification_confidence` is used in the overall-confidence
+        formula and the 0.9 / 0.7 cascade thresholds. It must be the
+        score from the source that actually matched."""
+        worker = _make_worker()
+        candidate = _make_candidate(file_count=10)
+
+        spotify_client = MagicMock()
+        spotify_client.search_albums.return_value = []
+        # Perfect match on Tidal — all 3 weights at max → score = 1.0
+        tidal_client = MagicMock()
+        tidal_client.search_albums.return_value = [
+            _make_album("Test Album", "Test Artist", total_tracks=10),
+        ]
+
+        def client_dispatch(source, **kwargs):
+            return {"spotify": spotify_client, "tidal": tidal_client}.get(source)
+
+        with patch("core.metadata_service.get_primary_source", return_value="spotify"), \
+             patch("core.metadata_service.get_source_priority",
+                   return_value=["spotify", "tidal"]), \
+             patch("core.metadata_service.get_client_for_source",
+                   side_effect=client_dispatch):
+            result = worker._search_metadata_source(
+                "Test Artist", "Test Album", "tags", candidate,
+            )
+
+        assert result["identification_confidence"] == pytest.approx(1.0, abs=0.01)

webui/static/helper.js

@@ -3416,6 +3416,7 @@ const WHATS_NEW = {
     '2.5.1': [
         // --- post-release patch work on the 2.5.1 line — entries hidden by _getLatestWhatsNewVersion until the build version bumps ---
         { date: 'Unreleased — 2.5.1 patch work' },
+        { title: 'Auto-Import: Falls Through To Other Metadata Sources When Primary Has No Match', desc: 'discord report (mushy): 16 bandcamp indie albums sat in staging because auto-import couldn\'t identify them. manual search at the bottom of the import music tab found the same albums fine — they just weren\'t on the user\'s primary metadata source (spotify) but existed on tidal/deezer. trace: `_search_metadata_source` in `core/auto_import_worker.py` only queried `get_primary_source()` — single source, no fallback. meanwhile `search_import_albums` (the manual search bar at the bottom of the tab) already iterated the full `get_source_priority(get_primary_source())` chain and broke on first source with results. asymmetric behavior — manual search worked, auto-import didn\'t, same album. fix: lift auto-import to use the same source-chain pattern. try primary first; if it returns nothing OR scores below the 0.4 threshold, fall through to next source in priority order. first source that produces a strong-enough match wins. result dict carries the `source` that actually matched (not the primary name), so downstream `_match_tracks` calls the right client to fetch the album\'s tracklist. defensive per-source try/except so a rate-limited or auth-failed source doesn\'t abort the chain. unconfigured sources (client=None) silently skipped. scoring math lifted to pure helper `_score_album_search_result` so weight tweaks (album 50% / artist 20% / track-count 30%) are pinned at the function boundary independent of the orchestrator. weight constants exposed at module level (`_ALBUM_NAME_WEIGHT`, `_ARTIST_NAME_WEIGHT`, `_TRACK_COUNT_WEIGHT`) — greppable, bumpable in one place. 9 integration tests + 18 scoring-helper tests. integration tests pin: primary-success path unchanged (no fallback fires, only primary client called), primary-empty falls through to next source, primary-weak-score falls through, first fallback success stops the chain (no wasted api calls on remaining sources), all-sources-fail returns None, per-source exception contained, unconfigured-source skipped gracefully, result `source` field reflects winning source, `identification_confidence` from winning source. backwards compatible — single-source users see no change (chain just has one entry).', page: 'import' },
         { title: 'Multi-Artist Tag Settings Now Actually Work (artist_separator + feat_in_title + write_multi_artist)', desc: 'three settings on settings → metadata → tags were partially or completely unimplemented. (1) `write_multi_artist` only worked because of a never-populated `_artists_list` field — `core/metadata/source.py` built `metadata["artist"]` as a hardcoded ", "-joined string but never assigned `metadata["_artists_list"]`, so `core/metadata/enrichment.py:114` always saw an empty list and silently no-op\'d the multi-value tag write. (2) `artist_separator` (default ", ") was referenced in the UI + settings.js save path but ZERO python code read the value — every multi-artist track ended up with hardcoded ", " regardless of what the user picked. (3) `feat_in_title` (when true: pull featured artists into the title as " (feat. X, Y)" and leave only primary in the ARTIST tag — picard convention) had no implementation at all. fix in source.py: populate `_artists_list` from the search response\'s artists array, then build the ARTIST string per the user\'s settings — primary-only when feat_in_title is on (with featured names appended to title; double-append guarded for source titles that already include "feat."), else joined with the configured separator. fix in enrichment.py id3 path: writing TPE1 twice (single-string then list) was overwriting the configured separator. now keeps TPE1 as the display string and writes a separate `TXXX:Artists` frame for the multi-value list (picard convention). vorbis path was already correct (separate "artist" + "artists" keys). deezer-specific upgrade path: deezer\'s `/search` endpoint only returns the primary artist — full contributors live on `/track/<id>`. when source==deezer AND the search response had a single artist AND a track_id is available, enrichment now fetches the per-track endpoint and upgrades the artists list before tag-write. one extra API call per affected deezer track (skipped when search already returned multiple). spotify, tidal, itunes search responses already include all artists so they\'re unaffected. 29 new tests pin: `_artists_list` populated for multi/single/no-artist cases, separator drives ARTIST string (default + custom), single-artist case unaffected by either setting, feat_in_title pulls featured to title + leaves primary in ARTIST, feat_in_title no-op for single artist, double-append guard recognizes 9 source-title variants ("(feat. X)", "(Feat. X)", "(FEAT X)", "(feat X)", "(Featuring X)", "[feat. X]", "ft. X", "(ft X)", "FT. X"), word-boundary regex doesn\'t false-match substrings ("Aftermath" still gets the append), combined-settings precedence (feat_in_title wins over separator for ARTIST string but `_artists_list` carries everyone for the multi-value tag), deezer upgrade fires only when search returned single artist + track_id available, no upgrade for non-deezer sources, upgrade failure falls through to search-result list, no false-positive when /track/<id> confirms single artist.', page: 'settings' },
         { title: 'AudioDB Enrichment: Track Worker No Longer Stuck In Infinite Retry Loop', desc: 'github issue #553: audiodb track enrichment "stuck" — constant requests, no progress, only error log was a 10s read-timeout from `lookup_track_by_id` repeating against the same track. trace: when an entity already has `audiodb_id` populated (from manual match or earlier scan) but `audiodb_match_status` is NULL, the worker tries a direct ID lookup. if it fails (returns None on timeout — audiodb\'s `track.php` endpoint is slow, 10s timeouts common), the prior code logged "preserving manual match" and returned WITHOUT marking status. row stayed NULL → queue picked it up next tick → tried direct lookup → timed out → returned → infinite loop. fix: (1) when direct lookup fails (None or exception), mark `audiodb_match_status="error"` so the queue\'s NULL-status filter stops re-picking the row on every tick. preserves the existing `audiodb_id` (no fallback to name-search guess that would overwrite a manual match). (2) extended the retry-after-cutoff queue priorities (4/5/6) to include `\'error\'` rows alongside `\'not_found\'` — same `retry_days=30` window. transient audiodb outages still recover automatically; permanently-broken IDs eventually get re-attempted once a month. only triggered for entities in the inconsistent state of `audiodb_id` set + `match_status` NULL — happy path and already-matched/already-not-found rows unchanged. 5 new tests pin: lookup-returns-none marks error (no infinite loop), lookup-raises-exception marks error, lookup-success preserves happy path, error-row-past-cutoff gets re-picked, error-row-within-cutoff stays skipped.', page: 'tools' },
         { title: 'Docker: Container No Longer Restart-Loops On Bind-Mounted Staging Folder', desc: 'after pulling latest, the container refused to start. logs showed `mkdir: cannot create directory \'/app/Staging\': Permission denied`. cause traced back to the 2026-05-08 image-bloat fix (commit 70e1750) which changed the Dockerfile from `chown -R /app` to a scoped chown on specific subdirs (the recursive chown was duplicating the whole /app tree into a new layer and ballooning image size). side effect: `/app` itself went from soulsync:soulsync to root:root (Docker WORKDIR default), AND `/app/Staging` was left out of both the Dockerfile mkdir + chown list and only created at runtime by the entrypoint script. on rootless Docker / Podman where in-container "root" maps to a host UID, the entrypoint mkdir on `/app/Staging` could fail with EACCES depending on the bind-mount path\'s host ownership — `set -e` then aborted the script and the container restart-looped. fix: (1) Dockerfile now pre-bakes `/app/Staging` into the image alongside the other runtime mount points (mkdir + scoped chown) so the entrypoint mkdir is a guaranteed no-op even when bind-mount perms are weird. (2) entrypoint mkdir + chown both have `|| true` now so any future bind-mount permission quirk surfaces as a log line, not a restart loop. (3) new writability audit at the end of entrypoint setup — `gosu soulsync test -w` on every bind-mountable dir, logs a loud warning with the exact `chown` command to run on the host if perms mismatch the configured PUID/PGID. catches the underlying bind-mount perm issue that the restart-loop fix would otherwise mask (container starts, but auto-import / downloads write into unwritable dirs and fail silently). zero behavior change for users whose containers were already starting fine; defensive against the rootless/podman config that broke after the image-bloat refactor.', page: 'tools' },