Merge pull request #531 from Nezreka/feat/candidates-modal-manual-search

Feat/candidates modal manual search
2026-05-08 15:18:02 -07:00 · 2026-05-08 15:18:02 -07:00 · 661f00cb35
commit 661f00cb35
parent 4277648734 e20994e1c7
9 changed files with 2013 additions and 103 deletions
--- a/core/downloads/monitor.py
+++ b/core/downloads/monitor.py
@ -332,6 +332,13 @@ class WebUIDownloadMonitor:
        live_info = live_transfers_lookup.get(lookup_key)

        if not live_info:
+            # User-initiated manual pick — skip auto-retry. The status
+            # engine fallback owns the terminal transition for non-Soulseek
+            # manual downloads. Yanking the task back to 'searching' here
+            # would defeat the user's explicit selection.
+            if task.get('_user_manual_pick'):
+                return False
+
            # Task not in live transfers but status is downloading/queued - likely stuck
            if current_time - task.get('status_change_time', current_time) > 90:
                retry_count = task.get('stuck_retry_count', 0)
@ -396,6 +403,11 @@ class WebUIDownloadMonitor:

        # IMMEDIATE ERROR RETRY: Check for errored/rejected/timed-out downloads first (no timeout needed)
        if 'Errored' in state_str or 'Failed' in state_str or 'Rejected' in state_str or 'TimedOut' in state_str:
+            # Same manual-pick guard as the not-in-live-transfers path —
+            # user explicitly selected this candidate, surface the failure.
+            if task.get('_user_manual_pick'):
+                return False
+
            retry_count = task.get('error_retry_count', 0)
            last_retry = task.get('last_error_retry_time', 0)

--- a/core/downloads/status.py
+++ b/core/downloads/status.py
@ -21,6 +21,7 @@ are passed via `StatusDeps` so the module is web_server-import-free.
 from __future__ import annotations

 import logging
+import threading
 import time
 from dataclasses import dataclass
 from typing import Any, Callable, Optional
