diff --git a/core/downloads/lifecycle.py b/core/downloads/lifecycle.py
new file mode 100644
index 00000000..bb3d7d4a
--- /dev/null
+++ b/core/downloads/lifecycle.py
@@ -0,0 +1,653 @@
+"""Batch lifecycle: start workers, on-completion accounting, completion check.
+
+Three deeply-coupled functions:
+
+- `start_next_batch_of_downloads(batch_id, deps)` — launches workers up to
+  the batch's max_concurrent. Skips cancelled tasks, sets searching status,
+  submits to the executor, decrement-safe on submit failures (no ghost
+  workers).
+
+- `on_download_completed(batch_id, task_id, success, deps)` — called when
+  a single track download finishes (good or bad). Tracks failed/cancelled
+  tracks for wishlist replay, decrements active count, then runs the full
+  batch-completion check — which is its own beast: stuck-task detection
+  (searching > 10min → not_found, post_processing > 5min → completed),
+  M3U regeneration, repair worker hand-off, album consistency pass,
+  wishlist failed-tracks processing.
+
+- `check_batch_completion_v2(batch_id, deps)` — same completion check
+  but called from the V2 atomic cancel path (which bypasses
+  on_download_completed). Duplicate logic preserved verbatim.
+
+Lifted verbatim from web_server.py. Dependencies injected via
+`LifecycleDeps` since the surface is wide (15+ callbacks/refs).
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+import traceback
+from dataclasses import dataclass
+from typing import Any, Callable, Optional
+
+from core.downloads.history import record_sync_history_completion
+from core.runtime_state import (
+    add_activity_item,
+    download_batches,
+    download_tasks,
+    tasks_lock,
+)
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class LifecycleDeps:
+    """Bundle of cross-cutting deps the batch lifecycle needs."""
+    config_manager: Any
+    automation_engine: Any
+    download_monitor: Any
+    repair_worker: Any
+    mb_worker: Any
+    is_shutting_down: Callable[[], bool]
+    get_batch_lock: Callable[[str], Any]                    # (batch_id) -> threading.Lock
+    submit_download_track_worker: Callable                  # (task_id, batch_id) -> None (submits to executor)
+    submit_failed_to_wishlist: Callable[[str], None]                  # async — submits to executor
+    submit_failed_to_wishlist_with_auto_completion: Callable[[str], None]  # async — submits to executor
+    process_failed_to_wishlist: Callable[[str], None]                 # sync — direct call (used by v2 path)
+    process_failed_to_wishlist_with_auto_completion: Callable[[str], None]  # sync — direct call (used by v2 path)
+    ensure_spotify_track_format: Callable
+    get_track_artist_name: Callable
+    check_and_remove_from_wishlist: Callable
+    regenerate_batch_m3u: Callable
+    youtube_playlist_states: dict
+    tidal_discovery_states: dict
+    deezer_discovery_states: dict
+    spotify_public_discovery_states: dict
+
+
+# ---------------------------------------------------------------------------
+# start_next_batch_of_downloads
+# ---------------------------------------------------------------------------
+
+def start_next_batch_of_downloads(batch_id: str, deps: LifecycleDeps) -> None:
+    """Start the next batch of downloads up to the concurrent limit (like GUI)."""
+    # ENHANCED: Use batch-specific lock to prevent race conditions when multiple threads
+    # try to start workers for the same batch concurrently
+    batch_lock = deps.get_batch_lock(batch_id)
+
+    with batch_lock:
+        # Prevent starting new tasks if shutting down
+        if deps.is_shutting_down():
+            logger.info(f"[Batch Manager] Server shutting down - skipping new tasks for batch {batch_id}")
+            return
+
+        with tasks_lock:
+            if batch_id not in download_batches:
+                return
+
+            batch = download_batches[batch_id]
+            max_concurrent = batch['max_concurrent']
+            queue = batch['queue']
+            queue_index = batch['queue_index']
+            active_count = batch['active_count']
+
+            logger.info(f"[Batch Lock] Starting workers for {batch_id}: active={active_count}, max={max_concurrent}, queue_pos={queue_index}/{len(queue)}")
+
+            # Start downloads up to the concurrent limit
+            while active_count < max_concurrent and queue_index < len(queue):
+                task_id = queue[queue_index]
+
+                # CRITICAL V2 FIX: Skip cancelled tasks instead of trying to restart them
+                if task_id in download_tasks:
+                    current_status = download_tasks[task_id]['status']
+                    if current_status == 'cancelled':
+                        logger.warning(f"[Batch Lock] Skipping cancelled task {task_id} (queue position {queue_index + 1})")
+                        download_batches[batch_id]['queue_index'] += 1
+                        queue_index += 1
+                        continue  # Skip to next task without consuming worker slot
+
+                    # IMPORTANT: Set status to 'searching' BEFORE starting worker (like GUI)
+                    # Must be done INSIDE the lock to prevent race conditions with status polling
+                    download_tasks[task_id]['status'] = 'searching'
+                    download_tasks[task_id]['status_change_time'] = time.time()
+                    logger.info(f"[Batch Manager] Set task {task_id} status to 'searching'")
+                else:
+                    logger.warning(f"[Batch Lock] Task {task_id} not found in download_tasks - skipping")
+                    download_batches[batch_id]['queue_index'] += 1
+                    queue_index += 1
+                    continue
+
+                # CRITICAL FIX: Submit to executor BEFORE incrementing counters to prevent ghost workers
+                try:
+                    # Submit to executor first - this can fail
+                    deps.submit_download_track_worker(task_id, batch_id)
+
+                    # Only increment counters AFTER successful submit
+                    download_batches[batch_id]['active_count'] += 1
+                    download_batches[batch_id]['queue_index'] += 1
+
+                    logger.info(f"[Batch Lock] Started download {queue_index + 1}/{len(queue)} - Active: {active_count + 1}/{max_concurrent}")
+
+                    # Update local counters for next iteration
+                    active_count += 1
+                    queue_index += 1
+
+                except Exception as submit_error:
+                    logger.error(f"[Batch Lock] CRITICAL: Failed to submit task {task_id} to executor: {submit_error}")
+                    logger.info("[Batch Lock] Worker slot NOT consumed - preventing ghost worker")
+
+                    # Reset task status since worker never started
+                    if task_id in download_tasks:
+                        download_tasks[task_id]['status'] = 'failed'
+                        logger.error(f"[Batch Lock] Set task {task_id} status to 'failed' due to submit failure")
+
+                    # Don't increment counters - no worker was actually started
+                    # This prevents the "ghost worker" issue where active_count is incremented but no actual worker runs
+                    break  # Stop trying to start more workers if executor is failing
+
+            logger.info(f"[Batch Lock] Finished starting workers for {batch_id}: final_active={download_batches[batch_id]['active_count']}, max={max_concurrent}")
+
+
+# ---------------------------------------------------------------------------
+# on_download_completed
+# ---------------------------------------------------------------------------
+
+def on_download_completed(batch_id: str, task_id: str, success: bool, deps: LifecycleDeps) -> None:
+    """Called when a download completes to start the next one in queue."""
+    with tasks_lock:
+        if batch_id not in download_batches:
+            logger.warning(f"[Batch Manager] Batch {batch_id} not found for completed task {task_id}")
+            return
+
+        # Guard against double-calling: track which tasks have already been completed
+        # This prevents active_count from being decremented multiple times for the same task
+        # (e.g. monitor detects completion AND post-processing calls this again)
+        # NOTE: On duplicate calls, we skip decrement/tracking but STILL check batch completion.
+        # This is critical because the first call may see the task in 'post_processing' (not finished),
+        # and the second call (from post-processing worker) arrives after the task is truly 'completed'.
+        # Without the fallthrough, batch_complete would never be emitted.
+        completed_tasks = download_batches[batch_id].setdefault('_completed_task_ids', set())
+        _is_duplicate_completion = task_id in completed_tasks
+        if _is_duplicate_completion:
+            logger.info(f"[Batch Manager] Task {task_id} already completed — skipping decrement, still checking batch completion")
+            # Set terminal status so the monitor loop stops re-processing this task
+            if task_id in download_tasks and download_tasks[task_id].get('status') in ('downloading', 'queued'):
+                download_tasks[task_id]['status'] = 'completed'
+            # Fall through to batch completion check below (don't return)
+        else:
+            completed_tasks.add(task_id)
+
+        if not _is_duplicate_completion:
+            # Track failed/cancelled tasks in batch state (replicating sync.py)
+            if not success and task_id in download_tasks:
+                task = download_tasks[task_id]
+                task_status = task.get('status', 'unknown')
+
+                # Build track_info structure matching sync.py's permanently_failed_tracks format
+                original_track_info = task.get('track_info', {})
+
+                # Ensure spotify_track has proper structure for wishlist service
+                spotify_track_data = deps.ensure_spotify_track_format(original_track_info)
+
+                track_info = {
+                    'download_index': task.get('track_index', 0),
+                    'table_index': task.get('track_index', 0),
+                    'track_name': original_track_info.get('name', 'Unknown Track'),
+                    'artist_name': deps.get_track_artist_name(original_track_info),
+                    'retry_count': task.get('retry_count', 0),
+                    'spotify_track': spotify_track_data,  # Properly formatted spotify track for wishlist
+                    'failure_reason': 'Download cancelled' if task_status == 'cancelled' else ('No matching track found' if task_status == 'not_found' else 'Download failed'),
+                    'candidates': task.get('cached_candidates', []),  # Include search results if available
+                }
+
+                if task_status == 'cancelled':
+                    download_batches[batch_id]['cancelled_tracks'].add(task.get('track_index', 0))
+                    logger.warning(f"[Batch Manager] Added cancelled track to batch tracking: {track_info['track_name']}")
+                    add_activity_item("", "Download Cancelled", f"'{track_info['track_name']}'", "Now")
+                elif task_status in ('failed', 'not_found'):
+                    download_batches[batch_id]['permanently_failed_tracks'].append(track_info)
+                    if task_status == 'not_found':
+                        logger.info(f"[Batch Manager] Added not-found track to batch tracking: {track_info['track_name']}")
+                        add_activity_item("", "Not Found", f"'{track_info['track_name']}'", "Now")
+                    else:
+                        logger.error(f"[Batch Manager] Added failed track to batch tracking: {track_info['track_name']}")
+                        add_activity_item("", "Download Failed", f"'{track_info['track_name']}'", "Now")
+
+                    try:
+                        if deps.automation_engine:
+                            deps.automation_engine.emit('download_failed', {
+                                'artist': track_info.get('artist_name', ''),
+                                'title': track_info.get('track_name', ''),
+                                'reason': track_info.get('failure_reason', 'Unknown'),
+                            })
+                    except Exception:
+                        pass
+
+            # WISHLIST REMOVAL: Handle successful downloads for wishlist removal
+            if success and task_id in download_tasks:
+                try:
+                    task = download_tasks[task_id]
+                    track_info = task.get('track_info', {})
+                    logger.info(f"[Batch Manager] Successful download - checking wishlist removal for task {task_id}")
+
+                    # Add activity for successful download
+                    track_name = track_info.get('name', 'Unknown Track')
+
+                    # Safely extract artist name (handle both list and string formats)
+                    artists = track_info.get('artists', [])
+                    if isinstance(artists, list) and len(artists) > 0:
+                        first_artist = artists[0]
+                        artist_name = first_artist.get('name', 'Unknown Artist') if isinstance(first_artist, dict) else str(first_artist)
+                    elif isinstance(artists, str):
+                        artist_name = artists
+                    else:
+                        artist_name = 'Unknown Artist'
+
+                    add_activity_item("", "Download Complete", f"'{track_name}' by {artist_name}", "Now")
+
+                    # Try to remove from wishlist using track info
+                    if track_info:
+                        # Create a context-like structure for the wishlist removal function
+                        context = {
+                            'track_info': track_info,
+                            'original_search_result': track_info,  # fallback
+                        }
+                        deps.check_and_remove_from_wishlist(context)
+                except Exception as wishlist_error:
+                    logger.error(f"[Batch Manager] Error checking wishlist removal for successful download: {wishlist_error}")
+
+            # Decrement active count
+            old_active = download_batches[batch_id]['active_count']
+            download_batches[batch_id]['active_count'] -= 1
+            new_active = download_batches[batch_id]['active_count']
+
+            logger.error(f"[Batch Manager] Task {task_id} completed ({'success' if success else 'failed/cancelled'}). Active workers: {old_active} → {new_active}/{download_batches[batch_id]['max_concurrent']}")
+
+        # ENHANCED: Always check batch completion after any task completes (including duplicate calls)
+        # This ensures completion is detected even when mixing normal downloads with cancelled tasks
+        logger.info(f"[Batch Manager] Checking batch completion after task {task_id} completed")
+
+        # FIXED: Check if batch is truly complete (all tasks finished, not just workers freed)
+        batch = download_batches[batch_id]
+        all_tasks_started = batch['queue_index'] >= len(batch['queue'])
+        no_active_workers = batch['active_count'] == 0
+
+        # Count actually finished tasks (completed, failed, or cancelled)
+        # CRITICAL: Don't include 'post_processing' as finished - it's still in progress (unless stuck)!
+        # CRITICAL: Don't include 'searching' as finished - task is being retried (unless stuck)!
+        finished_count = 0
+        retrying_count = 0
+        queue = batch.get('queue', [])
+        current_time = time.time()
+        for queue_task_id in queue:
+            if queue_task_id in download_tasks:
+                task = download_tasks[queue_task_id]
+                task_status = task['status']
+
+                # STUCK DETECTION: Force fail tasks that have been in transitional states too long
+                if task_status == 'searching':
+                    task_age = current_time - task.get('status_change_time', current_time)
+                    if task_age > 600:  # 10 minutes
+                        logger.info(f"⏰ [Stuck Detection] Task {queue_task_id} stuck in searching for {task_age:.0f}s - forcing not_found")
+                        task['status'] = 'not_found'
+                        task['error_message'] = f'Search stuck for {int(task_age // 60)} minutes with no results — timed out'
+                        finished_count += 1
+                    else:
+                        retrying_count += 1
+                elif task_status == 'post_processing':
+                    task_age = current_time - task.get('status_change_time', current_time)
+                    if task_age > 300:  # 5 minutes (post-processing should be fast)
+                        logger.info(f"⏰ [Stuck Detection] Task {queue_task_id} stuck in post_processing for {task_age:.0f}s - forcing completion")
+                        task['status'] = 'completed'  # Assume it worked if file verification is taking too long
+                        finished_count += 1
+                    else:
+                        retrying_count += 1
+                elif task_status in ['completed', 'failed', 'cancelled', 'not_found']:
+                    finished_count += 1
+            else:
+                # Task ID in queue but not in download_tasks - treat as completed to prevent blocking
+                logger.warning(f"[Orphaned Task] Task {queue_task_id} in queue but not in download_tasks - counting as finished")
+                finished_count += 1
+
+        all_tasks_truly_finished = finished_count >= len(queue)
+        has_retrying_tasks = retrying_count > 0
+
+        if all_tasks_started and no_active_workers and all_tasks_truly_finished and not has_retrying_tasks:
+            logger.error(f"[Batch Manager] Batch {batch_id} truly complete - all {finished_count}/{len(queue)} tasks finished - processing failed tracks to wishlist")
+        elif all_tasks_started and no_active_workers and has_retrying_tasks:
+            logger.warning(f"[Batch Manager] Batch {batch_id}: all workers free but {retrying_count} tasks retrying - continuing monitoring")
+        elif all_tasks_started and no_active_workers:
+            # This used to incorrectly mark batch as complete!
+            logger.info(f"[Batch Manager] Batch {batch_id}: all workers free but only {finished_count}/{len(queue)} tasks finished - continuing monitoring")
+
+        if all_tasks_started and no_active_workers and all_tasks_truly_finished and not has_retrying_tasks:
+
+            # Check if this is an auto-initiated batch
+            is_auto_batch = batch.get('auto_initiated', False)
+
+            # FIXED: Ensure batch is not already marked as complete to prevent duplicate processing
+            if batch.get('phase') != 'complete':
+                # Mark batch as complete and set completion timestamp for auto-cleanup
+                batch['phase'] = 'complete'
+                batch['completion_time'] = time.time()  # Track when batch completed
+
+                # Record sync history completion
+                from database.music_database import MusicDatabase
+                record_sync_history_completion(MusicDatabase(), batch_id, batch)
+
+                # Add activity for batch completion
+                playlist_name = batch.get('playlist_name', 'Unknown Playlist')
+                failed_count = len(batch.get('permanently_failed_tracks', []))
+                successful_downloads = finished_count - failed_count
+                add_activity_item("", "Download Batch Complete", f"'{playlist_name}' - {successful_downloads} tracks downloaded", "Now")
+
+                # Emit batch_complete event for automation engine (only if something downloaded)
+                if successful_downloads > 0:
+                    try:
+                        if deps.automation_engine:
+                            deps.automation_engine.emit('batch_complete', {
+                                'playlist_name': playlist_name,
+                                'total_tracks': str(len(queue)),
+                                'completed_tracks': str(successful_downloads),
+                                'failed_tracks': str(failed_count),
+                            })
+                    except Exception:
+                        pass
+
+                # Update YouTube playlist phase to 'download_complete' if this is a YouTube playlist
+                playlist_id = batch.get('playlist_id')
+                if playlist_id and playlist_id.startswith('youtube_'):
+                    url_hash = playlist_id.replace('youtube_', '')
+                    if url_hash in deps.youtube_playlist_states:
+                        deps.youtube_playlist_states[url_hash]['phase'] = 'download_complete'
+                        logger.info(f"Updated YouTube playlist {url_hash} to download_complete phase")
+
+                # Update Tidal playlist phase to 'download_complete' if this is a Tidal playlist
+                if playlist_id and playlist_id.startswith('tidal_'):
+                    tidal_playlist_id = playlist_id.replace('tidal_', '')
+                    if tidal_playlist_id in deps.tidal_discovery_states:
+                        deps.tidal_discovery_states[tidal_playlist_id]['phase'] = 'download_complete'
+                        logger.info(f"Updated Tidal playlist {tidal_playlist_id} to download_complete phase")
+
+                # Update Deezer playlist phase to 'download_complete' if this is a Deezer playlist
+                if playlist_id and playlist_id.startswith('deezer_'):
+                    deezer_playlist_id = playlist_id.replace('deezer_', '')
+                    if deezer_playlist_id in deps.deezer_discovery_states:
+                        deps.deezer_discovery_states[deezer_playlist_id]['phase'] = 'download_complete'
+                        logger.info(f"Updated Deezer playlist {deezer_playlist_id} to download_complete phase")
+
+                # Update Spotify Public playlist phase to 'download_complete' if this is a Spotify Public playlist
+                if playlist_id and playlist_id.startswith('spotify_public_'):
+                    spotify_public_url_hash = playlist_id.replace('spotify_public_', '')
+                    if spotify_public_url_hash in deps.spotify_public_discovery_states:
+                        deps.spotify_public_discovery_states[spotify_public_url_hash]['phase'] = 'download_complete'
+                        logger.info(f"Updated Spotify Public playlist {spotify_public_url_hash} to download_complete phase")
+
+                logger.info(f"[Batch Manager] Batch {batch_id} complete - stopping monitor")
+                deps.download_monitor.stop_monitoring(batch_id)
+
+                # M3U REGENERATION: Regenerate M3U with real library paths now that
+                # all post-processing (tagging, moving, DB writes) is complete.
+                # The frontend M3U save may fire too early — this ensures paths resolve.
+                if deps.config_manager.get('m3u_export.enabled', False):
+                    try:
+                        m3u_tracks = []
+                        for tid in queue:
+                            if tid in download_tasks and download_tasks[tid].get('status') == 'completed':
+                                ti = download_tasks[tid].get('track_info', {})
+                                artists = ti.get('artists', [])
+                                artist_str = artists[0] if isinstance(artists, list) and artists else ''
+                                if isinstance(artist_str, dict):
+                                    artist_str = artist_str.get('name', '')
+                                m3u_tracks.append({
+                                    'name': ti.get('name', ''),
+                                    'artist': artist_str,
+                                    'duration_ms': ti.get('duration_ms', 0),
+                                })
+                        if m3u_tracks:
+                            deps.regenerate_batch_m3u(batch, m3u_tracks)
+                    except Exception as m3u_err:
+                        logger.error(f"[M3U] Error regenerating M3U on batch complete: {m3u_err}")
+
+                # REPAIR: Scan all album folders from this batch for track number issues
+                if deps.repair_worker:
+                    deps.repair_worker.process_batch(batch_id)
+
+                # ALBUM CONSISTENCY: Picard-style post-batch pass — pick ONE MusicBrainz
+                # release and overwrite album-level tags on all files to guarantee consistency.
+                # This is the safety net: even if per-track MB lookups drifted (different cache
+                # keys, API hiccups), this pass forces every file to share the same release MBID,
+                # album artist ID, release group ID, etc. — preventing Navidrome album splits.
+                _cons_files = batch.get('_consistency_files', [])
+                if batch.get('is_album_download') and _cons_files and len(_cons_files) >= 2:
+                    _cons_album = batch.get('album_context', {})
+                    _cons_artist = batch.get('artist_context', {})
+                    _cons_album_name = _cons_album.get('name', '') if isinstance(_cons_album, dict) else ''
+                    _cons_artist_name = _cons_artist.get('name', '') if isinstance(_cons_artist, dict) else ''
+                    if _cons_album_name and _cons_artist_name:
+                        try:
+                            _cons_mb_svc = deps.mb_worker.mb_service if deps.mb_worker else None
+                            if _cons_mb_svc and deps.config_manager.get('musicbrainz.embed_tags', True):
+                                from core.album_consistency import run_album_consistency
+                                from core.metadata.common import get_file_lock
+                                _cons_result = run_album_consistency(
+                                    file_infos=_cons_files,
+                                    album_name=_cons_album_name,
+                                    artist_name=_cons_artist_name,
+                                    mb_service=_cons_mb_svc,
+                                    total_discs=_cons_album.get('total_discs', 1),
+                                    file_lock_fn=get_file_lock,
+                                )
+                                if _cons_result.get('success'):
+                                    logger.info(f"[Album Consistency] {_cons_result['tags_written']}/{_cons_result['total_files']} files "
+                                          f"harmonized to release {_cons_result.get('release_mbid', '')[:8]}...")
+                                elif _cons_result.get('error'):
+                                    logger.error(f"[Album Consistency] Skipped: {_cons_result['error']}")
+                        except Exception as cons_err:
+                            logger.error(f"[Album Consistency] Failed (non-fatal): {cons_err}")
+
+                # Mark that wishlist processing is starting (prevents premature cleanup)
+                batch['wishlist_processing_started'] = True
+
+                # Process wishlist outside of the lock to prevent threading issues
+                if is_auto_batch:
+                    # For auto-initiated batches, handle completion and schedule next cycle
+                    deps.submit_failed_to_wishlist_with_auto_completion(batch_id)
+                else:
+                    # For manual batches, use standard wishlist processing
+                    deps.submit_failed_to_wishlist(batch_id)
+            else:
+                logger.warning(f"[Batch Manager] Batch {batch_id} already marked complete - skipping duplicate processing")
+
+            return  # Don't start next batch if we're done
+
+    # Start next downloads in queue
+    logger.info(f"[Batch Manager] Starting next batch for {batch_id}")
+    start_next_batch_of_downloads(batch_id, deps)
+
+
+# ---------------------------------------------------------------------------
+# check_batch_completion_v2
+# ---------------------------------------------------------------------------
+
+def check_batch_completion_v2(batch_id: str, deps: LifecycleDeps) -> Optional[bool]:
+    """V2 SYSTEM: Check if batch is complete after worker slot changes.
+
+    This is needed because V2 atomic cancel bypasses on_download_completed,
+    so we need to manually check for batch completion.
+    """
+    try:
+        with tasks_lock:
+            if batch_id not in download_batches:
+                logger.warning(f"[Completion Check V2] Batch {batch_id} not found")
+                return
+
+            batch = download_batches[batch_id]
+            all_tasks_started = batch['queue_index'] >= len(batch['queue'])
+            no_active_workers = batch['active_count'] == 0
+
+            # Count actually finished tasks (completed, failed, or cancelled)
+            finished_count = 0
+            retrying_count = 0
+            queue = batch.get('queue', [])
+            current_time = time.time()
+
+            for task_id in queue:
+                if task_id in download_tasks:
+                    task = download_tasks[task_id]
+                    task_status = task['status']
+
+                    # STUCK DETECTION: Force fail tasks that have been in transitional states too long
+                    if task_status == 'searching':
+                        task_age = current_time - task.get('status_change_time', current_time)
+                        if task_age > 600:  # 10 minutes
+                            logger.info(f"⏰ [Stuck Detection V2] Task {task_id} stuck in searching for {task_age:.0f}s - forcing not_found")
+                            task['status'] = 'not_found'
+                            task['error_message'] = f'Search stuck for {int(task_age // 60)} minutes with no results — timed out'
+                            finished_count += 1
+                        else:
+                            retrying_count += 1
+                    elif task_status == 'post_processing':
+                        task_age = current_time - task.get('status_change_time', current_time)
+                        if task_age > 300:  # 5 minutes (post-processing should be fast)
+                            logger.info(f"⏰ [Stuck Detection V2] Task {task_id} stuck in post_processing for {task_age:.0f}s - forcing completion")
+                            task['status'] = 'completed'  # Assume it worked if file verification is taking too long
+                            finished_count += 1
+                        else:
+                            retrying_count += 1
+                    elif task_status in ['completed', 'failed', 'cancelled', 'not_found']:
+                        finished_count += 1
+                else:
+                    # Task ID in queue but not in download_tasks - treat as completed to prevent blocking
+                    logger.warning(f"[Orphaned Task V2] Task {task_id} in queue but not in download_tasks - counting as finished")
+                    finished_count += 1
+
+            all_tasks_truly_finished = finished_count >= len(queue)
+            has_retrying_tasks = retrying_count > 0
+
+            logger.warning(f"[Completion Check V2] Batch {batch_id}: tasks_started={all_tasks_started}, workers={no_active_workers}, finished={finished_count}/{len(queue)}, retrying={retrying_count}")
+
+            is_auto_batch = False
+            if all_tasks_started and no_active_workers and all_tasks_truly_finished and not has_retrying_tasks:
+                # FIXED: Ensure batch is not already marked as complete to prevent duplicate processing
+                if batch.get('phase') != 'complete':
+                    logger.info(f"[Completion Check V2] Batch {batch_id} is complete - marking as finished")
+
+                    # Check if this is an auto-initiated batch
+                    is_auto_batch = batch.get('auto_initiated', False)
+
+                    # Mark batch as complete and set completion timestamp for auto-cleanup
+                    batch['phase'] = 'complete'
+                    batch['completion_time'] = time.time()  # Track when batch completed
+
+                    # Add activity for batch completion
+                    playlist_name = batch.get('playlist_name', 'Unknown Playlist')
+                    failed_count = len(batch.get('permanently_failed_tracks', []))
+                    successful_downloads = finished_count - failed_count
+                    add_activity_item("", "Download Batch Complete", f"'{playlist_name}' - {successful_downloads} tracks downloaded", "Now")
+
+                    # Emit batch_complete event for automation engine (only if something downloaded)
+                    if successful_downloads > 0:
+                        try:
+                            if deps.automation_engine:
+                                deps.automation_engine.emit('batch_complete', {
+                                    'playlist_name': playlist_name,
+                                    'total_tracks': str(len(queue)),
+                                    'completed_tracks': str(successful_downloads),
+                                    'failed_tracks': str(failed_count),
+                                })
+                        except Exception:
+                            pass
+                else:
+                    logger.warning(f"[Completion Check V2] Batch {batch_id} already marked complete - skipping duplicate processing")
+                    return True  # Already complete
+
+                # Update YouTube playlist phase to 'download_complete' if this is a YouTube playlist
+                playlist_id = batch.get('playlist_id')
+                if playlist_id and playlist_id.startswith('youtube_'):
+                    url_hash = playlist_id.replace('youtube_', '')
+                    if url_hash in deps.youtube_playlist_states:
+                        deps.youtube_playlist_states[url_hash]['phase'] = 'download_complete'
+                        logger.info(f"[Completion Check V2] Updated YouTube playlist {url_hash} to download_complete phase")
+
+                # Update Tidal playlist phase to 'download_complete' if this is a Tidal playlist
+                if playlist_id and playlist_id.startswith('tidal_'):
+                    tidal_playlist_id = playlist_id.replace('tidal_', '')
+                    if tidal_playlist_id in deps.tidal_discovery_states:
+                        deps.tidal_discovery_states[tidal_playlist_id]['phase'] = 'download_complete'
+                        logger.info(f"[Completion Check V2] Updated Tidal playlist {tidal_playlist_id} to download_complete phase")
+
+                # Update Deezer playlist phase to 'download_complete' if this is a Deezer playlist
+                if playlist_id and playlist_id.startswith('deezer_'):
+                    deezer_playlist_id = playlist_id.replace('deezer_', '')
+                    if deezer_playlist_id in deps.deezer_discovery_states:
+                        deps.deezer_discovery_states[deezer_playlist_id]['phase'] = 'download_complete'
+                        logger.info(f"[Completion Check V2] Updated Deezer playlist {deezer_playlist_id} to download_complete phase")
+
+                # Update Spotify Public playlist phase to 'download_complete' if this is a Spotify Public playlist
+                if playlist_id and playlist_id.startswith('spotify_public_'):
+                    spotify_public_url_hash = playlist_id.replace('spotify_public_', '')
+                    if spotify_public_url_hash in deps.spotify_public_discovery_states:
+                        deps.spotify_public_discovery_states[spotify_public_url_hash]['phase'] = 'download_complete'
+                        logger.info(f"[Completion Check V2] Updated Spotify Public playlist {spotify_public_url_hash} to download_complete phase")
+
+                logger.info(f"[Completion Check V2] Batch {batch_id} complete - stopping monitor")
+                deps.download_monitor.stop_monitoring(batch_id)
+
+                # REPAIR: Scan all album folders from this batch for track number issues
+                if deps.repair_worker:
+                    deps.repair_worker.process_batch(batch_id)
+
+                # ALBUM CONSISTENCY: Same Picard-style pass as the primary completion path
+                _cons_files = batch.get('_consistency_files', [])
+                if batch.get('is_album_download') and _cons_files and len(_cons_files) >= 2:
+                    _cons_album = batch.get('album_context', {})
+                    _cons_artist = batch.get('artist_context', {})
+                    _cons_album_name = _cons_album.get('name', '') if isinstance(_cons_album, dict) else ''
+                    _cons_artist_name = _cons_artist.get('name', '') if isinstance(_cons_artist, dict) else ''
+                    if _cons_album_name and _cons_artist_name:
+                        try:
+                            _cons_mb_svc = deps.mb_worker.mb_service if deps.mb_worker else None
+                            if _cons_mb_svc and deps.config_manager.get('musicbrainz.embed_tags', True):
+                                from core.album_consistency import run_album_consistency
+                                from core.metadata.common import get_file_lock
+                                _cons_result = run_album_consistency(
+                                    file_infos=_cons_files,
+                                    album_name=_cons_album_name,
+                                    artist_name=_cons_artist_name,
+                                    mb_service=_cons_mb_svc,
+                                    total_discs=_cons_album.get('total_discs', 1),
+                                    file_lock_fn=get_file_lock,
+                                )
+                                if _cons_result.get('success'):
+                                    logger.info(f"[Album Consistency V2] {_cons_result['tags_written']}/{_cons_result['total_files']} files "
+                                          f"harmonized to release {_cons_result.get('release_mbid', '')[:8]}...")
+                                elif _cons_result.get('error'):
+                                    logger.error(f"[Album Consistency V2] Skipped: {_cons_result['error']}")
+                        except Exception as cons_err:
+                            logger.error(f"[Album Consistency V2] Failed (non-fatal): {cons_err}")
+
+        # Process wishlist outside of the lock to prevent threading issues
+        if all_tasks_started and no_active_workers and all_tasks_truly_finished and not has_retrying_tasks:
+            # Call wishlist processing outside the lock — DIRECT (synchronous) call
+            # to match original v2 behavior. The non-v2 path (on_download_completed)
+            # uses the async submit_* deps; v2 calls directly because v2 itself runs
+            # from a context where blocking is acceptable.
+            if is_auto_batch:
+                logger.info("[Completion Check V2] Processing auto-initiated batch completion")
+                deps.process_failed_to_wishlist_with_auto_completion(batch_id)
+            else:
+                logger.info("[Completion Check V2] Processing regular batch completion")
+                deps.process_failed_to_wishlist(batch_id)
+
+            return True  # Batch was completed
+        else:
+            logger.warning(f"[Completion Check V2] Batch {batch_id} not yet complete: finished={finished_count}/{len(queue)}, retrying={retrying_count}, workers={batch['active_count']}")
+            return False  # Batch still in progress
+
+    except Exception as e:
+        logger.error(f"[Completion Check V2] Error checking batch completion: {e}")
+        traceback.print_exc()
+        return False
diff --git a/tests/downloads/test_downloads_lifecycle.py b/tests/downloads/test_downloads_lifecycle.py
new file mode 100644
index 00000000..8f82cb72
--- /dev/null
+++ b/tests/downloads/test_downloads_lifecycle.py
@@ -0,0 +1,509 @@
+"""Tests for core/downloads/lifecycle.py — batch lifecycle (start, complete, check)."""
+
+from __future__ import annotations
+
+import threading
+
+import pytest
+
+from core.downloads import lifecycle as lc
+from core.runtime_state import (
+    download_batches,
+    download_tasks,
+)
+
+
+@pytest.fixture(autouse=True)
+def reset_state():
+    download_tasks.clear()
+    download_batches.clear()
+    yield
+    download_tasks.clear()
+    download_batches.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 _FakeAutoEngine:
+    def __init__(self):
+        self.events = []
+
+    def emit(self, event_type, data):
+        self.events.append((event_type, data))
+
+
+class _FakeMonitor:
+    def __init__(self):
+        self.stopped = []
+
+    def stop_monitoring(self, batch_id):
+        self.stopped.append(batch_id)
+
+
+class _FakeRepair:
+    def __init__(self):
+        self.batches = []
+
+    def process_batch(self, batch_id):
+        self.batches.append(batch_id)
+
+
+class _FakeConfig:
+    def __init__(self, values=None):
+        self._v = values or {}
+
+    def get(self, key, default=None):
+        return self._v.get(key, default)
+
+
+def _build_deps(
+    *,
+    automation=None,
+    monitor=None,
+    repair=None,
+    config=None,
+    is_shutting_down=lambda: False,
+    submit_dl_worker=None,
+    submit_failed=None,
+    submit_failed_auto=None,
+    process_failed=None,
+    process_failed_auto=None,
+    yt_states=None,
+    tidal_states=None,
+    deezer_states=None,
+    spotify_states=None,
+):
+    rec = _Recorder()
+    return lc.LifecycleDeps(
+        config_manager=config or _FakeConfig(),
+        automation_engine=automation,
+        download_monitor=monitor or _FakeMonitor(),
+        repair_worker=repair,
+        mb_worker=None,
+        is_shutting_down=is_shutting_down,
+        get_batch_lock=lambda bid: threading.Lock(),
+        submit_download_track_worker=submit_dl_worker or rec('submit_dl'),
+        submit_failed_to_wishlist=submit_failed or rec('submit_failed'),
+        submit_failed_to_wishlist_with_auto_completion=submit_failed_auto or rec('submit_failed_auto'),
+        process_failed_to_wishlist=process_failed or rec('process_failed'),
+        process_failed_to_wishlist_with_auto_completion=process_failed_auto or rec('process_failed_auto'),
+        ensure_spotify_track_format=lambda track: track,
+        get_track_artist_name=lambda track: 'Artist',
+        check_and_remove_from_wishlist=rec('check_wishlist'),
+        regenerate_batch_m3u=rec('regen_m3u'),
+        youtube_playlist_states=yt_states or {},
+        tidal_discovery_states=tidal_states or {},
+        deezer_discovery_states=deezer_states or {},
+        spotify_public_discovery_states=spotify_states or {},
+    ), rec
+
+
+# ---------------------------------------------------------------------------
+# start_next_batch_of_downloads
+# ---------------------------------------------------------------------------
+
+def test_start_next_returns_silently_for_missing_batch():
+    deps, rec = _build_deps()
+    lc.start_next_batch_of_downloads('absent', deps)
+    assert rec.calls == []
+
+
+def test_start_next_skipped_when_shutting_down():
+    download_batches['b1'] = {'queue': ['t1'], 'queue_index': 0, 'active_count': 0, 'max_concurrent': 1}
+    deps, rec = _build_deps(is_shutting_down=lambda: True)
+    lc.start_next_batch_of_downloads('b1', deps)
+    assert rec.calls == []  # no submit
+
+
+def test_start_next_submits_up_to_max_concurrent():
+    download_tasks['t1'] = {'status': 'queued'}
+    download_tasks['t2'] = {'status': 'queued'}
+    download_tasks['t3'] = {'status': 'queued'}
+    download_batches['b1'] = {
+        'queue': ['t1', 't2', 't3'], 'queue_index': 0,
+        'active_count': 0, 'max_concurrent': 2,
+    }
+    deps, rec = _build_deps()
+    lc.start_next_batch_of_downloads('b1', deps)
+    submits = [c for c in rec.calls if c[0] == 'submit_dl']
+    assert len(submits) == 2
+    assert download_batches['b1']['active_count'] == 2
+    assert download_batches['b1']['queue_index'] == 2
+
+
+def test_start_next_skips_cancelled_tasks_without_consuming_slots():
+    download_tasks['t1'] = {'status': 'cancelled'}
+    download_tasks['t2'] = {'status': 'queued'}
+    download_batches['b1'] = {
+        'queue': ['t1', 't2'], 'queue_index': 0,
+        'active_count': 0, 'max_concurrent': 1,
+    }
+    deps, rec = _build_deps()
+    lc.start_next_batch_of_downloads('b1', deps)
+    submits = [c for c in rec.calls if c[0] == 'submit_dl']
+    assert len(submits) == 1
+    # t2 should be the one submitted (t1 skipped)
+    assert submits[0][1] == ('t2', 'b1')
+
+
+def test_start_next_sets_searching_status_before_submit():
+    download_tasks['t1'] = {'status': 'queued'}
+    download_batches['b1'] = {
+        'queue': ['t1'], 'queue_index': 0,
+        'active_count': 0, 'max_concurrent': 1,
+    }
+    deps, _ = _build_deps()
+    lc.start_next_batch_of_downloads('b1', deps)
+    assert download_tasks['t1']['status'] == 'searching'
+    assert download_tasks['t1']['status_change_time'] is not None
+
+
+def test_start_next_submit_failure_marks_task_failed_no_ghost_worker():
+    download_tasks['t1'] = {'status': 'queued'}
+    download_batches['b1'] = {
+        'queue': ['t1'], 'queue_index': 0,
+        'active_count': 0, 'max_concurrent': 1,
+    }
+
+    def _broken_submit(task_id, batch_id):
+        raise RuntimeError("executor dead")
+
+    deps, _ = _build_deps(submit_dl_worker=_broken_submit)
+    lc.start_next_batch_of_downloads('b1', deps)
+    # No counters incremented
+    assert download_batches['b1']['active_count'] == 0
+    assert download_batches['b1']['queue_index'] == 0
+    # Task marked failed
+    assert download_tasks['t1']['status'] == 'failed'
+
+
+def test_start_next_orphan_task_in_queue_skipped():
+    download_batches['b1'] = {
+        'queue': ['absent', 't2'], 'queue_index': 0,
+        'active_count': 0, 'max_concurrent': 2,
+    }
+    download_tasks['t2'] = {'status': 'queued'}
+    deps, rec = _build_deps()
+    lc.start_next_batch_of_downloads('b1', deps)
+    submits = [c for c in rec.calls if c[0] == 'submit_dl']
+    # Only t2 submitted
+    assert len(submits) == 1
+    assert submits[0][1] == ('t2', 'b1')
+
+
+# ---------------------------------------------------------------------------
+# on_download_completed
+# ---------------------------------------------------------------------------
+
+def test_on_complete_missing_batch_returns_silently():
+    deps, rec = _build_deps()
+    lc.on_download_completed('absent', 't1', True, deps)
+    assert rec.calls == []
+
+
+def test_on_complete_decrements_active_count():
+    download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X'}}
+    download_batches['b1'] = {
+        'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
+        'max_concurrent': 1, 'permanently_failed_tracks': [],
+        'cancelled_tracks': set(),
+    }
+    deps, _ = _build_deps()
+    lc.on_download_completed('b1', 't1', True, deps)
+    assert download_batches['b1']['active_count'] == 0
+
+
+def test_on_complete_duplicate_call_skips_decrement_but_checks_completion():
+    download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X'}}
+    download_batches['b1'] = {
+        'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
+        'max_concurrent': 1, 'permanently_failed_tracks': [],
+        'cancelled_tracks': set(),
+    }
+    deps, _ = _build_deps()
+    # First call: decrements
+    lc.on_download_completed('b1', 't1', True, deps)
+    assert download_batches['b1']['active_count'] == 0
+    # Mark as complete to prevent batch completion path
+    download_batches['b1']['phase'] = 'complete'
+    # Second call: should NOT decrement again (would go negative)
+    lc.on_download_completed('b1', 't1', True, deps)
+    assert download_batches['b1']['active_count'] == 0  # still 0, not -1
+
+
+def test_on_complete_failed_task_appended_to_permanently_failed_tracks():
+    download_tasks['t1'] = {
+        'status': 'failed', 'track_info': {'name': 'Money'},
+        'track_index': 0, 'retry_count': 0,
+    }
+    download_batches['b1'] = {
+        'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
+        'max_concurrent': 1, 'permanently_failed_tracks': [],
+        'cancelled_tracks': set(),
+    }
+    deps, _ = _build_deps()
+    lc.on_download_completed('b1', 't1', False, deps)
+    assert len(download_batches['b1']['permanently_failed_tracks']) == 1
+    assert download_batches['b1']['permanently_failed_tracks'][0]['track_name'] == 'Money'
+
+
+def test_on_complete_cancelled_task_added_to_cancelled_tracks():
+    download_tasks['t1'] = {
+        'status': 'cancelled', 'track_info': {'name': 'X'},
+        'track_index': 5,
+    }
+    download_batches['b1'] = {
+        'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
+        'max_concurrent': 1, 'permanently_failed_tracks': [],
+        'cancelled_tracks': set(),
+    }
+    deps, _ = _build_deps()
+    lc.on_download_completed('b1', 't1', False, deps)
+    assert 5 in download_batches['b1']['cancelled_tracks']
+
+
+def test_on_complete_emits_download_failed_for_not_found():
+    download_tasks['t1'] = {
+        'status': 'not_found', 'track_info': {'name': 'X'},
+        'track_index': 0, 'retry_count': 0,
+    }
+    download_batches['b1'] = {
+        'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
+        'max_concurrent': 1, 'permanently_failed_tracks': [],
+        'cancelled_tracks': set(),
+    }
+    auto = _FakeAutoEngine()
+    deps, _ = _build_deps(automation=auto)
+    lc.on_download_completed('b1', 't1', False, deps)
+    events = [e for e in auto.events if e[0] == 'download_failed']
+    assert len(events) == 1
+
+
+def test_on_complete_success_calls_check_and_remove_wishlist():
+    download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X', 'artists': ['A']}}
+    download_batches['b1'] = {
+        'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
+        'max_concurrent': 1, 'permanently_failed_tracks': [],
+        'cancelled_tracks': set(),
+    }
+    deps, rec = _build_deps()
+    lc.on_download_completed('b1', 't1', True, deps)
+    assert any(c[0] == 'check_wishlist' for c in rec.calls)
+
+
+# ---------------------------------------------------------------------------
+# Batch completion (via on_download_completed)
+# ---------------------------------------------------------------------------
+
+def test_batch_completion_emits_batch_complete_when_all_done():
+    download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X'}}
+    download_tasks['t2'] = {'status': 'completed', 'track_info': {'name': 'Y'}}
+    download_batches['b1'] = {
+        'queue': ['t1', 't2'], 'queue_index': 2, 'active_count': 1,
+        'max_concurrent': 2, 'permanently_failed_tracks': [],
+        'cancelled_tracks': set(),
+        'playlist_name': 'PL',
+    }
+    auto = _FakeAutoEngine()
+    monitor = _FakeMonitor()
+    deps, _ = _build_deps(automation=auto, monitor=monitor)
+    # Final task completes
+    lc.on_download_completed('b1', 't2', True, deps)
+    assert download_batches['b1']['phase'] == 'complete'
+    assert ('batch_complete', {'playlist_name': 'PL', 'total_tracks': '2', 'completed_tracks': '2', 'failed_tracks': '0'}) in auto.events
+    assert 'b1' in monitor.stopped
+
+
+def test_batch_completion_skips_emit_when_zero_successful():
+    """Don't emit batch_complete if nothing actually downloaded."""
+    download_tasks['t1'] = {'status': 'failed', 'track_info': {'name': 'X'}, 'track_index': 0}
+    download_batches['b1'] = {
+        'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
+        'max_concurrent': 1,
+        'permanently_failed_tracks': [{'track_name': 'X'}],  # 1 failed
+        'cancelled_tracks': set(),
+        'playlist_name': 'PL',
+    }
+    auto = _FakeAutoEngine()
+    deps, _ = _build_deps(automation=auto)
+    lc.on_download_completed('b1', 't1', False, deps)
+    # Pre-existing failed already counted, so successful = 1 (count) - 1 (already failed) = 0
+    # but on_download_completed appends another failure for t1, so failed count = 2 > finished = 1
+    # the emit is only triggered if successful_downloads > 0
+    events = [e for e in auto.events if e[0] == 'batch_complete']
+    # successful = finished_count(1) - failed_count(2) = -1, which is not > 0 → no emit
+    assert events == []
+
+
+def test_batch_completion_routes_auto_batch_to_auto_completion_handler():
+    download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X'}}
+    download_batches['b1'] = {
+        'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
+        'max_concurrent': 1,
+        'permanently_failed_tracks': [], 'cancelled_tracks': set(),
+        'auto_initiated': True,
+        'playlist_name': 'PL',
+    }
+    deps, rec = _build_deps()
+    lc.on_download_completed('b1', 't1', True, deps)
+    assert any(c[0] == 'submit_failed_auto' and c[1] == ('b1',) for c in rec.calls)
+
+
+def test_batch_completion_routes_manual_batch_to_regular_handler():
+    download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X'}}
+    download_batches['b1'] = {
+        'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
+        'max_concurrent': 1,
+        'permanently_failed_tracks': [], 'cancelled_tracks': set(),
+        # no auto_initiated → manual
+    }
+    deps, rec = _build_deps()
+    lc.on_download_completed('b1', 't1', True, deps)
+    assert any(c[0] == 'submit_failed' and c[1] == ('b1',) for c in rec.calls)
+
+
+def test_batch_completion_does_not_complete_when_tasks_still_searching():
+    download_tasks['t1'] = {'status': 'searching', 'track_info': {'name': 'X'},
+                             'status_change_time': lc.time.time()}  # fresh
+    download_tasks['t2'] = {'status': 'completed', 'track_info': {'name': 'Y'}}
+    download_batches['b1'] = {
+        'queue': ['t1', 't2'], 'queue_index': 2, 'active_count': 1,
+        'max_concurrent': 2, 'permanently_failed_tracks': [],
+        'cancelled_tracks': set(),
+    }
+    auto = _FakeAutoEngine()
+    deps, _ = _build_deps(automation=auto)
+    lc.on_download_completed('b1', 't2', True, deps)
+    # Batch NOT marked complete (t1 still searching)
+    assert download_batches['b1'].get('phase') != 'complete'
+
+
+def test_stuck_searching_task_forced_to_not_found():
+    """Task searching > 10min gets forced to not_found."""
+    download_tasks['t1'] = {
+        'status': 'searching', 'track_info': {'name': 'X'},
+        'status_change_time': 0,  # very ancient
+    }
+    download_batches['b1'] = {
+        'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
+        'max_concurrent': 1, 'permanently_failed_tracks': [],
+        'cancelled_tracks': set(),
+    }
+    deps, _ = _build_deps()
+    # Trigger completion check
+    lc.on_download_completed('b1', 't1', False, deps)
+    # t1 forced to not_found
+    assert download_tasks['t1']['status'] == 'not_found'
+
+
+def test_stuck_post_processing_task_forced_to_completed():
+    """Task post_processing > 5min gets forced to completed."""
+    download_tasks['t1'] = {
+        'status': 'post_processing', 'track_info': {'name': 'X'},
+        'status_change_time': 0,
+    }
+    download_batches['b1'] = {
+        'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
+        'max_concurrent': 1, 'permanently_failed_tracks': [],
+        'cancelled_tracks': set(),
+    }
+    deps, _ = _build_deps()
+    lc.on_download_completed('b1', 't1', True, deps)
+    assert download_tasks['t1']['status'] == 'completed'
+
+
+def test_youtube_playlist_phase_updated_on_completion():
+    download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X'}}
+    download_batches['b1'] = {
+        'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
+        'max_concurrent': 1, 'permanently_failed_tracks': [],
+        'cancelled_tracks': set(),
+        'playlist_id': 'youtube_abc123',
+    }
+    yt = {'abc123': {'phase': 'downloading'}}
+    deps, _ = _build_deps(yt_states=yt)
+    lc.on_download_completed('b1', 't1', True, deps)
+    assert yt['abc123']['phase'] == 'download_complete'
+
+
+# ---------------------------------------------------------------------------
+# check_batch_completion_v2
+# ---------------------------------------------------------------------------
+
+def test_check_v2_returns_none_for_missing_batch():
+    deps, _ = _build_deps()
+    result = lc.check_batch_completion_v2('absent', deps)
+    assert result is None
+
+
+def test_check_v2_returns_false_when_not_complete():
+    download_tasks['t1'] = {'status': 'downloading', 'track_info': {}}
+    download_batches['b1'] = {
+        'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
+        'max_concurrent': 1, 'permanently_failed_tracks': [],
+    }
+    deps, _ = _build_deps()
+    result = lc.check_batch_completion_v2('b1', deps)
+    assert result is False
+
+
+def test_check_v2_returns_true_when_complete():
+    download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X'}}
+    download_batches['b1'] = {
+        'queue': ['t1'], 'queue_index': 1, 'active_count': 0,
+        'max_concurrent': 1, 'permanently_failed_tracks': [],
+    }
+    deps, _ = _build_deps()
+    result = lc.check_batch_completion_v2('b1', deps)
+    assert result is True
+    assert download_batches['b1']['phase'] == 'complete'
+
+
+def test_check_v2_already_complete_returns_true_without_reprocessing():
+    download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X'}}
+    download_batches['b1'] = {
+        'queue': ['t1'], 'queue_index': 1, 'active_count': 0,
+        'max_concurrent': 1, 'permanently_failed_tracks': [],
+        'phase': 'complete',  # already marked
+    }
+    deps, rec = _build_deps()
+    result = lc.check_batch_completion_v2('b1', deps)
+    assert result is True
+    # Wishlist NOT submitted again
+    assert not any(c[0] in ('process_failed', 'process_failed_auto') for c in rec.calls)
+
+
+def test_check_v2_routes_auto_batch_to_auto_handler():
+    """v2 calls wishlist processing DIRECTLY (sync), not via executor submit.
+    Different from on_download_completed which uses async submit."""
+    download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X'}}
+    download_batches['b1'] = {
+        'queue': ['t1'], 'queue_index': 1, 'active_count': 0,
+        'max_concurrent': 1, 'permanently_failed_tracks': [],
+        'auto_initiated': True,
+    }
+    deps, rec = _build_deps()
+    lc.check_batch_completion_v2('b1', deps)
+    assert any(c[0] == 'process_failed_auto' for c in rec.calls)
+
+
+def test_check_v2_exception_returns_false():
+    download_batches['b1'] = {'queue': ['t1'], 'queue_index': 1, 'active_count': 0}
+    # Force an exception by putting invalid type in batch
+    download_batches['b1']['queue'] = None  # will raise TypeError on len()
+    deps, _ = _build_deps()
+    result = lc.check_batch_completion_v2('b1', deps)
+    assert result is False
diff --git a/web_server.py b/web_server.py
index 64d447a5..f81dfac8 100644
--- a/web_server.py
+++ b/web_server.py
@@ -21129,83 +21129,47 @@ def _get_batch_lock(batch_id):
             batch_locks[batch_id] = threading.Lock()
         return batch_locks[batch_id]
 
