Merge pull request #801 from nick2000713/feature/retry-next-candidate-on-mismatch

Downloads: complete retry overhaul,  exhaustive multi-source retry, MusicBrainz kanji fix, version-mismatch last-resort fallback

config/settings.py

@@ -494,6 +494,19 @@ class ConfigManager:
                 "album_bundle_poll_interval_seconds": 2.0,
                 "album_bundle_timeout_seconds": 6 * 60 * 60,    # 6 hours
             },
+            "post_processing": {
+                # When a download is quarantined (AcoustID mismatch, integrity /
+                # duration failure), retry the next-best candidate instead of
+                # failing outright. Default off — opt-in.
+                "retry_next_candidate_on_mismatch": False,
+                # Opt-in exhaustive retry: budget retries PER SOURCE so every
+                # source (Soulseek, then HiFi/Tidal/…) gets its own attempts
+                # before the track gives up. Default off (single global cap).
+                "retry_exhaustive": False,
+                # Retries per search query per source in exhaustive mode. The
+                # per-source budget is query_count × this value.
+                "retries_per_query": 5,
+            },
             "tidal_download": {
                 "quality": "lossless",  # Options: "low", "high", "lossless", "hires"
                 "session": {

core/downloads/monitor.py

@@ -37,11 +37,228 @@ missing_download_executor = None
 download_orchestrator = None
 _RELEASE_SOURCE_NAMES = frozenset(('torrent', 'usenet'))
 
+# Hard ceiling on automatic next-candidate retries after a download was
+# quarantined (AcoustID mismatch / integrity / duration). The natural
+# terminator is used_sources exhaustion — once every candidate the worker can
+# find has been tried, attempt_download_with_candidates returns False and the
+# worker reports a clean failure. This cap is a safety net against a pathological
+# quarantine→retry→quarantine loop (e.g. a source that keeps returning fresh
+# wrong files).
+#
+# Default (non-exhaustive) mode uses this single global cap. The opt-in
+# exhaustive mode (post_processing.retry_exhaustive) instead budgets retries
+# PER SOURCE — see requeue_quarantined_task_for_retry.
+MAX_QUARANTINE_RETRIES = 5
+
+# Absolute runaway guard for exhaustive mode. Per-source budgets are already
+# finite (query_count × retries_per_query, and Soulseek peers all collapse to
+# one 'soulseek' bucket), but this ceiling caps the TOTAL retries across every
+# source so a misbehaving source-resolution can never loop forever.
+MAX_TOTAL_QUARANTINE_RETRIES = 100
+
+# Streaming plugins report their source name as the download's "username"
+# (see download_orchestrator._streaming_sources). Soulseek uses the peer name
+# instead, so anything not in this set is bucketed under 'soulseek' for the
+# per-source retry budget.
+_STREAMING_SOURCE_NAMES = frozenset((
+    'youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud', 'amazon',
+))
+
+
+def _resolve_download_source(username):
+    """Map a download's username to its logical source for per-source budgeting.
+
+    Streaming sources use the source name as username; Soulseek uses the peer
+    name, so every Soulseek peer collapses to a single 'soulseek' bucket.
+    """
+    if username and username in _STREAMING_SOURCE_NAMES:
+        return username
+    return 'soulseek'
+
+
+def _remaining_fallback_sources(exhausted):
+    """Sources in the configured hybrid chain that haven't exhausted their
+    per-source budget yet.
+
+    When a source spends its whole budget (exhaustive mode), the task switches
+    to the next source instead of failing — but only if there *is* another
+    source. Single-source mode has nothing to fall back to, so this returns
+    empty there (and when the orchestrator isn't wired). The returned list
+    drives both the give-up decision here and the worker's search-exclusion on
+    the next attempt (see task_worker: exhausted_download_sources).
+    """
+    orch = download_orchestrator
+    if orch is None or getattr(orch, 'mode', None) != 'hybrid':
+        return []
+    chain = getattr(orch, 'hybrid_order', None) or []
+    blocked = {str(s).lower() for s in exhausted}
+    return [s for s in chain if str(s).lower() not in blocked]
+
 
 def _download_id_key(download_id):
     return f"download_id::{download_id}" if download_id else None
 
 
+def requeue_quarantined_task_for_retry(task_id, batch_id, trigger):
+    """Re-queue a task whose download was just quarantined so the worker tries
+    the NEXT best candidate instead of failing outright.
+
+    Called from the post-processing verification wrapper when AcoustID
+    verification or the integrity/duration check quarantines a file. It mirrors
+    the monitor's transfer-error retry path: mark the bad source as used, clear
+    the stale download identity, reset the task to ``searching`` and resubmit
+    the download worker. Because ``used_sources`` is preserved across the
+    re-run, the worker skips the quarantined source and picks the next-best
+    candidate (see ``attempt_download_with_candidates``).
+
+    Returns True if a retry was queued — the caller must then NOT mark the task
+    failed or notify batch completion, since the task is going around again.
+    Returns False when no retry is possible (retry engine unwired, manual pick,
+    cancelled, or retry budget exhausted); the caller falls through to its
+    existing failure handling.
+    """
+    # Opt-out escape hatch — default on. Lets users restore the old
+    # quarantine-and-fail behaviour without a code change.
+    if not config_manager.get('post_processing.retry_next_candidate_on_mismatch', True):
+        return False
+
+    # Retry engine not wired (e.g. manual-import path that never started a
+    # download worker). Nothing to re-run.
+    if missing_download_executor is None or _download_track_worker is None:
+        return False
+
+    with tasks_lock:
+        task = download_tasks.get(task_id)
+        if not task:
+            return False
+        # The user explicitly picked this candidate via the candidates modal —
+        # honour their choice rather than silently swapping in another file.
+        # (Matches the monitor's transfer-retry guards.)
+        if task.get('_user_manual_pick'):
+            return False
+        if task.get('status') == 'cancelled':
+            return False
+
+        username = task.get('username')
+        filename = task.get('filename')
+        # No source identity means this wasn't a worker-dispatched download we
+        # can retry — without the "{username}_{filename}" key we can't flag the
+        # bad source as used, so a re-run could re-pick the same file and loop.
+        # Bail and let the caller fail it normally.
+        if not username or not filename:
+            return False
+
+        total_count = task.get('quarantine_retry_count', 0)
+
+        if config_manager.get('post_processing.retry_exhaustive', False):
+            # Exhaustive mode: a SEPARATE budget per source. The budget scales
+            # with the track's own query count (the worker generates a variable
+            # number of search queries per track) × the configured retries per
+            # query. Soulseek candidates are walked first (one per retry), then
+            # the worker's hybrid fallback moves to the next source — each source
+            # spending its own budget. The natural terminator (used_sources
+            # exhaustion → worker clean-fail) still ends most tracks well before
+            # any budget is reached; the budget is the per-source safety ceiling.
+            source = _resolve_download_source(username)
+            retries_per_query = config_manager.get('post_processing.retries_per_query', 5)
+            try:
+                retries_per_query = int(retries_per_query)
+            except (TypeError, ValueError):
+                retries_per_query = 5
+            if retries_per_query < 1:
+                retries_per_query = 1
+
+            query_count = task.get('query_count') or 1
+            if query_count < 1:
+                query_count = 1
+            budget = query_count * retries_per_query
+
+            counts = task.get('quarantine_retry_counts_by_source')
+            if not isinstance(counts, dict):
+                counts = {}
+            source_count = counts.get(source, 0)
+
+            if source_count >= budget:
+                # This source spent its whole budget. Rather than fail the
+                # track outright, mark the source exhausted and fall through to
+                # the next source in the hybrid chain (the worker excludes
+                # exhausted sources from its next search). Only give up once no
+                # fallback source remains — or the absolute ceiling trips.
+                exhausted = set(task.get('exhausted_download_sources') or ())
+                exhausted.add(source)
+                remaining = _remaining_fallback_sources(exhausted)
+                if not remaining:
+                    logger.warning(
+                        f"[Retry:{trigger}] Task {task_id} exhausted its retry "
+                        f"budget for source '{source}' ({source_count}/{budget}) "
+                        f"and no fallback source remains — giving up, marking failed"
+                    )
+                    return False
+                if total_count >= MAX_TOTAL_QUARANTINE_RETRIES:
+                    logger.warning(
+                        f"[Retry:{trigger}] Task {task_id} hit the absolute retry "
+                        f"ceiling ({MAX_TOTAL_QUARANTINE_RETRIES}) — giving up, "
+                        f"marking failed"
+                    )
+                    return False
+                task['exhausted_download_sources'] = exhausted
+                # Don't push this source's counter past its budget — it's done.
+                # The next source starts spending its own fresh budget when its
+                # first candidate fails verification.
+                attempt_desc = (
+                    f"source '{source}' budget spent ({source_count}/{budget}) "
+                    f"— switching sources (remaining: {', '.join(remaining)})"
+                )
+            else:
+                if total_count >= MAX_TOTAL_QUARANTINE_RETRIES:
+                    logger.warning(
+                        f"[Retry:{trigger}] Task {task_id} hit the absolute retry "
+                        f"ceiling ({MAX_TOTAL_QUARANTINE_RETRIES}) — giving up, "
+                        f"marking failed"
+                    )
+                    return False
+                counts[source] = source_count + 1
+                task['quarantine_retry_counts_by_source'] = counts
+                attempt_desc = f"source '{source}' {source_count + 1}/{budget}"
+        else:
+            # Default mode: a single global cap, conservative and predictable.
+            if total_count >= MAX_QUARANTINE_RETRIES:
+                logger.warning(
+                    f"[Retry:{trigger}] Task {task_id} hit the quarantine-retry cap "
+                    f"({MAX_QUARANTINE_RETRIES}) — giving up, marking failed"
+                )
+                return False
+            attempt_desc = f"{total_count + 1}/{MAX_QUARANTINE_RETRIES}"
+
+        # Mark the quarantined source as used so the re-run won't pick it again.
+        # Uses the same "{username}_{filename}" key the worker dedups against.
+        used_sources = task.get('used_sources', set())
+        used_sources.add(f"{username}_{filename}")
+        task['used_sources'] = used_sources
+
+        task['quarantine_retry_count'] = total_count + 1
+        # Flag the re-run as a quarantine retry so the worker walks the
+        # already-found candidates (cached-first) before re-searching — the
+        # connection was fine, the content was just wrong. Dead-connection /
+        # stuck retries (handled elsewhere in the monitor) deliberately do NOT
+        # set this, so they re-search fresh.
+        task['_quarantine_retry'] = True
+        # Drop the stale download identity + the prior attempt's quarantine link.
+        task.pop('download_id', None)
+        task.pop('username', None)
+        task.pop('filename', None)
+        task.pop('quarantine_entry_id', None)
+        task['status'] = 'searching'
+        task['status_change_time'] = time.time()
+
+    logger.info(
+        f"[Retry:{trigger}] Re-queuing task {task_id} for next-best candidate "
+        f"(attempt {attempt_desc})"
+    )
+    missing_download_executor.submit(_download_track_worker, task_id, batch_id)
+    return True
+
+
 def _is_release_task(task):
     ti = task.get('track_info') if isinstance(task.get('track_info'), dict) else {}
     username = task.get('username') or ti.get('username')

core/downloads/task_worker.py

@@ -31,6 +31,65 @@ from core.spotify_client import Track as SpotifyTrack
 logger = logging.getLogger(__name__)
 
 
+def _resolve_worker_source(username):
+    """Logical source bucket for a candidate's username (Soulseek peers all
+    collapse to 'soulseek'; streaming sources keep their name). Mirrors the
+    monitor's resolver — imported lazily to avoid an import cycle."""
+    try:
+        from core.downloads.monitor import _resolve_download_source
+        return _resolve_download_source(username)
+    except Exception:
+        return 'soulseek'
+
+
+def _cand_user_file(candidate):
+    """Read (username, filename) from a candidate that may be a TrackResult
+    object or a plain dict (tests / cached raw rows)."""
+    if isinstance(candidate, dict):
+        return candidate.get('username'), candidate.get('filename')
+    return getattr(candidate, 'username', None), getattr(candidate, 'filename', None)
+
+
+def _try_cached_candidates(task_id, batch_id, track, deps):
+    """Quarantine-retry fast path: attempt the already-found candidates before
+    re-searching anything.
+
+    When a verified-bad file is re-queued, the connection was fine (the file
+    downloaded, it was just the wrong/broken content) — so the next-best pick is
+    almost always already sitting in ``cached_candidates``. Walk those (skipping
+    sources already tried or budget-exhausted) and hand them to the normal
+    download path. Returns True if a download was started; False to fall through
+    to a fresh search (which only happens for a not-yet-searched source).
+    """
+    with tasks_lock:
+        task = download_tasks.get(task_id)
+        if not task:
+            return False
+        cached = list(task.get('cached_candidates') or [])
+        used = set(task.get('used_sources') or ())
+        exhausted = {str(s).lower() for s in (task.get('exhausted_download_sources') or ())}
+
+    remaining = []
+    for c in cached:
+        uname, fname = _cand_user_file(c)
+        if not uname or not fname:
+            continue
+        if f"{uname}_{fname}" in used:
+            continue
+        if _resolve_worker_source(uname).lower() in exhausted:
+            continue
+        remaining.append(c)
+
+    if not remaining:
+        return False
+
+    logger.info(
+        f"[Modal Worker] Quarantine retry: trying {len(remaining)} cached "
+        f"candidate(s) before re-searching (task {task_id})"
+    )
+    return deps.attempt_download_with_candidates(task_id, remaining, track, batch_id)
+
+
 def _private_album_bundle_staging_miss_reason(batch_id: Optional[str], deps: Any) -> Optional[str]:
     """Return a user-facing miss reason when per-track search should stop.
 
@@ -92,6 +151,7 @@ class TaskWorkerDeps:
     attempt_download_with_candidates: Callable    # (task_id, candidates, track, batch_id) -> bool
     on_download_completed: Callable               # (batch_id, task_id, success) -> None
     recover_worker_slot: Callable                 # (batch_id, task_id) -> None
+    try_version_mismatch_fallback: Optional[Callable] = None  # (title, artist, task_id, batch_id) -> bool
 
 
 def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorkerDeps) -> None:
@@ -206,6 +266,26 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke
                     download_tasks[task_id]['used_sources'] = set()
                 # Else: keep existing used_sources to avoid retrying same failed hosts
 
+        # Cached-first quarantine retry. The monitor sets ``_quarantine_retry``
+        # when a verified-bad file is re-queued; in that case we walk the
+        # already-found candidates before re-searching (the connection was fine,
+        # just the content was wrong). A NON-quarantine entry (fresh download, or
+        # the monitor's dead-connection/stuck retry) instead starts a new search
+        # generation: clear the searched-source memory so each source can be
+        # searched fresh again.
+        with tasks_lock:
+            _t = download_tasks.get(task_id, {})
+            is_quarantine_retry = bool(_t.pop('_quarantine_retry', False))
+            if not is_quarantine_retry:
+                _t.pop('searched_queries', None)
+        if is_quarantine_retry and _try_cached_candidates(task_id, batch_id, track, deps):
+            with tasks_lock:
+                used_filename = download_tasks.get(task_id, {}).get('filename')
+                used_username = download_tasks.get(task_id, {}).get('username')
+            if used_filename and used_username:
+                deps.store_batch_source(batch_id, used_username, used_filename)
+            return
+
         # 1. Generate multiple search queries (like GUI's generate_smart_search_queries)
         artist_name = track.artists[0] if track.artists else None
         track_name = track.name
@@ -277,12 +357,40 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke
                 seen.add(query.lower())
 
         search_queries = unique_queries
+        # Expose the query count so the quarantine-retry budget (exhaustive mode)
+        # can size each source's budget as query_count × retries_per_query.
+        with tasks_lock:
+            if task_id in download_tasks:
+                download_tasks[task_id]['query_count'] = len(search_queries)
         logger.info(f"[Modal Worker] Generated {len(search_queries)} smart search queries for '{track.name}': {search_queries}")
         logger.info(f"[Modal Worker] About to start search loop for task {task_id} (track: '{track.name}')")
 
         # 2. Sequential Query Search (matches GUI's start_search_worker_parallel logic)
         search_diagnostics = []  # Track what happened per query for detailed error messages
         all_raw_results = []  # Collect raw results across queries for candidate review modal
+        # Sources whose per-source quarantine-retry budget is spent (exhaustive
+        # mode). The monitor sets this when a source gives up; we exclude those
+        # sources from the hybrid search so the chain falls through to the next
+        # source instead of re-fetching the same exhausted one (e.g. Soulseek
+        # keeps returning fresh wrong peers — once its budget is gone, switch to
+        # HiFi/Tidal/…). See monitor.requeue_quarantined_task_for_retry.
+        #
+        # On a quarantine retry we do NOT exclude a source just because it was
+        # searched once: the first run only ran ONE query before starting a
+        # download, so the later queries (e.g. "artist + album") have never hit
+        # that source yet and may surface the correct upload. Instead we remember
+        # which QUERIES already ran (``searched_queries``) and skip re-running
+        # only those — their candidates are walked via the cached-first path
+        # above. The not-yet-searched queries still search the same source, so
+        # every query is exhausted per source before the chain switches sources.
+        # Fresh / dead-connection runs cleared searched_queries above, so they
+        # search everything again.
+        with tasks_lock:
+            _t = download_tasks.get(task_id, {})
+            _exhausted_sources = [str(s) for s in (_t.get('exhausted_download_sources') or ())]
+            _searched_queries = (
+                set(_t.get('searched_queries') or ()) if is_quarantine_retry else set()
+            )
         for query_index, query in enumerate(search_queries):
             # Cancellation check before each query
             with tasks_lock:
@@ -295,6 +403,17 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke
                     return
                 download_tasks[task_id]['current_query_index'] = query_index
 
+            # Cached-first: a query already run last generation has its candidates
+            # sitting in cache (walked above) — re-searching it is the wasteful
+            # repeat the cached-first design removes. Skip it; the not-yet-run
+            # queries below still search this source.
+            if is_quarantine_retry and query in _searched_queries:
+                logger.debug(
+                    f"[Modal Worker] Skipping already-searched query '{query}' "
+                    f"(candidates served from cache) for task {task_id}"
+                )
+                continue
+
             logger.debug(f"[Modal Worker] Query {query_index + 1}/{len(search_queries)}: '{query}'")
             logger.debug(f"About to call soulseek search for task {task_id}")
 
@@ -319,9 +438,13 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke
                         _exclude_for_hybrid_album = ['torrent', 'usenet']
                 except Exception as _exc_filter_err:
                     logger.debug("[Modal Worker] album-source-exclusion check failed: %s", _exc_filter_err)
+                # Fold in budget-exhausted sources (per-source quarantine retry).
+                _exclude_sources = list(_exhausted_sources)
+                if _exclude_for_hybrid_album:
+                    _exclude_sources.extend(_exclude_for_hybrid_album)
                 # Perform search with timeout
                 tracks_result, _ = deps.run_async(deps.download_orchestrator.search(
-                    query, timeout=30, exclude_sources=_exclude_for_hybrid_album,
+                    query, timeout=30, exclude_sources=_exclude_sources or None,
                 ))
                 logger.debug(f"Search completed for task {task_id}, got {len(tracks_result) if tracks_result else 0} results")
 
@@ -330,6 +453,16 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke
                     if task_id not in download_tasks:
                         logger.info(f"[Modal Worker] Task {task_id} was deleted after search returned")
                         return
+                    # Remember this query ran so a later quarantine retry skips
+                    # re-searching it (its candidates are walked via cached-first).
+                    # Recorded regardless of result count: re-running a query is
+                    # deterministic, so a query that returned nothing won't return
+                    # anything new next time either.
+                    _sq = download_tasks[task_id].get('searched_queries')
+                    if not isinstance(_sq, set):
+                        _sq = set()
+                    _sq.add(query)
+                    download_tasks[task_id]['searched_queries'] = _sq
                     if download_tasks[task_id]['status'] == 'cancelled':
                         logger.warning(f"[Modal Worker] Task {task_id} cancelled after search returned - ignoring results")
                         # Don't call _on_download_completed for cancelled tasks as it can stop monitoring
@@ -352,7 +485,9 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke
                                 logger.warning(f"[Modal Worker] Task {task_id} cancelled before processing candidates")
                                 # Don't call _on_download_completed for cancelled tasks as it can stop monitoring
                                 return
-                            # Store candidates for retry fallback (like GUI)
+                            # Store candidates for retry fallback (like GUI). A
+                            # later quarantine retry walks these via cached-first
+                            # and skips re-searching this query (searched_queries).
                             download_tasks[task_id]['cached_candidates'] = candidates
 
                         # Try to download with these candidates
@@ -414,7 +549,12 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke
                 # (which was definitely tried). If the first was skipped (unconfigured),
                 # the orchestrator would have tried the second — but trying it again is
                 # harmless (streaming sources return fast).
-                remaining_sources = [s for s in hybrid_order[1:] if s in source_clients and source_clients[s]]
+                _exhausted_lower = {s.lower() for s in _exhausted_sources}
+                remaining_sources = [
+                    s for s in hybrid_order[1:]
+                    if s in source_clients and source_clients[s]
+                    and s.lower() not in _exhausted_lower
+                ]
                 if remaining_sources:
                     logger.warning(f"[Hybrid Fallback] Primary source had no valid matches. Trying fallback sources: {remaining_sources}")
 
@@ -433,6 +573,9 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke
                             fb_candidates = deps.get_valid_candidates(fb_results, track, fb_query)
                             if fb_candidates:
                                 logger.warning(f"[Hybrid Fallback] {fallback_source} found {len(fb_candidates)} valid candidates!")
+                                with tasks_lock:
+                                    if task_id in download_tasks:
+                                        download_tasks[task_id]['cached_candidates'] = fb_candidates
                                 success = deps.attempt_download_with_candidates(task_id, fb_candidates, track, batch_id)
                                 if success:
                                     return
@@ -447,6 +590,15 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke
 
         # If we get here, all search queries and hybrid fallbacks failed
         logger.warning(f"[Modal Worker] No valid candidates found for '{track.name}' after trying all {len(search_queries)} queries.")
+
+        # Last-resort: quarantine retry with no new candidates — the retry search
+        # exhausted all sources.  If the setting is enabled, accept the best
+        # already-quarantined candidate rather than leaving the track missing.
+        if is_quarantine_retry and deps.try_version_mismatch_fallback:
+            _fallback_artist = track.artists[0] if track.artists else ''
+            if deps.try_version_mismatch_fallback(track.name, _fallback_artist, task_id, batch_id):
+                return  # fallback re-dispatched; batch completion handled by reprocess thread
+
         with tasks_lock:
             if task_id in download_tasks:
                 download_tasks[task_id]['status'] = 'not_found'

core/imports/pipeline.py

@@ -35,7 +35,12 @@ from core.imports.context import (
 from core.imports.file_integrity import check_audio_integrity, resolve_duration_tolerance
 from core.imports.filename import extract_track_number_from_filename
 from core.imports.guards import check_flac_bit_depth, move_to_quarantine
-from core.imports.quarantine import entry_id_from_quarantined_filename
+from core.imports.quarantine import (
+    approve_quarantine_entry,
+    entry_id_from_quarantined_filename,
+    list_quarantine_entries,
+)
+from core.imports.version_mismatch_fallback import try_accept_version_mismatch_fallback
 from core.imports.side_effects import (
     emit_track_downloaded,
     record_download_provenance,
@@ -109,6 +114,30 @@ def _mark_task_quarantined(context: dict, quarantine_path: str | None) -> None:
             download_tasks[task_id]['quarantine_entry_id'] = entry_id
 
 
+def _requeue_quarantined_task_for_retry(task_id, batch_id, trigger) -> bool:
+    """Ask the download monitor to re-run this task on its next-best candidate.
+
+    Thin lazy-import wrapper around
+    ``core.downloads.monitor.requeue_quarantined_task_for_retry``. Imported
+    lazily (and defensively) so the post-processing pipeline stays importable on
+    its own — the monitor's retry globals are wired by web_server at startup, and
+    manual-import callers that never started a download worker simply get False.
+    Returns True when a retry was queued (caller must not mark the task failed).
+    """
+    if not task_id:
+        return False
+    try:
+        from core.downloads.monitor import requeue_quarantined_task_for_retry
+    except Exception as exc:  # pragma: no cover - defensive
+        logger.debug(f"next-candidate retry unavailable ({trigger}): {exc}")
+        return False
+    try:
+        return requeue_quarantined_task_for_retry(task_id, batch_id, trigger)
+    except Exception as exc:  # pragma: no cover - defensive
+        logger.error(f"next-candidate retry failed ({trigger}): {exc}")
+        return False
+
+
 def import_rejection_reason(context: dict) -> str | None:
     """Human-readable reason if post-processing terminally rejected the file
     (quarantine or race-guard), else ``None`` for a clean import.
@@ -178,6 +207,19 @@ def post_process_matched_download(context_key, context, file_path, runtime, meta
                     f"{os.path.basename(existing_final)}"
                 )
                 return
+            # File was intentionally moved to quarantine by a concurrent/earlier
+            # post-process call — this is a stale duplicate dispatch, not a race.
+            # _mark_task_quarantined sets _quarantine_entry_id for every quarantine
+            # trigger (AcoustID, integrity, bit-depth).  The quarantine entry and
+            # its retry are already in flight; don't overwrite the task state with
+            # a spurious race-guard failure.
+            if context.get('_quarantine_entry_id'):
+                logger.debug(
+                    f"[Race Guard] Source gone but already quarantined (entry %s) — stale duplicate call, ignoring: "
+                    f"{os.path.basename(file_path)}",
+                    context['_quarantine_entry_id'],
+                )
+                return
             logger.error(
                 f"[Race Guard] Source file gone and no known destination — marking as failed: "
                 f"{os.path.basename(file_path)}"
@@ -966,6 +1008,53 @@ def post_process_matched_download(context_key, context, file_path, runtime, meta
             post_process_locks.pop(context_key, None)
 
 
+def _attempt_version_mismatch_fallback(context, task_id, batch_id, runtime, metadata_runtime):
+    """Opt-in last resort once AcoustID retries are exhausted: accept the best
+    quarantined version-mismatch candidate for this track instead of failing.
+
+    Delegates the decision + safety rules to
+    ``core.imports.version_mismatch_fallback`` (version-mismatch only, all the
+    same matched version, >= min_count, AcoustID-only bypass). Returns True when
+    a candidate was accepted and re-dispatched — the caller then skips marking
+    the task failed.
+    """
+    try:
+        download_path = docker_resolve_path(
+            config_manager.get('soulseek.download_path', './downloads')
+        )
+        quarantine_dir = os.path.join(download_path, 'ss_quarantine')
+        restore_dir = os.path.join(download_path, 'Transfer')
+        expected_title = get_import_clean_title(context, default='')
+        expected_artist = get_import_clean_artist(context, default='')
+        if not expected_title or not expected_artist:
+            return False
+
+        def _reprocess(restored_path, ctx, tid, bid):
+            new_key = f"vmfallback_{tid}_{int(time.time())}"
+            threading.Thread(
+                target=lambda: post_process_matched_download_with_verification(
+                    new_key, ctx, restored_path, tid, bid, runtime, metadata_runtime
+                ),
+                daemon=True,
+            ).start()
+
+        return try_accept_version_mismatch_fallback(
+            quarantine_dir=quarantine_dir,
+            restore_dir=restore_dir,
+            expected_title=expected_title,
+            expected_artist=expected_artist,
+            task_id=task_id,
+            batch_id=batch_id,
+            config_get=config_manager.get,
+            list_entries=list_quarantine_entries,
+            approve_entry=approve_quarantine_entry,
+            reprocess=_reprocess,
+        )
+    except Exception as exc:
+        pp_logger.debug("[Version-Mismatch Fallback] skipped due to error: %s", exc)
+        return False
+
+
 def post_process_matched_download_with_verification(context_key, context, file_path, task_id, batch_id, runtime, metadata_runtime=None):
     on_download_completed = getattr(runtime, "on_download_completed", None)
 
@@ -997,6 +1086,21 @@ def post_process_matched_download_with_verification(context_key, context, file_p
 
         if context.get('_acoustid_quarantined'):
             failure_msg = context.get('_acoustid_failure_msg', 'AcoustID verification failed')
+            # Before failing outright, try the next-best candidate. The wrong
+            # file was just quarantined; re-running the worker (with the bad
+            # source flagged used) picks the runner-up match instead.
+            with matched_context_lock:
+                matched_downloads_context.pop(context_key, None)
+            if _requeue_quarantined_task_for_retry(task_id, batch_id, 'acoustid'):
+                logger.info(
+                    f"AcoustID mismatch for task {task_id} — retrying next-best candidate: {failure_msg}"
+                )
+                return
+            # Retries exhausted. Opt-in last resort: if every quarantined
+            # candidate for this track failed the SAME version mismatch (e.g. all
+            # instrumental), accept the best one rather than leaving it missing.
+            if _attempt_version_mismatch_fallback(context, task_id, batch_id, runtime, metadata_runtime):
+                return
             logger.info(f"File was quarantined by AcoustID verification (task={task_id}): {failure_msg}")
             with tasks_lock:
                 if task_id in download_tasks:
@@ -1005,9 +1109,6 @@ def post_process_matched_download_with_verification(context_key, context, file_p
                     _eid = context.get('_quarantine_entry_id')
                     if _eid:
                         download_tasks[task_id]['quarantine_entry_id'] = _eid
-            with matched_context_lock:
-                if context_key in matched_downloads_context:
-                    del matched_downloads_context[context_key]
             _notify_download_completed(batch_id, task_id, success=False)
             return
 
@@ -1049,6 +1150,16 @@ def post_process_matched_download_with_verification(context_key, context, file_p
         # source files failed integrity and were quarantined.
         if context.get('_integrity_failure_msg'):
             failure_msg = context.get('_integrity_failure_msg', 'unknown')
+            # Integrity/duration mismatch (truncated transfer, wrong-length cut,
+            # etc). Same treatment as an AcoustID mismatch: quarantine the bad
+            # file and retry the next-best candidate before failing.
+            with matched_context_lock:
+                matched_downloads_context.pop(context_key, None)
+            if _requeue_quarantined_task_for_retry(task_id, batch_id, 'integrity'):
+                logger.info(
+                    f"Integrity check failed for task {task_id} — retrying next-best candidate: {failure_msg}"
+                )
+                return
             logger.error(
                 f"Task {task_id} failed integrity check — marking failed: {failure_msg}"
             )
@@ -1061,9 +1172,6 @@ def post_process_matched_download_with_verification(context_key, context, file_p
                     _eid = context.get('_quarantine_entry_id')
                     if _eid:
                         download_tasks[task_id]['quarantine_entry_id'] = _eid
-            with matched_context_lock:
-                if context_key in matched_downloads_context:
-                    del matched_downloads_context[context_key]
             _notify_download_completed(batch_id, task_id, success=False)
             return
 

core/imports/version_mismatch_fallback.py

@@ -0,0 +1,185 @@
+"""Last-resort acceptance of a version-mismatched download.
+
+Some tracks simply don't exist on the configured sources in the wanted cut —
+every copy is, say, the instrumental. The retry engine correctly rejects each
+one (version mismatch) and eventually gives up, leaving the track missing.
+
+This module provides an OPT-IN fallback: once a track's retries are fully
+exhausted, if every quarantined candidate for it failed the *same* way (same
+matched version, e.g. all ``instrumental``) and there are at least ``min_count``
+of them, accept the best (first-tried) one rather than failing outright.
+
+Hard safety rules:
+- Only ``Version mismatch`` quarantines qualify. Audio/artist mismatches
+  (a genuinely different recording) and integrity/duration failures
+  (truncated or wrong file) never participate.
+- All qualifying entries must share the same matched version. A mix
+  (instrumental + live) is ambiguous → no acceptance.
+- The chosen candidate is re-imported with only the AcoustID gate bypassed;
+  the integrity / duration / bit-depth gates still run, so a truncated or
+  corrupt file is never let through by this path.
+
+``select_version_mismatch_fallback`` is the pure decision core (no I/O) so it
+can be tested directly. ``try_accept_version_mismatch_fallback`` wires it to the
+quarantine store + re-import dispatch via injected callables.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from typing import Any, Callable, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+# Matches the reason string written by acoustid_verification's version gate:
+#   "Version mismatch: expected '<title>' (<exp>) but file is '<title>' (<got>)"
+# We only need the matched (<got>) version to test cross-entry consistency.
+_VERSION_MISMATCH_RE = re.compile(
+    r"^Version mismatch:.*\bbut file is\b.*\(([^()]+)\)\s*$"
+)
+
+
+def _norm(text: Optional[str]) -> str:
+    return (text or "").strip().casefold()
+
+
+def matched_version(reason: Optional[str]) -> Optional[str]:
+    """Return the matched version token (e.g. ``'instrumental'``) for a
+    Version-mismatch reason string, or None if the reason isn't a version
+    mismatch / can't be parsed."""
+    if not reason:
+        return None
+    m = _VERSION_MISMATCH_RE.match(reason.strip())
+    if not m:
+        return None
+    return m.group(1).strip().casefold()
+
+
+def select_version_mismatch_fallback(
+    entries: List[Dict[str, Any]],
+    expected_title: str,
+    expected_artist: str,
+    min_count: int,
+) -> Optional[Dict[str, Any]]:
+    """Pick the quarantine entry to accept as a last resort, or None.
+
+    ``entries`` are dicts as produced by
+    :func:`core.imports.quarantine.list_quarantine_entries` (needs ``id``,
+    ``reason``, ``expected_track``, ``expected_artist``, ``has_full_context``).
+
+    Returns the chosen entry (the first-tried = oldest = best, by ascending
+    ``id`` whose timestamp prefix sorts chronologically) when, for this track,
+    there are at least ``min_count`` version-mismatch entries that all share the
+    same matched version and carry full context. Otherwise None.
+    """
+    title = _norm(expected_title)
+    artist = _norm(expected_artist)
+
+    candidates = []
+    for e in entries:
+        if not e.get("has_full_context"):
+            continue
+        if _norm(e.get("expected_track")) != title:
+            continue
+        if _norm(e.get("expected_artist")) != artist:
+            continue
+        version = matched_version(e.get("reason"))
+        if version is None:
+            continue
+        candidates.append((version, e))
+
+    if len(candidates) < max(1, int(min_count or 1)):
+        return None
+
+    versions = {v for v, _ in candidates}
+    if len(versions) != 1:
+        # Inconsistent wrong versions (e.g. instrumental + live) — ambiguous,
+        # don't guess which the user wants.
+        return None
+
+    # First tried = oldest = highest-confidence (the retry walks candidates
+    # best-first). The id is a "<date>_<time>_<name>" timestamp prefix, so the
+    # lexicographically smallest id is the earliest attempt.
+    return min((e for _, e in candidates), key=lambda e: e["id"])
+
+
+def try_accept_version_mismatch_fallback(
+    *,
+    quarantine_dir: str,
+    restore_dir: str,
+    expected_title: str,
+    expected_artist: str,
+    task_id: str,
+    batch_id: Optional[str],
+    config_get: Callable[[str, Any], Any],
+    list_entries: Callable[[str], List[Dict[str, Any]]],
+    approve_entry: Callable[..., Optional[Any]],
+    reprocess: Callable[..., None],
+) -> bool:
+    """Orchestrate the last-resort acceptance. Returns True if a candidate was
+    accepted and re-dispatched (caller must then NOT mark the task failed).
+
+    All I/O is injected so this is testable without a filesystem or the
+    web_server pipeline:
+      - ``config_get(key, default)`` — settings lookup.
+      - ``list_entries(quarantine_dir)`` — quarantine.list_quarantine_entries.
+      - ``approve_entry(quarantine_dir, entry_id, restore_dir)`` ->
+        ``(restored_path, context, trigger)`` or None — quarantine.approve_quarantine_entry.
+      - ``reprocess(restored_path, context, task_id, batch_id)`` — re-run the
+        verification pipeline on the restored file.
+    """
+    if not config_get("post_processing.accept_version_mismatch_fallback", False):
+        return False
+
+    try:
+        min_count = int(config_get("post_processing.version_mismatch_min_count", 2))
+    except (TypeError, ValueError):
+        min_count = 2
+    if min_count < 1:
+        min_count = 1
+
+    try:
+        entries = list_entries(quarantine_dir) or []
+    except Exception as exc:  # never let the fallback break the failure path
+        logger.debug("[Version-Mismatch Fallback] listing quarantine failed: %s", exc)
+        return False
+
+    chosen = select_version_mismatch_fallback(
+        entries, expected_title, expected_artist, min_count
+    )
+    if not chosen:
+        return False
+
+    version = matched_version(chosen.get("reason")) or "?"
+    try:
+        result = approve_entry(quarantine_dir, chosen["id"], restore_dir)
+    except Exception as exc:
+        logger.error("[Version-Mismatch Fallback] approve failed for %s: %s", chosen["id"], exc)
+        return False
+    if not result:
+        return False
+
+    restored_path, context, _trigger = result
+    if not isinstance(context, dict):
+        return False
+    # Bypass ONLY the AcoustID gate — integrity / duration / bit-depth still run,
+    # so a truncated or genuinely wrong file is still caught.
+    context["_skip_quarantine_check"] = "acoustid"
+    context["_version_mismatch_fallback"] = version
+    context["task_id"] = task_id
+    if batch_id:
+        context["batch_id"] = batch_id
+
+    logger.warning(
+        "[Version-Mismatch Fallback] retries exhausted for '%s - %s'; accepting "
+        "best quarantined candidate (%s, entry %s) as last resort",
+        expected_artist, expected_title, version, chosen["id"],
+    )
+
+    try:
+        reprocess(restored_path, context, task_id, batch_id)
+    except Exception as exc:
+        logger.error("[Version-Mismatch Fallback] re-import dispatch failed: %s", exc)
+        return False
+    return True

core/musicbrainz_service.py

@@ -448,31 +448,50 @@ class MusicBrainzService:
         scored.sort(key=lambda x: -x[0])
         best_score, best_mbid, best_mb_score = scored[0]
 
+        # The genuine cross-script match (romaji↔kanji, latin↔cyrillic)
+        # has near-zero LOCAL similarity, so its COMBINED score sinks
+        # below an unrelated same-script decoy — even though MB itself is
+        # certain. "Sawano Hiroyuki": a decoy entity led on combined
+        # (sim 0.82, mb_score 83, combined 0.82 — just under the 0.85 bar)
+        # while the real artist '澤野弘之' had mb_score 100 but combined
+        # 0.30, sorted last. So evaluate the MB-SCORE leader independently
+        # of the combined ranking for the mb-only escape, not scored[0].
+        mb_leader = max(scored, key=lambda x: x[2])  # (combined, mbid, raw_mb)
+        mb_scores_desc = sorted((x[2] for x in scored), reverse=True)
+        mb_unambiguous = len(mb_scores_desc) < 2 or (mb_scores_desc[0] - mb_scores_desc[1]) >= 5
+
         # Trust gate. Two ways to pass:
         #   1. Combined score >= 0.85 (the historical strict bar that
-        #      catches same-script matches)
-        #   2. MB's OWN score is very high (>= 95) AND the result is
-        #      unambiguous (top result clearly leads). Bridges the
-        #      cross-script case where local similarity is near zero
-        #      ("Dmitry Yablonsky" vs "Дмитрий Яблонский" sim ~0)
-        #      but MB's index found a high-confidence match.
+        #      catches same-script matches) → trust the combined leader.
+        #   2. MB's OWN score is very high (>= 95) AND that MB-score leader
+        #      is unambiguous → trust IT. Bridges the cross-script case
+        #      where local similarity is near zero ("Dmitry Yablonsky" vs
+        #      "Дмитрий Яблонский" sim ~0) but MB's index is confident.
         passes_combined = best_score >= 0.85
-        passes_mb_only = best_mb_score >= 95 and (
-            len(scored) < 2 or (scored[0][2] - scored[1][2]) >= 5
-        )
+        passes_mb_only = mb_leader[2] >= 95 and mb_unambiguous
         if not (passes_combined or passes_mb_only):
             logger.debug(
                 "lookup_artist_aliases: best match for %r below trust "
-                "threshold (combined=%.2f, mb_score=%d)",
-                artist_name, best_score, best_mb_score,
+                "threshold (combined=%.2f, best_mb=%d, leader_mb=%d)",
+                artist_name, best_score, best_mb_score, mb_leader[2],
             )
             self._save_to_cache('artist_aliases', artist_name, None, None, {'aliases': []}, 0)
             return []
 
+        # Pick the entity to pull aliases from. Combined-strong matches use
+        # the combined leader; the mb-only escape uses the MB-score leader
+        # (which may differ from scored[0] in the cross-script case above).
+        if passes_combined:
+            chosen_mbid, chosen_conf = best_mbid, best_score
+        else:
+            chosen_mbid, chosen_conf = mb_leader[1], mb_leader[2] / 100.0
+
         # Ambiguity detection: when 2+ results both score high (within
         # 0.1 of the best combined), the search hit multiple distinct
         # artists with similar names. Pulling aliases for one could
-        # produce wrong matches. Skip + cache empty.
+        # produce wrong matches. Skip + cache empty. The unambiguous
+        # MB-score leader (passes_mb_only) is exempt — its decisiveness
+        # was already checked via mb_unambiguous.
         if len(scored) >= 2 and (scored[0][0] - scored[1][0]) < 0.1 and not passes_mb_only:
             logger.debug(
                 "lookup_artist_aliases: ambiguous match for %r — top "
@@ -482,10 +501,10 @@ class MusicBrainzService:
             self._save_to_cache('artist_aliases', artist_name, None, None, {'aliases': []}, 0)
             return []
 
-        aliases = self.fetch_artist_aliases(best_mbid)
+        aliases = self.fetch_artist_aliases(chosen_mbid)
         self._save_to_cache(
-            'artist_aliases', artist_name, None, best_mbid,
-            {'aliases': aliases}, int(best_score * 100),
+            'artist_aliases', artist_name, None, chosen_mbid,
+            {'aliases': aliases}, int(chosen_conf * 100),
         )
         return aliases
 

tests/downloads/test_downloads_task_worker.py

@@ -45,6 +45,7 @@ class _FakeClient:
         self._results = results if results is not None else []
         self.mode = mode
         self.search_calls = []
+        self.exclude_calls = []  # exclude_sources arg per search() call
         self._client_map = {}
         for k, v in (subclients or {}).items():
             if k in self._CLIENT_NAMES:
@@ -58,6 +59,7 @@ class _FakeClient:
 
     async def search(self, query, timeout=30, exclude_sources=None):
         self.search_calls.append((query, timeout))
+        self.exclude_calls.append(exclude_sources)
         return (self._results, None)
 
 
@@ -506,6 +508,190 @@ def test_hybrid_fallback_skipped_when_mode_not_hybrid():
     assert yt.search_calls == []
 
 
+# ---------------------------------------------------------------------------
+# Cached-first quarantine retry: walk already-found candidates before any
+# re-search; never re-search a source whose candidates are spent (only switch
+# to a not-yet-searched source). See task_worker cached-first phase.
+# ---------------------------------------------------------------------------
+
+class _Cand:
+    def __init__(self, username, filename):
+        self.username = username
+        self.filename = filename
+
+
+def test_quarantine_retry_tries_cached_candidates_without_searching():
+    _seed_task(
+        _quarantine_retry=True,
+        cached_candidates=[_Cand('peerA', 'f1.flac'), _Cand('peerB', 'f2.flac')],
+        used_sources={'peerA_f1.flac'},  # peerA already tried
+    )
+    attempted = []
+
+    def _attempt(task_id, candidates, track, batch_id):
+        attempted.append([getattr(c, 'filename', None) for c in candidates])
+        return True
+
+    sk = _FakeClient(results=['should-not-be-used'])
+    deps, _ = _build_deps(
+        soulseek=sk,
+        matching=_FakeMatchEngine(queries=['q1']),
+        attempt_download_with_candidates=_attempt,
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    # No search performed — cached candidates used directly.
+    assert sk.search_calls == []
+    # Only the unused candidate (peerB) was passed to attempt.
+    assert attempted == [['f2.flac']]
+
+
+def test_quarantine_retry_skips_cached_from_exhausted_source():
+    # hifi is budget-exhausted; its cached candidate must be skipped, soulseek's tried.
+    _seed_task(
+        _quarantine_retry=True,
+        cached_candidates=[_Cand('hifi', 'h.flac'), _Cand('peerB', 'f2.flac')],
+        used_sources=set(),
+        exhausted_download_sources={'hifi'},
+    )
+    attempted = []
+
+    def _attempt(task_id, candidates, track, batch_id):
+        attempted.append([getattr(c, 'username', None) for c in candidates])
+        return True
+
+    sk = _FakeClient(results=['x'])
+    deps, _ = _build_deps(
+        soulseek=sk,
+        matching=_FakeMatchEngine(queries=['q1']),
+        attempt_download_with_candidates=_attempt,
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    assert sk.search_calls == []
+    assert attempted == [['peerB']]  # hifi candidate excluded
+
+
+# A track_info that yields exactly the engine queries (no artist → no
+# first-word legacy query; name equals a query → track-only query dedupes away),
+# so the generated query set is deterministic for cached-first assertions.
+_SOLO_TRACK = {'id': 'sp-1', 'name': 'Solo', 'artists': [],
+               'album': 'A', 'duration_ms': 1000}
+
+
+def test_quarantine_retry_searches_unsearched_query_without_excluding_source():
+    # Cache spent + 'Solo' already searched, but 'q2' is NOT yet searched. The
+    # retry must search q2 against the SAME source (no source-level exclusion) so
+    # every query is exhausted per source before the chain switches sources
+    # (lazy multi-query retry).
+    _seed_task(
+        track_info=dict(_SOLO_TRACK),
+        _quarantine_retry=True,
+        cached_candidates=[_Cand('peerA', 'f1.flac')],
+        used_sources={'peerA_f1.flac'},
+        searched_queries={'Solo'},
+    )
+    sk = _FakeClient(results=[], mode='hybrid',
+                     subclients={'hybrid_order': ['soulseek', 'hifi']})
+    deps, _ = _build_deps(
+        soulseek=sk,
+        matching=_FakeMatchEngine(queries=['Solo', 'q2']),
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    # Only q2 was searched — 'Solo' skipped because its candidates were cached.
+    assert [c[0] for c in sk.search_calls] == ['q2']
+    # Soulseek NOT excluded: the not-yet-searched query still hits it.
+    assert all((not ex) or 'soulseek' not in ex for ex in sk.exclude_calls)
+
+
+def test_quarantine_retry_skips_already_searched_query_no_research():
+    # The only generated query is already searched and cache spent → it is NOT
+    # re-searched (its candidates live in cache). This is the wasteful repeat the
+    # cached-first design removes.
+    _seed_task(
+        track_info=dict(_SOLO_TRACK),
+        _quarantine_retry=True,
+        cached_candidates=[_Cand('peerA', 'f1.flac')],
+        used_sources={'peerA_f1.flac'},
+        searched_queries={'Solo'},
+    )
+    sk = _FakeClient(results=[], mode='hybrid',
+                     subclients={'hybrid_order': ['soulseek', 'hifi']})
+    deps, _ = _build_deps(
+        soulseek=sk,
+        matching=_FakeMatchEngine(queries=['Solo']),
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    assert sk.search_calls == []  # 'Solo' not re-searched
+
+
+def test_quarantine_retry_still_excludes_budget_exhausted_source():
+    # A source whose per-source budget is spent (exhaustive mode) stays excluded
+    # so the chain falls through to the next source.
+    _seed_task(
+        _quarantine_retry=True,
+        cached_candidates=[],
+        searched_queries=set(),
+        exhausted_download_sources={'soulseek'},
+    )
+    sk = _FakeClient(results=[], mode='hybrid',
+                     subclients={'hybrid_order': ['soulseek', 'hifi']})
+    deps, _ = _build_deps(
+        soulseek=sk,
+        matching=_FakeMatchEngine(queries=['q1']),
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    assert sk.search_calls
+    assert all(ex and 'soulseek' in ex for ex in sk.exclude_calls)
+
+
+def test_search_records_searched_queries():
+    # Every query the worker actually runs is recorded so a later quarantine
+    # retry can skip re-searching it.
+    _seed_task(status='pending')
+    sk = _FakeClient(results=['r1'])
+    deps, _ = _build_deps(
+        soulseek=sk,
+        matching=_FakeMatchEngine(queries=['q1']),
+        get_valid_candidates=lambda r, t, q: [],  # no candidates → loop completes
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    assert 'q1' in download_tasks['t1'].get('searched_queries', set())
+
+
+def test_non_quarantine_run_resets_searched_queries():
+    # A fresh / dead-connection retry starts a new search generation: the stale
+    # searched-query memory is cleared so every query can be searched again.
+    _seed_task(
+        track_info=dict(_SOLO_TRACK),
+        cached_candidates=[],
+        searched_queries={'stale-q'},
+    )
+    sk = _FakeClient(results=[])
+    deps, _ = _build_deps(
+        soulseek=sk,
+        matching=_FakeMatchEngine(queries=['Solo']),
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    sq = download_tasks['t1'].get('searched_queries', set())
+    assert 'stale-q' not in sq   # stale generation cleared
+    assert 'Solo' in sq          # current generation recorded fresh
+
+
+def test_non_quarantine_run_ignores_cached_first_and_searches():
+    # A fresh (non-quarantine) run must NOT use cached-first — it searches.
+    _seed_task(
+        cached_candidates=[_Cand('peerA', 'f1.flac')],
+        used_sources=set(),
+    )
+    sk = _FakeClient(results=[])
+    deps, _ = _build_deps(
+        soulseek=sk,
+        matching=_FakeMatchEngine(queries=['q1']),
+        attempt_download_with_candidates=lambda *a, **kw: False,
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    assert sk.search_calls  # searched, did not short-circuit on cache
+
+
 # ---------------------------------------------------------------------------
 # Top-level exception path
 # ---------------------------------------------------------------------------

tests/imports/test_import_pipeline.py

@@ -304,3 +304,356 @@ def test_verification_wrapper_applies_quarantine_entry_id_on_integrity_failure(m
         runtime_state.download_tasks.update(original)
         runtime_state.matched_downloads_context.clear()
         runtime_state.matched_downloads_context.update(original_ctx)
+
+
+# ---------------------------------------------------------------------------
+# Next-best-candidate retry on AcoustID / integrity quarantine. When a
+# verification or integrity check quarantines the wrong/broken file, the wrapper
+# asks the monitor to re-run the worker on the next candidate (skipping the bad
+# source) instead of failing the task outright.
+# ---------------------------------------------------------------------------
+
+def _wire_retry_engine(monkeypatch):
+    """Wire monitor's retry globals to capture the worker re-submission."""
+    import core.downloads.monitor as monitor
+
+    submitted = []
+
+    class _Exec:
+        def submit(self, fn, *args):
+            submitted.append(args)
+
+    monkeypatch.setattr(monitor, "missing_download_executor", _Exec())
+    monkeypatch.setattr(monitor, "_download_track_worker", lambda task_id, batch_id: None)
+    monkeypatch.setattr(monitor, "MAX_QUARANTINE_RETRIES", 5)
+    return submitted
+
+
+def _patch_config(monkeypatch, overrides):
+    """Override specific config keys for the monitor's config_manager reads."""
+    import core.downloads.monitor as monitor
+
+    real_get = monitor.config_manager.get
+
+    def fake_get(key, default=None):
+        if key in overrides:
+            return overrides[key]
+        return real_get(key, default)
+
+    monkeypatch.setattr(monitor.config_manager, "get", fake_get)
+
+
+def _run_wrapper_with_quarantine(monkeypatch, flag_setter, task_extra=None):
+    task_id, batch_id, context_key = "rtask", "rbatch", "rctx"
+    context = {"track_info": {}, "task_id": task_id, "batch_id": batch_id}
+
+    monkeypatch.setattr(import_pipeline, "post_process_matched_download", flag_setter)
+
+    original = dict(runtime_state.download_tasks)
+    original_ctx = dict(runtime_state.matched_downloads_context)
+    try:
+        runtime_state.download_tasks.clear()
+        task = {
+            "track_info": {}, "status": "downloading",
+            "username": "hifi", "filename": "123||A - B", "used_sources": set(),
+        }
+        if task_extra:
+            task.update(task_extra)
+        runtime_state.download_tasks[task_id] = task
+        runtime_state.matched_downloads_context.clear()
+        runtime_state.matched_downloads_context[context_key] = context
+
+        completion = []
+        runtime = types.SimpleNamespace(
+            automation_engine=None,
+            on_download_completed=lambda b, t, success: completion.append((b, t, success)),
+            web_scan_manager=None,
+            repair_worker=None,
+        )
+        import_pipeline.post_process_matched_download_with_verification(
+            context_key, context, "/tmp/source.flac", task_id, batch_id, runtime,
+        )
+        return dict(runtime_state.download_tasks[task_id]), completion, context_key
+    finally:
+        runtime_state.download_tasks.clear()
+        runtime_state.download_tasks.update(original)
+        runtime_state.matched_downloads_context.clear()
+        runtime_state.matched_downloads_context.update(original_ctx)
+
+
+def test_acoustid_mismatch_requeues_next_candidate(monkeypatch):
+    submitted = _wire_retry_engine(monkeypatch)
+
+    def _fake_inner(ck, ctx, fp, runtime, metadata_runtime=None):
+        ctx["_acoustid_quarantined"] = True
+        ctx["_acoustid_failure_msg"] = "wrong song"
+
+    task, completion, context_key = _run_wrapper_with_quarantine(monkeypatch, _fake_inner)
+
+    # Task goes back to searching for the next candidate — NOT failed.
+    assert task["status"] == "searching"
+    assert task["quarantine_retry_count"] == 1
+    # The quarantined source is flagged so the re-run won't re-pick it.
+    assert "hifi_123||A - B" in task["used_sources"]
+    # Stale download identity cleared; worker re-submitted; no batch failure.
+    assert "download_id" not in task and "username" not in task
+    assert submitted == [("rtask", "rbatch")]
+    assert completion == []
+    # Old context cleaned up (the re-run builds a fresh one for the new pick).
+    assert context_key not in runtime_state.matched_downloads_context
+
+
+def test_requeue_flags_quarantine_retry_for_cached_first(monkeypatch):
+    _wire_retry_engine(monkeypatch)
+
+    def _fake_inner(ck, ctx, fp, runtime, metadata_runtime=None):
+        ctx["_acoustid_quarantined"] = True
+        ctx["_acoustid_failure_msg"] = "wrong song"
+
+    task, _, _ = _run_wrapper_with_quarantine(monkeypatch, _fake_inner)
+
+    # The re-run is flagged so the worker walks cached candidates before
+    # re-searching (cached-first), rather than re-running the full search.
+    assert task["_quarantine_retry"] is True
+
+
+def test_integrity_mismatch_requeues_next_candidate(monkeypatch):
+    submitted = _wire_retry_engine(monkeypatch)
+
+    def _fake_inner(ck, ctx, fp, runtime, metadata_runtime=None):
+        ctx["_integrity_failure_msg"] = "Duration mismatch: file is 231.0s, expected 271.0s"
+
+    task, completion, _ = _run_wrapper_with_quarantine(monkeypatch, _fake_inner)
+
+    assert task["status"] == "searching"
+    assert task["quarantine_retry_count"] == 1
+    assert submitted == [("rtask", "rbatch")]
+    assert completion == []
+
+
+def test_manual_pick_does_not_requeue_on_mismatch(monkeypatch):
+    submitted = _wire_retry_engine(monkeypatch)
+
+    def _fake_inner(ck, ctx, fp, runtime, metadata_runtime=None):
+        ctx["_integrity_failure_msg"] = "Duration mismatch"
+
+    task, completion, _ = _run_wrapper_with_quarantine(
+        monkeypatch, _fake_inner, task_extra={"_user_manual_pick": True},
+    )
+
+    # User explicitly chose this file — fail it, don't silently swap.
+    assert task["status"] == "failed"
+    assert submitted == []
+    assert completion == [("rbatch", "rtask", False)]
+
+
+def test_retry_budget_exhausted_fails_task(monkeypatch):
+    submitted = _wire_retry_engine(monkeypatch)
+    import core.downloads.monitor as monitor
+    monkeypatch.setattr(monitor, "MAX_QUARANTINE_RETRIES", 2)
+
+    def _fake_inner(ck, ctx, fp, runtime, metadata_runtime=None):
+        ctx["_acoustid_quarantined"] = True
+        ctx["_acoustid_failure_msg"] = "wrong song"
+
+    task, completion, _ = _run_wrapper_with_quarantine(
+        monkeypatch, _fake_inner, task_extra={"quarantine_retry_count": 2},
+    )
+
+    # Cap reached — fall through to normal failure handling.
+    assert task["status"] == "failed"
+    assert submitted == []
+    assert completion == [("rbatch", "rtask", False)]
+
+
+def _acoustid_quarantine(ck, ctx, fp, runtime, metadata_runtime=None):
+    ctx["_acoustid_quarantined"] = True
+    ctx["_acoustid_failure_msg"] = "wrong song"
+
+
+def test_exhaustive_mode_uses_per_source_budget(monkeypatch):
+    submitted = _wire_retry_engine(monkeypatch)
+    _patch_config(monkeypatch, {
+        "post_processing.retry_exhaustive": True,
+        "post_processing.retries_per_query": 5,
+    })
+
+    # query_count=2 → budget for source 'hifi' = 2 * 5 = 10; first failure retries.
+    task, completion, _ = _run_wrapper_with_quarantine(
+        monkeypatch, _acoustid_quarantine, task_extra={"query_count": 2},
+    )
+
+    assert task["status"] == "searching"
+    # Per-source budget tracked separately from the legacy global counter.
+    assert task["quarantine_retry_counts_by_source"] == {"hifi": 1}
+    assert task["quarantine_retry_count"] == 1
+    assert "hifi_123||A - B" in task["used_sources"]
+    assert submitted == [("rtask", "rbatch")]
+    assert completion == []
+
+
+def test_exhaustive_source_budget_exhausted_fails(monkeypatch):
+    submitted = _wire_retry_engine(monkeypatch)
+    _patch_config(monkeypatch, {
+        "post_processing.retry_exhaustive": True,
+        "post_processing.retries_per_query": 5,
+    })
+
+    # hifi already at its full budget (query_count 2 * 5 = 10) → fail, no retry.
+    task, completion, _ = _run_wrapper_with_quarantine(
+        monkeypatch, _acoustid_quarantine,
+        task_extra={"query_count": 2, "quarantine_retry_counts_by_source": {"hifi": 10}},
+    )
+
+    assert task["status"] == "failed"
+    assert submitted == []
+    assert completion == [("rbatch", "rtask", False)]
+
+
+def test_exhaustive_budget_is_separate_per_source(monkeypatch):
+    submitted = _wire_retry_engine(monkeypatch)
+    _patch_config(monkeypatch, {
+        "post_processing.retry_exhaustive": True,
+        "post_processing.retries_per_query": 5,
+    })
+
+    # soulseek is already maxed, but the failing download is on hifi — hifi has
+    # its own fresh budget, so the task still retries.
+    task, completion, _ = _run_wrapper_with_quarantine(
+        monkeypatch, _acoustid_quarantine,
+        task_extra={"query_count": 1, "quarantine_retry_counts_by_source": {"soulseek": 5}},
+    )
+
+    assert task["status"] == "searching"
+    assert task["quarantine_retry_counts_by_source"] == {"soulseek": 5, "hifi": 1}
+    assert submitted == [("rtask", "rbatch")]
+
+
+def test_exhaustive_soulseek_peer_resolves_to_soulseek(monkeypatch):
+    submitted = _wire_retry_engine(monkeypatch)
+    _patch_config(monkeypatch, {
+        "post_processing.retry_exhaustive": True,
+        "post_processing.retries_per_query": 5,
+    })
+
+    # A Soulseek peer name (not a streaming source) is bucketed under 'soulseek'.
+    task, completion, _ = _run_wrapper_with_quarantine(
+        monkeypatch, _acoustid_quarantine,
+        task_extra={"username": "DjPeer", "filename": "f.flac", "query_count": 1},
+    )
+
+    assert task["status"] == "searching"
+    assert task["quarantine_retry_counts_by_source"] == {"soulseek": 1}
+
+
+def test_exhaustive_budget_defaults_query_count_to_one(monkeypatch):
+    submitted = _wire_retry_engine(monkeypatch)
+    _patch_config(monkeypatch, {
+        "post_processing.retry_exhaustive": True,
+        "post_processing.retries_per_query": 1,
+    })
+
+    # No query_count on the task → budget defaults to 1 * 1 = 1; hifi already at 1.
+    task, completion, _ = _run_wrapper_with_quarantine(
+        monkeypatch, _acoustid_quarantine,
+        task_extra={"quarantine_retry_counts_by_source": {"hifi": 1}},
+    )
+
+    assert task["status"] == "failed"
+    assert submitted == []
+
+
+def test_exhaustive_absolute_ceiling_guards_runaway(monkeypatch):
+    submitted = _wire_retry_engine(monkeypatch)
+    import core.downloads.monitor as monitor
+    monkeypatch.setattr(monitor, "MAX_TOTAL_QUARANTINE_RETRIES", 3)
+    _patch_config(monkeypatch, {
+        "post_processing.retry_exhaustive": True,
+        "post_processing.retries_per_query": 1000,  # per-source budget effectively unbounded
+    })
+
+    # Per-source budget is huge, but the absolute total ceiling (3) still fires.
+    task, completion, _ = _run_wrapper_with_quarantine(
+        monkeypatch, _acoustid_quarantine,
+        task_extra={"query_count": 1, "quarantine_retry_count": 3,
+                    "quarantine_retry_counts_by_source": {"hifi": 0}},
+    )
+
+    assert task["status"] == "failed"
+    assert submitted == []
+
+
+def _wire_orchestrator(monkeypatch, mode, hybrid_order):
+    """Wire monitor's download_orchestrator so per-source budget exhaustion can
+    decide whether another source remains to fall back to."""
+    import core.downloads.monitor as monitor
+    orch = types.SimpleNamespace(mode=mode, hybrid_order=list(hybrid_order))
+    monkeypatch.setattr(monitor, "download_orchestrator", orch)
+    return orch
+
+
+def test_exhaustive_exhausted_source_switches_in_hybrid(monkeypatch):
+    submitted = _wire_retry_engine(monkeypatch)
+    _wire_orchestrator(monkeypatch, "hybrid", ["soulseek", "hifi"])
+    _patch_config(monkeypatch, {
+        "post_processing.retry_exhaustive": True,
+        "post_processing.retries_per_query": 5,
+    })
+
+    # soulseek's budget (query_count 2 * 5 = 10) is spent. In hybrid mode the
+    # task switches to the next source instead of failing the whole track.
+    task, completion, _ = _run_wrapper_with_quarantine(
+        monkeypatch, _acoustid_quarantine,
+        task_extra={"username": "DjPeer", "query_count": 2,
+                    "quarantine_retry_counts_by_source": {"soulseek": 10}},
+    )
+
+    assert task["status"] == "searching"
+    # The spent source is flagged so the worker excludes it from the next search.
+    assert task["exhausted_download_sources"] == {"soulseek"}
+    # Its per-source counter is NOT pushed past budget — the source is simply done.
+    assert task["quarantine_retry_counts_by_source"]["soulseek"] == 10
+    assert submitted == [("rtask", "rbatch")]
+    assert completion == []
+
+
+def test_exhaustive_all_sources_exhausted_fails_in_hybrid(monkeypatch):
+    submitted = _wire_retry_engine(monkeypatch)
+    _wire_orchestrator(monkeypatch, "hybrid", ["soulseek", "hifi"])
+    _patch_config(monkeypatch, {
+        "post_processing.retry_exhaustive": True,
+        "post_processing.retries_per_query": 5,
+    })
+
+    # soulseek was exhausted on an earlier attempt; now hifi spends its last
+    # budget too — no fallback source remains, so the task finally fails.
+    task, completion, _ = _run_wrapper_with_quarantine(
+        monkeypatch, _acoustid_quarantine,
+        task_extra={"username": "hifi", "query_count": 2,
+                    "exhausted_download_sources": {"soulseek"},
+                    "quarantine_retry_counts_by_source": {"hifi": 10}},
+    )
+
+    assert task["status"] == "failed"
+    assert submitted == []
+    assert completion == [("rbatch", "rtask", False)]
+
+
+def test_exhaustive_single_source_exhausted_fails(monkeypatch):
+    submitted = _wire_retry_engine(monkeypatch)
+    # Single-source mode: nothing to fall back to once the budget is spent.
+    _wire_orchestrator(monkeypatch, "soulseek", [])
+    _patch_config(monkeypatch, {
+        "post_processing.retry_exhaustive": True,
+        "post_processing.retries_per_query": 5,
+    })
+
+    task, completion, _ = _run_wrapper_with_quarantine(
+        monkeypatch, _acoustid_quarantine,
+        task_extra={"username": "DjPeer", "query_count": 2,
+                    "quarantine_retry_counts_by_source": {"soulseek": 10}},
+    )
+
+    assert task["status"] == "failed"
+    assert submitted == []
+    assert completion == [("rbatch", "rtask", False)]

tests/imports/test_version_mismatch_fallback.py

@@ -0,0 +1,207 @@
+"""Last-resort acceptance of a version-mismatched quarantine candidate.
+
+When a track's retries are fully exhausted and EVERY quarantined candidate for
+it failed the same way (same wrong version, e.g. all instrumental), the only
+available version is that one — accept the best (first-tried) one instead of
+leaving the track missing. Strict guards: version-mismatch only, all the same
+matched version, and a minimum count.
+"""
+
+from __future__ import annotations
+
+from core.imports.version_mismatch_fallback import (
+    select_version_mismatch_fallback,
+    try_accept_version_mismatch_fallback,
+)
+
+
+def _entry(eid, reason, *, track="Barricades (Movie Ver.)", artist="Hiroyuki Sawano",
+           ctx=True):
+    return {
+        "id": eid,
+        "reason": reason,
+        "expected_track": track,
+        "expected_artist": artist,
+        "has_full_context": ctx,
+    }
+
+
+_VM = "Version mismatch: expected 'Barricades (Movie Ver.)' (original) but file is 'Barricades <MOVIEver.> ({v})' ({v})"
+
+
+def _vm(version):
+    return _VM.format(v=version)
+
+
+def test_picks_oldest_when_all_same_version_and_count_met():
+    # 3 instrumental mismatches → all same kind → pick first-tried (smallest id).
+    entries = [
+        _entry("20260605_120300", _vm("instrumental")),
+        _entry("20260605_120100", _vm("instrumental")),  # oldest = first tried
+        _entry("20260605_120200", _vm("instrumental")),
+    ]
+    chosen = select_version_mismatch_fallback(entries, "Barricades (Movie Ver.)",
+                                              "Hiroyuki Sawano", min_count=2)
+    assert chosen is not None
+    assert chosen["id"] == "20260605_120100"
+
+
+def test_none_when_below_min_count():
+    entries = [_entry("20260605_120100", _vm("instrumental"))]
+    assert select_version_mismatch_fallback(
+        entries, "Barricades (Movie Ver.)", "Hiroyuki Sawano", min_count=2) is None
+
+
+def test_none_when_mixed_versions():
+    # instrumental + live → inconsistent → never auto-accept (ambiguous).
+    entries = [
+        _entry("20260605_120100", _vm("instrumental")),
+        _entry("20260605_120200", _vm("live")),
+    ]
+    assert select_version_mismatch_fallback(
+        entries, "Barricades (Movie Ver.)", "Hiroyuki Sawano", min_count=2) is None
+
+
+def test_ignores_non_version_mismatch_reasons():
+    # Audio mismatch (wrong artist/song) and integrity must NOT count.
+    entries = [
+        _entry("20260605_120100",
+               "Audio mismatch: file identified as 'X' by 'Y' (artist=0%)"),
+        _entry("20260605_120200",
+               "Integrity check failed: Duration mismatch: file is 175s, expected 182s"),
+    ]
+    assert select_version_mismatch_fallback(
+        entries, "Barricades (Movie Ver.)", "Hiroyuki Sawano", min_count=1) is None
+
+
+def test_only_counts_entries_for_this_track():
+    entries = [
+        _entry("20260605_120100", _vm("instrumental")),
+        _entry("20260605_120200", _vm("instrumental"), track="Call Your Name (Gv)"),
+    ]
+    # Only one entry matches this track → below min_count of 2.
+    assert select_version_mismatch_fallback(
+        entries, "Barricades (Movie Ver.)", "Hiroyuki Sawano", min_count=2) is None
+
+
+def test_excludes_thin_sidecar_entries_without_context():
+    # Can't approve without embedded context → exclude from the candidate pool.
+    entries = [
+        _entry("20260605_120100", _vm("instrumental"), ctx=False),
+        _entry("20260605_120200", _vm("instrumental"), ctx=False),
+    ]
+    assert select_version_mismatch_fallback(
+        entries, "Barricades (Movie Ver.)", "Hiroyuki Sawano", min_count=2) is None
+
+
+def test_track_match_is_case_and_space_insensitive():
+    entries = [
+        _entry("20260605_120100", _vm("instrumental")),
+        _entry("20260605_120200", _vm("instrumental")),
+    ]
+    chosen = select_version_mismatch_fallback(
+        entries, "  barricades (movie ver.) ", "HIROYUKI SAWANO", min_count=2)
+    assert chosen is not None
+    assert chosen["id"] == "20260605_120100"
+
+
+# ── Orchestration (try_accept_version_mismatch_fallback) ──────────────────────
+
+def _cfg(enabled=True, min_count=2):
+    values = {
+        "post_processing.accept_version_mismatch_fallback": enabled,
+        "post_processing.version_mismatch_min_count": min_count,
+    }
+    return lambda key, default=None: values.get(key, default)
+
+
+def _two_instrumental_entries():
+    return [
+        _entry("20260605_120100", _vm("instrumental")),
+        _entry("20260605_120200", _vm("instrumental")),
+    ]
+
+
+def test_orchestration_disabled_does_nothing():
+    calls = {"approve": 0, "reprocess": 0}
+
+    def approve(*a, **k):
+        calls["approve"] += 1
+        return ("/restored.flac", {}, "acoustid")
+
+    def reprocess(*a, **k):
+        calls["reprocess"] += 1
+
+    ok = try_accept_version_mismatch_fallback(
+        quarantine_dir="/q", restore_dir="/r",
+        expected_title="Barricades (Movie Ver.)", expected_artist="Hiroyuki Sawano",
+        task_id="t1", batch_id="b1",
+        config_get=_cfg(enabled=False),
+        list_entries=lambda d: _two_instrumental_entries(),
+        approve_entry=approve, reprocess=reprocess,
+    )
+    assert ok is False
+    assert calls == {"approve": 0, "reprocess": 0}
+
+
+def test_orchestration_accepts_and_reprocesses_with_acoustid_bypass():
+    captured = {}
+
+    def approve(qdir, entry_id, rdir):
+        captured["entry_id"] = entry_id
+        return ("/restored.flac", {"existing": 1}, "acoustid")
+
+    def reprocess(path, context, task_id, batch_id):
+        captured["path"] = path
+        captured["context"] = context
+        captured["task_id"] = task_id
+
+    ok = try_accept_version_mismatch_fallback(
+        quarantine_dir="/q", restore_dir="/r",
+        expected_title="Barricades (Movie Ver.)", expected_artist="Hiroyuki Sawano",
+        task_id="t1", batch_id="b1",
+        config_get=_cfg(),
+        list_entries=lambda d: _two_instrumental_entries(),
+        approve_entry=approve, reprocess=reprocess,
+    )
+    assert ok is True
+    assert captured["entry_id"] == "20260605_120100"      # oldest/best
+    assert captured["path"] == "/restored.flac"
+    assert captured["task_id"] == "t1"
+    # Only AcoustID bypassed — integrity/bit-depth gates still run.
+    assert captured["context"]["_skip_quarantine_check"] == "acoustid"
+    assert captured["context"]["_version_mismatch_fallback"] == "instrumental"
+    assert captured["context"]["task_id"] == "t1"
+    assert captured["context"]["batch_id"] == "b1"
+
+
+def test_orchestration_no_candidate_does_not_reprocess():
+    calls = {"reprocess": 0}
+
+    def reprocess(*a, **k):
+        calls["reprocess"] += 1
+
+    ok = try_accept_version_mismatch_fallback(
+        quarantine_dir="/q", restore_dir="/r",
+        expected_title="Barricades (Movie Ver.)", expected_artist="Hiroyuki Sawano",
+        task_id="t1", batch_id="b1",
+        config_get=_cfg(),
+        list_entries=lambda d: [_entry("20260605_120100", _vm("instrumental"))],  # 1 < min 2
+        approve_entry=lambda *a, **k: ("/x", {}, "acoustid"),
+        reprocess=reprocess,
+    )
+    assert ok is False
+    assert calls["reprocess"] == 0
+
+
+def test_orchestration_approve_failure_returns_false():
+    ok = try_accept_version_mismatch_fallback(
+        quarantine_dir="/q", restore_dir="/r",
+        expected_title="Barricades (Movie Ver.)", expected_artist="Hiroyuki Sawano",
+        task_id="t1", batch_id="b1",
+        config_get=_cfg(),
+        list_entries=lambda d: _two_instrumental_entries(),
+        approve_entry=lambda *a, **k: None,   # thin sidecar / move failed
+        reprocess=lambda *a, **k: (_ for _ in ()).throw(AssertionError("must not reprocess")),
+    )
+    assert ok is False

tests/matching/test_artist_alias_lookup_586.py

@@ -191,6 +191,35 @@ def test_trust_gate_rejects_when_two_high_mb_scores_tie(service):
     assert aliases == []
 
 
+def test_trust_gate_uses_mb_score_leader_not_combined_leader(service):
+    # Production case "Sawano Hiroyuki": a same-script DECOY entity leads on
+    # COMBINED score (high local sim, mb_score 83) but sits just under the
+    # 0.85 combined bar, while the genuine cross-script artist has mb_score
+    # 100 and ~0 local sim → lowest combined, sorted last. The mb-only escape
+    # must evaluate the MB-SCORE leader, not scored[0] (the combined leader),
+    # otherwise it inspects mb_score 83 < 95 and wrongly returns [].
+    service._calculate_similarity = (
+        lambda a, b: 0.82 if b == 'SawanoHiroyuki[nZk]' else 0.0
+    )
+    service.mb_client.search_artist.return_value = [
+        {'id': 'mbid-decoy', 'name': 'SawanoHiroyuki[nZk]', 'score': 83},
+        {'id': 'mbid-canonical', 'name': '澤野弘之', 'score': 100},
+    ]
+
+    def get_artist(mbid, **kwargs):
+        if mbid == 'mbid-canonical':
+            return {'name': '澤野弘之', 'aliases': [{'name': 'Hiroyuki Sawano'}]}
+        return {'name': 'SawanoHiroyuki[nZk]', 'aliases': []}
+    service.mb_client.get_artist.side_effect = get_artist
+
+    aliases = service.lookup_artist_aliases('Sawano Hiroyuki')
+    # The canonical kanji name must come back (its alias set was fetched).
+    assert '澤野弘之' in aliases
+    # And we must have fetched the MB-score leader, not the decoy.
+    service.mb_client.get_artist.assert_called_once()
+    assert service.mb_client.get_artist.call_args.args[0] == 'mbid-canonical'
+
+
 def test_trust_gate_passes_combined_score_when_local_sim_strong(service):
     # Same-script case from #442 — local sim high. Should still pass
     # (no regression on the existing path).

web_server.py

@@ -17000,6 +17000,53 @@ def _run_post_processing_worker(task_id, batch_id):
 from core.downloads import task_worker as _downloads_task_worker
 
 
+def _try_version_mismatch_fallback_for_worker(expected_title, expected_artist, task_id, batch_id):
+    """Called by the download worker when a quarantine-retry search finds no
+    candidates.  Delegates to version_mismatch_fallback so the best already-
+    quarantined version-mismatch candidate is accepted rather than giving up."""
+    import threading
+    import time
+    from core.imports.version_mismatch_fallback import try_accept_version_mismatch_fallback
+    from core.imports.quarantine import approve_quarantine_entry, list_quarantine_entries
+    from config.settings import config_manager
+    from core.imports.paths import docker_resolve_path
+    import os
+    try:
+        download_path = docker_resolve_path(
+            config_manager.get('soulseek.download_path', './downloads')
+        )
+        quarantine_dir = os.path.join(download_path, 'ss_quarantine')
+        restore_dir = os.path.join(download_path, 'Transfer')
+
+        def _reprocess(restored_path, ctx, tid, bid):
+            new_key = f"vmfallback_{tid}_{int(time.time())}"
+            threading.Thread(
+                target=lambda: _post_process_matched_download_with_verification(
+                    new_key, ctx, restored_path, tid, bid
+                ),
+                daemon=True,
+            ).start()
+
+        return try_accept_version_mismatch_fallback(
+            quarantine_dir=quarantine_dir,
+            restore_dir=restore_dir,
+            expected_title=expected_title,
+            expected_artist=expected_artist,
+            task_id=task_id,
+            batch_id=batch_id,
+            config_get=config_manager.get,
+            list_entries=list_quarantine_entries,
+            approve_entry=approve_quarantine_entry,
+            reprocess=_reprocess,
+        )
+    except Exception as exc:
+        import logging
+        logging.getLogger(__name__).debug(
+            "[Version-Mismatch Fallback] worker-path skipped due to error: %s", exc
+        )
+        return False
+
+
 def _build_task_worker_deps():
     """Build TaskWorkerDeps bundle from web_server.py globals on each call."""
     return _downloads_task_worker.TaskWorkerDeps(
@@ -17013,6 +17060,7 @@ def _build_task_worker_deps():
         attempt_download_with_candidates=_attempt_download_with_candidates,
         on_download_completed=lambda b, t, success: _on_download_completed(b, t, success=success),
         recover_worker_slot=_recover_worker_slot,
+        try_version_mismatch_fallback=_try_version_mismatch_fallback_for_worker,
     )
 
 

webui/index.html

@@ -4967,6 +4967,18 @@
 
                             </div>
 
+                            <!-- ═══ QUALITY PROFILE ═══ -->
+                            <!-- Whole tile is gated as a unit (Soulseek-only + downloads tab)
+                                 by settings.js updateSourceVisibility — gating only the inner
+                                 group would leave an empty expandable shell. -->
+                            <div id="quality-profile-tile" data-stg="downloads">
+                            <div class="settings-section-header collapsed" onclick="this.classList.toggle('collapsed'); const b=this.nextElementSibling; b.classList.toggle('collapsed'); b.style.display=b.classList.contains('collapsed')?'none':''">
+                                <span class="settings-section-arrow">&#9660;</span>
+                                <h3>Quality Profile</h3>
+                                <span class="settings-section-hint">Format priorities, bitrate, bit depth</span>
+                            </div>
+                            <div class="settings-section-body collapsed" data-stg="downloads">
+
                             <!-- Quality Profile Settings (Soulseek only) -->
                             <div class="settings-group" id="quality-profile-section" data-stg="downloads">
                                 <h3>🎵 Quality Profile</h3>
@@ -5148,6 +5160,55 @@
                                 </div>
                             </div>
 
+                            </div><!-- end Quality Profile body -->
+                            </div><!-- end Quality Profile tile -->
+
+                            <!-- ═══ RETRY LOGIC ═══ -->
+                            <div class="settings-section-header collapsed" data-stg="downloads" onclick="this.classList.toggle('collapsed'); const b=this.nextElementSibling; b.classList.toggle('collapsed'); b.style.display=b.classList.contains('collapsed')?'none':''">
+                                <span class="settings-section-arrow">&#9660;</span>
+                                <h3>Retry Logic</h3>
+                                <span class="settings-section-hint">Next-best candidate, exhaustive per-source retry</span>
+                            </div>
+                            <div class="settings-section-body collapsed" data-stg="downloads">
+
+                            <div class="settings-group" data-stg="downloads">
+                                <h3>🔁 Retry Logic</h3>
+                                <small class="settings-hint" style="margin-bottom: 10px; display: block;">Controls what happens when a downloaded file is rejected (AcoustID mismatch, wrong version, or duration/integrity failure). Independent of post-processing.</small>
+                                <div class="form-group">
+                                    <label class="checkbox-label">
+                                        <input type="checkbox" id="retry-next-candidate" checked>
+                                        Retry next-best candidate on mismatch
+                                    </label>
+                                    <small class="settings-hint">When a download is quarantined (AcoustID mismatch or duration/integrity failure), automatically try the next-best candidate instead of failing the track. Off = quarantine and fail immediately.</small>
+                                </div>
+                                <div class="form-group">
+                                    <label class="checkbox-label">
+                                        <input type="checkbox" id="retry-exhaustive">
+                                        Exhaustive retry (separate budget per source)
+                                    </label>
+                                    <small class="settings-hint">Give every source (Soulseek, then HiFi/Tidal/…) its own retry budget. Each source spends <em>queries × retries-per-query</em> attempts before the track moves on. Worst case across two sources can mean many downloads — use for hard-to-match tracks (e.g. CJK artist names). Requires the option above.</small>
+                                </div>
+                                <div class="form-group">
+                                    <label for="retries-per-query">Retries per query (per source)</label>
+                                    <input type="number" id="retries-per-query" min="1" max="20" step="1" value="5" style="width: 100px;" autocomplete="off" data-bwignore="" data-1p-ignore="" data-lpignore="true">
+                                    <small class="settings-hint">In exhaustive mode, how many candidates to try per search query per source. The per-source budget is <em>number of search queries × this value</em>. Only used when Exhaustive retry is on.</small>
+                                </div>
+                                <div class="form-group">
+                                    <label class="checkbox-label">
+                                        <input type="checkbox" id="accept-version-mismatch-fallback">
+                                        Accept best version mismatch as last resort
+                                    </label>
+                                    <small class="settings-hint">When retries are fully exhausted and <em>every</em> candidate for a track failed the <strong>same</strong> version mismatch (e.g. only an instrumental exists), accept the best (first-tried) one instead of leaving the track missing. Only AcoustID is bypassed — integrity/duration/bit-depth checks still run, so truncated or genuinely wrong files are never let through. Off = leave such tracks failed.</small>
+                                </div>
+                                <div class="form-group">
+                                    <label for="version-mismatch-min-count">Minimum matching mismatches before accepting</label>
+                                    <input type="number" id="version-mismatch-min-count" min="1" max="20" step="1" value="2" style="width: 100px;" autocomplete="off" data-bwignore="" data-1p-ignore="" data-lpignore="true">
+                                    <small class="settings-hint">How many quarantined candidates must have failed the <em>same</em> version mismatch before the last-resort acceptance kicks in. Higher = more confirmation that no correct version exists. Only used when the option above is on.</small>
+                                </div>
+                            </div>
+
+                            </div><!-- end Retry Logic body -->
+
                             <!-- ═══ INDEXERS & DOWNLOADERS ═══ -->
                             <!-- Intro hero — explains the search → fetch flow in one glance -->
                             <div class="settings-group ind-hero" data-stg="indexers">

webui/static/settings.js

@@ -1168,6 +1168,11 @@ async function loadSettingsData() {
         document.getElementById('lrclib-enabled').checked = settings.metadata_enhancement?.lrclib_enabled !== false;
         document.getElementById('replaygain-enabled').checked = settings.post_processing?.replaygain_enabled === true;
         document.getElementById('duration-tolerance-seconds').value = settings.post_processing?.duration_tolerance_seconds ?? 0;
+        document.getElementById('retry-next-candidate').checked = settings.post_processing?.retry_next_candidate_on_mismatch !== false;
+        document.getElementById('retry-exhaustive').checked = settings.post_processing?.retry_exhaustive === true;
+        document.getElementById('retries-per-query').value = settings.post_processing?.retries_per_query ?? 5;
+        document.getElementById('accept-version-mismatch-fallback').checked = settings.post_processing?.accept_version_mismatch_fallback === true;
+        document.getElementById('version-mismatch-min-count').value = settings.post_processing?.version_mismatch_min_count ?? 2;
         // Load service master toggles
         document.getElementById('embed-spotify').checked = settings.spotify?.embed_tags !== false;
         document.getElementById('embed-itunes').checked = settings.itunes?.embed_tags !== false;
@@ -1708,12 +1713,16 @@ function updateDownloadSourceUI() {
         prowlarrRedirect.style.display = showProwlarr ? 'block' : 'none';
     }
 
-    // Quality profile is Soulseek-only and downloads-tab-only
-    const qualityProfileSection = document.getElementById('quality-profile-section');
-    if (qualityProfileSection) {
+    // Quality profile is Soulseek-only (it only affects Soulseek downloads) and
+    // downloads-tab-only. Gate the WHOLE collapsible tile (#quality-profile-tile
+    // = header + body) as a unit, so it either fully shows (Soulseek active) or
+    // fully hides — never an empty expandable shell (the earlier bug came from
+    // gating only the inner #quality-profile-section).
+    const qualityProfileTile = document.getElementById('quality-profile-tile');
+    if (qualityProfileTile) {
         const activeTab = document.querySelector('.stg-tab.active');
         const onDownloadsTab = activeTab && activeTab.dataset.tab === 'downloads';
-        qualityProfileSection.style.display = (activeSources.has('soulseek') && onDownloadsTab) ? '' : 'none';
+        qualityProfileTile.style.display = (activeSources.has('soulseek') && onDownloadsTab) ? '' : 'none';
     }
 
     if (activeSources.has('tidal')) {
@@ -2996,6 +3005,11 @@ async function saveSettings(quiet = false) {
         post_processing: {
             replaygain_enabled: document.getElementById('replaygain-enabled').checked,
             duration_tolerance_seconds: parseFloat(document.getElementById('duration-tolerance-seconds').value) || 0,
+            retry_next_candidate_on_mismatch: document.getElementById('retry-next-candidate').checked,
+            retry_exhaustive: document.getElementById('retry-exhaustive').checked,
+            retries_per_query: Math.max(1, parseInt(document.getElementById('retries-per-query').value, 10) || 5),
+            accept_version_mismatch_fallback: document.getElementById('accept-version-mismatch-fallback').checked,
+            version_mismatch_min_count: Math.max(1, parseInt(document.getElementById('version-mismatch-min-count').value, 10) || 2),
         },
         library: {
             music_paths: collectMusicPaths(),

webui/static/style.css

@@ -471,6 +471,22 @@ body:not(.reduce-effects) .nav-button.active:hover {
         inset 0 1px 0 rgba(255, 255, 255, 0.12);
 }
 
+/* Reduce-effects mode keeps the sidebar hover/active highlight — the
+   feedback matters for navigation — but only the CHEAP parts: a flat
+   background + border-color change (the base already reserves a 1px
+   transparent border, so no layout shift). The expensive full-effects
+   bits (gradient, transform/translateX, multi-layer box-shadow) stay
+   off so hovering doesn't trigger compositing/repaint churn. */
+body.reduce-effects .nav-button:hover {
+    background: rgba(255, 255, 255, 0.05);
+    border-color: rgba(255, 255, 255, 0.08);
+}
+
+body.reduce-effects .nav-button.active:hover {
+    background: rgba(var(--accent-rgb), 0.18);
+    border-color: rgba(var(--accent-rgb), 0.3);
+}
+
 .nav-icon {
     width: 40px;
     height: 40px;