@ -34,6 +35,37 @@ from core.runtime_state import (
 logger = logging.getLogger(__name__)


+def _schedule_completion_callback(deps, batch_id: str, task_id: str, success: bool) -> None:
+    """Fire ``deps.on_download_completed`` on a one-shot daemon thread so
+    the caller can hold ``tasks_lock`` without deadlocking.
+
+    ``on_download_completed`` re-acquires ``tasks_lock`` (it removes the
+    completed task from the batch's active set, decrements active_count,
+    and may submit the next queued worker). Calling it synchronously
+    from within ``build_batch_status_data`` — which is invoked under the
+    same Lock — would self-deadlock since ``threading.Lock`` is not
+    reentrant. A daemon thread defers the call until after the lock is
+    released.
+    """
+    if deps.on_download_completed is None:
+        return
+
+    def _run():
+        try:
+            deps.on_download_completed(batch_id, task_id, success)
+        except Exception as exc:
+            logger.error(
+                "[Status] deferred on_download_completed raised for task %s: %s",
+                task_id, exc,
+            )
+
+    threading.Thread(
+        target=_run,
+        name=f"on-completed-{task_id[:8]}",
+        daemon=True,
+    ).start()
+
+
@dataclass
 class StatusDeps:
    """Cross-cutting deps the status helpers need."""
@ -43,6 +75,162 @@ class StatusDeps:
    make_context_key: Callable[[str, str], str]
    submit_post_processing: Callable[[str, str], None]  # (task_id, batch_id) -> None
    get_cached_transfer_data: Callable[[], dict]
+    # Engine-state fallback for non-Soulseek (streaming) downloads.
+    # Without these, YouTube/Tidal/Qobuz/HiFi/Deezer/SoundCloud/Lidarr
+    # tasks never appear in live_transfers_lookup so their status never
+    # advances out of 'downloading 0%'.
+    download_orchestrator: Any = None
+    run_async: Optional[Callable] = None
+    on_download_completed: Optional[Callable[[str, str, bool], None]] = None
+
+
+# Streaming sources the engine fallback applies to. Soulseek goes through
+# slskd's live_transfers path and must NOT hit the engine fallback.
+_STREAMING_SOURCE_NAMES = frozenset((
+    'youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud',
+))
+
+# Keep these in sync with the engine plugins' state strings.
+_ENGINE_FAILURE_STATES = ('Errored', 'Failed', 'Rejected', 'TimedOut', 'Aborted')
+_ENGINE_CANCELLED_STATES = ('Cancelled', 'Canceled')
+_ENGINE_SUCCESS_STATES = ('Succeeded', 'Completed, Succeeded')
+
+
+def _engine_state_str(record: Any) -> str:
+    if record is None:
+        return ''
+    state = getattr(record, 'state', None)
+    if state is None and isinstance(record, dict):
+        state = record.get('state')
+    return str(state) if state is not None else ''
+
+
+def _engine_progress_pct(record: Any) -> float:
+    if record is None:
+        return 0
+    progress = getattr(record, 'progress', None)
+    if progress is None and isinstance(record, dict):
+        progress = record.get('progress')
+    try:
+        progress = float(progress)
+    except (TypeError, ValueError):
+        return 0
+    if progress <= 1.0:
+        progress *= 100
+    return progress
+
+
+def _apply_engine_state_fallback(
+    task_id: str,
+    task: dict,
+    task_status: dict,
+    batch_id: str,
+    deps: StatusDeps,
+) -> None:
+    """Populate ``task_status`` from the download engine's per-source
+    record when the task isn't in ``live_transfers_lookup`` — i.e. it's
+    a non-Soulseek streaming source. Mirrors the Soulseek branch's
+    Cancelled → Failed → Succeeded → InProgress priority order so
+    compound states like ``"Completed, Errored"`` hit the failure branch
+    first.
+
+    Mutates ``task`` in place (status / error_message) the same way the
+    Soulseek branch does, so the next status poll sees the new state.
+    Submits post-processing on terminal success and fires
+    ``on_download_completed`` on terminal failure to free the worker
+    slot.
+    """
+    if deps.download_orchestrator is None or deps.run_async is None:
+        return
+    if task.get('status') in ('completed', 'failed', 'cancelled', 'not_found', 'post_processing'):
+        return
+    # Scope this fallback to user-initiated manual picks. Auto attempts
+    # already flow through the live_transfers_lookup IF branch (the engine
+    # pre-populates non-Soulseek records via get_all_downloads), and on
+    # failure the monitor's existing retry path picks the next candidate.
+    # Marking auto attempts failed here would short-circuit that fallback.
+    if not task.get('_user_manual_pick'):
+        return
+    download_id = task.get('download_id')
+    if not download_id:
+        return
+    ti = task.get('track_info') if isinstance(task.get('track_info'), dict) else {}
+    username = task.get('username') or ti.get('username')
+    if username not in _STREAMING_SOURCE_NAMES:
+        return
+
+    try:
+        record = deps.run_async(
+            deps.download_orchestrator.get_download_status(download_id)
+        )
+    except Exception as exc:
+        logger.debug(
+            "[Engine Fallback] get_download_status(%s) raised: %s",
+            download_id, exc,
+        )
+        return
+
+    if record is None:
+        return
+
+    state_str = _engine_state_str(record)
+    if not state_str:
+        return
+
+    if any(s in state_str for s in _ENGINE_CANCELLED_STATES):
+        if task['status'] != 'cancelled':
+            task['status'] = 'cancelled'
+            err = getattr(record, 'error_message', None) or getattr(record, 'error', None) or ''
+            if err:
+                task['error_message'] = str(err)
+            _schedule_completion_callback(deps, batch_id, task_id, False)
+        task_status['status'] = 'cancelled'
+        task_status['progress'] = _engine_progress_pct(record)
+        return
+
+    if any(s in state_str for s in _ENGINE_FAILURE_STATES):
+        if task['status'] != 'failed':
+            task['status'] = 'failed'
+            err = getattr(record, 'error_message', None) or getattr(record, 'error', None) or ''
+            task['error_message'] = (
+                str(err) if err
+                else f'{username} download failed (engine state: {state_str})'
+            )
+            logger.info(
+                "[Engine Fallback] Task %s engine reports '%s' — marking failed",
+                task_id, state_str,
+            )
+            _schedule_completion_callback(deps, batch_id, task_id, False)
+        task_status['status'] = 'failed'
+        task_status['error_message'] = task.get('error_message')
+        task_status['progress'] = _engine_progress_pct(record)
+        return
+
+    if any(s in state_str for s in _ENGINE_SUCCESS_STATES):
+        if task['status'] != 'post_processing':
+            task['status'] = 'post_processing'
+            logger.info(
+                "[Engine Fallback] Task %s engine reports '%s' — starting post-processing verification",
+                task_id, state_str,
+            )
+            try:
+                deps.submit_post_processing(task_id, batch_id)
+            except Exception as exc:
+                logger.error(
+                    "[Engine Fallback] submit_post_processing raised for task %s: %s",
+                    task_id, exc,
+                )
+        task_status['status'] = 'post_processing'
+        task_status['progress'] = 95
+        return
+
+    if 'InProgress' in state_str:
+        task_status['status'] = 'downloading'
+        if task['status'] in ('searching', 'queued'):
+            task['status'] = 'downloading'
+    elif 'Queued' in state_str:
+        task_status['status'] = 'queued'
+    task_status['progress'] = _engine_progress_pct(record)


 def build_batch_status_data(batch_id: str, batch: dict, live_transfers_lookup: dict, deps: StatusDeps) -> dict:
@ -144,17 +332,41 @@ def build_batch_status_data(batch_id: str, batch: dict, live_transfers_lookup: d
                            task_status['status'] = 'cancelled'
                            task['status'] = 'cancelled'
                        elif 'Failed' in state_str or 'Errored' in state_str or 'Rejected' in state_str or 'TimedOut' in state_str:
-                            # UNIFIED ERROR HANDLING: Let monitor handle errors for consistency
-                            # Monitor will detect errored state and trigger retry within 5 seconds
-                            logger.error(f"Task {task_id} API shows error state: {state_str} - letting monitor handle retry")
-
-                            # Keep task in current status (downloading/queued) so monitor can detect error
-                            # Don't mark as failed here - let the unified retry system handle it
-                            if task['status'] in ['searching', 'downloading', 'queued']:
-                                task_status['status'] = task['status']  # Keep current status for monitor
+                            # User-initiated manual pick — surface the failure
+                            # immediately. The monitor's auto-retry path is gated
+                            # on `_user_manual_pick` and won't fire, so deferring
+                            # to it would leave the task stuck at 'downloading 0%'
+                            # forever. Mark failed here and free the worker slot.
+                            if task.get('_user_manual_pick'):
+                                err_msg = live_info.get('errorMessage') or live_info.get('error') or ''
+                                task['status'] = 'failed'
+                                task['error_message'] = (
+                                    str(err_msg) if err_msg
+                                    else f'Manual pick failed (state: {state_str})'
+                                )
+                                task_status['status'] = 'failed'
+                                task_status['error_message'] = task['error_message']
+                                logger.info(
+                                    f"[Manual Pick] Task {task_id} engine reports '{state_str}' — marking failed"
+                                )
+                                # NOTE: caller (build_batched_status) holds
+                                # tasks_lock. on_download_completed re-acquires
+                                # the same Lock — synchronous call would
+                                # deadlock. Spawn a thread so it runs after we
+                                # release the lock.
+                                _schedule_completion_callback(deps, batch_id, task_id, False)
                            else:
-                                task_status['status'] = 'downloading'  # Default to downloading for error detection
-                                task['status'] = 'downloading'
+                                # UNIFIED ERROR HANDLING: Let monitor handle errors for consistency
+                                # Monitor will detect errored state and trigger retry within 5 seconds
+                                logger.error(f"Task {task_id} API shows error state: {state_str} - letting monitor handle retry")
+
+                                # Keep task in current status (downloading/queued) so monitor can detect error
+                                # Don't mark as failed here - let the unified retry system handle it
+                                if task['status'] in ['searching', 'downloading', 'queued']:
+                                    task_status['status'] = task['status']  # Keep current status for monitor
+                                else:
+                                    task_status['status'] = 'downloading'  # Default to downloading for error detection
+                                    task['status'] = 'downloading'
                        elif 'Completed' in state_str or 'Succeeded' in state_str:
                            # Verify bytes actually transferred before trusting state string
                            expected_size = live_info.get('size', 0)
@ -193,6 +405,15 @@ def build_batch_status_data(batch_id: str, batch: dict, live_transfers_lookup: d
                        task_status['progress'] = 100
                    elif task['status'] == 'post_processing':
                        task_status['progress'] = 95  # Nearly complete, just verifying
+                    else:
+                        # Non-Soulseek (streaming) sources don't appear in
+                        # slskd's live_transfers_lookup — poll the engine
+                        # directly so YouTube/Tidal/Qobuz/HiFi/Deezer/
+                        # SoundCloud/Lidarr tasks actually advance out of
+                        # 'downloading 0%' instead of staying there forever.
+                        _apply_engine_state_fallback(
+                            task_id, task, task_status, batch_id, deps,
+                        )
            batch_tasks.append(task_status)
        batch_tasks.sort(key=lambda x: x['track_index'])
        response_data['tasks'] = batch_tasks
--- a/tests/test_manual_pick_no_auto_retry.py
+++ b/tests/test_manual_pick_no_auto_retry.py
@ -0,0 +1,95 @@
+"""Pin the ``_user_manual_pick`` flag — auto-retry monitor must not
+yank a manually-picked download back to 'searching' when it fails.
+
+User report: "I manually searched and selected a candidate. It went
+to 0% downloading, then suddenly switched back to searching." The
+download monitor's ``_should_retry_task`` was treating the failed
+manual pick the same as a normal auto-attempt failure — fresh search,
+new query, new candidates. From the user's perspective: "I picked
+THIS one, why is it searching for something else now?"
+
+Fix: when the user explicitly selects a candidate via
+``/api/downloads/task/<id>/download-candidate``, the task is flagged
+``_user_manual_pick=True``. The monitor's retry decision checks that
+flag and bails — letting the failure surface to the user instead of
+auto-falling-back.
+
+These tests exercise ``_should_retry_task`` directly with the flag
+set + various engine/transfer states.
+"""
+
+from __future__ import annotations
+
+import time
+
+import pytest
+
+from core.downloads import monitor as dm
+
+
+@pytest.fixture
+def fake_monitor(monkeypatch):
+    """Monitor with patched make_context_key so tests don't pull in the
+    real one."""
+    monkeypatch.setattr(dm, '_make_context_key', lambda u, f: f"{u}::{f}")
+    return dm.WebUIDownloadMonitor()
+
+
+def _task(*, username='youtube', filename='vid.mp3', status='downloading',
+          download_id='dl-1', manual_pick=False, status_change_time=None):
+    return {
+        'track_info': {'name': 'Test Track'},
+        'username': username,
+        'filename': filename,
+        'status': status,
+        'download_id': download_id,
+        '_user_manual_pick': manual_pick,
+        'status_change_time': status_change_time or (time.time() - 1000),
+    }
+
+
+def test_manual_pick_skips_retry_when_not_in_live_transfers(fake_monitor):
+    """The not-in-live-transfers stuck-task path normally triggers
+    retry after 90s. Manual picks must bypass it — no retry submitted,
+    no status reset to 'searching'."""
+    task = _task(manual_pick=True)
+    deferred_ops = []
+
+    result = fake_monitor._should_retry_task(
+        task_id='t1',
+        task=task,
+        live_transfers_lookup={},
+        current_time=time.time(),
+        deferred_ops=deferred_ops,
+    )
+
+    assert result is False
+    assert deferred_ops == []
+    assert task['status'] == 'downloading'
+
+
+def test_manual_pick_skips_retry_on_errored_state(fake_monitor):
+    """When the task IS in live_transfers but the engine reports
+    Errored, the monitor would normally retry. Manual picks bypass
+    that path too."""
+    task = _task(manual_pick=True)
+    deferred_ops = []
+    live_lookup = {
+        'youtube::vid.mp3': {
+            'state': 'Completed, Errored',
+            'percentComplete': 0,
+        }
+    }
+
+    fake_monitor._should_retry_task(
+        task_id='t1',
+        task=task,
+        live_transfers_lookup=live_lookup,
+        current_time=time.time(),
+        deferred_ops=deferred_ops,
+    )
+
+    assert deferred_ops == []
+    assert task['status'] == 'downloading'
+
+
--- a/tests/test_manual_search_endpoint.py
+++ b/tests/test_manual_search_endpoint.py
@ -0,0 +1,498 @@
+"""Tests for the /api/downloads/task/<id>/manual-search endpoint.
+
+The candidates modal lets the user click an auto-found candidate to retry
+a failed download. Manual search adds a second avenue — type a query, hit
+search, get fresh results from the configured download source(s) without
+having to leave the modal.
+
+The endpoint streams results as NDJSON — one JSON object per line — so
+the modal can render rows from each source as that source's search
+completes, instead of blocking the whole UI on the slowest source. The
+``_consume_ndjson`` helper below replays the stream as a list of message
+dicts so test assertions stay readable.
+
+These tests cover the new endpoint's validation + dispatch behavior:
+
+- Query length / source whitelist validation
+- Hybrid mode 'all' fans out across every configured source in parallel
+- Per-source request hits only the named source
+- Unconfigured sources in hybrid mode are silently skipped
+- Task lookup gates the endpoint with a 404 when the task isn't known
+- Per-source exceptions emit ``source_error`` events but don't fail the
+  overall stream
+
+The existing ``/download-candidate`` retry path is unchanged — manual-
+search results POST through the same endpoint so AcoustID + post-download
+safety nets stay in the loop.
+"""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+def _consume_ndjson(resp) -> list:
+    """Parse a Flask test-client streaming response into a list of
+    NDJSON message dicts. The endpoint emits one JSON object per line,
+    each terminated by ``\\n``.
+    """
+    raw = resp.get_data(as_text=True)
+    msgs = []
+    for line in raw.splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        msgs.append(json.loads(line))
+    return msgs
+
+
+def _flatten_candidates(msgs: list) -> list:
+    """Pull all candidate dicts out of every ``source_results`` message —
+    same flat list the old single-shot response used to return."""
+    out = []
+    for m in msgs:
+        if m.get('type') == 'source_results':
+            out.extend(m.get('candidates', []))
+    return out
+
+
+# ---------------------------------------------------------------------------
+# Fixture — Flask test client + mocked plugin registry.
+# ---------------------------------------------------------------------------
+
+
+def _async_return(value):
+    """Return a coroutine that resolves to ``value`` — passed to plugins'
+    .search() so run_async() can drive it like a real awaitable."""
+    async def _coro():
+        return value
+    return _coro()
+
+
+def _fake_track_result(filename: str, source: str, username_override: str = None):
+    """Build a TrackResult-shaped MagicMock that the serializer can read."""
+    mock = MagicMock()
+    mock.filename = filename
+    mock.username = username_override or source  # streaming sources stamp the source name
+    mock.size = 1024 * 1024 * 4
+    mock.bitrate = 320
+    mock.duration = 200_000
+    mock.quality = 'mp3'
+    mock.free_upload_slots = 1
+    mock.upload_speed = 100_000
+    mock.queue_length = 0
+    mock.artist = 'Test Artist'
+    mock.title = 'Test Title'
+    mock.album = 'Test Album'
+    # `hasattr(c, '__dict__')` is what _serialize_candidate uses to detect
+    # dataclass-shaped inputs. MagicMock has __dict__, so this works.
+    return mock
+
+
+def _make_plugin(search_results=None, configured=True):
+    """Stand-in for a download-source plugin. Records calls to .search()
+    so tests can assert which sources got dispatched."""
+    plugin = MagicMock()
+    plugin.is_configured = MagicMock(return_value=configured)
+    # Each call returns a fresh coroutine — async functions can't be
+    # awaited twice, so the side_effect rebuilds the awaitable each time.
+    plugin.search = MagicMock(
+        side_effect=lambda *args, **kwargs: _async_return((search_results or [], []))
+    )
+    return plugin
+
+
+@pytest.fixture
+def manual_search_client(monkeypatch):
+    """Flask test client with a fully mocked download_orchestrator + a
+    seeded download_tasks entry. Each test reaches into the plugin mocks
+    via the returned ``ctx`` dict to assert dispatch behavior."""
+    # Avoid the real activity-feed side effects.
+    with patch("web_server.add_activity_item"):
+        # Mock external service singletons so import doesn't try to spin up
+        # real clients / hit real APIs at module-load time.
+        with patch("web_server.SpotifyClient"):
+            with patch("core.tidal_client.TidalClient"):
+                from web_server import app as flask_app
+                import web_server
+
+                flask_app.config['TESTING'] = True
+
+                # Build a fresh registry-like object with deterministic plugins
+                # — bypasses the eight real clients the orchestrator instantiates.
+                plugins = {
+                    'soulseek': _make_plugin(),
+                    'youtube':  _make_plugin(),
+                    'tidal':    _make_plugin(configured=False),  # not configured
+                    'qobuz':    _make_plugin(),
+                    'hifi':     _make_plugin(),
+                    'deezer':   _make_plugin(),
+                }
+
+                class _FakeSpec:
+                    def __init__(self, name):
+                        self.name = name
+                        self.display_name = name.title()
+                        self.aliases = ()
+
+                class _FakeRegistry:
+                    def __init__(self, plugins_dict):
+                        self._plugins = plugins_dict
+
+                    def get(self, name):
+                        return self._plugins.get(name)
+
+                    def get_spec(self, name):
+                        return _FakeSpec(name) if name in self._plugins else None
+
+                    def names(self):
+                        return list(self._plugins.keys())
+
+                    def all_plugins(self):
+                        return list(self._plugins.items())
+
+                fake_orch = MagicMock()
+                fake_orch.registry = _FakeRegistry(plugins)
+                fake_orch.client = MagicMock(side_effect=lambda name: plugins.get(name))
+
+                monkeypatch.setattr(web_server, 'download_orchestrator', fake_orch)
+
+                # run_async drives the awaitable each plugin.search() returns —
+                # the real one schedules onto the asyncio loop. The default
+                # implementation in utils.async_helpers handles this fine,
+                # but force a deterministic synchronous version so test
+                # ordering is predictable.
+                def _sync_run_async(coro):
+                    import asyncio
+                    loop = asyncio.new_event_loop()
+                    try:
+                        return loop.run_until_complete(coro)
+                    finally:
+                        loop.close()
+                monkeypatch.setattr(web_server, 'run_async', _sync_run_async)
+
+                # Seed download_tasks so the endpoint finds a real task.
+                from core.runtime_state import download_tasks, tasks_lock
+                with tasks_lock:
+                    download_tasks.clear()
+                    download_tasks['task-abc'] = {
+                        'status': 'failed',
+                        'track_info': {
+                            'name': 'Test Track',
+                            'artists': [{'name': 'Test Artist'}],
+                            'duration_ms': 200_000,
+                        },
+                        'cached_candidates': [],
+                    }
+
+                # Default config: hybrid mode with all six in hybrid_order.
+                # Individual tests override this.
+                from config.settings import config_manager
+                original_get = config_manager.get
+
+                def _fake_config_get(key, default=None):
+                    if key == 'download_source.mode':
+                        return 'hybrid'
+                    if key == 'download_source.hybrid_order':
+                        return ['soulseek', 'youtube', 'tidal', 'qobuz', 'hifi', 'deezer']
+                    return original_get(key, default)
+                monkeypatch.setattr(config_manager, 'get', _fake_config_get)
+
+                ctx = {
+                    'plugins': plugins,
+                    'config_get_setter': lambda fn: monkeypatch.setattr(config_manager, 'get', fn),
+                }
+
+                yield flask_app.test_client(), ctx
+
+                with tasks_lock:
+                    download_tasks.clear()
+
+
+# ---------------------------------------------------------------------------
+# Validation
+# ---------------------------------------------------------------------------
+
+
+def test_manual_search_validates_query_length(manual_search_client):
+    """Empty / 1-char query returns 400 — frontend hint says ≥2 chars."""
+    client, _ctx = manual_search_client
+
+    for bad_query in ['', ' ', 'a', '  a  ']:
+        resp = client.post(
+            '/api/downloads/task/task-abc/manual-search',
+            json={'query': bad_query, 'source': 'all'},
+        )
+        assert resp.status_code == 400, f"query={bad_query!r} should 400"
+        body = resp.get_json()
+        assert 'error' in body
+
+
+def test_manual_search_validates_source(manual_search_client):
+    """Source must be 'all' or one of the configured source ids — anything
+    else returns 400. Prevents users from triggering searches against a
+    source the user explicitly disabled."""
+    client, _ctx = manual_search_client
+
+    resp = client.post(
+        '/api/downloads/task/task-abc/manual-search',
+        json={'query': 'drake feelings', 'source': 'made_up_source'},
+    )
+    assert resp.status_code == 400
+    assert 'error' in resp.get_json()
+
+
+def test_manual_search_handles_task_not_found(manual_search_client):
+    """Unknown task_id returns 404 — same gate as the existing /candidates
+    and /download-candidate endpoints."""
+    client, _ctx = manual_search_client
+
+    resp = client.post(
+        '/api/downloads/task/no-such-task/manual-search',
+        json={'query': 'drake feelings', 'source': 'all'},
+    )
+    assert resp.status_code == 404
+    assert 'error' in resp.get_json()
+
+
+# ---------------------------------------------------------------------------
+# Dispatch behavior — single source vs. parallel "all"
+# ---------------------------------------------------------------------------
+
+
+def test_manual_search_dispatches_to_configured_source_only(manual_search_client):
+    """In hybrid mode, source='youtube' should hit only the youtube plugin's
+    search — not soulseek, hifi, etc."""
+    client, ctx = manual_search_client
+    plugins = ctx['plugins']
+
+    plugins['youtube'] = _make_plugin(
+        search_results=[_fake_track_result('youtube_song.mp3', 'youtube')]
+    )
+    import web_server
+    web_server.download_orchestrator.registry._plugins['youtube'] = plugins['youtube']
+
+    resp = client.post(
+        '/api/downloads/task/task-abc/manual-search',
+        json={'query': 'drake feelings', 'source': 'youtube'},
+    )
+
+    assert resp.status_code == 200
+    msgs = _consume_ndjson(resp)
+    candidates = _flatten_candidates(msgs)
+    assert len(candidates) == 1
+    assert plugins['youtube'].search.call_count == 1
+    assert plugins['soulseek'].search.call_count == 0
+    assert plugins['hifi'].search.call_count == 0
+    assert plugins['qobuz'].search.call_count == 0
+
+
+def test_manual_search_all_dispatches_parallel(manual_search_client):
+    """source='all' with hybrid mode → searches every CONFIGURED source.
+    Tidal is unconfigured so it's filtered out — 5 of 6 sources hit."""
+    client, ctx = manual_search_client
+    plugins = ctx['plugins']
+
+    import web_server
+    for src_name in ('soulseek', 'youtube', 'qobuz', 'hifi', 'deezer'):
+        new_plugin = _make_plugin(
+            search_results=[_fake_track_result(f'{src_name}_song.mp3', src_name)]
+        )
+        plugins[src_name] = new_plugin
+        web_server.download_orchestrator.registry._plugins[src_name] = new_plugin
+
+    resp = client.post(
+        '/api/downloads/task/task-abc/manual-search',
+        json={'query': 'drake feelings', 'source': 'all'},
+    )
+
+    assert resp.status_code == 200
+    msgs = _consume_ndjson(resp)
+    candidates = _flatten_candidates(msgs)
+    assert len(candidates) == 5
+    for src_name in ('soulseek', 'youtube', 'qobuz', 'hifi', 'deezer'):
+        assert plugins[src_name].search.call_count == 1
+    assert plugins['tidal'].search.call_count == 0
+
+
+def test_manual_search_streams_one_event_per_source(manual_search_client):
+    """source='all' must emit one ``source_results`` event per configured
+    source — not a single batched event. That's what lets the frontend
+    render rows as they arrive instead of waiting for the slowest source."""
+    client, ctx = manual_search_client
+    plugins = ctx['plugins']
+
+    import web_server
+    for src_name in ('soulseek', 'youtube', 'qobuz', 'hifi', 'deezer'):
+        new_plugin = _make_plugin(
+            search_results=[_fake_track_result(f'{src_name}_song.mp3', src_name)]
+        )
+        plugins[src_name] = new_plugin
+        web_server.download_orchestrator.registry._plugins[src_name] = new_plugin
+
+    resp = client.post(
+        '/api/downloads/task/task-abc/manual-search',
+        json={'query': 'drake feelings', 'source': 'all'},
+    )
+
+    msgs = _consume_ndjson(resp)
+    source_events = [m for m in msgs if m.get('type') == 'source_results']
+    seen_sources = {m['source'] for m in source_events}
+    assert seen_sources == {'soulseek', 'youtube', 'qobuz', 'hifi', 'deezer'}
+    # One header + one per source + one done terminator
+    assert msgs[0]['type'] == 'header'
+    assert msgs[-1]['type'] == 'done'
+    assert msgs[-1]['total'] == 5
+
+
+def test_manual_search_skips_unconfigured_sources(manual_search_client):
+    """Sources whose is_configured() returns False are excluded from the
+    'all' dispatch list."""
+    client, ctx = manual_search_client
+    plugins = ctx['plugins']
+
+    resp = client.post(
+        '/api/downloads/task/task-abc/manual-search',
+        json={'query': 'something', 'source': 'all'},
+    )
+
+    assert resp.status_code == 200
+    msgs = _consume_ndjson(resp)
+    header = msgs[0]
+    assert header['type'] == 'header'
+    available_ids = {s['id'] for s in header['available_sources']}
+    assert 'tidal' not in available_ids
+    assert {'soulseek', 'youtube', 'qobuz', 'hifi', 'deezer'} <= available_ids
+    assert plugins['tidal'].search.call_count == 0
+
+
+def test_manual_search_rejects_unconfigured_source_explicitly(manual_search_client):
+    """User can't bypass the 'all' filter by naming an unconfigured source
+    directly."""
+    client, _ctx = manual_search_client
+
+    resp = client.post(
+        '/api/downloads/task/task-abc/manual-search',
+        json={'query': 'something', 'source': 'tidal'},
+    )
+
+    assert resp.status_code == 400
+
+
+# ---------------------------------------------------------------------------
+# Response shape
+# ---------------------------------------------------------------------------
+
+
+def test_manual_search_header_carries_track_and_source_metadata(manual_search_client):
+    """The first NDJSON line is a ``header`` event carrying track_info,
+    download_mode, available_sources, and the echoed query — everything
+    the frontend needs to render the modal shell before any results
+    arrive."""
+    client, ctx = manual_search_client
+    plugins = ctx['plugins']
+
+    import web_server
+    plugins['youtube'] = _make_plugin(
+        search_results=[_fake_track_result('youtube_song.mp3', 'youtube')]
+    )
+    web_server.download_orchestrator.registry._plugins['youtube'] = plugins['youtube']
+
+    resp = client.post(
+        '/api/downloads/task/task-abc/manual-search',
+        json={'query': 'drake feelings', 'source': 'youtube'},
+    )
+
+    assert resp.status_code == 200
+    msgs = _consume_ndjson(resp)
+    header = msgs[0]
+    assert header['type'] == 'header'
+    assert header['task_id'] == 'task-abc'
+    assert header['track_info']['name'] == 'Test Track'
+    assert header['download_mode'] == 'hybrid'
+    assert isinstance(header['available_sources'], list)
+    assert header['query'] == 'drake feelings'
+    assert header['sources_queried'] == ['youtube']
+
+    candidates = _flatten_candidates(msgs)
+    assert len(candidates) == 1
+    for candidate in candidates:
+        assert candidate['source'] == 'youtube'
+        for field in ('username', 'filename', 'size', 'quality',
+                      'duration', 'bitrate', 'queue_length',
+                      'free_upload_slots'):
+            assert field in candidate, f"missing {field}"
+
+
+def test_manual_search_single_source_mode_only_offers_one_source(monkeypatch, manual_search_client):
+    """When download_source.mode is a single source, available_sources
+    contains just that one entry. Frontend swaps the dropdown for a static
+    label in this case."""
+    client, _ctx = manual_search_client
+
+    from config.settings import config_manager
+    monkeypatch.setattr(
+        config_manager, 'get',
+        lambda key, default=None: (
+            'soulseek' if key == 'download_source.mode'
+            else (['soulseek', 'youtube'] if key == 'download_source.hybrid_order' else default)
+        ),
+    )
+
+    resp = client.post(
+        '/api/downloads/task/task-abc/manual-search',
+        json={'query': 'something', 'source': 'soulseek'},
+    )
+
+    assert resp.status_code == 200
+    msgs = _consume_ndjson(resp)
+    header = msgs[0]
+    assert header['download_mode'] == 'soulseek'
+    available_ids = [s['id'] for s in header['available_sources']]
+    assert available_ids == ['soulseek']
+
+
+def test_manual_search_handles_plugin_exception_gracefully(manual_search_client):
+    """If one source's .search() raises, the endpoint emits a
+    ``source_error`` event for it but other sources' results still come
+    through. The whole stream doesn't fail."""
+    client, ctx = manual_search_client
+    plugins = ctx['plugins']
+
+    import web_server
+
+    flaky_plugin = MagicMock()
+    flaky_plugin.is_configured = MagicMock(return_value=True)
+
+    def _raise(*args, **kwargs):
+        async def _coro():
+            raise RuntimeError("network blip")
+        return _coro()
+
+    flaky_plugin.search = MagicMock(side_effect=_raise)
+    plugins['soulseek'] = flaky_plugin
+    web_server.download_orchestrator.registry._plugins['soulseek'] = flaky_plugin
+
+    plugins['youtube'] = _make_plugin(
+        search_results=[_fake_track_result('youtube_song.mp3', 'youtube')]
+    )
+    web_server.download_orchestrator.registry._plugins['youtube'] = plugins['youtube']
+
+    resp = client.post(
+        '/api/downloads/task/task-abc/manual-search',
+        json={'query': 'something', 'source': 'all'},
+    )
+
+    assert resp.status_code == 200
+    msgs = _consume_ndjson(resp)
+
+    error_events = [m for m in msgs if m.get('type') == 'source_error']
+    assert any(m['source'] == 'soulseek' for m in error_events)
+    assert any('network blip' in m.get('error', '') for m in error_events)
+
+    candidates = _flatten_candidates(msgs)
+    yt_results = [c for c in candidates if c.get('source') == 'youtube']
+    assert len(yt_results) == 1
--- a/tests/test_status_engine_fallback.py
+++ b/tests/test_status_engine_fallback.py
@ -0,0 +1,469 @@
+"""Pin the engine-state fallback that drives non-Soulseek (streaming)
+download status forward.
+
+Soulseek downloads land in slskd's ``live_transfers_lookup``, so their
+status updates flow through the existing slskd-state branch. Streaming
+sources (YouTube, Tidal, Qobuz, HiFi, Deezer, SoundCloud, Lidarr) never
+appear there — without these tests' code path, a manually-picked
+SoundCloud download stays at "downloading 0%" forever, even after the
+engine logs an Errored terminal state.
+
+These tests exercise ``_apply_engine_state_fallback`` directly with a
+fake ``download_orchestrator`` so we don't have to spin up the real
+engine. The real fix relies on the per-source plugin storing the
+terminal state via ``_mark_terminal`` (state='Errored' / 'Completed,
+Succeeded'), which our fake mirrors.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Optional
+from unittest.mock import MagicMock
+
+import pytest
+
+from core.downloads import status as status_mod
+
+
+@dataclass
+class _FakeDownloadStatus:
+    """Mirror the DownloadStatus shape that engine plugins return — only
+    the fields the fallback reads."""
+    id: str
+    state: str
+    progress: float = 0
+    error_message: Optional[str] = None
+
+
+def _make_deps(record_for_id: dict, on_completed=None, submit_pp=None):
+    """StatusDeps with a fake orchestrator that returns whatever
+    DownloadStatus the test provides for a given download_id."""
+
+    fake_orch = MagicMock()
+
+    async def _fake_get_status(download_id):
+        return record_for_id.get(download_id)
+
+    fake_orch.get_download_status = _fake_get_status
+
+    def _sync_run_async(coro):
+        import asyncio
+        loop = asyncio.new_event_loop()
+        try:
+            return loop.run_until_complete(coro)
+        finally:
+            loop.close()
+
+    return status_mod.StatusDeps(
+        config_manager=MagicMock(),
+        docker_resolve_path=lambda p: p,
+        find_completed_file=lambda *args, **kwargs: (None, None),
+        make_context_key=lambda u, f: f"{u}::{f}",
+        submit_post_processing=submit_pp or (lambda task_id, batch_id: None),
+        get_cached_transfer_data=lambda: {},
+        download_orchestrator=fake_orch,
+        run_async=_sync_run_async,
+        on_download_completed=on_completed,
+    )
+
+
+def _task(*, status='downloading', username='soundcloud', filename='1234||https://sc/x',
+          download_id='dl-1', manual_pick=True):
+    """Default to manual_pick=True because the engine fallback is
+    deliberately scoped to manual picks — auto attempts go through the
+    live_transfers branch + monitor retry path. Tests opt out of the
+    flag explicitly when exercising the auto-attempt skip."""
+    return {
+        'status': status,
+        'username': username,
+        'filename': filename,
+        'download_id': download_id,
+        '_user_manual_pick': manual_pick,
+        'track_info': {'name': 'Test Track'},
+    }
+
+
+# ---------------------------------------------------------------------------
+# Failure / cancel / success transitions
+# ---------------------------------------------------------------------------
+
+
+def test_errored_state_marks_task_failed():
+    import time as _t
+    task = _task()
+    task_status = {'status': 'downloading', 'progress': 0}
+    completed_calls = []
+    deps = _make_deps(
+        {'dl-1': _FakeDownloadStatus(id='dl-1', state='Errored', error_message='HTTP 404')},
+        on_completed=lambda batch_id, task_id, success: completed_calls.append((batch_id, task_id, success)),
+    )
+
+    status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps)
+
+    assert task['status'] == 'failed'
+    assert task['error_message'] == 'HTTP 404'
+    assert task_status['status'] == 'failed'
+    # on_download_completed is deferred to a daemon thread to avoid the
+    # tasks_lock self-deadlock — give it a beat to fire.
+    for _ in range(50):
+        if completed_calls:
+            break
+        _t.sleep(0.01)
+    assert completed_calls == [('b1', 't1', False)]
+
+
+def test_compound_completed_errored_hits_failure_branch_first():
+    """``"Completed, Errored"`` must be treated as failure, not success.
+    Order of state-substring checks matters."""
+    task = _task()
+    task_status = {'status': 'downloading', 'progress': 0}
+    deps = _make_deps(
+        {'dl-1': _FakeDownloadStatus(id='dl-1', state='Completed, Errored')},
+        on_completed=lambda *args: None,
+    )
+
+    status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps)
+
+    assert task['status'] == 'failed'
+
+
+def test_cancelled_state_marks_task_cancelled():
+    import time as _t
+    task = _task()
+    task_status = {'status': 'downloading', 'progress': 0}
+    completed_calls = []
+    deps = _make_deps(
+        {'dl-1': _FakeDownloadStatus(id='dl-1', state='Cancelled')},
+        on_completed=lambda *args: completed_calls.append(args),
+    )
+
+    status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps)
+
+    assert task['status'] == 'cancelled'
+    assert task_status['status'] == 'cancelled'
+    for _ in range(50):
+        if completed_calls:
+            break
+        _t.sleep(0.01)
+    assert len(completed_calls) == 1
+
+
+def test_succeeded_state_submits_post_processing():
+    task = _task()
+    task_status = {'status': 'downloading', 'progress': 0}
+    pp_calls = []
+    deps = _make_deps(
+        {'dl-1': _FakeDownloadStatus(id='dl-1', state='Completed, Succeeded', progress=100)},
+        submit_pp=lambda task_id, batch_id: pp_calls.append((task_id, batch_id)),
+    )
+
+    status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps)
+
+    assert task['status'] == 'post_processing'
+    assert task_status['status'] == 'post_processing'
+    assert pp_calls == [('t1', 'b1')]
+
+
+def test_inprogress_reflects_progress_without_changing_status():
+    task = _task()
+    task_status = {'status': 'downloading', 'progress': 0}
+    deps = _make_deps(
+        {'dl-1': _FakeDownloadStatus(id='dl-1', state='InProgress, Downloading', progress=42.5)},
+    )
+
+    status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps)
+
+    assert task['status'] == 'downloading'
+    assert task_status['status'] == 'downloading'
+    assert task_status['progress'] == 42.5
+
+
+# ---------------------------------------------------------------------------
+# Gates — bail without mutating state
+# ---------------------------------------------------------------------------
+
+
+def test_skips_when_orchestrator_missing():
+    task = _task()
+    task_status = {'status': 'downloading', 'progress': 0}
+    deps = status_mod.StatusDeps(
+        config_manager=MagicMock(),
+        docker_resolve_path=lambda p: p,
+        find_completed_file=lambda *a, **k: (None, None),
+        make_context_key=lambda u, f: f"{u}::{f}",
+        submit_post_processing=lambda *a: None,
+        get_cached_transfer_data=lambda: {},
+        download_orchestrator=None,
+    )
+
+    status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps)
+
+    assert task['status'] == 'downloading'  # unchanged
+
+
+def test_skips_terminal_states():
+    """Already-failed / completed / cancelled tasks must not be touched —
+    they may have been marked by another path (e.g. the live_transfers
+    branch on a slskd Errored state for a Soulseek manual pick)."""
+    for terminal in ('completed', 'failed', 'cancelled', 'not_found', 'post_processing'):
+        task = _task(status=terminal)
+        task_status = {'status': terminal, 'progress': 0}
+        deps = _make_deps({'dl-1': _FakeDownloadStatus(id='dl-1', state='Errored')})
+
+        status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps)
+
+        assert task['status'] == terminal
+
+
+def test_skips_soulseek_username():
+    """Soulseek goes through live_transfers_lookup — never the engine
+    fallback. Otherwise we'd double-process its terminal state."""
+    task = _task(username='peer-username-xyz')  # not in _STREAMING_SOURCE_NAMES
+    task_status = {'status': 'downloading', 'progress': 0}
+    deps = _make_deps({'dl-1': _FakeDownloadStatus(id='dl-1', state='Errored')})
+
+    status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps)
+
+    assert task['status'] == 'downloading'
+
+
+def test_skips_when_download_id_missing():
+    task = _task()
+    task.pop('download_id')
+    task_status = {'status': 'downloading', 'progress': 0}
+    deps = _make_deps({})
+
+    status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps)
+
+    assert task['status'] == 'downloading'
+
+
+def test_engine_returning_none_leaves_task_alone():
+    """Engine doesn't know about this download_id (worker hasn't registered
+    yet, or the record was cleaned). The fallback must not falsely mark
+    the task failed in this case — the safety valve covers stuck-forever."""
+    task = _task()
+    task_status = {'status': 'downloading', 'progress': 0}
+    deps = _make_deps({'dl-1': None})
+
+    status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps)
+
+    assert task['status'] == 'downloading'
+
+
+def test_skips_auto_attempts_without_manual_pick_flag():
+    """Auto attempts (no _user_manual_pick flag) must NOT hit the engine
+    fallback even if they end up in the else branch. The monitor's retry
+    path owns auto-attempt failure handling — short-circuiting it here
+    would skip the fallback-to-next-candidate behavior."""
+    task = _task(manual_pick=False)
+    task_status = {'status': 'downloading', 'progress': 0}
+    deps = _make_deps({'dl-1': _FakeDownloadStatus(id='dl-1', state='Errored')})
+
+    status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps)
+
+    # Untouched — auto retry path will handle it.
+    assert task['status'] == 'downloading'
+
+
+# ---------------------------------------------------------------------------
+# Live-transfers IF branch — manual-pick failure path
+# ---------------------------------------------------------------------------
+#
+# Streaming-source records are pre-populated into ``live_transfers_lookup``
+# via ``download_orchestrator.engine.get_all_downloads(exclude=('soulseek',))``,
+# so a manually-picked SoundCloud / YouTube / Tidal / etc. download whose
+# engine record reports ``state='Errored'`` arrives via the IF branch
+# (lookup_key IS in live_transfers_lookup), NOT the engine-fallback else
+# branch. Without the manual-pick guard inside that elif, the live-
+# transfers branch would defer to the monitor — which itself bails on
+# manual picks — and the task would sit at "downloading 0%" forever.
+#
+# These tests exercise ``build_batch_status_data`` end-to-end so the
+# guard is pinned by behavior rather than by the unit-level fallback
+# tests above.
+
+
+def _seed_runtime(batch_id, task_id, *, manual_pick: bool):
+    import time as _t
+    from core.runtime_state import download_batches, download_tasks, tasks_lock
+
+    with tasks_lock:
+        download_tasks[task_id] = {
+            'status': 'downloading',
+            'username': 'soundcloud',
+            'filename': '1234||https://sc/x||Display Name',
+            'download_id': 'dl-1',
+            '_user_manual_pick': manual_pick,
+            'track_info': {'name': 'Test Track'},
+            'track_index': 0,
+            # Recent so the safety-valve "stuck-too-long" branch doesn't fire.
+            'status_change_time': _t.time(),
+            'cached_candidates': [],
+        }
+        download_batches[batch_id] = {
+            'phase': 'downloading',
+            'queue': [task_id],
+            'analysis_results': [],
+            'active_count': 1,
+            'max_concurrent': 3,
+        }
+
+
+def _clear_runtime():
+    from core.runtime_state import download_batches, download_tasks, tasks_lock
+    with tasks_lock:
+        download_tasks.clear()
+        download_batches.clear()
+
+
+@pytest.fixture
+def runtime():
+    """Seeded download_batches + download_tasks; cleared after each test."""
+    _clear_runtime()
+    yield
+    _clear_runtime()
+
+
+def _build_batch_deps(completed_calls):
+    """StatusDeps with a long timeout so the safety valve doesn't fire +
+    a captured ``on_download_completed`` we can assert against."""
+    fake_config = MagicMock()
+    fake_config.get = lambda key, default=None: 99999 if key == 'soulseek.download_timeout' else default
+
+    return status_mod.StatusDeps(
+        config_manager=fake_config,
+        docker_resolve_path=lambda p: p,
+        find_completed_file=lambda *a, **k: (None, None),
+        make_context_key=lambda u, f: f"{u}::{f}",
+        submit_post_processing=lambda *a: None,
+        get_cached_transfer_data=lambda: {},
+        download_orchestrator=None,  # IF branch doesn't need engine
+        run_async=None,
+        on_download_completed=lambda batch_id, task_id, success: completed_calls.append(
+            (batch_id, task_id, success)
+        ),
+    )
+
+
+def test_if_branch_manual_pick_marks_failed_on_errored(runtime):
+    """Manual-pick task whose live_transfers entry reports Errored —
+    must transition to 'failed' synchronously, not defer to the monitor."""
+    import time as _t
+    from core.runtime_state import download_batches
+
+    batch_id = 'b1'
+    task_id = 't1'
+    _seed_runtime(batch_id, task_id, manual_pick=True)
+
+    completed_calls = []
+    deps = _build_batch_deps(completed_calls)
+
+    live_transfers_lookup = {
+        'soundcloud::1234||https://sc/x||Display Name': {
+            'state': 'Errored',
+            'percentComplete': 0,
+            'errorMessage': 'HTTP 404 Not Found',
+        }
+    }
+
+    response = status_mod.build_batch_status_data(
+        batch_id, download_batches[batch_id], live_transfers_lookup, deps,
+    )
+
+    task_status = response['tasks'][0]
+    assert task_status['status'] == 'failed'
+    assert 'HTTP 404' in (task_status.get('error_message') or '')
+
+    from core.runtime_state import download_tasks
+    assert download_tasks[task_id]['status'] == 'failed'
+
+    # on_download_completed is deferred to a daemon thread — wait briefly.
+    for _ in range(50):
+        if completed_calls:
+            break
+        _t.sleep(0.01)
+    assert completed_calls == [(batch_id, task_id, False)]
+
+
+def test_if_branch_compound_completed_errored_hits_manual_pick_failure(runtime):
+    """``"Completed, Errored"`` must trigger the failure branch, not the
+    success branch. Slskd / engine pluginscan emit compound states when
+    a download technically completes but the file is corrupt / partial."""
+    from core.runtime_state import download_batches
+
+    batch_id = 'b1'
+    task_id = 't1'
+    _seed_runtime(batch_id, task_id, manual_pick=True)
+
+    deps = _build_batch_deps([])
+    live_transfers_lookup = {
+        'soundcloud::1234||https://sc/x||Display Name': {
+            'state': 'Completed, Errored',
+            'percentComplete': 100,
+        }
+    }
+
+    response = status_mod.build_batch_status_data(
+        batch_id, download_batches[batch_id], live_transfers_lookup, deps,
+    )
+
+    assert response['tasks'][0]['status'] == 'failed'
+
+
+def test_if_branch_auto_attempt_defers_to_monitor(runtime):
+    """Auto attempts (no manual-pick flag) keep the original "let monitor
+    handle retry" behavior — task stays in its current pre-error status
+    so the monitor's retry path can detect the Errored live_info on its
+    next tick. This is the byte-identical pre-fix behavior; the guard is
+    additive."""
+    from core.runtime_state import download_batches, download_tasks
+
+    batch_id = 'b1'
+    task_id = 't1'
+    _seed_runtime(batch_id, task_id, manual_pick=False)
+
+    deps = _build_batch_deps([])
+    live_transfers_lookup = {
+        'soundcloud::1234||https://sc/x||Display Name': {
+            'state': 'Errored',
+            'percentComplete': 0,
+        }
+    }
+
+    status_mod.build_batch_status_data(
+        batch_id, download_batches[batch_id], live_transfers_lookup, deps,
+    )
+
+    # Auto retry path keeps the task in 'downloading' so the monitor can
+    # observe the Errored state on its own poll. NOT marked failed here.
+    assert download_tasks[task_id]['status'] == 'downloading'
+
+
+def test_orchestrator_exception_swallowed():
+    """If get_download_status raises, the fallback logs + bails — it must
+    not propagate and crash the whole status response."""
+    task = _task()
+    task_status = {'status': 'downloading', 'progress': 0}
+
+    fake_orch = MagicMock()
+
+    async def _boom(_):
+        raise RuntimeError("network blip")
+
+    fake_orch.get_download_status = _boom
+    deps = status_mod.StatusDeps(
+        config_manager=MagicMock(),
+        docker_resolve_path=lambda p: p,
+        find_completed_file=lambda *a, **k: (None, None),
+        make_context_key=lambda u, f: f"{u}::{f}",
+        submit_post_processing=lambda *a: None,
+        get_cached_transfer_data=lambda: {},
+        download_orchestrator=fake_orch,
+        run_async=lambda coro: __import__('asyncio').new_event_loop().run_until_complete(coro),
+    )
+
+    status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps)
+
+    assert task['status'] == 'downloading'
--- a/web_server.py
+++ b/web_server.py
@ -3015,35 +3015,6 @@ atexit.register(_atexit_silence_shutdown_logger_errors)
 signal.signal(signal.SIGINT, signal_handler)
 signal.signal(signal.SIGTERM, signal_handler)