-def _start_next_batch_of_downloads(batch_id):
-    """Start the next batch of downloads up to the concurrent limit (like GUI)"""
-    # ENHANCED: Use batch-specific lock to prevent race conditions when multiple threads
-    # try to start workers for the same batch concurrently
-    batch_lock = _get_batch_lock(batch_id)
-    
-    with batch_lock:
-        # Prevent starting new tasks if shutting down
-        if IS_SHUTTING_DOWN:
-            logger.info(f"[Batch Manager] Server shutting down - skipping new tasks for batch {batch_id}")
-            return
+# Batch lifecycle logic lives in core/downloads/lifecycle.py.
+from core.downloads import lifecycle as _downloads_lifecycle
+
+
+def _build_lifecycle_deps():
+    """Build LifecycleDeps bundle from web_server.py globals on each call."""
+    return _downloads_lifecycle.LifecycleDeps(
+        config_manager=config_manager,
+        automation_engine=automation_engine,
+        download_monitor=download_monitor,
+        repair_worker=repair_worker,
+        mb_worker=mb_worker,
+        is_shutting_down=lambda: IS_SHUTTING_DOWN,
+        get_batch_lock=_get_batch_lock,
+        submit_download_track_worker=lambda task_id, batch_id: missing_download_executor.submit(
+            _download_track_worker, task_id, batch_id,
+        ),
+        submit_failed_to_wishlist=lambda batch_id: missing_download_executor.submit(
+            _process_failed_tracks_to_wishlist_exact, batch_id,
+        ),
+        submit_failed_to_wishlist_with_auto_completion=lambda batch_id: missing_download_executor.submit(
+            _process_failed_tracks_to_wishlist_exact_with_auto_completion, batch_id,
+        ),
+        process_failed_to_wishlist=_process_failed_tracks_to_wishlist_exact,
+        process_failed_to_wishlist_with_auto_completion=_process_failed_tracks_to_wishlist_exact_with_auto_completion,
+        ensure_spotify_track_format=_ensure_spotify_track_format,
+        get_track_artist_name=_get_track_artist_name,
+        check_and_remove_from_wishlist=_check_and_remove_from_wishlist,
+        regenerate_batch_m3u=_regenerate_batch_m3u,
+        youtube_playlist_states=youtube_playlist_states,
+        tidal_discovery_states=tidal_discovery_states,
+        deezer_discovery_states=deezer_discovery_states,
+        spotify_public_discovery_states=spotify_public_discovery_states,
+    )
+
+
+def _start_next_batch_of_downloads(batch_id):
+    """Start the next batch of downloads up to the concurrent limit."""
+    _downloads_lifecycle.start_next_batch_of_downloads(batch_id, _build_lifecycle_deps())
+
 
