diff --git a/core/downloads/task_worker.py b/core/downloads/task_worker.py
new file mode 100644
index 00000000..d7708154
--- /dev/null
+++ b/core/downloads/task_worker.py
@@ -0,0 +1,380 @@
+"""Per-task download worker.
+
+Runs as a background thread (one per task) that:
+1. Tries source-reuse (use the batch's last good slskd peer if available)
+2. Tries staging-match (file already in staging folder, no download needed)
+3. Generates smart search queries via the matching engine + legacy fallbacks
+4. Iterates queries sequentially against the soulseek client
+5. For each query: validates results, attempts download with fallback candidates
+6. If hybrid mode: falls back to remaining sources (youtube/tidal/qobuz/hifi/deezer_dl)
+7. On total failure: marks task not_found + records search diagnostics
+8. On any uncaught exception: marks failed + emergency worker-slot recovery
+
+Lifted verbatim from web_server.py's `_download_track_worker`. The helpers
+this calls into (try_source_reuse, store_batch_source, try_staging_match,
+get_valid_candidates, attempt_download_with_candidates, on_download_completed,
+recover_worker_slot) are passed via `TaskWorkerDeps` since each is itself
+a large web_server.py helper that will get its own lift in subsequent PRs.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+import traceback
+from dataclasses import dataclass
+from typing import Any, Callable, Optional
+
+from core.runtime_state import download_tasks, tasks_lock
+from core.spotify_client import Track as SpotifyTrack
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class TaskWorkerDeps:
+    """Bundle of cross-cutting deps the per-task download worker needs."""
+    soulseek_client: Any
+    matching_engine: Any
+    run_async: Callable
+    try_source_reuse: Callable                    # (task_id, batch_id, track) -> bool
+    store_batch_source: Callable                  # (batch_id, username, filename) -> None
+    try_staging_match: Callable                   # (task_id, batch_id, track) -> bool
+    get_valid_candidates: Callable                # (results, spotify_track, query) -> list
+    attempt_download_with_candidates: Callable    # (task_id, candidates, track, batch_id) -> bool
+    on_download_completed: Callable               # (batch_id, task_id, success) -> None
+    recover_worker_slot: Callable                 # (batch_id, task_id) -> None
+
+
+def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorkerDeps) -> None:
+    """Enhanced download worker that matches the GUI's exact retry logic.
+
+    Implements sequential query retry, fallback candidates, and download
+    failure retry.
+    """
+    try:
+        # Retrieve task details from global state
+        with tasks_lock:
+            if task_id not in download_tasks:
+                logger.warning(f"[Modal Worker] Task {task_id} not found in download_tasks")
+                return
+            task = download_tasks[task_id].copy()
+
+        # Cancellation Checkpoint 1: Before doing anything
+        with tasks_lock:
+            if task_id not in download_tasks:
+                logger.info(f"[Modal Worker] Task {task_id} was deleted before starting")
+                return
+            if download_tasks[task_id]['status'] == 'cancelled':
+                logger.warning(f"[Modal Worker] Task {task_id} cancelled before starting")
+                # V2 FIX: Don't call _on_download_completed for cancelled V2 tasks
+                # V2 system handles worker slot freeing in atomic cancel function
+                task_playlist_id = download_tasks[task_id].get('playlist_id')
+                if task_playlist_id:
+                    logger.warning(f"[Modal Worker] V2 task {task_id} cancelled - worker slot already freed by V2 system")
+                    return  # V2 system already handled worker slot management
+                elif batch_id:
+                    # Legacy system - use old completion callback
+                    logger.warning(f"[Modal Worker] Legacy task {task_id} cancelled - using legacy completion callback")
+                    deps.on_download_completed(batch_id, task_id, False)
+                return
+
+        track_data = task['track_info']
+        track_name = track_data.get('name', 'Unknown Track')
+
+        logger.info(f"[Modal Worker] Task {task_id} starting search for track: '{track_name}'")
+
+        # Recreate a SpotifyTrack object for the matching engine
+        # Handle both string format and Spotify API format for artists
+        raw_artists = track_data.get('artists', [])
+        processed_artists = []
+        for artist in raw_artists:
+            if isinstance(artist, str):
+                processed_artists.append(artist)
+            elif isinstance(artist, dict) and 'name' in artist:
+                processed_artists.append(artist['name'])
+            else:
+                processed_artists.append(str(artist))
+
+        # Handle album field - extract name if it's a dictionary
+        raw_album = track_data.get('album', '')
+        if isinstance(raw_album, dict) and 'name' in raw_album:
+            album_name = raw_album['name']
+        elif isinstance(raw_album, str):
+            album_name = raw_album
+        else:
+            album_name = str(raw_album)
+
+        track = SpotifyTrack(
+            id=track_data.get('id', ''),
+            name=track_data.get('name', ''),
+            artists=processed_artists,
+            album=album_name,
+            duration_ms=track_data.get('duration_ms', 0),
+            popularity=track_data.get('popularity', 0),
+        )
+        logger.info(f"[Modal Worker] Starting download task for: {track.name} by {track.artists[0] if track.artists else 'Unknown'}")
+
+        # === SOURCE REUSE: Check batch's last good source before searching ===
+        if deps.try_source_reuse(task_id, batch_id, track):
+            # Store source for next worker (cascading reuse)
+            with tasks_lock:
+                used_filename = download_tasks.get(task_id, {}).get('filename')
+                used_username = download_tasks.get(task_id, {}).get('username')
+            if used_filename and used_username:
+                deps.store_batch_source(batch_id, used_username, used_filename)
+            return
+
+        # === STAGING CHECK: Check staging folder for existing file before searching ===
+        if deps.try_staging_match(task_id, batch_id, track):
+            return
+
+        # Initialize task state tracking (like GUI's parallel_search_tracking)
+        with tasks_lock:
+            if task_id in download_tasks:
+                download_tasks[task_id]['status'] = 'searching'  # Now actively being processed
+                download_tasks[task_id]['current_query_index'] = 0
+                download_tasks[task_id]['current_candidate_index'] = 0
+                download_tasks[task_id]['retry_count'] = 0
+                download_tasks[task_id]['candidates'] = []
+                # CRITICAL: Preserve used_sources from previous retry attempts (don't reset to empty set)
+                # If this is a retry, the monitor will have already marked failed sources
+                if 'used_sources' not in download_tasks[task_id]:
+                    download_tasks[task_id]['used_sources'] = set()
+                # Else: keep existing used_sources to avoid retrying same failed hosts
+
+        # 1. Generate multiple search queries (like GUI's generate_smart_search_queries)
+        artist_name = track.artists[0] if track.artists else None
+        track_name = track.name
+
+        # Start with matching engine queries
+        search_queries = deps.matching_engine.generate_download_queries(track)
+
+        # Add legacy fallback queries (like GUI does)
+        legacy_queries = []
+
+        if artist_name:
+            # Add first word of artist approach (legacy compatibility)
+            artist_words = artist_name.split()
+            if artist_words:
+                first_word = artist_words[0]
+                if first_word.lower() == 'the' and len(artist_words) > 1:
+                    first_word = artist_words[1]
+
+                if len(first_word) > 1:
+                    legacy_queries.append(f"{track_name} {first_word}".strip())
+
+        # Add track-only query
+        if track_name.strip():
+            legacy_queries.append(track_name.strip())
+
+        # Add traditional cleaned queries
+        cleaned_name = re.sub(r'\s*\([^)]*\)', '', track_name).strip()
+        cleaned_name = re.sub(r'\s*\[[^\]]*\]', '', cleaned_name).strip()
+
+        if cleaned_name and cleaned_name.lower() != track_name.lower():
+            legacy_queries.append(cleaned_name.strip())
+
+        # Combine enhanced queries with legacy fallbacks
+        all_queries = search_queries + legacy_queries
+
+        # Remove duplicates while preserving order
+        unique_queries = []
+        seen = set()
+        for query in all_queries:
+            if query and query.lower() not in seen:
+                unique_queries.append(query)
+                seen.add(query.lower())
+
+        search_queries = unique_queries
+        logger.info(f"[Modal Worker] Generated {len(search_queries)} smart search queries for '{track.name}': {search_queries}")
+        logger.info(f"[Modal Worker] About to start search loop for task {task_id} (track: '{track.name}')")
+
+        # 2. Sequential Query Search (matches GUI's start_search_worker_parallel logic)
+        search_diagnostics = []  # Track what happened per query for detailed error messages
+        all_raw_results = []  # Collect raw results across queries for candidate review modal
+        for query_index, query in enumerate(search_queries):
+            # Cancellation check before each query
+            with tasks_lock:
+                if task_id not in download_tasks:
+                    logger.debug(f"[Modal Worker] Task {task_id} was deleted during query {query_index + 1}")
+                    return
+                if download_tasks[task_id]['status'] == 'cancelled':
+                    logger.debug(f"[Modal Worker] Task {task_id} cancelled during query {query_index + 1}")
+                    # Don't call _on_download_completed for cancelled tasks as it can stop monitoring
+                    return
+                download_tasks[task_id]['current_query_index'] = query_index
+
+            logger.debug(f"[Modal Worker] Query {query_index + 1}/{len(search_queries)}: '{query}'")
+            logger.debug(f"About to call soulseek search for task {task_id}")
+
+            try:
+                # Perform search with timeout
+                tracks_result, _ = deps.run_async(deps.soulseek_client.search(query, timeout=30))
+                logger.debug(f"Search completed for task {task_id}, got {len(tracks_result) if tracks_result else 0} results")
+
+                # CRITICAL: Check cancellation immediately after search returns
+                with tasks_lock:
+                    if task_id not in download_tasks:
+                        logger.info(f"[Modal Worker] Task {task_id} was deleted after search returned")
+                        return
+                    if download_tasks[task_id]['status'] == 'cancelled':
+                        logger.warning(f"[Modal Worker] Task {task_id} cancelled after search returned - ignoring results")
+                        # Don't call _on_download_completed for cancelled tasks as it can stop monitoring
+                        # The cancellation endpoint already handles batch management properly
+                        return
+
+                if tracks_result:
+                    result_count = len(tracks_result)
+                    # Validate candidates using GUI's get_valid_candidates logic
+                    candidates = deps.get_valid_candidates(tracks_result, track, query)
+                    if candidates:
+                        logger.debug(f"[Modal Worker] Found {len(candidates)} valid candidates for query '{query}'")
+
+                        # CRITICAL: Check cancellation before processing candidates
+                        with tasks_lock:
+                            if task_id not in download_tasks:
+                                logger.info(f"[Modal Worker] Task {task_id} was deleted before processing candidates")
+                                return
+                            if download_tasks[task_id]['status'] == 'cancelled':
+                                logger.warning(f"[Modal Worker] Task {task_id} cancelled before processing candidates")
+                                # Don't call _on_download_completed for cancelled tasks as it can stop monitoring
+                                return
+                            # Store candidates for retry fallback (like GUI)
+                            download_tasks[task_id]['cached_candidates'] = candidates
+
+                        # Try to download with these candidates
+                        success = deps.attempt_download_with_candidates(task_id, candidates, track, batch_id)
+                        if success:
+                            # Download initiated successfully - let the download monitoring system handle completion
+                            if batch_id:
+                                logger.info(f"[Modal Worker] Download initiated successfully for task {task_id} - monitoring will handle completion")
+                            # Store this source for batch reuse
+                            with tasks_lock:
+                                used_filename = download_tasks.get(task_id, {}).get('filename')
+                                used_username = download_tasks.get(task_id, {}).get('username')
+                            if used_filename and used_username:
+                                deps.store_batch_source(batch_id, used_username, used_filename)
+                            return  # Success, exit the worker
+                        else:
+                            search_diagnostics.append(f'"{query}": {result_count} results, {len(candidates)} passed filters but download failed to start')
+                    else:
+                        search_diagnostics.append(f'"{query}": {result_count} results but none passed quality/artist filters')
+                        all_raw_results.extend(tracks_result[:20])  # Keep top results for review
+                else:
+                    search_diagnostics.append(f'"{query}": no results found')
+
+            except Exception as e:
+                logger.debug(f"[Modal Worker] Search failed for query '{query}': {e}")
+                search_diagnostics.append(f'"{query}": search error — {e}')
+                continue
+
+        # === HYBRID FALLBACK: If primary source failed, try remaining sources directly ===
+        # The orchestrator's hybrid search stops at the first source with results, even if
+        # those results all fail quality filtering. Try remaining sources individually.
+        if getattr(deps.soulseek_client, 'mode', '') == 'hybrid':
+            try:
+                orch = deps.soulseek_client
+                hybrid_order = getattr(orch, 'hybrid_order', None) or []
+                if not hybrid_order:
+                    primary = getattr(orch, 'hybrid_primary', 'soulseek')
+                    secondary = getattr(orch, 'hybrid_secondary', '')
+                    hybrid_order = [primary, secondary] if secondary and secondary != primary else [primary]
+
+                source_clients = {
+                    'soulseek': getattr(orch, 'soulseek', None),
+                    'youtube': getattr(orch, 'youtube', None),
+                    'tidal': getattr(orch, 'tidal', None),
+                    'qobuz': getattr(orch, 'qobuz', None),
+                    'hifi': getattr(orch, 'hifi', None),
+                    'deezer_dl': getattr(orch, 'deezer_dl', None),
+                }
+
+                # The orchestrator tried sources in order but stopped at the first with results.
+                # We don't know which it stopped at, so try ALL sources except the first
+                # (which was definitely tried). If the first was skipped (unconfigured),
+                # the orchestrator would have tried the second — but trying it again is
+                # harmless (streaming sources return fast).
+                remaining_sources = [s for s in hybrid_order[1:] if s in source_clients and source_clients[s]]
+                if remaining_sources:
+                    logger.warning(f"[Hybrid Fallback] Primary source had no valid matches. Trying fallback sources: {remaining_sources}")
+
+                for fallback_source in remaining_sources:
+                    fb_client = source_clients[fallback_source]
+                    if hasattr(fb_client, 'is_configured') and not fb_client.is_configured():
+                        continue
+
+                    # Use first 2 queries only for speed
+                    for fb_query in search_queries[:2]:
+                        try:
+                            logger.warning(f"[Hybrid Fallback] Trying {fallback_source}: '{fb_query}'")
+                            fb_results, _ = deps.run_async(fb_client.search(fb_query, timeout=20))
+                            if not fb_results:
+                                continue
+                            fb_candidates = deps.get_valid_candidates(fb_results, track, fb_query)
+                            if fb_candidates:
+                                logger.warning(f"[Hybrid Fallback] {fallback_source} found {len(fb_candidates)} valid candidates!")
+                                success = deps.attempt_download_with_candidates(task_id, fb_candidates, track, batch_id)
+                                if success:
+                                    return
+                        except Exception as e:
+                            logger.error(f"[Hybrid Fallback] {fallback_source} search failed: {e}")
+                            continue
+
+                    logger.warning(f"[Hybrid Fallback] {fallback_source} returned no valid candidates")
+
+            except Exception as e:
+                logger.error(f"[Hybrid Fallback] Error in fallback logic: {e}")
+
+        # If we get here, all search queries and hybrid fallbacks failed
+        logger.warning(f"[Modal Worker] No valid candidates found for '{track.name}' after trying all {len(search_queries)} queries.")
+        with tasks_lock:
+            if task_id in download_tasks:
+                download_tasks[task_id]['status'] = 'not_found'
+                _diag_summary = ' | '.join(search_diagnostics) if search_diagnostics else 'no queries attempted'
+                download_tasks[task_id]['error_message'] = f'No match found for "{track_name}" by {artist_name or "Unknown"} after {len(search_queries)} queries. Breakdown: {_diag_summary}'
+                # Store raw results so the user can review what Soulseek returned
+                if all_raw_results and not download_tasks[task_id].get('cached_candidates'):
+                    download_tasks[task_id]['cached_candidates'] = all_raw_results
+
+        # Notify batch manager that this task completed (failed) - THREAD SAFE
+        if batch_id:
+            try:
+                deps.on_download_completed(batch_id, task_id, False)
+            except Exception as completion_error:
+                logger.error(f"Error in batch completion callback for {task_id}: {completion_error}")
+
+    except Exception as e:
+        track_name_safe = locals().get('track_name', 'unknown')  # Safe fallback for track_name
+        logger.error(f"CRITICAL ERROR in download task for '{track_name_safe}' (task_id: {task_id}): {e}")
+        traceback.print_exc()
+
+        # Update task status safely with timeout
+        try:
+            lock_acquired = tasks_lock.acquire(timeout=2.0)
+            if lock_acquired:
+                try:
+                    if task_id in download_tasks:
+                        download_tasks[task_id]['status'] = 'failed'
+                        download_tasks[task_id]['error_message'] = f'Unexpected error during download: {type(e).__name__}: {e}'
+                        logger.error(f"[Exception Recovery] Set task {task_id} status to 'failed'")
+                finally:
+                    tasks_lock.release()
+            else:
+                logger.error(f"[Exception Recovery] Could not acquire lock to update task {task_id} status")
+        except Exception as status_error:
+            logger.error(f"Error updating task status in exception handler: {status_error}")
+
+        # Notify batch manager that this task completed (failed) - THREAD SAFE with RECOVERY
+        if batch_id:
+            try:
+                deps.on_download_completed(batch_id, task_id, False)
+                logger.error(f"[Exception Recovery] Successfully freed worker slot for task {task_id}")
+            except Exception as completion_error:
+                logger.error(f"[Exception Recovery] Error in batch completion callback for {task_id}: {completion_error}")
+                # CRITICAL: If batch completion fails, we need to manually recover the worker slot
+                try:
+                    logger.error(f"[Exception Recovery] Attempting manual worker slot recovery for batch {batch_id}")
+                    deps.recover_worker_slot(batch_id, task_id)
+                except Exception as recovery_error:
+                    logger.error(f"[Exception Recovery] FATAL: Could not recover worker slot: {recovery_error}")
diff --git a/tests/downloads/test_downloads_task_worker.py b/tests/downloads/test_downloads_task_worker.py
new file mode 100644
index 00000000..fca7b10b
--- /dev/null
+++ b/tests/downloads/test_downloads_task_worker.py
@@ -0,0 +1,421 @@
+"""Tests for core/downloads/task_worker.py — per-task download worker."""
+
+from __future__ import annotations
+
+import pytest
+
+from core.downloads import task_worker as tw
+from core.runtime_state import download_tasks
+
+
+@pytest.fixture(autouse=True)
+def reset_state():
+    download_tasks.clear()
+    yield
+    download_tasks.clear()
+
+
+# ---------------------------------------------------------------------------
+# Fakes
+# ---------------------------------------------------------------------------
+
+class _Recorder:
+    def __init__(self):
+        self.calls = []
+
+    def __call__(self, name):
+        def _inner(*args, **kwargs):
+            self.calls.append((name, args, kwargs))
+            return None
+        return _inner
+
+
+class _FakeClient:
+    """Stub soulseek client. `mode` defaults to non-hybrid."""
+    def __init__(self, results=None, mode='soulseek', subclients=None):
+        self._results = results if results is not None else []
+        self.mode = mode
+        self.search_calls = []
+        for k, v in (subclients or {}).items():
+            setattr(self, k, v)
+
+    async def search(self, query, timeout=30):
+        self.search_calls.append((query, timeout))
+        return (self._results, None)
+
+
+class _FakeMatchEngine:
+    def __init__(self, queries=None):
+        self._queries = queries or []
+
+    def generate_download_queries(self, track):
+        return list(self._queries)
+
+
+def _sync_run_async(coro):
+    """Drain a coroutine on a fresh loop."""
+    import asyncio
+    loop = asyncio.new_event_loop()
+    try:
+        return loop.run_until_complete(coro)
+    finally:
+        loop.close()
+
+
+def _build_deps(
+    *,
+    soulseek=None,
+    matching=None,
+    try_source_reuse=lambda *a, **kw: False,
+    store_batch_source=None,
+    try_staging_match=lambda *a, **kw: False,
+    get_valid_candidates=lambda r, t, q: [],
+    attempt_download_with_candidates=lambda *a, **kw: False,
+    on_download_completed=None,
+    recover_worker_slot=None,
+):
+    rec = _Recorder()
+    return tw.TaskWorkerDeps(
+        soulseek_client=soulseek or _FakeClient(),
+        matching_engine=matching or _FakeMatchEngine(),
+        run_async=_sync_run_async,
+        try_source_reuse=try_source_reuse,
+        store_batch_source=store_batch_source or rec('store_batch_source'),
+        try_staging_match=try_staging_match,
+        get_valid_candidates=get_valid_candidates,
+        attempt_download_with_candidates=attempt_download_with_candidates,
+        on_download_completed=on_download_completed or rec('on_download_completed'),
+        recover_worker_slot=recover_worker_slot or rec('recover_worker_slot'),
+    ), rec
+
+
+def _seed_task(task_id='t1', status='pending', track_info=None, **extra):
+    download_tasks[task_id] = {
+        'status': status,
+        'track_info': track_info if track_info is not None else {
+            'id': 'sp-1', 'name': 'Money', 'artists': ['Pink Floyd'],
+            'album': 'DSOTM', 'duration_ms': 383000,
+        },
+        **extra,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Early-return guards
+# ---------------------------------------------------------------------------
+
+def test_missing_task_returns_silently():
+    deps, rec = _build_deps()
+    tw.download_track_worker('absent', 'b1', deps)
+    assert rec.calls == []
+
+
+def test_cancelled_v2_task_returns_without_completion_callback():
+    """V2 tasks (with playlist_id) handle worker slot freeing themselves."""
+    _seed_task(status='cancelled', playlist_id='pl1')
+    deps, rec = _build_deps()
+    tw.download_track_worker('t1', 'b1', deps)
+    # NOT called — V2 system frees its own slots
+    assert ('on_download_completed', ('b1', 't1', False), {}) not in rec.calls
+
+
+def test_cancelled_legacy_task_calls_completion_callback():
+    """Legacy tasks (no playlist_id) need on_download_completed to free slot."""
+    _seed_task(status='cancelled')  # no playlist_id
+    deps, rec = _build_deps()
+    tw.download_track_worker('t1', 'b1', deps)
+    assert ('on_download_completed', ('b1', 't1', False), {}) in rec.calls
+
+
+def test_cancelled_no_batch_id_just_returns():
+    _seed_task(status='cancelled')
+    deps, rec = _build_deps()
+    tw.download_track_worker('t1', None, deps)
+    # Nothing called — no batch to notify
+    assert rec.calls == []
+
+
+# ---------------------------------------------------------------------------
+# Source reuse + staging shortcuts
+# ---------------------------------------------------------------------------
+
+def test_source_reuse_hit_records_filename_and_returns():
+    _seed_task()
+
+    def _reuse_hit(task_id, batch_id, track):
+        download_tasks[task_id]['filename'] = 'reused.flac'
+        download_tasks[task_id]['username'] = 'u1'
+        return True
+
+    rec = _Recorder()
+    deps, _ = _build_deps(
+        try_source_reuse=_reuse_hit,
+        store_batch_source=rec('store_batch_source'),
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    assert ('store_batch_source', ('b1', 'u1', 'reused.flac'), {}) in rec.calls
+
+
+def test_source_reuse_hit_skips_store_when_no_filename():
+    """Reuse returns True but doesn't write filename → no store call."""
+    _seed_task()
+    rec = _Recorder()
+    deps, _ = _build_deps(
+        try_source_reuse=lambda *a, **kw: True,
+        store_batch_source=rec('store_batch_source'),
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    assert rec.calls == []  # neither store nor on_complete
+
+
+def test_staging_match_hit_returns_immediately():
+    _seed_task()
+    rec = _Recorder()
+    deps, _ = _build_deps(
+        try_staging_match=lambda *a, **kw: True,
+        store_batch_source=rec('store_batch_source'),
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    assert rec.calls == []
+
+
+# ---------------------------------------------------------------------------
+# Search loop happy path
+# ---------------------------------------------------------------------------
+
+def test_first_query_success_returns_after_storing_source():
+    _seed_task()
+    rec = _Recorder()
+
+    def _attempt_success(task_id, candidates, track, batch_id):
+        download_tasks[task_id]['filename'] = 'song.flac'
+        download_tasks[task_id]['username'] = 'u1'
+        return True
+
+    deps, _ = _build_deps(
+        soulseek=_FakeClient(results=['raw1', 'raw2']),
+        matching=_FakeMatchEngine(queries=['Pink Floyd Money']),
+        get_valid_candidates=lambda r, t, q: [{'username': 'u1', 'filename': 'song.flac'}],
+        attempt_download_with_candidates=_attempt_success,
+        store_batch_source=rec('store'),
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    assert ('store', ('b1', 'u1', 'song.flac'), {}) in rec.calls
+    assert download_tasks['t1']['status'] == 'searching'
+
+
+def test_no_results_marks_not_found_and_calls_completion():
+    _seed_task()
+    rec = _Recorder()
+    deps, _ = _build_deps(
+        soulseek=_FakeClient(results=[]),
+        matching=_FakeMatchEngine(queries=['q1']),
+        on_download_completed=rec('done'),
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    assert download_tasks['t1']['status'] == 'not_found'
+    assert 'No match found' in download_tasks['t1']['error_message']
+    assert ('done', ('b1', 't1', False), {}) in rec.calls
+
+
+def test_results_but_no_valid_candidates_stores_raw_for_review():
+    """Each query that returns results contributes top 20 to cached_candidates.
+    With legacy fallback queries (track-only, cleaned), multiple queries fire."""
+    _seed_task()
+    deps, _ = _build_deps(
+        soulseek=_FakeClient(results=[f'raw{i}' for i in range(30)]),
+        matching=_FakeMatchEngine(queries=['q1']),
+        get_valid_candidates=lambda r, t, q: [],  # nothing passes filter
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    # Status: not_found
+    assert download_tasks['t1']['status'] == 'not_found'
+    # Raw results stored (top 20 PER query that returned results)
+    assert len(download_tasks['t1']['cached_candidates']) >= 20
+    assert len(download_tasks['t1']['cached_candidates']) % 20 == 0  # multiple of 20
+
+
+def test_attempt_download_failure_falls_through_to_next_query():
+    _seed_task()
+    deps, _ = _build_deps(
+        soulseek=_FakeClient(results=['r1']),
+        matching=_FakeMatchEngine(queries=['q1', 'q2']),
+        get_valid_candidates=lambda r, t, q: [{'x': 1}],  # both queries get candidates
+        attempt_download_with_candidates=lambda *a, **kw: False,  # but download fails
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    # Both queries tried, failed → not_found
+    assert download_tasks['t1']['status'] == 'not_found'
+
+
+# ---------------------------------------------------------------------------
+# Cancellation mid-flight
+# ---------------------------------------------------------------------------
+
+def test_cancellation_mid_query_returns_without_completion():
+    _seed_task()
+    rec = _Recorder()
+
+    def _cancel_during_search(query, timeout=30):
+        download_tasks['t1']['status'] = 'cancelled'
+
+        async def _empty():
+            return ([], None)
+        return _empty()
+
+    sk = _FakeClient(results=[])
+    sk.search = _cancel_during_search
+
+    deps, _ = _build_deps(
+        soulseek=sk,
+        matching=_FakeMatchEngine(queries=['q1']),
+        on_download_completed=rec('done'),
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    # No completion callback (cancellation prevents it)
+    assert ('done', ('b1', 't1', False), {}) not in rec.calls
+
+
+# ---------------------------------------------------------------------------
+# Hybrid fallback
+# ---------------------------------------------------------------------------
+
+def test_hybrid_fallback_tries_secondary_sources():
+    _seed_task()
+    youtube_client = _FakeClient(results=['yt-r1'])
+    sk = _FakeClient(
+        results=[],  # primary source returns nothing
+        mode='hybrid',
+        subclients={
+            'hybrid_order': ['soulseek', 'youtube'],
+            'soulseek': _FakeClient(results=[]),
+            'youtube': youtube_client,
+            'tidal': None, 'qobuz': None, 'hifi': None, 'deezer_dl': None,
+        },
+    )
+
+    def _attempt_yt_success(task_id, candidates, track, batch_id):
+        return True
+
+    deps, _ = _build_deps(
+        soulseek=sk,
+        matching=_FakeMatchEngine(queries=['q1']),
+        get_valid_candidates=lambda r, t, q: [{'x': 1}] if r else [],
+        attempt_download_with_candidates=_attempt_yt_success,
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    # YouTube was searched
+    assert len(youtube_client.search_calls) >= 1
+
+
+def test_hybrid_fallback_skipped_when_mode_not_hybrid():
+    _seed_task()
+    yt = _FakeClient(results=['r1'])
+    sk = _FakeClient(
+        results=[], mode='soulseek',  # not hybrid
+        subclients={'youtube': yt},
+    )
+    deps, _ = _build_deps(
+        soulseek=sk,
+        matching=_FakeMatchEngine(queries=['q1']),
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    # Fallback didn't run — youtube never searched
+    assert yt.search_calls == []
+
+
+# ---------------------------------------------------------------------------
+# Top-level exception path
+# ---------------------------------------------------------------------------
+
+def test_critical_exception_marks_failed_and_calls_completion():
+    _seed_task()
+    rec = _Recorder()
+
+    def _broken_engine(track):
+        raise RuntimeError("matching engine dead")
+
+    me = _FakeMatchEngine()
+    me.generate_download_queries = _broken_engine
+
+    deps, _ = _build_deps(
+        matching=me,
+        on_download_completed=rec('done'),
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    assert download_tasks['t1']['status'] == 'failed'
+    assert 'Unexpected error during download' in download_tasks['t1']['error_message']
+    assert ('done', ('b1', 't1', False), {}) in rec.calls
+
+
+def test_critical_exception_with_completion_failure_attempts_recovery():
+    _seed_task()
+    rec = _Recorder()
+
+    def _broken_engine(track):
+        raise RuntimeError("dead")
+
+    def _broken_completion(*a, **kw):
+        raise RuntimeError("completion dead")
+
+    me = _FakeMatchEngine()
+    me.generate_download_queries = _broken_engine
+
+    deps, _ = _build_deps(
+        matching=me,
+        on_download_completed=_broken_completion,
+        recover_worker_slot=rec('recover'),
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    # Recovery attempted after completion callback failed
+    assert ('recover', ('b1', 't1'), {}) in rec.calls
+
+
+# ---------------------------------------------------------------------------
+# Query generation edge cases
+# ---------------------------------------------------------------------------
+
+def test_artist_starting_with_the_uses_second_word():
+    """Legacy fallback: 'The Beatles' → first_word becomes 'Beatles'."""
+    _seed_task(track_info={
+        'id': 'sp1', 'name': 'Help', 'artists': ['The Beatles'],
+        'album': 'Help', 'duration_ms': 100000,
+    })
+    sk = _FakeClient(results=[])
+    deps, _ = _build_deps(soulseek=sk, matching=_FakeMatchEngine(queries=[]))
+    tw.download_track_worker('t1', 'b1', deps)
+    # Searched queries should contain 'Help Beatles' (track + second word)
+    queries = [q for q, _ in sk.search_calls]
+    assert any('Beatles' in q for q in queries)
+
+
+def test_track_with_parens_generates_cleaned_variant():
+    """`Money (Remastered)` → also tries `Money` as fallback query."""
+    _seed_task(track_info={
+        'id': 'sp1', 'name': 'Money (Remastered)', 'artists': ['Pink Floyd'],
+        'album': 'DSOTM', 'duration_ms': 100000,
+    })
+    sk = _FakeClient(results=[])
+    deps, _ = _build_deps(soulseek=sk, matching=_FakeMatchEngine(queries=[]))
+    tw.download_track_worker('t1', 'b1', deps)
+    queries = [q for q, _ in sk.search_calls]
+    # Cleaned variant included
+    assert 'Money' in queries
+
+
+def test_duplicate_queries_deduplicated_case_insensitive():
+    """Generated + legacy queries dedupe by lowercase."""
+    _seed_task(track_info={
+        'id': 'sp1', 'name': 'X', 'artists': ['Y'],
+        'album': '', 'duration_ms': 0,
+    })
+    sk = _FakeClient(results=[])
+    deps, _ = _build_deps(
+        soulseek=sk,
+        # Engine generates same query as legacy 'track-only' fallback
+        matching=_FakeMatchEngine(queries=['x', 'X']),
+    )
+    tw.download_track_worker('t1', 'b1', deps)
+    # 'x' and 'X' dedupe to one search per case-insensitive match
+    queries_lower = [q.lower() for q, _ in sk.search_calls]
+    assert queries_lower.count('x') == 1
diff --git a/web_server.py b/web_server.py
index 1a6008c8..64d447a5 100644
--- a/web_server.py
+++ b/web_server.py
@@ -22537,338 +22537,31 @@ def _run_post_processing_worker(task_id, batch_id):
 
 
 
+# Per-task download worker logic lives in core/downloads/task_worker.py.
+from core.downloads import task_worker as _downloads_task_worker
+
+
+def _build_task_worker_deps():
+    """Build TaskWorkerDeps bundle from web_server.py globals on each call."""
+    return _downloads_task_worker.TaskWorkerDeps(
+        soulseek_client=soulseek_client,
+        matching_engine=matching_engine,
+        run_async=run_async,
+        try_source_reuse=_try_source_reuse,
+        store_batch_source=_store_batch_source,
+        try_staging_match=_try_staging_match,
+        get_valid_candidates=get_valid_candidates,
+        attempt_download_with_candidates=_attempt_download_with_candidates,
+        on_download_completed=lambda b, t, success: _on_download_completed(b, t, success=success),
+        recover_worker_slot=_recover_worker_slot,
+    )
+
+
 def _download_track_worker(task_id, batch_id=None):
-    """
-    Enhanced download worker that matches the GUI's exact retry logic.
-    Implements sequential query retry, fallback candidates, and download failure retry.
-    """
-    try:
-        # Retrieve task details from global state
-        with tasks_lock:
-            if task_id not in download_tasks:
-                logger.warning(f"[Modal Worker] Task {task_id} not found in download_tasks")
-                return
-            task = download_tasks[task_id].copy()
-            
-        # Cancellation Checkpoint 1: Before doing anything
-        with tasks_lock:
-            if task_id not in download_tasks:
-                logger.info(f"[Modal Worker] Task {task_id} was deleted before starting")
-                return
-            if download_tasks[task_id]['status'] == 'cancelled':
-                logger.warning(f"[Modal Worker] Task {task_id} cancelled before starting")
-                # V2 FIX: Don't call _on_download_completed for cancelled V2 tasks
-                # V2 system handles worker slot freeing in atomic cancel function
-                task_playlist_id = download_tasks[task_id].get('playlist_id')
-                if task_playlist_id:
-                    logger.warning(f"[Modal Worker] V2 task {task_id} cancelled - worker slot already freed by V2 system")
-                    return  # V2 system already handled worker slot management
-                elif batch_id:
-                    # Legacy system - use old completion callback
-                    logger.warning(f"[Modal Worker] Legacy task {task_id} cancelled - using legacy completion callback")
-                    _on_download_completed(batch_id, task_id, success=False)
-                return
+    """Per-task download worker — see core/downloads/task_worker.py."""
+    _downloads_task_worker.download_track_worker(task_id, batch_id, _build_task_worker_deps())
 
-        track_data = task['track_info']
-        track_name = track_data.get('name', 'Unknown Track')
-        
-        logger.info(f"[Modal Worker] Task {task_id} starting search for track: '{track_name}'")
-        
-        # Recreate a SpotifyTrack object for the matching engine
-        # Handle both string format and Spotify API format for artists
-        raw_artists = track_data.get('artists', [])
-        processed_artists = []
-        for artist in raw_artists:
-            if isinstance(artist, str):
-                processed_artists.append(artist)
-            elif isinstance(artist, dict) and 'name' in artist:
-                processed_artists.append(artist['name'])
-            else:
-                processed_artists.append(str(artist))
-        
-        # Handle album field - extract name if it's a dictionary
-        raw_album = track_data.get('album', '')
-        if isinstance(raw_album, dict) and 'name' in raw_album:
-            album_name = raw_album['name']
-        elif isinstance(raw_album, str):
-            album_name = raw_album
-        else:
-            album_name = str(raw_album)
-        
-        track = SpotifyTrack(
-            id=track_data.get('id', ''),
-            name=track_data.get('name', ''),
-            artists=processed_artists,
-            album=album_name,
-            duration_ms=track_data.get('duration_ms', 0),
-            popularity=track_data.get('popularity', 0)
-        )
-        logger.info(f"[Modal Worker] Starting download task for: {track.name} by {track.artists[0] if track.artists else 'Unknown'}")
 
-        # === SOURCE REUSE: Check batch's last good source before searching ===
-        if _try_source_reuse(task_id, batch_id, track):
-            # Store source for next worker (cascading reuse)
-            with tasks_lock:
-                used_filename = download_tasks.get(task_id, {}).get('filename')
-                used_username = download_tasks.get(task_id, {}).get('username')
-            if used_filename and used_username:
-                _store_batch_source(batch_id, used_username, used_filename)
-            return
-
-        # === STAGING CHECK: Check staging folder for existing file before searching ===
-        if _try_staging_match(task_id, batch_id, track):
-            return
-
-        # Initialize task state tracking (like GUI's parallel_search_tracking)
-        with tasks_lock:
-            if task_id in download_tasks:
-                download_tasks[task_id]['status'] = 'searching'  # Now actively being processed
-                download_tasks[task_id]['current_query_index'] = 0
-                download_tasks[task_id]['current_candidate_index'] = 0
-                download_tasks[task_id]['retry_count'] = 0
-                download_tasks[task_id]['candidates'] = []
-                # CRITICAL: Preserve used_sources from previous retry attempts (don't reset to empty set)
-                # If this is a retry, the monitor will have already marked failed sources
-                if 'used_sources' not in download_tasks[task_id]:
-                    download_tasks[task_id]['used_sources'] = set()
-                # Else: keep existing used_sources to avoid retrying same failed hosts
-
-        # 1. Generate multiple search queries (like GUI's generate_smart_search_queries)
-        artist_name = track.artists[0] if track.artists else None
-        track_name = track.name
-        
-        # Start with matching engine queries
-        search_queries = matching_engine.generate_download_queries(track)
-        
-        # Add legacy fallback queries (like GUI does)
-        legacy_queries = []
-        
-        if artist_name:
-            # Add first word of artist approach (legacy compatibility)
-            artist_words = artist_name.split()
-            if artist_words:
-                first_word = artist_words[0]
-                if first_word.lower() == 'the' and len(artist_words) > 1:
-                    first_word = artist_words[1]
-                
-                if len(first_word) > 1:
-                    legacy_queries.append(f"{track_name} {first_word}".strip())
-        
-        # Add track-only query
-        if track_name.strip():
-            legacy_queries.append(track_name.strip())
-        
-        # Add traditional cleaned queries
-        cleaned_name = re.sub(r'\s*\([^)]*\)', '', track_name).strip()
-        cleaned_name = re.sub(r'\s*\[[^\]]*\]', '', cleaned_name).strip()
-        
-        if cleaned_name and cleaned_name.lower() != track_name.lower():
-            legacy_queries.append(cleaned_name.strip())
-        
-        # Combine enhanced queries with legacy fallbacks
-        all_queries = search_queries + legacy_queries
-        
-        # Remove duplicates while preserving order
-        unique_queries = []
-        seen = set()
-        for query in all_queries:
-            if query and query.lower() not in seen:
-                unique_queries.append(query)
-                seen.add(query.lower())
-        
-        search_queries = unique_queries
-        logger.info(f"[Modal Worker] Generated {len(search_queries)} smart search queries for '{track.name}': {search_queries}")
-        logger.info(f"[Modal Worker] About to start search loop for task {task_id} (track: '{track.name}')")
-
-        # 2. Sequential Query Search (matches GUI's start_search_worker_parallel logic)
-        search_diagnostics = []  # Track what happened per query for detailed error messages
-        all_raw_results = []  # Collect raw results across queries for candidate review modal
-        for query_index, query in enumerate(search_queries):
-            # Cancellation check before each query
-            with tasks_lock:
-                if task_id not in download_tasks:
-                    logger.debug(f"[Modal Worker] Task {task_id} was deleted during query {query_index + 1}")
-                    return
-                if download_tasks[task_id]['status'] == 'cancelled':
-                    logger.debug(f"[Modal Worker] Task {task_id} cancelled during query {query_index + 1}")
-                    # Don't call _on_download_completed for cancelled tasks as it can stop monitoring
-                    return
-                download_tasks[task_id]['current_query_index'] = query_index
-                    
-            logger.debug(f"[Modal Worker] Query {query_index + 1}/{len(search_queries)}: '{query}'")
-            logger.debug(f"About to call soulseek search for task {task_id}")
-            
-            try:
-                # Perform search with timeout
-                tracks_result, _ = run_async(soulseek_client.search(query, timeout=30))
-                logger.debug(f"Search completed for task {task_id}, got {len(tracks_result) if tracks_result else 0} results")
-                
-                # CRITICAL: Check cancellation immediately after search returns
-                with tasks_lock:
-                    if task_id not in download_tasks:
-                        logger.info(f"[Modal Worker] Task {task_id} was deleted after search returned")
-                        return
-                    if download_tasks[task_id]['status'] == 'cancelled':
-                        logger.warning(f"[Modal Worker] Task {task_id} cancelled after search returned - ignoring results")
-                        # Don't call _on_download_completed for cancelled tasks as it can stop monitoring
-                        # The cancellation endpoint already handles batch management properly
-                        return
-                
-                if tracks_result:
-                    result_count = len(tracks_result)
-                    # Validate candidates using GUI's get_valid_candidates logic
-                    candidates = get_valid_candidates(tracks_result, track, query)
-                    if candidates:
-                        logger.debug(f"[Modal Worker] Found {len(candidates)} valid candidates for query '{query}'")
-
-                        # CRITICAL: Check cancellation before processing candidates
-                        with tasks_lock:
-                            if task_id not in download_tasks:
-                                logger.info(f"[Modal Worker] Task {task_id} was deleted before processing candidates")
-                                return
-                            if download_tasks[task_id]['status'] == 'cancelled':
-                                logger.warning(f"[Modal Worker] Task {task_id} cancelled before processing candidates")
-                                # Don't call _on_download_completed for cancelled tasks as it can stop monitoring
-                                return
-                            # Store candidates for retry fallback (like GUI)
-                            download_tasks[task_id]['cached_candidates'] = candidates
-
-                        # Try to download with these candidates
-                        success = _attempt_download_with_candidates(task_id, candidates, track, batch_id)
-                        if success:
-                            # Download initiated successfully - let the download monitoring system handle completion
-                            if batch_id:
-                                logger.info(f"[Modal Worker] Download initiated successfully for task {task_id} - monitoring will handle completion")
-                            # Store this source for batch reuse
-                            with tasks_lock:
-                                used_filename = download_tasks.get(task_id, {}).get('filename')
-                                used_username = download_tasks.get(task_id, {}).get('username')
-                            if used_filename and used_username:
-                                _store_batch_source(batch_id, used_username, used_filename)
-                            return  # Success, exit the worker
-                        else:
-                            search_diagnostics.append(f'"{query}": {result_count} results, {len(candidates)} passed filters but download failed to start')
-                    else:
-                        search_diagnostics.append(f'"{query}": {result_count} results but none passed quality/artist filters')
-                        all_raw_results.extend(tracks_result[:20])  # Keep top results for review
-                else:
-                    search_diagnostics.append(f'"{query}": no results found')
-
-            except Exception as e:
-                logger.debug(f"[Modal Worker] Search failed for query '{query}': {e}")
-                search_diagnostics.append(f'"{query}": search error — {e}')
-                continue
-
-        # === HYBRID FALLBACK: If primary source failed, try remaining sources directly ===
-        # The orchestrator's hybrid search stops at the first source with results, even if
-        # those results all fail quality filtering. Try remaining sources individually.
-        if getattr(soulseek_client, 'mode', '') == 'hybrid':
-            try:
-                orch = soulseek_client
-                hybrid_order = getattr(orch, 'hybrid_order', None) or []
-                if not hybrid_order:
-                    primary = getattr(orch, 'hybrid_primary', 'soulseek')
-                    secondary = getattr(orch, 'hybrid_secondary', '')
-                    hybrid_order = [primary, secondary] if secondary and secondary != primary else [primary]
-
-                source_clients = {
-                    'soulseek': getattr(orch, 'soulseek', None),
-                    'youtube': getattr(orch, 'youtube', None),
-                    'tidal': getattr(orch, 'tidal', None),
-                    'qobuz': getattr(orch, 'qobuz', None),
-                    'hifi': getattr(orch, 'hifi', None),
-                    'deezer_dl': getattr(orch, 'deezer_dl', None),
-                }
-
-                # The orchestrator tried sources in order but stopped at the first with results.
-                # We don't know which it stopped at, so try ALL sources except the first
-                # (which was definitely tried). If the first was skipped (unconfigured),
-                # the orchestrator would have tried the second — but trying it again is
-                # harmless (streaming sources return fast).
-                remaining_sources = [s for s in hybrid_order[1:] if s in source_clients and source_clients[s]]
-                if remaining_sources:
-                    logger.warning(f"[Hybrid Fallback] Primary source had no valid matches. Trying fallback sources: {remaining_sources}")
-
-                for fallback_source in remaining_sources:
-                    fb_client = source_clients[fallback_source]
-                    if hasattr(fb_client, 'is_configured') and not fb_client.is_configured():
-                        continue
-
-                    # Use first 2 queries only for speed
-                    for fb_query in search_queries[:2]:
-                        try:
-                            logger.warning(f"[Hybrid Fallback] Trying {fallback_source}: '{fb_query}'")
-                            fb_results, _ = run_async(fb_client.search(fb_query, timeout=20))
-                            if not fb_results:
-                                continue
-                            fb_candidates = get_valid_candidates(fb_results, track, fb_query)
-                            if fb_candidates:
-                                logger.warning(f"[Hybrid Fallback] {fallback_source} found {len(fb_candidates)} valid candidates!")
-                                success = _attempt_download_with_candidates(task_id, fb_candidates, track, batch_id)
-                                if success:
-                                    return
-                        except Exception as e:
-                            logger.error(f"[Hybrid Fallback] {fallback_source} search failed: {e}")
-                            continue
-
-                    logger.warning(f"[Hybrid Fallback] {fallback_source} returned no valid candidates")
-
-            except Exception as e:
-                logger.error(f"[Hybrid Fallback] Error in fallback logic: {e}")
-
-        # If we get here, all search queries and hybrid fallbacks failed
-        logger.warning(f"[Modal Worker] No valid candidates found for '{track.name}' after trying all {len(search_queries)} queries.")
-        with tasks_lock:
-            if task_id in download_tasks:
-                download_tasks[task_id]['status'] = 'not_found'
-                _diag_summary = ' | '.join(search_diagnostics) if search_diagnostics else 'no queries attempted'
-                download_tasks[task_id]['error_message'] = f'No match found for "{track_name}" by {artist_name or "Unknown"} after {len(search_queries)} queries. Breakdown: {_diag_summary}'
-                # Store raw results so the user can review what Soulseek returned
-                if all_raw_results and not download_tasks[task_id].get('cached_candidates'):
-                    download_tasks[task_id]['cached_candidates'] = all_raw_results
-        
-        # Notify batch manager that this task completed (failed) - THREAD SAFE
-        if batch_id:
-            try:
-                _on_download_completed(batch_id, task_id, success=False)
-            except Exception as completion_error:
-                logger.error(f"Error in batch completion callback for {task_id}: {completion_error}")
-
-    except Exception as e:
-        import traceback
-        track_name_safe = locals().get('track_name', 'unknown')  # Safe fallback for track_name
-        logger.error(f"CRITICAL ERROR in download task for '{track_name_safe}' (task_id: {task_id}): {e}")
-        traceback.print_exc()
-        
-        # Update task status safely with timeout
-        try:
-            lock_acquired = tasks_lock.acquire(timeout=2.0)
-            if lock_acquired:
-                try:
-                    if task_id in download_tasks:
-                        download_tasks[task_id]['status'] = 'failed'
-                        download_tasks[task_id]['error_message'] = f'Unexpected error during download: {type(e).__name__}: {e}'
-                        logger.error(f"[Exception Recovery] Set task {task_id} status to 'failed'")
-                finally:
-                    tasks_lock.release()
-            else:
-                logger.error(f"[Exception Recovery] Could not acquire lock to update task {task_id} status")
-        except Exception as status_error:
-            logger.error(f"Error updating task status in exception handler: {status_error}")
-        
-        # Notify batch manager that this task completed (failed) - THREAD SAFE with RECOVERY
-        if batch_id:
-            try:
-                _on_download_completed(batch_id, task_id, success=False)
-                logger.error(f"[Exception Recovery] Successfully freed worker slot for task {task_id}")
-            except Exception as completion_error:
-                logger.error(f"[Exception Recovery] Error in batch completion callback for {task_id}: {completion_error}")
-                # CRITICAL: If batch completion fails, we need to manually recover the worker slot
-                try:
-                    logger.error(f"[Exception Recovery] Attempting manual worker slot recovery for batch {batch_id}")
-                    _recover_worker_slot(batch_id, task_id)
-                except Exception as recovery_error:
-                    logger.error(f"[Exception Recovery] FATAL: Could not recover worker slot: {recovery_error}")
 
 def _attempt_download_with_candidates(task_id, candidates, track, batch_id=None):
     """