-def _handle_failed_download(batch_id, task_id, task, task_status):
-    """Handle failed download by triggering retry logic like GUI"""
-    try:
-        with tasks_lock:
-            if task_id not in download_tasks:
-                return
-                
-            retry_count = task.get('retry_count', 0)
-            task['retry_count'] = retry_count + 1
-            
-            if task['retry_count'] > 2:  # Max 3 attempts total (matches GUI)
-                # All retries exhausted, mark as permanently failed
-                logger.error(f"Task {task_id} failed after 3 retry attempts")
-                task_status['status'] = 'failed'
-                task['status'] = 'failed'
-                return
-            
-            # Show retrying status while we process retry
-            task_status['status'] = 'pending'  # Will show as pending until retry kicks in
-            logger.error(f"Triggering retry {task['retry_count']}/3 for failed task {task_id}")
-            
-        # Trigger retry with next candidate (matches GUI retry_parallel_download_with_fallback)
-        missing_download_executor.submit(download_monitor._retry_task_with_fallback, batch_id, task_id, task)
-        
-    except Exception as e:
-        logger.error(f"Error handling failed download {task_id}: {e}")
-        task_status['status'] = 'failed'
-        task['status'] = 'failed'
-
 def _update_task_status(task_id, new_status):
    """Helper to update task status and timestamp for timeout tracking"""
    with tasks_lock:
@ -7742,6 +7713,109 @@ def clear_finished_downloads():
        logger.error(f"Error clearing finished downloads: {e}")
        return jsonify({"success": False, "error": str(e)}), 500

+# Streaming sources where the candidate's `username` field IS the source name
+# (Soulseek uses a real peer username; everything else stamps the source string).
+_STREAMING_SOURCE_NAMES = frozenset((
+    'youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud'
+))
+
+
+def _infer_candidate_source(username: str) -> str:
+    """Infer which download source a candidate came from based on its
+    `username` field. Streaming sources stamp their canonical name there;
+    everything else is Soulseek."""
+    if not username:
+        return 'soulseek'
+    return username if username in _STREAMING_SOURCE_NAMES else 'soulseek'
+
+
+def _serialize_candidate(c, source_override: str = None) -> dict:
+    """Convert a TrackResult (or dict) into the JSON shape the candidates
+    modal expects. ``source_override`` lets manual-search callers stamp
+    the source explicitly when the dispatcher knows it; otherwise we
+    infer from the username."""
+    if hasattr(c, '__dict__'):
+        username = getattr(c, 'username', '')
+        return {
+            'username': username,
+            'filename': getattr(c, 'filename', ''),
+            'size': getattr(c, 'size', 0),
+            'bitrate': getattr(c, 'bitrate', None),
+            'duration': getattr(c, 'duration', None),
+            'quality': getattr(c, 'quality', ''),
+            'free_upload_slots': getattr(c, 'free_upload_slots', 0),
+            'upload_speed': getattr(c, 'upload_speed', 0),
+            'queue_length': getattr(c, 'queue_length', 0),
+            'artist': getattr(c, 'artist', None),
+            'title': getattr(c, 'title', None),
+            'album': getattr(c, 'album', None),
+            'source': source_override or _infer_candidate_source(username),
+        }
+    if isinstance(c, dict):
+        out = dict(c)
+        out.setdefault('source', source_override or _infer_candidate_source(out.get('username', '')))
+        return out
+    return {}
+
+
+def _list_available_download_sources() -> tuple:
+    """Return ``(download_mode, available_sources)`` for the current
+    download configuration. ``download_mode`` is the value of
+    ``download_source.mode`` (one of 'soulseek'/'youtube'/.../'hybrid').
+    ``available_sources`` is a list of ``{id, label}`` dicts — the
+    sources the manual-search dropdown should offer.
+
+    In single-source mode: returns just that one source if it's
+    initialized + configured (the user picked it, so we expose it
+    even if is_configured() doesn't fully approve — they may still
+    want to retry).
+
+    In hybrid mode: filters ``hybrid_order`` down to sources that are
+    BOTH initialized and ``is_configured()`` — same gate hybrid-mode
+    fallback already uses.
+    """
+    if not download_orchestrator:
+        return 'soulseek', []
+
+    download_mode = config_manager.get('download_source.mode', 'soulseek')
+    sources = []
+
+    def _make_entry(name: str) -> dict:
+        spec = download_orchestrator.registry.get_spec(name) if hasattr(download_orchestrator, 'registry') else None
+        return {
+            'id': name,
+            'label': spec.display_name if spec else name.title(),
+        }
+
+    if download_mode == 'hybrid':
+        hybrid_order = config_manager.get('download_source.hybrid_order',
+                                          ['hifi', 'youtube', 'soulseek']) or []
+        seen = set()
+        for raw_name in hybrid_order:
+            spec = download_orchestrator.registry.get_spec(raw_name) if hasattr(download_orchestrator, 'registry') else None
+            canonical = spec.name if spec else raw_name
+            if canonical in seen:
+                continue
+            client = download_orchestrator.client(canonical)
+            if not client:
+                continue
+            try:
+                if not client.is_configured():
+                    continue
+            except Exception:
+                continue
+            seen.add(canonical)
+            sources.append(_make_entry(canonical))
+    else:
+        # Single-source mode — just expose the configured mode (the user
+        # picked it, so they expect manual search to hit that source).
+        client = download_orchestrator.client(download_mode)
+        if client:
+            sources.append(_make_entry(download_mode))
+
+    return download_mode, sources
+
+
@app.route('/api/downloads/task/<task_id>/candidates', methods=['GET'])
 def get_task_candidates(task_id):
    """Returns the cached search candidates for a download task so the UI can show what was found."""