-        with tasks_lock:
-            if batch_id not in download_batches:
-                return
-                
-            batch = download_batches[batch_id]
-            max_concurrent = batch['max_concurrent']
-            queue = batch['queue']
-            queue_index = batch['queue_index']
-            active_count = batch['active_count']
-            
-            logger.info(f"[Batch Lock] Starting workers for {batch_id}: active={active_count}, max={max_concurrent}, queue_pos={queue_index}/{len(queue)}")
-            
-            # Start downloads up to the concurrent limit
-            while active_count < max_concurrent and queue_index < len(queue):
-                task_id = queue[queue_index]
-                
-                # CRITICAL V2 FIX: Skip cancelled tasks instead of trying to restart them
-                if task_id in download_tasks:
-                    current_status = download_tasks[task_id]['status']
-                    if current_status == 'cancelled':
-                        logger.warning(f"[Batch Lock] Skipping cancelled task {task_id} (queue position {queue_index + 1})")
-                        download_batches[batch_id]['queue_index'] += 1
-                        queue_index += 1
-                        continue  # Skip to next task without consuming worker slot
-                    
-                    # IMPORTANT: Set status to 'searching' BEFORE starting worker (like GUI)
-                    # Must be done INSIDE the lock to prevent race conditions with status polling
-                    download_tasks[task_id]['status'] = 'searching'
-                    download_tasks[task_id]['status_change_time'] = time.time()
-                    logger.info(f"[Batch Manager] Set task {task_id} status to 'searching'")
-                else:
-                    logger.warning(f"[Batch Lock] Task {task_id} not found in download_tasks - skipping")
-                    download_batches[batch_id]['queue_index'] += 1
-                    queue_index += 1
-                    continue
-                
-                # CRITICAL FIX: Submit to executor BEFORE incrementing counters to prevent ghost workers
-                try:
-                    # Submit to executor first - this can fail
-                    future = missing_download_executor.submit(_download_track_worker, task_id, batch_id)
-                    
-                    # Only increment counters AFTER successful submit
-                    download_batches[batch_id]['active_count'] += 1
-                    download_batches[batch_id]['queue_index'] += 1
-                    
-                    logger.info(f"[Batch Lock] Started download {queue_index + 1}/{len(queue)} - Active: {active_count + 1}/{max_concurrent}")
-                    
-                    # Update local counters for next iteration
-                    active_count += 1
-                    queue_index += 1
-                    
-                except Exception as submit_error:
-                    logger.error(f"[Batch Lock] CRITICAL: Failed to submit task {task_id} to executor: {submit_error}")
-                    logger.info("[Batch Lock] Worker slot NOT consumed - preventing ghost worker")
-                    
-                    # Reset task status since worker never started
-                    if task_id in download_tasks:
-                        download_tasks[task_id]['status'] = 'failed'
-                        logger.error(f"[Batch Lock] Set task {task_id} status to 'failed' due to submit failure")
-                    
-                    # Don't increment counters - no worker was actually started
-                    # This prevents the "ghost worker" issue where active_count is incremented but no actual worker runs
-                    break  # Stop trying to start more workers if executor is failing
-            
-            logger.info(f"[Batch Lock] Finished starting workers for {batch_id}: final_active={download_batches[batch_id]['active_count']}, max={max_concurrent}")
 
 def _get_track_artist_name(track_info):
     """Extract artist name from track info, handling different data formats (replicating sync.py)"""
@@ -21613,315 +21577,10 @@ def _process_failed_tracks_to_wishlist_exact_with_auto_completion(batch_id):
         return {'tracks_added': 0, 'errors': 1, 'total_failed': 0}
 
 def _on_download_completed(batch_id, task_id, success=True):
-    """Called when a download completes to start the next one in queue"""
-    with tasks_lock:
-        if batch_id not in download_batches:
-            logger.warning(f"[Batch Manager] Batch {batch_id} not found for completed task {task_id}")
-            return
+    """Called when a download completes to start the next one in queue."""
+    _downloads_lifecycle.on_download_completed(batch_id, task_id, success, _build_lifecycle_deps())
 
-        # Guard against double-calling: track which tasks have already been completed
-        # This prevents active_count from being decremented multiple times for the same task
-        # (e.g. monitor detects completion AND post-processing calls this again)
-        # NOTE: On duplicate calls, we skip decrement/tracking but STILL check batch completion.
-        # This is critical because the first call may see the task in 'post_processing' (not finished),
-        # and the second call (from post-processing worker) arrives after the task is truly 'completed'.
-        # Without the fallthrough, batch_complete would never be emitted.
-        completed_tasks = download_batches[batch_id].setdefault('_completed_task_ids', set())
-        _is_duplicate_completion = task_id in completed_tasks
-        if _is_duplicate_completion:
-            logger.info(f"[Batch Manager] Task {task_id} already completed — skipping decrement, still checking batch completion")
-            # Set terminal status so the monitor loop stops re-processing this task
-            if task_id in download_tasks and download_tasks[task_id].get('status') in ('downloading', 'queued'):
-                download_tasks[task_id]['status'] = 'completed'
-            # Fall through to batch completion check below (don't return)
-        else:
-            completed_tasks.add(task_id)
 