@ -7755,25 +7829,10 @@ def get_task_candidates(task_id):
            track_info = task.get('track_info', {})
            error_message = task.get('error_message', '')

-            serialized = []
-            for c in candidates:
-                if hasattr(c, '__dict__'):
-                    serialized.append({
-                        'username': getattr(c, 'username', ''),
-                        'filename': getattr(c, 'filename', ''),
-                        'size': getattr(c, 'size', 0),
-                        'bitrate': getattr(c, 'bitrate', None),
-                        'duration': getattr(c, 'duration', None),
-                        'quality': getattr(c, 'quality', ''),
-                        'free_upload_slots': getattr(c, 'free_upload_slots', 0),
-                        'upload_speed': getattr(c, 'upload_speed', 0),
-                        'queue_length': getattr(c, 'queue_length', 0),
-                        'artist': getattr(c, 'artist', None),
-                        'title': getattr(c, 'title', None),
-                        'album': getattr(c, 'album', None),
-                    })
-                elif isinstance(c, dict):
-                    serialized.append(c)
+            serialized = [_serialize_candidate(c) for c in candidates if c is not None]
+            serialized = [s for s in serialized if s]
+
+        download_mode, available_sources = _list_available_download_sources()

        return jsonify({
            "task_id": task_id,
@ -7784,6 +7843,8 @@ def get_task_candidates(task_id):
            "error_message": error_message,
            "candidates": serialized,
            "candidate_count": len(serialized),
+            "download_mode": download_mode,
+            "available_sources": available_sources,
        })
    except Exception as e:
        logger.error(f"[Candidates] Error fetching candidates for task {task_id}: {e}")