-        if not _is_duplicate_completion:
-            # Track failed/cancelled tasks in batch state (replicating sync.py)
-            if not success and task_id in download_tasks:
-                task = download_tasks[task_id]
-                task_status = task.get('status', 'unknown')
-
-                # Build track_info structure matching sync.py's permanently_failed_tracks format
-                original_track_info = task.get('track_info', {})
-
-                # Ensure spotify_track has proper structure for wishlist service
-                spotify_track_data = _ensure_spotify_track_format(original_track_info)
-
-                track_info = {
-                    'download_index': task.get('track_index', 0),
-                    'table_index': task.get('track_index', 0),
-                    'track_name': original_track_info.get('name', 'Unknown Track'),
-                    'artist_name': _get_track_artist_name(original_track_info),
-                    'retry_count': task.get('retry_count', 0),
-                    'spotify_track': spotify_track_data,  # Properly formatted spotify track for wishlist
-                    'failure_reason': 'Download cancelled' if task_status == 'cancelled' else ('No matching track found' if task_status == 'not_found' else 'Download failed'),
-                    'candidates': task.get('cached_candidates', [])  # Include search results if available
-                }
-
-                if task_status == 'cancelled':
-                    download_batches[batch_id]['cancelled_tracks'].add(task.get('track_index', 0))
-                    logger.warning(f"[Batch Manager] Added cancelled track to batch tracking: {track_info['track_name']}")
-                    add_activity_item("", "Download Cancelled", f"'{track_info['track_name']}'", "Now")
-                elif task_status in ('failed', 'not_found'):
-                    download_batches[batch_id]['permanently_failed_tracks'].append(track_info)
-                    if task_status == 'not_found':
-                        logger.info(f"[Batch Manager] Added not-found track to batch tracking: {track_info['track_name']}")
-                        add_activity_item("", "Not Found", f"'{track_info['track_name']}'", "Now")
-                    else:
-                        logger.error(f"[Batch Manager] Added failed track to batch tracking: {track_info['track_name']}")
-                        add_activity_item("", "Download Failed", f"'{track_info['track_name']}'", "Now")
-
-                    try:
-                        if automation_engine:
-                            automation_engine.emit('download_failed', {
-                                'artist': track_info.get('artist_name', ''),
-                                'title': track_info.get('track_name', ''),
-                                'reason': track_info.get('failure_reason', 'Unknown'),
-                            })
-                    except Exception:
-                        pass
-
-            # WISHLIST REMOVAL: Handle successful downloads for wishlist removal
-            if success and task_id in download_tasks:
-                try:
-                    task = download_tasks[task_id]
-                    track_info = task.get('track_info', {})
-                    logger.info(f"[Batch Manager] Successful download - checking wishlist removal for task {task_id}")
-
-                    # Add activity for successful download
-                    track_name = track_info.get('name', 'Unknown Track')
-
-                    # Safely extract artist name (handle both list and string formats)
-                    artists = track_info.get('artists', [])
-                    if isinstance(artists, list) and len(artists) > 0:
-                        first_artist = artists[0]
-                        artist_name = first_artist.get('name', 'Unknown Artist') if isinstance(first_artist, dict) else str(first_artist)
-                    elif isinstance(artists, str):
-                        artist_name = artists
-                    else:
-                        artist_name = 'Unknown Artist'
-
-                    add_activity_item("", "Download Complete", f"'{track_name}' by {artist_name}", "Now")
-
-                    # Try to remove from wishlist using track info
-                    if track_info:
-                        # Create a context-like structure for the wishlist removal function
-                        context = {
-                            'track_info': track_info,
-                            'original_search_result': track_info  # fallback
-                        }
-                        _check_and_remove_from_wishlist(context)
-                except Exception as wishlist_error:
-                    logger.error(f"[Batch Manager] Error checking wishlist removal for successful download: {wishlist_error}")
-
-            # Decrement active count
-            old_active = download_batches[batch_id]['active_count']
-            download_batches[batch_id]['active_count'] -= 1
-            new_active = download_batches[batch_id]['active_count']
-
-            logger.error(f"[Batch Manager] Task {task_id} completed ({'success' if success else 'failed/cancelled'}). Active workers: {old_active} → {new_active}/{download_batches[batch_id]['max_concurrent']}")
-
-        # ENHANCED: Always check batch completion after any task completes (including duplicate calls)
-        # This ensures completion is detected even when mixing normal downloads with cancelled tasks
-        logger.info(f"[Batch Manager] Checking batch completion after task {task_id} completed")
-        
-        # FIXED: Check if batch is truly complete (all tasks finished, not just workers freed)
-        batch = download_batches[batch_id]
-        all_tasks_started = batch['queue_index'] >= len(batch['queue'])
-        no_active_workers = batch['active_count'] == 0
-        
-        # Count actually finished tasks (completed, failed, or cancelled)
-        # CRITICAL: Don't include 'post_processing' as finished - it's still in progress (unless stuck)!
-        # CRITICAL: Don't include 'searching' as finished - task is being retried (unless stuck)!
-        finished_count = 0
-        retrying_count = 0
-        queue = batch.get('queue', [])
-        current_time = time.time()
-        for task_id in queue:
-            if task_id in download_tasks:
-                task = download_tasks[task_id]
-                task_status = task['status']
-
-                # STUCK DETECTION: Force fail tasks that have been in transitional states too long
-                if task_status == 'searching':
-                    task_age = current_time - task.get('status_change_time', current_time)
-                    if task_age > 600:  # 10 minutes
-                        logger.info(f"⏰ [Stuck Detection] Task {task_id} stuck in searching for {task_age:.0f}s - forcing not_found")
-                        task['status'] = 'not_found'
-                        task['error_message'] = f'Search stuck for {int(task_age // 60)} minutes with no results — timed out'
-                        finished_count += 1
-                    else:
-                        retrying_count += 1
-                elif task_status == 'post_processing':
-                    task_age = current_time - task.get('status_change_time', current_time)
-                    if task_age > 300:  # 5 minutes (post-processing should be fast)
-                        logger.info(f"⏰ [Stuck Detection] Task {task_id} stuck in post_processing for {task_age:.0f}s - forcing completion")
-                        task['status'] = 'completed'  # Assume it worked if file verification is taking too long
-                        finished_count += 1
-                    else:
-                        retrying_count += 1
-                elif task_status in ['completed', 'failed', 'cancelled', 'not_found']:
-                    finished_count += 1
-            else:
-                # Task ID in queue but not in download_tasks - treat as completed to prevent blocking
-                logger.warning(f"[Orphaned Task] Task {task_id} in queue but not in download_tasks - counting as finished")
-                finished_count += 1
-        
-        all_tasks_truly_finished = finished_count >= len(queue)
-        has_retrying_tasks = retrying_count > 0
-        
-        if all_tasks_started and no_active_workers and all_tasks_truly_finished and not has_retrying_tasks:
-            logger.error(f"[Batch Manager] Batch {batch_id} truly complete - all {finished_count}/{len(queue)} tasks finished - processing failed tracks to wishlist")
-        elif all_tasks_started and no_active_workers and has_retrying_tasks:
-            logger.warning(f"[Batch Manager] Batch {batch_id}: all workers free but {retrying_count} tasks retrying - continuing monitoring")
-        elif all_tasks_started and no_active_workers:
-            # This used to incorrectly mark batch as complete!
-            logger.info(f"[Batch Manager] Batch {batch_id}: all workers free but only {finished_count}/{len(queue)} tasks finished - continuing monitoring")
-        
-        if all_tasks_started and no_active_workers and all_tasks_truly_finished and not has_retrying_tasks:
-            
-            # Check if this is an auto-initiated batch
-            is_auto_batch = batch.get('auto_initiated', False)
-            
-            # FIXED: Ensure batch is not already marked as complete to prevent duplicate processing
-            if batch.get('phase') != 'complete':
-                # Mark batch as complete and set completion timestamp for auto-cleanup
-                batch['phase'] = 'complete'
-                batch['completion_time'] = time.time()  # Track when batch completed
-
-                # Record sync history completion
-                _record_sync_history_completion(batch_id, batch)
-
-                # Add activity for batch completion
-                playlist_name = batch.get('playlist_name', 'Unknown Playlist')
-                failed_count = len(batch.get('permanently_failed_tracks', []))
-                successful_downloads = finished_count - failed_count
-                add_activity_item("", "Download Batch Complete", f"'{playlist_name}' - {successful_downloads} tracks downloaded", "Now")
-
-                # Emit batch_complete event for automation engine (only if something downloaded)
-                if successful_downloads > 0:
-                    try:
-                        if automation_engine:
-                            automation_engine.emit('batch_complete', {
-                                'playlist_name': playlist_name,
-                                'total_tracks': str(len(queue)),
-                                'completed_tracks': str(successful_downloads),
-                                'failed_tracks': str(failed_count),
-                            })
-                    except Exception:
-                        pass
-
-                # Update YouTube playlist phase to 'download_complete' if this is a YouTube playlist
-                playlist_id = batch.get('playlist_id')
-                if playlist_id and playlist_id.startswith('youtube_'):
-                    url_hash = playlist_id.replace('youtube_', '')
-                    if url_hash in youtube_playlist_states:
-                        youtube_playlist_states[url_hash]['phase'] = 'download_complete'
-                        logger.info(f"Updated YouTube playlist {url_hash} to download_complete phase")
-
-                # Update Tidal playlist phase to 'download_complete' if this is a Tidal playlist
-                if playlist_id and playlist_id.startswith('tidal_'):
-                    tidal_playlist_id = playlist_id.replace('tidal_', '')
-                    if tidal_playlist_id in tidal_discovery_states:
-                        tidal_discovery_states[tidal_playlist_id]['phase'] = 'download_complete'
-                        logger.info(f"Updated Tidal playlist {tidal_playlist_id} to download_complete phase")
-
-                # Update Deezer playlist phase to 'download_complete' if this is a Deezer playlist
-                if playlist_id and playlist_id.startswith('deezer_'):
-                    deezer_playlist_id = playlist_id.replace('deezer_', '')
-                    if deezer_playlist_id in deezer_discovery_states:
-                        deezer_discovery_states[deezer_playlist_id]['phase'] = 'download_complete'
-                        logger.info(f"Updated Deezer playlist {deezer_playlist_id} to download_complete phase")
-
-                # Update Spotify Public playlist phase to 'download_complete' if this is a Spotify Public playlist
-                if playlist_id and playlist_id.startswith('spotify_public_'):
-                    spotify_public_url_hash = playlist_id.replace('spotify_public_', '')
-                    if spotify_public_url_hash in spotify_public_discovery_states:
-                        spotify_public_discovery_states[spotify_public_url_hash]['phase'] = 'download_complete'
-                        logger.info(f"Updated Spotify Public playlist {spotify_public_url_hash} to download_complete phase")
-
-                logger.info(f"[Batch Manager] Batch {batch_id} complete - stopping monitor")
-                download_monitor.stop_monitoring(batch_id)
-
-                # M3U REGENERATION: Regenerate M3U with real library paths now that
-                # all post-processing (tagging, moving, DB writes) is complete.
-                # The frontend M3U save may fire too early — this ensures paths resolve.
-                if config_manager.get('m3u_export.enabled', False):
-                    try:
-                        m3u_tracks = []
-                        for tid in queue:
-                            if tid in download_tasks and download_tasks[tid].get('status') == 'completed':
-                                ti = download_tasks[tid].get('track_info', {})
-                                artists = ti.get('artists', [])
-                                artist_str = artists[0] if isinstance(artists, list) and artists else ''
-                                if isinstance(artist_str, dict):
-                                    artist_str = artist_str.get('name', '')
-                                m3u_tracks.append({
-                                    'name': ti.get('name', ''),
-                                    'artist': artist_str,
-                                    'duration_ms': ti.get('duration_ms', 0),
-                                })
-                        if m3u_tracks:
-                            _regenerate_batch_m3u(batch, m3u_tracks)
-                    except Exception as m3u_err:
-                        logger.error(f"[M3U] Error regenerating M3U on batch complete: {m3u_err}")
-
-                # REPAIR: Scan all album folders from this batch for track number issues
-                if repair_worker:
-                    repair_worker.process_batch(batch_id)
-
-                # ALBUM CONSISTENCY: Picard-style post-batch pass — pick ONE MusicBrainz
-                # release and overwrite album-level tags on all files to guarantee consistency.
-                # This is the safety net: even if per-track MB lookups drifted (different cache
-                # keys, API hiccups), this pass forces every file to share the same release MBID,
-                # album artist ID, release group ID, etc. — preventing Navidrome album splits.
-                _cons_files = batch.get('_consistency_files', [])
-                if batch.get('is_album_download') and _cons_files and len(_cons_files) >= 2:
-                    _cons_album = batch.get('album_context', {})
-                    _cons_artist = batch.get('artist_context', {})
-                    _cons_album_name = _cons_album.get('name', '') if isinstance(_cons_album, dict) else ''
-                    _cons_artist_name = _cons_artist.get('name', '') if isinstance(_cons_artist, dict) else ''
-                    if _cons_album_name and _cons_artist_name:
-                        try:
-                            _cons_mb_svc = mb_worker.mb_service if mb_worker else None
-                            if _cons_mb_svc and config_manager.get('musicbrainz.embed_tags', True):
-                                from core.album_consistency import run_album_consistency
-                                _cons_result = run_album_consistency(
-                                    file_infos=_cons_files,
-                                    album_name=_cons_album_name,
-                                    artist_name=_cons_artist_name,
-                                    mb_service=_cons_mb_svc,
-                                    total_discs=_cons_album.get('total_discs', 1),
-                                    file_lock_fn=get_file_lock,
-                                )
-                                if _cons_result.get('success'):
-                                    logger.info(f"[Album Consistency] {_cons_result['tags_written']}/{_cons_result['total_files']} files "
-                                          f"harmonized to release {_cons_result.get('release_mbid', '')[:8]}...")
-                                elif _cons_result.get('error'):
-                                    logger.error(f"[Album Consistency] Skipped: {_cons_result['error']}")
-                        except Exception as cons_err:
-                            logger.error(f"[Album Consistency] Failed (non-fatal): {cons_err}")
-
-                # Mark that wishlist processing is starting (prevents premature cleanup)
-                batch['wishlist_processing_started'] = True
-
-                # Process wishlist outside of the lock to prevent threading issues
-                if is_auto_batch:
-                    # For auto-initiated batches, handle completion and schedule next cycle
-                    missing_download_executor.submit(_process_failed_tracks_to_wishlist_exact_with_auto_completion, batch_id)
-                else:
-                    # For manual batches, use standard wishlist processing
-                    missing_download_executor.submit(_process_failed_tracks_to_wishlist_exact, batch_id)
-            else:
-                logger.warning(f"[Batch Manager] Batch {batch_id} already marked complete - skipping duplicate processing")
-
-            return  # Don't start next batch if we're done
-    
-    # Start next downloads in queue
-    logger.info(f"[Batch Manager] Starting next batch for {batch_id}")
-    _start_next_batch_of_downloads(batch_id)
 
 def _run_full_missing_tracks_process(batch_id, playlist_id, tracks_json):
     """
@@ -23909,183 +23568,10 @@ def cancel_task_v2():
         return jsonify({"success": False, "error": str(e)}), 500
 
 def _check_batch_completion_v2(batch_id):
-    """
-    V2 SYSTEM: Check if batch is complete after worker slot changes.
-    
-    This is needed because V2 atomic cancel bypasses _on_download_completed,
-    so we need to manually check for batch completion.
-    """
-    try:
-        with tasks_lock:
-            if batch_id not in download_batches:
-                logger.warning(f"[Completion Check V2] Batch {batch_id} not found")
-                return
-            
-            batch = download_batches[batch_id]
-            all_tasks_started = batch['queue_index'] >= len(batch['queue'])
-            no_active_workers = batch['active_count'] == 0
-            
-            # Count actually finished tasks (completed, failed, or cancelled)
-            finished_count = 0
-            retrying_count = 0
-            queue = batch.get('queue', [])
-            current_time = time.time()
+    """V2 SYSTEM: Check if batch is complete after worker slot changes."""
+    return _downloads_lifecycle.check_batch_completion_v2(batch_id, _build_lifecycle_deps())
 
-            for task_id in queue:
-                if task_id in download_tasks:
-                    task = download_tasks[task_id]
-                    task_status = task['status']
 
-                    # STUCK DETECTION: Force fail tasks that have been in transitional states too long
-                    if task_status == 'searching':
-                        task_age = current_time - task.get('status_change_time', current_time)
-                        if task_age > 600:  # 10 minutes
-                            logger.info(f"⏰ [Stuck Detection V2] Task {task_id} stuck in searching for {task_age:.0f}s - forcing not_found")
-                            task['status'] = 'not_found'
-                            task['error_message'] = f'Search stuck for {int(task_age // 60)} minutes with no results — timed out'
-                            finished_count += 1
-                        else:
-                            retrying_count += 1
-                    elif task_status == 'post_processing':
-                        task_age = current_time - task.get('status_change_time', current_time)
-                        if task_age > 300:  # 5 minutes (post-processing should be fast)
-                            logger.info(f"⏰ [Stuck Detection V2] Task {task_id} stuck in post_processing for {task_age:.0f}s - forcing completion")
-                            task['status'] = 'completed'  # Assume it worked if file verification is taking too long
-                            finished_count += 1
-                        else:
-                            retrying_count += 1
-                    elif task_status in ['completed', 'failed', 'cancelled', 'not_found']:
-                        finished_count += 1
-                else:
-                    # Task ID in queue but not in download_tasks - treat as completed to prevent blocking
-                    logger.warning(f"[Orphaned Task V2] Task {task_id} in queue but not in download_tasks - counting as finished")
-                    finished_count += 1
-            
-            all_tasks_truly_finished = finished_count >= len(queue)
-            has_retrying_tasks = retrying_count > 0
-            
-            logger.warning(f"[Completion Check V2] Batch {batch_id}: tasks_started={all_tasks_started}, workers={no_active_workers}, finished={finished_count}/{len(queue)}, retrying={retrying_count}")
-            
-            if all_tasks_started and no_active_workers and all_tasks_truly_finished and not has_retrying_tasks:
-                # FIXED: Ensure batch is not already marked as complete to prevent duplicate processing
-                if batch.get('phase') != 'complete':
-                    logger.info(f"[Completion Check V2] Batch {batch_id} is complete - marking as finished")
-
-                    # Check if this is an auto-initiated batch
-                    is_auto_batch = batch.get('auto_initiated', False)
-
-                    # Mark batch as complete and set completion timestamp for auto-cleanup
-                    batch['phase'] = 'complete'
-                    batch['completion_time'] = time.time()  # Track when batch completed
-
-                    # Add activity for batch completion
-                    playlist_name = batch.get('playlist_name', 'Unknown Playlist')
-                    failed_count = len(batch.get('permanently_failed_tracks', []))
-                    successful_downloads = finished_count - failed_count
-                    add_activity_item("", "Download Batch Complete", f"'{playlist_name}' - {successful_downloads} tracks downloaded", "Now")
-
-                    # Emit batch_complete event for automation engine (only if something downloaded)
-                    if successful_downloads > 0:
-                        try:
-                            if automation_engine:
-                                automation_engine.emit('batch_complete', {
-                                    'playlist_name': playlist_name,
-                                    'total_tracks': str(len(queue)),
-                                    'completed_tracks': str(successful_downloads),
-                                    'failed_tracks': str(failed_count),
-                                })
-                        except Exception:
-                            pass
-                else:
-                    logger.warning(f"[Completion Check V2] Batch {batch_id} already marked complete - skipping duplicate processing")
-                    return True  # Already complete
-
-                # Update YouTube playlist phase to 'download_complete' if this is a YouTube playlist
-                playlist_id = batch.get('playlist_id')
-                if playlist_id and playlist_id.startswith('youtube_'):
-                    url_hash = playlist_id.replace('youtube_', '')
-                    if url_hash in youtube_playlist_states:
-                        youtube_playlist_states[url_hash]['phase'] = 'download_complete'
-                        logger.info(f"[Completion Check V2] Updated YouTube playlist {url_hash} to download_complete phase")
-                
-                # Update Tidal playlist phase to 'download_complete' if this is a Tidal playlist
-                if playlist_id and playlist_id.startswith('tidal_'):
-                    tidal_playlist_id = playlist_id.replace('tidal_', '')
-                    if tidal_playlist_id in tidal_discovery_states:
-                        tidal_discovery_states[tidal_playlist_id]['phase'] = 'download_complete'
-                        logger.info(f"[Completion Check V2] Updated Tidal playlist {tidal_playlist_id} to download_complete phase")
-
-                # Update Deezer playlist phase to 'download_complete' if this is a Deezer playlist
-                if playlist_id and playlist_id.startswith('deezer_'):
-                    deezer_playlist_id = playlist_id.replace('deezer_', '')
-                    if deezer_playlist_id in deezer_discovery_states:
-                        deezer_discovery_states[deezer_playlist_id]['phase'] = 'download_complete'
-                        logger.info(f"[Completion Check V2] Updated Deezer playlist {deezer_playlist_id} to download_complete phase")
-
-                # Update Spotify Public playlist phase to 'download_complete' if this is a Spotify Public playlist
-                if playlist_id and playlist_id.startswith('spotify_public_'):
-                    spotify_public_url_hash = playlist_id.replace('spotify_public_', '')
-                    if spotify_public_url_hash in spotify_public_discovery_states:
-                        spotify_public_discovery_states[spotify_public_url_hash]['phase'] = 'download_complete'
-                        logger.info(f"[Completion Check V2] Updated Spotify Public playlist {spotify_public_url_hash} to download_complete phase")
-
-                logger.info(f"[Completion Check V2] Batch {batch_id} complete - stopping monitor")
-                download_monitor.stop_monitoring(batch_id)
-
-                # REPAIR: Scan all album folders from this batch for track number issues
-                if repair_worker:
-                    repair_worker.process_batch(batch_id)
-
-                # ALBUM CONSISTENCY: Same Picard-style pass as the primary completion path
-                _cons_files = batch.get('_consistency_files', [])
-                if batch.get('is_album_download') and _cons_files and len(_cons_files) >= 2:
-                    _cons_album = batch.get('album_context', {})
-                    _cons_artist = batch.get('artist_context', {})
-                    _cons_album_name = _cons_album.get('name', '') if isinstance(_cons_album, dict) else ''
-                    _cons_artist_name = _cons_artist.get('name', '') if isinstance(_cons_artist, dict) else ''
-                    if _cons_album_name and _cons_artist_name:
-                        try:
-                            _cons_mb_svc = mb_worker.mb_service if mb_worker else None
-                            if _cons_mb_svc and config_manager.get('musicbrainz.embed_tags', True):
-                                from core.album_consistency import run_album_consistency
-                                _cons_result = run_album_consistency(
-                                    file_infos=_cons_files,
-                                    album_name=_cons_album_name,
-                                    artist_name=_cons_artist_name,
-                                    mb_service=_cons_mb_svc,
-                                    total_discs=_cons_album.get('total_discs', 1),
-                                    file_lock_fn=get_file_lock,
-                                )
-                                if _cons_result.get('success'):
-                                    logger.info(f"[Album Consistency V2] {_cons_result['tags_written']}/{_cons_result['total_files']} files "
-                                          f"harmonized to release {_cons_result.get('release_mbid', '')[:8]}...")
-                                elif _cons_result.get('error'):
-                                    logger.error(f"[Album Consistency V2] Skipped: {_cons_result['error']}")
-                        except Exception as cons_err:
-                            logger.error(f"[Album Consistency V2] Failed (non-fatal): {cons_err}")
-
-        # Process wishlist outside of the lock to prevent threading issues
-        if all_tasks_started and no_active_workers and all_tasks_truly_finished and not has_retrying_tasks:
-            # Call wishlist processing outside the lock
-            if is_auto_batch:
-                logger.info("[Completion Check V2] Processing auto-initiated batch completion")
-                # Use the existing auto-completion function
-                _process_failed_tracks_to_wishlist_exact_with_auto_completion(batch_id)
-            else:
-                logger.info("[Completion Check V2] Processing regular batch completion")
-                # Use the regular completion function
-                _process_failed_tracks_to_wishlist_exact(batch_id)
-            
-            return True  # Batch was completed
-        else:
-            logger.warning(f"[Completion Check V2] Batch {batch_id} not yet complete: finished={finished_count}/{len(queue)}, retrying={retrying_count}, workers={batch['active_count']}")
-            return False  # Batch still in progress
-                
-    except Exception as e:
-        logger.error(f"[Completion Check V2] Error checking batch completion: {e}")
-        import traceback
-        traceback.print_exc()
-        return False
 
 def _add_cancelled_task_to_wishlist(task):
     """