@ -7821,6 +7882,21 @@ def download_selected_candidate(task_id):
            task.pop('download_id', None)
            task.pop('username', None)
            task.pop('filename', None)
+            # Mark this as a user-initiated manual pick. The auto-retry
+            # monitor (`_should_retry_task`) and the engine-state status
+            # fallback both check this flag and skip the "fall back to
+            # another candidate via fresh search" behavior. When the user
+            # explicitly chose THIS file, the mental model is "try this
+            # one and tell me if it failed", not "try this, then auto-
+            # pick something else if it fails". Stays set until the task
+            # reaches a terminal state.
+            task['_user_manual_pick'] = True
+            # Reset retry counters so previous auto-attempts don't
+            # immediately exhaust the manual pick.
+            task.pop('stuck_retry_count', None)
+            task.pop('error_retry_count', None)
+            task.pop('last_retry_time', None)
+            task.pop('last_error_retry_time', None)
            # Clear the selected candidate from used_sources so it won't be skipped
            used_sources = task.get('used_sources', set())
            source_key = f"{username}_{filename}"
@ -7880,20 +7956,44 @@ def download_selected_candidate(task_id):
            popularity=0,
        )

-        # Submit to thread pool — don't block the request
+        track_name = track_info.get('name', 'Unknown')
+
+        # Run on a dedicated thread instead of `missing_download_executor`
+        # — that pool is shared with the batch's other in-flight tracks
+        # (3 workers total) and a saturated pool would queue the manual
+        # pick indefinitely, leaving the user stuck at "downloading 0%".
+        # Manual picks are user-initiated and infrequent; a fresh thread
+        # per pick is cheaper than starving them behind background work.
        def _run_manual_download():
-            success = _attempt_download_with_candidates(task_id, [candidate], track, batch_id)
-            if not success:
+            logger.info(f"[Manual Download] worker started for task {task_id} ({username} / {track_name})")
+            try:
+                success = _attempt_download_with_candidates(task_id, [candidate], track, batch_id)
+                logger.info(f"[Manual Download] worker finished for task {task_id} success={success}")
+                if not success:
+                    with tasks_lock:
+                        if task_id in download_tasks:
+                            download_tasks[task_id]['status'] = 'failed'
+                            download_tasks[task_id]['error_message'] = 'Manual download failed to start — source may be unavailable'
+                    if batch_id:
+                        _on_download_completed(batch_id, task_id, success=False)
+            except Exception as exc:
+                logger.exception(f"[Manual Download] worker crashed for task {task_id}: {exc}")
                with tasks_lock:
                    if task_id in download_tasks:
                        download_tasks[task_id]['status'] = 'failed'
-                        download_tasks[task_id]['error_message'] = 'Manual download failed to start — user may be offline'
+                        download_tasks[task_id]['error_message'] = f'Manual download crashed: {exc}'
                if batch_id:
-                    _on_download_completed(batch_id, task_id, success=False)
+                    try:
+                        _on_download_completed(batch_id, task_id, success=False)
+                    except Exception:
+                        logger.exception("[Manual Download] _on_download_completed cleanup also failed")

-        missing_download_executor.submit(_run_manual_download)
+        threading.Thread(
+            target=_run_manual_download,
+            name=f"manual-download-{task_id[:8]}",
+            daemon=True,
+        ).start()

-        track_name = track_info.get('name', 'Unknown')
        logger.info(f"[Manual Download] User selected candidate for '{track_name}' from {username}")
        return jsonify({"success": True, "message": f"Download initiated for '{track_name}'"})

@ -7903,6 +8003,136 @@ def download_selected_candidate(task_id):
        traceback.print_exc()
        return jsonify({"error": str(e)}), 500

+
+@app.route('/api/downloads/task/<task_id>/manual-search', methods=['POST'])
+def manual_search_for_task(task_id):
+    """Run a user-driven search against one (or all) configured download
+    sources and stream candidate results as NDJSON — one JSON object per
+    line, terminated by ``\\n``. Streaming lets the modal render results as
+    each source completes instead of blocking on the slowest source.
+
+    The candidates modal lets the user pick a result; that retry still
+    goes through ``/download-candidate``, so all AcoustID +
+    post-download safety nets stay in the loop.
+
+    Stream shape (one JSON object per line):
+    - ``{"type": "header", ...}`` — emitted first; carries ``track_info``,
+      ``download_mode``, ``available_sources``, ``query``,
+      ``sources_queried``.
+    - ``{"type": "source_results", "source": "<name>", "candidates": [...]}``
+      — one per source, emitted as that source's search completes.
+    - ``{"type": "source_error", "source": "<name>", "error": "<msg>"}``
+      — when a source's search raised.
+    - ``{"type": "done", "total": <int>}`` — terminator.
+    """
+    try:
+        data = request.get_json(silent=True) or {}
+        raw_query = data.get('query', '')
+        query = raw_query.strip() if isinstance(raw_query, str) else ''
+        source = data.get('source', 'all')
+
+        if len(query) < 2:
+            return jsonify({"error": "Query must be at least 2 characters"}), 400
+
+        with tasks_lock:
+            task = download_tasks.get(task_id)
+            if not task:
+                return jsonify({"error": "Task not found"}), 404
+            track_info = dict(task.get('track_info', {}))
+
+        download_mode, available_sources = _list_available_download_sources()
+        valid_source_ids = {s['id'] for s in available_sources}
+
+        if source != 'all':
+            if source not in valid_source_ids:
+                return jsonify({
+                    "error": f"Source '{source}' is not configured or available"
+                }), 400
+            sources_to_query = [source]
+        else:
+            sources_to_query = list(valid_source_ids)
+
+        track_payload = {
+            "name": track_info.get('name', 'Unknown'),
+            "artist": _get_track_artist_name(track_info) if isinstance(track_info, dict) else 'Unknown',
+        }
+
+        from concurrent.futures import ThreadPoolExecutor, as_completed
+
+        def _search_one(src_name: str):
+            client = download_orchestrator.client(src_name) if download_orchestrator else None
+            if not client:
+                return src_name, [], None
+            try:
+                result = run_async(client.search(query))
+                if isinstance(result, tuple):
+                    tracks = result[0] if result else []
+                else:
+                    tracks = result or []
+                return src_name, tracks, None
+            except Exception as exc:
+                logger.warning(f"[Manual Search] {src_name} search failed for query '{query}': {exc}")
+                return src_name, [], str(exc)
+
+        def _generate():
+            yield json.dumps({
+                "type": "header",
+                "task_id": task_id,
+                "track_info": track_payload,
+                "download_mode": download_mode,
+                "available_sources": available_sources,
+                "query": query,
+                "sources_queried": sources_to_query,
+            }) + "\n"
+
+            if not sources_to_query:
+                yield json.dumps({"type": "done", "total": 0}) + "\n"
+                return
+
+            total = 0
+            max_workers = min(8, max(1, len(sources_to_query)))
+            with ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix='manual-search') as executor:
+                futures = [executor.submit(_search_one, name) for name in sources_to_query]
+                for future in as_completed(futures):
+                    src_name, tracks, error = future.result()
+                    if error is not None:
+                        yield json.dumps({
+                            "type": "source_error",
+                            "source": src_name,
+                            "error": error,
+                        }) + "\n"
+                        continue
+                    serialized = []
+                    for t in tracks:
+                        s = _serialize_candidate(t, source_override=src_name)
+                        if s:
+                            serialized.append(s)
+                    total += len(serialized)
+                    yield json.dumps({
+                        "type": "source_results",
+                        "source": src_name,
+                        "candidates": serialized,
+                    }) + "\n"
+
+            logger.info(
+                f"[Manual Search] task={task_id} query='{query}' source={source} "
+                f"sources_queried={sources_to_query} results={total}"
+            )
+            yield json.dumps({"type": "done", "total": total}) + "\n"
+
+        return Response(
+            _generate(),
+            mimetype='application/x-ndjson',
+            headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'},
+        )
+
+    except Exception as e:
+        logger.error(f"[Manual Search] {e}")
+        import traceback
+        traceback.print_exc()
+        return jsonify({"error": str(e)}), 500
+
+
@app.route('/api/quarantine/clear', methods=['POST'])
 def clear_quarantine():
    """Delete all files and folders inside the ss_quarantine directory."""
@ -17495,6 +17725,9 @@ def _build_status_deps():
            _run_post_processing_worker, task_id, batch_id
        ),
        get_cached_transfer_data=get_cached_transfer_data,
+        download_orchestrator=download_orchestrator,
+        run_async=run_async,
+        on_download_completed=_on_download_completed,
    )


--- a/webui/static/downloads.js
+++ b/webui/static/downloads.js
@ -2903,6 +2903,45 @@ async function showCandidatesModal(taskId) {
    }
 }

+// Format helpers used by both auto-candidates and manual-search rendering.
+function _candidatesFmtSize(bytes) {
+    if (!bytes) return '-';
+    const units = ['B', 'KB', 'MB', 'GB'];
+    let s = bytes, u = 0;
+    while (s >= 1024 && u < units.length - 1) { s /= 1024; u++; }
+    return `${s.toFixed(1)} ${units[u]}`;
+}
+
+function _candidatesFmtDur(ms) {
+    if (!ms) return '-';
+    const sec = Math.floor(ms / 1000);
+    return `${Math.floor(sec / 60)}:${(sec % 60).toString().padStart(2, '0')}`;
+}
+
+// Build a single <tr> for the candidates table. ``rowClass`` lets the
+// manual-search renderer distinguish its rows from the auto-candidates
+// rows (different click binding scope). ``showSourceBadge`` adds a small
+// per-row source pill — used in hybrid "All sources" mode where the user
+// otherwise can't tell which source a row came from.
+function _renderCandidateRow(c, index, rowClass, showSourceBadge) {
+    const shortFile = c.filename ? c.filename.split(/[/\\]/).pop() : '-';
+    const qBadge = c.quality
+        ? `<span class="candidates-quality-badge candidates-quality-${c.quality.toLowerCase()}">${c.quality.toUpperCase()}</span>`
+        : '';
+    const sourceBadge = (showSourceBadge && c.source)
+        ? `<span class="candidates-source-badge" title="Source">${escapeHtml(c.source)}</span> `
+        : '';
+    return `<tr class="${rowClass}">
+        <td class="candidates-col-index">${index + 1}</td>
+        <td class="candidates-col-file" title="${escapeHtml(c.filename || '')}">${sourceBadge}${escapeHtml(shortFile)}</td>
+        <td class="candidates-col-quality">${qBadge}${c.bitrate ? ` ${c.bitrate}kbps` : ''}</td>
+        <td class="candidates-col-size">${_candidatesFmtSize(c.size)}</td>
+        <td class="candidates-col-duration">${_candidatesFmtDur(c.duration)}</td>
+        <td class="candidates-col-user" title="Queue: ${c.queue_length || 0} | Slots: ${c.free_upload_slots || 0}">${escapeHtml(c.username || '-')}</td>
+        <td class="candidates-col-action"><button class="candidates-download-btn" data-index="${index}" title="Download this file">⬇</button></td>
+    </tr>`;
+}
+
 function _renderCandidatesModal(data) {
    let overlay = document.getElementById('candidates-modal-overlay');
    if (overlay) overlay.remove();
@ -2911,42 +2950,57 @@ function _renderCandidatesModal(data) {
    const trackArtist = data.track_info?.artist || 'Unknown Artist';
    const candidates = data.candidates || [];
    const errorMsg = data.error_message || '';
-
-    const fmtSize = (bytes) => {
-        if (!bytes) return '-';
-        const units = ['B', 'KB', 'MB', 'GB'];
-        let s = bytes, u = 0;
-        while (s >= 1024 && u < units.length - 1) { s /= 1024; u++; }
-        return `${s.toFixed(1)} ${units[u]}`;
-    };
-    const fmtDur = (ms) => {
-        if (!ms) return '-';
-        const sec = Math.floor(ms / 1000);
-        return `${Math.floor(sec / 60)}:${(sec % 60).toString().padStart(2, '0')}`;
-    };
+    const downloadMode = data.download_mode || 'soulseek';
+    const availableSources = Array.isArray(data.available_sources) ? data.available_sources : [];
+    // Hybrid mode shows the dropdown; everything else implies a single source.
+    const isHybrid = downloadMode === 'hybrid';

    let tableRows = '';
    if (candidates.length === 0) {
        tableRows = `<tr><td colspan="7" style="text-align:center; color: rgba(255,255,255,0.5); padding: 30px;">
            No candidates were found during search.</td></tr>`;
    } else {
+        // Auto-candidates only show source badges in hybrid mode (where the
+        // user can't infer source from the dropdown).
        candidates.forEach((c, i) => {
-            const shortFile = c.filename ? c.filename.split(/[/\\]/).pop() : '-';
-            const qBadge = c.quality
-                ? `<span class="candidates-quality-badge candidates-quality-${c.quality.toLowerCase()}">${c.quality.toUpperCase()}</span>`
-                : '';
-            tableRows += `<tr>
-                <td class="candidates-col-index">${i + 1}</td>
-                <td class="candidates-col-file" title="${escapeHtml(c.filename || '')}">${escapeHtml(shortFile)}</td>
-                <td class="candidates-col-quality">${qBadge}${c.bitrate ? ` ${c.bitrate}kbps` : ''}</td>
-                <td class="candidates-col-size">${fmtSize(c.size)}</td>
-                <td class="candidates-col-duration">${fmtDur(c.duration)}</td>
-                <td class="candidates-col-user" title="Queue: ${c.queue_length || 0} | Slots: ${c.free_upload_slots || 0}">${escapeHtml(c.username || '-')}</td>
-                <td class="candidates-col-action"><button class="candidates-download-btn" data-index="${i}" title="Download this file">⬇</button></td>
-            </tr>`;
+            tableRows += _renderCandidateRow(c, i, 'candidates-row-auto', isHybrid);
        });
    }

+    // ----- Manual search bar -----
+    let sourceControl;
+    if (isHybrid && availableSources.length > 0) {
+        const optionsHtml = ['<option value="all">All sources</option>']
+            .concat(availableSources.map(s =>
+                `<option value="${escapeHtml(s.id)}">${escapeHtml(s.label)}</option>`
+            ))
+            .join('');
+        sourceControl = `<select class="candidates-manual-source" id="candidates-manual-source">${optionsHtml}</select>`;
+    } else {
+        // Single-source mode — render a small static label, not a dropdown.
+        const onlySrc = availableSources[0];
+        const label = onlySrc ? onlySrc.label : (downloadMode || 'configured source');
+        sourceControl = `<span class="candidates-manual-source-label">Searching ${escapeHtml(label)}</span>`;
+    }
+
+    const manualSearchHtml = `
+        <div class="candidates-manual-search">
+            <div class="candidates-manual-search-header">Manual search</div>
+            <div class="candidates-manual-search-controls">
+                <input type="text"
+                       class="candidates-manual-search-input"
+                       id="candidates-manual-search-input"
+                       placeholder="Search for a different track..."
+                       maxlength="200" />
+                ${sourceControl}
+                <button class="candidates-manual-search-btn"
+                        id="candidates-manual-search-btn"
+                        disabled>Search</button>
+            </div>
+            <div class="candidates-manual-search-hint" id="candidates-manual-search-hint">Type at least 2 characters</div>
+            <div class="candidates-manual-search-results" id="candidates-manual-search-results"></div>
+        </div>`;
+
    overlay = document.createElement('div');
    overlay.id = 'candidates-modal-overlay';
    overlay.className = 'candidates-modal-overlay';
@ -2962,14 +3016,17 @@ function _renderCandidatesModal(data) {
            </div>
            <div class="candidates-modal-body">
                ${errorMsg ? `<div class="candidates-error-summary">${escapeHtml(errorMsg)}</div>` : ''}
-                <div class="candidates-count">${candidates.length} candidate${candidates.length !== 1 ? 's' : ''} found${candidates.length > 0 ? ' but none passed filters' : ''}</div>
-                <div class="candidates-table-wrapper">
-                    <table class="candidates-table">
-                        <thead><tr>
-                            <th>#</th><th>File</th><th>Quality</th><th>Size</th><th>Duration</th><th>User</th><th></th>
-                        </tr></thead>
-                        <tbody>${tableRows}</tbody>
-                    </table>
+                ${manualSearchHtml}
+                <div class="candidates-auto-section">
+                    <div class="candidates-count">${candidates.length} candidate${candidates.length !== 1 ? 's' : ''} found${candidates.length > 0 ? ' but none passed filters' : ''}</div>
+                    <div class="candidates-table-wrapper">
+                        <table class="candidates-table">
+                            <thead><tr>
+                                <th>#</th><th>File</th><th>Quality</th><th>Size</th><th>Duration</th><th>User</th><th></th>
+                            </tr></thead>
+                            <tbody>${tableRows}</tbody>
+                        </table>
+                    </div>
                </div>
            </div>
        </div>`;
@ -2977,14 +3034,196 @@ function _renderCandidatesModal(data) {
    document.body.appendChild(overlay);
    requestAnimationFrame(() => overlay.classList.add('visible'));

-    // Bind download buttons
-    overlay.querySelectorAll('.candidates-download-btn').forEach(btn => {
+    // Bind auto-candidate download buttons (existing behavior, scoped to
+    // .candidates-row-auto so manual-search rows don't double-trigger).
+    overlay.querySelectorAll('.candidates-row-auto .candidates-download-btn').forEach(btn => {
        btn.addEventListener('click', () => {
            const idx = parseInt(btn.dataset.index);
            const c = candidates[idx];
            if (c) downloadCandidate(data.task_id, c, trackName);
        });
    });
+
+    // Wire manual search controls.
+    _wireManualSearch(overlay, data.task_id, trackName, isHybrid);
+}
+
+// Manual-search wiring — input/button/dropdown. Kept separate from
+// _renderCandidatesModal so the existing render path stays readable and
+// any future refactor can lift this into its own module.
+function _wireManualSearch(overlay, taskId, trackName, isHybrid) {
+    const input = overlay.querySelector('#candidates-manual-search-input');
+    const button = overlay.querySelector('#candidates-manual-search-btn');
+    const hint = overlay.querySelector('#candidates-manual-search-hint');
+    const resultsContainer = overlay.querySelector('#candidates-manual-search-results');
+    const sourceSelect = overlay.querySelector('#candidates-manual-source');
+    if (!input || !button || !resultsContainer) return;
+
+    // Aggregated results across all source streams for the current query.
+    // Cleared at the start of each new search.
+    let currentResults = [];
+    let inFlight = false;
+    let abortController = null;
+
+    const updateButtonState = () => {
+        const q = (input.value || '').trim();
+        const tooShort = q.length < 2;
+        button.disabled = tooShort || inFlight;
+        if (hint) {
+            if (tooShort) {
+                hint.textContent = 'Type at least 2 characters';
+                hint.style.display = '';
+            } else {
+                hint.style.display = 'none';
+            }
+        }
+    };
+
+    const _renderTableShell = (query) => {
+        resultsContainer.innerHTML = `
+            <div class="candidates-manual-search-status" id="candidates-manual-search-status">Searching...</div>
+            <div class="candidates-table-wrapper" style="display: none;" id="candidates-manual-table-wrapper">
+                <table class="candidates-table">
+                    <thead><tr>
+                        <th>#</th><th>File</th><th>Quality</th><th>Size</th><th>Duration</th><th>User</th><th></th>
+                    </tr></thead>
+                    <tbody id="candidates-manual-tbody"></tbody>
+                </table>
+            </div>`;
+    };
+
+    const _appendRows = (newCandidates, query) => {
+        if (!newCandidates || newCandidates.length === 0) return;
+        const startIdx = currentResults.length;
+        currentResults = currentResults.concat(newCandidates);
+
+        const wrapper = resultsContainer.querySelector('#candidates-manual-table-wrapper');
+        const tbody = resultsContainer.querySelector('#candidates-manual-tbody');
+        const statusEl = resultsContainer.querySelector('#candidates-manual-search-status');
+        if (!tbody || !wrapper) return;
+
+        let rowsHtml = '';
+        newCandidates.forEach((c, i) => {
+            rowsHtml += _renderCandidateRow(c, startIdx + i, 'candidates-row-manual', isHybrid);
+        });
+        tbody.insertAdjacentHTML('beforeend', rowsHtml);
+        wrapper.style.display = '';
+        if (statusEl) {
+            statusEl.textContent = `${currentResults.length} result${currentResults.length !== 1 ? 's' : ''} so far...`;
+        }
+
+        // Wire newly-appended buttons
+        tbody.querySelectorAll('.candidates-download-btn').forEach(btn => {
+            if (btn._candidatesWired) return;
+            btn._candidatesWired = true;
+            btn.addEventListener('click', () => {
+                const idx = parseInt(btn.dataset.index);
+                const c = currentResults[idx];
+                if (c) downloadCandidate(taskId, c, trackName);
+            });
+        });
+    };
+
+    const _setStatus = (text) => {
+        const statusEl = resultsContainer.querySelector('#candidates-manual-search-status');
+        if (statusEl) statusEl.textContent = text;
+    };
+
+    const _setError = (msg) => {
+        resultsContainer.innerHTML = `<div class="candidates-manual-search-error">${escapeHtml(msg)}</div>`;
+    };
+
+    const runSearch = async () => {
+        const q = (input.value || '').trim();
+        if (q.length < 2 || inFlight) return;
+
+        if (abortController) {
+            try { abortController.abort(); } catch (_) { }
+        }
+        abortController = new AbortController();
+
+        const source = sourceSelect ? sourceSelect.value : 'all';
+        inFlight = true;
+        button.disabled = true;
+        const originalLabel = button.textContent;
+        button.textContent = 'Searching...';
+        currentResults = [];
+        _renderTableShell(q);
+
+        try {
+            const resp = await fetch(`/api/downloads/task/${encodeURIComponent(taskId)}/manual-search`, {
+                method: 'POST',
+                headers: { 'Content-Type': 'application/json' },
+                body: JSON.stringify({ query: q, source: source }),
+                signal: abortController.signal,
+            });
+            if (!resp.ok) {
+                let errMsg = 'Search failed';
+                try {
+                    const payload = await resp.json();
+                    if (payload && payload.error) errMsg = payload.error;
+                } catch (_) { }
+                _setError(errMsg);
+                return;
+            }
+
+            const reader = resp.body.getReader();
+            const decoder = new TextDecoder();
+            let buffer = '';
+            const errors = [];
+
+            while (true) {
+                const { value, done } = await reader.read();
+                if (done) break;
+                buffer += decoder.decode(value, { stream: true });
+
+                let lineEnd;
+                while ((lineEnd = buffer.indexOf('\n')) >= 0) {
+                    const line = buffer.slice(0, lineEnd).trim();
+                    buffer = buffer.slice(lineEnd + 1);
+                    if (!line) continue;
+                    let msg;
+                    try { msg = JSON.parse(line); } catch (_) { continue; }
+
+                    if (msg.type === 'source_results') {
+                        _appendRows(msg.candidates || [], q);
+                    } else if (msg.type === 'source_error') {
+                        errors.push(`${msg.source}: ${msg.error}`);
+                    } else if (msg.type === 'done') {
+                        if (currentResults.length === 0) {
+                            const errorNote = errors.length
+                                ? `<div class="candidates-manual-search-empty-note">${errors.length} source${errors.length !== 1 ? 's' : ''} failed</div>`
+                                : '';
+                            resultsContainer.innerHTML = `
+                                <div class="candidates-manual-search-empty">No manual search results for "${escapeHtml(q)}"</div>
+                                ${errorNote}`;
+                        } else {
+                            _setStatus(`${currentResults.length} result${currentResults.length !== 1 ? 's' : ''}`);
+                        }
+                    }
+                }
+            }
+        } catch (err) {
+            if (err.name === 'AbortError') return;
+            console.error('Manual search failed:', err);
+            _setError('Search request failed');
+        } finally {
+            inFlight = false;
+            button.textContent = originalLabel;
+            updateButtonState();
+        }
+    };
+
+    input.addEventListener('input', updateButtonState);
+    input.addEventListener('keydown', (e) => {
+        if (e.key === 'Enter' && !button.disabled) {
+            e.preventDefault();
+            runSearch();
+        }
+    });
+    button.addEventListener('click', runSearch);
+
+    updateButtonState();
 }

 async function downloadCandidate(taskId, candidate, trackName) {
@ -3160,8 +3399,11 @@ function processModalStatusUpdate(playlistId, data) {
                    statusEl.dataset.errorMsg = task.error_message;
                    _ensureErrorTooltipListeners(statusEl);
                }
-                // Make not_found and failed cells clickable to review search candidates
-                if ((task.status === 'not_found' || task.status === 'failed') && task.has_candidates) {
+                // Make not_found / failed / cancelled cells clickable to open
+                // the candidates modal. Always bind — even when no auto-search
+                // candidates were cached — because the modal carries the manual
+                // search bar, which is the user's recourse for empty results.
+                if (task.status === 'not_found' || task.status === 'failed' || task.status === 'cancelled') {
                    statusEl.classList.add('has-candidates');
                    statusEl.dataset.taskId = task.task_id;
                    _ensureCandidatesClickListener(statusEl);
--- a/webui/static/helper.js
+++ b/webui/static/helper.js
@ -3413,6 +3413,12 @@ function closeHelperSearch() {
 // projects that span multiple commits before shipping. Strip the flag at
 // release time and add a real `date:` line at the top of the version block.
 const WHATS_NEW = {
+    '2.4.4': [
+        // --- post-2.4.3 dev work — entries hidden by _getLatestWhatsNewVersion until the build version bumps ---
+        { date: 'Unreleased — 2.4.4 dev cycle' },
+        { title: 'Manual Search In The Failed-Track Candidates Modal', desc: 'when a download fails or returns "not found" the user can already click the status cell to open a modal showing whatever search candidates the auto-search left over and pick a different one. that modal now ALSO has a manual search bar. type any query, hit search, get a fresh round of results from the download sources without having to start the whole download flow over from the search page. solves the case where the auto-query was bad (featured artist not in title, parentheticals like "(remastered 2019)" tripping the matcher, slight artist-name variants) but the file genuinely exists on the source. source picker is smart per download mode: single-source mode (soulseek-only / youtube-only / etc) shows a "searching X" label, no dropdown; hybrid mode shows a dropdown with "all sources" default plus every configured source — picking "all" runs parallel searches across all of them and tags each result row with its source badge. only configured sources show up; unconfigured ones are hidden. results stream in as each source completes via NDJSON instead of blocking on the slowest source — the table starts populating the moment the first source returns. clicking a result reuses the existing retry-download flow → same path, same acoustid verification on the file when it lands, no shortcut around the safety net. additive in the truest sense: the existing modal layout / candidates table / download buttons are byte-identical when the user doesn\'t use manual search. backend extends the candidates endpoint with `download_mode` + `available_sources` + a `source` field per candidate (purely additive — old fields untouched), and adds a new `POST /api/downloads/task/<id>/manual-search` that streams NDJSON (one header line, one source_results line per source as completed, one done terminator) so the frontend renderer can append rows incrementally. 11 tests pin the streaming contract: query length / source whitelist / task 404 validation, single-source dispatch, parallel "all" dispatch, one-event-per-source streaming shape, unconfigured-source skip + reject, header metadata, and per-source exception isolation (one source raising emits a `source_error` event but doesn\'t fail the stream).', page: 'downloads' },
+        { title: 'Manual Picks Don\'t Auto-Retry Anymore (And The Modal Always Opens)', desc: 'three follow-on fixes to the manual-search feature once people started actually using it. (1) when the user picked a candidate and that download failed (e.g. soundcloud 404 on a stale track url), the auto-retry monitor would treat it like any other failed auto-attempt — yank the task back to "searching" and pick a different candidate. felt completely wrong from the user\'s perspective: "i picked THIS one, why is it searching for something else?" now manual picks are tagged with a `_user_manual_pick` flag and the auto-retry path bails on it. failure surfaces to the user instead of getting silently fallen-back. (2) non-soulseek manual picks (youtube / tidal / qobuz / hifi / deezer / soundcloud / lidarr) were getting stuck at "downloading 0%" forever even after their engine reported terminal failure. cause: status polling went into a "let monitor handle retry" branch that never fired because manual picks bail on retry — task was orphaned in downloading state. fix: when the engine reports Errored on a manual pick, mark the task failed directly, don\'t defer to the monitor. plus an engine-state fallback path covers the rare race where the orchestrator\'s pre-populated transfer lookup is missing the entry. (3) failed / not_found rows were only clickable when the auto-search had cached candidates — but the whole point of opening the modal now is to RUN a manual search, which doesn\'t need pre-existing candidates. now every failed / not_found / cancelled row opens the modal regardless. (4) one nasty deadlock fix in the process: the new "mark failed" path was synchronously calling `on_download_completed` while holding `tasks_lock`, which itself re-acquires the same lock — `threading.Lock` is non-reentrant so the polling thread wedged forever. while wedged the lock was held → every other endpoint that needed it (including /candidates → can\'t open OTHER modals) hung waiting. moved completion callbacks onto a daemon thread so the lock releases first. (5) manual download worker now runs on its own dedicated thread instead of competing with the batch\'s 3-worker `missing_download_executor` pool — saturated batches no longer queue manual picks indefinitely. all changes are scoped to manual picks only via the `_user_manual_pick` flag — auto-attempt flow is byte-identical to before. 17 unit tests pin the gate behavior (status engine fallback / monitor retry skip / IF-branch failure transition / auto-attempt skip).', page: 'downloads' },
+    ],
    '2.4.3': [
        // --- May 8, 2026 — patch release ---
        { date: 'May 8, 2026 — 2.4.3 release' },
--- a/webui/static/style.css
+++ b/webui/static/style.css
@ -40845,6 +40845,140 @@ div.artist-hero-badge {
    border-color: rgb(var(--accent-rgb));
 }

+/* Manual search section inside the candidates modal — sits above the
+   auto-candidates table; visually separated by a subtle border + spacing. */
+.candidates-manual-search {
+    background: rgba(255, 255, 255, 0.03);
+    border: 1px solid rgba(255, 255, 255, 0.08);
+    border-radius: 8px;
+    padding: 14px 16px;
+    margin-bottom: 18px;
+}
+.candidates-manual-search-header {
+    color: rgba(255, 255, 255, 0.85);
+    font-size: 13px;
+    font-weight: 600;
+    text-transform: uppercase;
+    letter-spacing: 0.5px;
+    margin-bottom: 10px;
+}
+.candidates-manual-search-controls {
+    display: flex;
+    gap: 8px;
+    align-items: center;
+    flex-wrap: wrap;
+}
+.candidates-manual-search-input {
+    flex: 1;
+    min-width: 200px;
+    background: rgba(0, 0, 0, 0.25);
+    border: 1px solid rgba(255, 255, 255, 0.12);
+    border-radius: 6px;
+    padding: 8px 12px;
+    color: rgba(255, 255, 255, 0.9);
+    font-size: 13px;
+    outline: none;
+    transition: border-color 0.15s ease;
+}
+.candidates-manual-search-input:focus {
+    border-color: rgba(var(--accent-rgb), 0.6);
+}
+.candidates-manual-source {
+    background: rgba(0, 0, 0, 0.25);
+    border: 1px solid rgba(255, 255, 255, 0.12);
+    border-radius: 6px;
+    padding: 8px 10px;
+    color: rgba(255, 255, 255, 0.9);
+    font-size: 13px;
+    cursor: pointer;
+    outline: none;
+}
+.candidates-manual-source:focus {
+    border-color: rgba(var(--accent-rgb), 0.6);
+}
+.candidates-manual-source-label {
+    color: rgba(255, 255, 255, 0.6);
+    font-size: 12px;
+    padding: 0 8px;
+    white-space: nowrap;
+}
+.candidates-manual-search-btn {
+    background: rgba(var(--accent-rgb), 0.2);
+    border: 1px solid rgba(var(--accent-rgb), 0.4);
+    color: rgb(var(--accent-rgb));
+    border-radius: 6px;
+    padding: 8px 16px;
+    font-size: 13px;
+    font-weight: 600;
+    cursor: pointer;
+    transition: all 0.15s ease;
+}
+.candidates-manual-search-btn:hover:not(:disabled) {
+    background: rgba(var(--accent-rgb), 0.35);
+    border-color: rgb(var(--accent-rgb));
+}
+.candidates-manual-search-btn:disabled {
+    opacity: 0.4;
+    cursor: not-allowed;
+}
+.candidates-manual-search-hint {
+    color: rgba(255, 255, 255, 0.4);
+    font-size: 11px;
+    margin-top: 6px;
+    font-style: italic;
+}
+.candidates-manual-search-results {
+    margin-top: 12px;
+}
+.candidates-manual-search-results:empty {
+    display: none;
+}
+.candidates-manual-search-loading,
+.candidates-manual-search-empty,
+.candidates-manual-search-error {
+    color: rgba(255, 255, 255, 0.5);
+    font-size: 13px;
+    padding: 14px;
+    text-align: center;
+    background: rgba(0, 0, 0, 0.15);
+    border-radius: 6px;
+}
+.candidates-manual-search-error {
+    color: rgba(255, 100, 100, 0.85);
+}
+.candidates-manual-search-count {
+    color: rgba(255, 255, 255, 0.55);
+    font-size: 12px;
+    margin-bottom: 8px;
+}
+.candidates-manual-search-status {
+    color: rgba(255, 255, 255, 0.6);
+    font-size: 12px;
+    margin-bottom: 8px;
+    font-style: italic;
+}
+.candidates-manual-search-empty-note {
+    color: rgba(255, 180, 100, 0.7);
+    font-size: 11px;
+    margin-top: 6px;
+    text-align: center;
+    font-style: italic;
+}
+.candidates-source-badge {
+    display: inline-block;
+    background: rgba(255, 255, 255, 0.08);
+    border: 1px solid rgba(255, 255, 255, 0.12);
+    color: rgba(255, 255, 255, 0.7);
+    border-radius: 4px;
+    padding: 1px 6px;
+    font-size: 10px;
+    font-weight: 600;
+    letter-spacing: 0.3px;
+    text-transform: uppercase;
+    margin-right: 4px;
+    vertical-align: middle;
+}
+
 /* =====================================
   HYDRABASE PAGE
   ===================================== */