Merge pull request #744 from Nezreka/refactor/wishlist-orchestration-unify

Refactor/wishlist orchestration unify

core/wishlist/processing.py

@@ -7,7 +7,7 @@ from dataclasses import dataclass
 from datetime import datetime
 from contextlib import AbstractContextManager
 from types import SimpleNamespace
-from typing import Any, Callable, Dict
+from typing import Any, Callable, Dict, Optional
 
 from core.wishlist.payloads import build_failed_track_wishlist_context
 from core.wishlist.selection import filter_wishlist_tracks_by_category, sanitize_and_dedupe_wishlist_tracks
@@ -100,6 +100,202 @@ def remove_completed_tracks_from_wishlist(
     return removed_count
 
 
+def make_wishlist_batch_row(
+    *,
+    playlist_id: str,
+    playlist_name: str,
+    track_count: int,
+    max_concurrent: int,
+    profile_id: int,
+    phase: str,
+    run_id: str | None = None,
+    is_album: bool = False,
+    album_context: Optional[Dict[str, Any]] = None,
+    artist_context: Optional[Dict[str, Any]] = None,
+    extra_fields: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    """Single source of truth for a wishlist ``download_batches`` row.
+
+    The auto and manual wishlist flows used to build this ~20-field dict in four
+    separate places, which let their batch shapes silently drift apart. They now
+    all go through here so every wishlist batch has an IDENTICAL field shape; the
+    genuinely per-flow differences (initial ``phase``, the auto-only
+    ``auto_initiated`` / ``current_cycle`` fields, album vs residual contexts) are
+    explicit arguments / ``extra_fields``.
+
+    NOTE: this builds the row only — it does NOT decide grouping, batch-id
+    allocation, or dispatch (parallel-submit vs serial), which legitimately
+    differ between the flows and stay in their callers.
+    """
+    row: Dict[str, Any] = {
+        'phase': phase,
+        'playlist_id': playlist_id,
+        'playlist_name': playlist_name,
+        'queue': [],
+        'active_count': 0,
+        'max_concurrent': max_concurrent,
+        'queue_index': 0,
+        'analysis_total': track_count,
+        'analysis_processed': 0,
+        'analysis_results': [],
+        'permanently_failed_tracks': [],
+        'cancelled_tracks': set(),
+        'force_download_all': True,
+        'profile_id': profile_id,
+        'is_album_download': is_album,
+        'album_context': album_context,
+        'artist_context': artist_context,
+        'wishlist_run_id': run_id,
+    }
+    if extra_fields:
+        row.update(extra_fields)
+    return row
+
+
+def _run_wishlist_cycle(
+    runtime,
+    *,
+    playlist_id: str,
+    cycle: str,
+    tracks: list,
+    run_id: str,
+    auto_initiated: bool,
+    first_batch_id: Optional[str] = None,
+) -> Dict[str, Any]:
+    """THE single wishlist orchestration engine — both the auto timer and the
+    manual trigger call this, so a manual scan runs the exact same code path as
+    an auto scan (group → per-album + residual batches → register → dispatch).
+
+    Per-flow differences are arguments, not separate code:
+      * ``auto_initiated`` stamps the auto-only fields (auto_initiated /
+        auto_processing_timestamp / current_cycle, which also drives the
+        once-per-run cycle toggle on completion) and selects the auto vs manual
+        display-name + log style.
+      * ``first_batch_id`` lets the manual flow reuse its synchronously-created
+        placeholder batch so the modal's existing poll target stays valid.
+
+    Album batches block their worker for the whole search+download, so they run
+    on the dedicated album pool; the residual per-track batch runs on the shared
+    pool. Returns a summary dict (submitted ids + album / residual counts).
+    """
+    logger = runtime.logger
+    from core.wishlist.album_grouping import group_wishlist_tracks_by_album
+
+    # Albums cycle splits into per-album bundles; singles keep the single
+    # per-track batch shape (Spotify already classifies them away from albums).
+    grouping = (
+        group_wishlist_tracks_by_album(
+            tracks, min_tracks_per_album=_resolve_album_bundle_threshold(),
+        )
+        if cycle == 'albums' else None
+    )
+
+    extra_fields = None
+    if auto_initiated:
+        extra_fields = {
+            'auto_initiated': True,
+            'auto_processing_timestamp': runtime.current_time_fn(),
+            'current_cycle': cycle,
+        }
+
+    # Reuse the caller-provided placeholder id for the FIRST batch created; every
+    # other batch gets a fresh uuid.
+    _reuse_id = first_batch_id
+
+    def _alloc_id() -> str:
+        nonlocal _reuse_id
+        if _reuse_id is not None:
+            bid, _reuse_id = _reuse_id, None
+            return bid
+        return str(uuid.uuid4())
+
+    album_executor = runtime.album_bundle_executor or runtime.missing_download_executor
+    submitted: list = []
+
+    album_groups = grouping.album_groups if grouping else []
+    for album_idx, group in enumerate(album_groups):
+        album_batch_id = _alloc_id()
+        album_name = group.album_context.get('name', 'Unknown')
+        batch_name = (
+            f"Wishlist (Auto - Album: {album_name})" if auto_initiated
+            else f"Wishlist (Album: {album_name})"
+        )
+        with runtime.tasks_lock:
+            runtime.download_batches[album_batch_id] = make_wishlist_batch_row(
+                playlist_id=playlist_id,
+                playlist_name=batch_name,
+                track_count=len(group.tracks),
+                max_concurrent=runtime.get_batch_max_concurrent(),
+                profile_id=runtime.profile_id,
+                phase='queued',
+                run_id=run_id,
+                is_album=True,
+                album_context=group.album_context,
+                artist_context=group.artist_context,
+                extra_fields=extra_fields,
+            )
+        if auto_initiated:
+            logger.info(
+                f"[Auto-Wishlist] Album sub-batch {album_idx + 1}/{len(album_groups)}: "
+                f"'{album_name}' by '{group.artist_context.get('name')}' "
+                f"({len(group.tracks)} tracks) → {album_batch_id} [run {run_id[:8]}]"
+            )
+        else:
+            logger.info(
+                f"[Manual-Wishlist] Album sub-batch {album_idx + 1}/{len(album_groups)}: "
+                f"'{album_name}' ({len(group.tracks)} tracks) → {album_batch_id}"
+            )
+        submitted.append(album_batch_id)
+        # Album bundles block their worker for the whole search+download → dedicated
+        # pool (falls back to the shared pool when unset). See #740.
+        album_executor.submit(
+            runtime.run_full_missing_tracks_process,
+            album_batch_id, playlist_id, group.tracks,
+        )
+
+    residual_tracks = grouping.residual_tracks if grouping is not None else tracks
+    residual_count = len(residual_tracks) if residual_tracks else 0
+    if residual_tracks:
+        residual_batch_id = _alloc_id()
+        residual_name = (
+            f"Wishlist (Auto - {cycle.capitalize()})" if auto_initiated
+            else "Wishlist (Residual)"
+        )
+        with runtime.tasks_lock:
+            runtime.download_batches[residual_batch_id] = make_wishlist_batch_row(
+                playlist_id=playlist_id,
+                playlist_name=residual_name,
+                track_count=residual_count,
+                max_concurrent=runtime.get_batch_max_concurrent(),
+                profile_id=runtime.profile_id,
+                phase='queued',
+                run_id=run_id,
+                extra_fields=extra_fields,
+            )
+        submitted.append(residual_batch_id)
+        runtime.missing_download_executor.submit(
+            runtime.run_full_missing_tracks_process,
+            residual_batch_id, playlist_id, residual_tracks,
+        )
+        if auto_initiated:
+            logger.info(
+                f"Starting wishlist residual batch {residual_batch_id} with {residual_count} tracks "
+                f"({'singles' if cycle == 'singles' else 'unbucketed albums'}) "
+                f"[run {run_id[:8]}]"
+            )
+        else:
+            logger.info(
+                f"[Manual-Wishlist] Residual per-track batch {residual_batch_id} "
+                f"with {residual_count} tracks"
+            )
+
+    return {
+        'submitted': submitted,
+        'album_batches': len(album_groups),
+        'residual_count': residual_count,
+    }
+
+
 def add_cancelled_tracks_to_failed_tracks(
     batch: Dict[str, Any],
     download_tasks: Dict[str, Dict[str, Any]],
@@ -431,6 +627,9 @@ class WishlistManualDownloadRuntime:
     active_server: str
     profile_id: int
     logger: Any = module_logger
+    # Dedicated album-bundle pool, shared with the auto flow via
+    # _run_wishlist_cycle. Falls back to missing_download_executor when unset.
+    album_bundle_executor: Any = None
 
 
 def start_manual_wishlist_download_batch(
@@ -455,24 +654,16 @@ def start_manual_wishlist_download_batch(
         playlist_name = "Wishlist"
 
         with runtime.tasks_lock:
-            runtime.download_batches[batch_id] = {
+            # analysis_total starts at 0; the bg job updates it after cleanup
-                'phase': 'analysis',
+            # finishes and the real track count is known.
-                'playlist_id': playlist_id,
+            runtime.download_batches[batch_id] = make_wishlist_batch_row(
-                'playlist_name': playlist_name,
+                playlist_id=playlist_id,
-                'queue': [],
+                playlist_name=playlist_name,
-                'active_count': 0,
+                track_count=0,
-                'max_concurrent': runtime.get_batch_max_concurrent(),
+                max_concurrent=runtime.get_batch_max_concurrent(),
-                'queue_index': 0,
+                profile_id=runtime.profile_id,
-                # analysis_total starts at 0; the bg job updates it after cleanup
+                phase='analysis',
-                # finishes and the real track count is known.
+            )
-                'analysis_total': 0,
-                'analysis_processed': 0,
-                'analysis_results': [],
-                'permanently_failed_tracks': [],
-                'cancelled_tracks': set(),
-                'force_download_all': True,
-                'profile_id': runtime.profile_id,
-            }
 
         runtime.missing_download_executor.submit(
             _prepare_and_run_manual_wishlist_batch,
@@ -560,118 +751,34 @@ def _prepare_and_run_manual_wishlist_batch(
 
         runtime.add_activity_item("", "Wishlist Download Started", f"{len(wishlist_tracks)} tracks", "Now")
 
-        # Try to split into per-album sub-batches so each album fires
+        if not wishlist_tracks:
-        # ONE slskd / torrent / usenet album-bundle search (gates on
+            # Nothing to download — clear out the placeholder batch.
-        # ``is_album_download`` + populated album/artist context).
-        # When a single category was requested (or no category filter)
-        # we apply the same grouping the auto-wishlist path uses.
-        # Tracks the grouper can't bucket fall through to a residual
-        # batch with the classic per-track flow.
-        from core.wishlist.album_grouping import group_wishlist_tracks_by_album
-        grouping = group_wishlist_tracks_by_album(
-            wishlist_tracks,
-            min_tracks_per_album=_resolve_album_bundle_threshold(),
-        )
-
-        # Build the final payload list (batch_id, tracks, album_context,
-        # artist_context, is_album). The first payload re-uses the
-        # caller-allocated ``batch_id`` so the frontend's existing poll
-        # against it keeps working. Subsequent payloads get fresh ids.
-        payloads = []
-        for group in grouping.album_groups:
-            payloads.append({
-                'tracks': group.tracks,
-                'is_album': True,
-                'album_context': group.album_context,
-                'artist_context': group.artist_context,
-                'display_name': f"Wishlist (Album: {group.album_context.get('name', 'Unknown')})",
-            })
-        if grouping.residual_tracks:
-            payloads.append({
-                'tracks': grouping.residual_tracks,
-                'is_album': False,
-                'album_context': None,
-                'artist_context': None,
-                'display_name': "Wishlist (Residual)",
-            })
-
-        if not payloads:
-            # Nothing to download — clear out the original batch.
             with runtime.tasks_lock:
                 if batch_id in runtime.download_batches:
                     runtime.download_batches[batch_id]['analysis_total'] = 0
                     runtime.download_batches[batch_id]['phase'] = 'complete'
             return
 
-        # Attach the original batch_id to the first payload; allocate
+        # Run the selection through the SHARED engine — the exact code path the
-        # fresh batch_ids for the rest.
+        # auto timer uses (group → album bundles + per-track residual → parallel
-        payloads[0]['batch_id'] = batch_id
+        # dispatch on the album / shared pools). cycle='albums' bundles whatever
-        for payload in payloads[1:]:
+        # forms an album and drops the rest (singles / ungroupable) into the
-            payload['batch_id'] = str(uuid.uuid4())
+        # per-track residual, so this single call covers the whole selection.
-
+        # The placeholder batch_id is reused as the first sub-batch so the
-        # Reify "wishlist run" — one shared id stamped on every sub-
+        # modal's existing poll target stays valid.
-        # batch this manual invocation produces. Mirrors the auto
+        result = _run_wishlist_cycle(
-        # path. Note manual wishlist completion currently doesn't
+            runtime,
-        # toggle the cycle (only auto does), but the id is set anyway
+            playlist_id='wishlist',
-        # so future code + UI grouping have a consistent hook.
+            cycle='albums',
-        wishlist_run_id = str(uuid.uuid4())
+            tracks=wishlist_tracks,
-
+            run_id=str(uuid.uuid4()),
-        # Materialize each sub-batch's row state up-front so the
+            auto_initiated=False,
-        # frontend's polling can see them all under the original
+            first_batch_id=batch_id,
-        # batch's flow.
+        )
-        with runtime.tasks_lock:
+        logger.info(
-            if batch_id in runtime.download_batches:
+            f"[Manual-Wishlist] Dispatched {result['album_batches']} album batch(es) + "
-                # Re-purpose the existing row for the first payload.
+            f"{result['residual_count']} residual track(s) via the shared engine"
-                first = payloads[0]
-                runtime.download_batches[batch_id]['analysis_total'] = len(first['tracks'])
-                runtime.download_batches[batch_id]['wishlist_run_id'] = wishlist_run_id
-                if first['is_album']:
-                    runtime.download_batches[batch_id]['is_album_download'] = True
-                    runtime.download_batches[batch_id]['album_context'] = first['album_context']
-                    runtime.download_batches[batch_id]['artist_context'] = first['artist_context']
-                runtime.download_batches[batch_id]['playlist_name'] = first['display_name']
-            for payload in payloads[1:]:
-                runtime.download_batches[payload['batch_id']] = {
-                    'phase': 'analysis',
-                    'playlist_id': 'wishlist',
-                    'playlist_name': payload['display_name'],
-                    'queue': [],
-                    'active_count': 0,
-                    'max_concurrent': runtime.get_batch_max_concurrent(),
-                    'queue_index': 0,
-                    'analysis_total': len(payload['tracks']),
-                    'analysis_processed': 0,
-                    'analysis_results': [],
-                    'permanently_failed_tracks': [],
-                    'cancelled_tracks': set(),
-                    'force_download_all': True,
-                    'profile_id': runtime.profile_id,
-                    'is_album_download': bool(payload['is_album']),
-                    'album_context': payload['album_context'],
-                    'artist_context': payload['artist_context'],
-                    'wishlist_run_id': wishlist_run_id,
-                }
-
-        logger.info(
-            f"[Manual-Wishlist] Split into {len(payloads)} sub-batch(es) "
-            f"({sum(1 for p in payloads if p['is_album'])} album + "
-            f"{sum(1 for p in payloads if not p['is_album'])} residual)"
         )
-        # Serial dispatch — each album-bundle search happens one at a
-        # time so the slskd / Prowlarr pipeline doesn't fan out across
-        # multiple parallel release searches.
-        for payload in payloads:
-            label = (
-                f"album '{payload['album_context'].get('name')}'"
-                if payload['is_album'] else 'residual per-track'
-            )
-            logger.info(
-                f"[Manual-Wishlist] Running sub-batch {payload['batch_id']} "
-                f"({label}, {len(payload['tracks'])} tracks)"
-            )
-            runtime.run_full_missing_tracks_process(
-                payload['batch_id'], "wishlist", payload['tracks'],
-            )
 
     except Exception as exc:
         logger.error(f"Error preparing manual wishlist batch {batch_id}: {exc}")
@@ -833,142 +940,24 @@ def process_wishlist_automatically(runtime: WishlistAutoProcessingRuntime, autom
                 for i, track in enumerate(wishlist_tracks):
                     track['_original_index'] = i
 
-                # When the cycle is 'albums', try to split the wishlist
+                # Reify one "wishlist run" id (the completion handler gates the
-                # into per-album sub-batches so each album fires ONE
+                # once-per-run cycle toggle on it) and hand off to the SHARED
-                # album-bundle search (slskd / torrent / usenet) instead
+                # wishlist engine — the same code path the manual trigger uses.
-                # of N per-track searches. Residual tracks (no resolvable
-                # album metadata) fall through to a normal per-track
-                # batch. Singles cycle keeps its original single-batch
-                # shape — Spotify already classifies them away from
-                # albums.
-                _submitted_batches: list[str] = []
-                if current_cycle == 'albums':
-                    from core.wishlist.album_grouping import group_wishlist_tracks_by_album
-                    grouping = group_wishlist_tracks_by_album(
-                        wishlist_tracks,
-                        min_tracks_per_album=_resolve_album_bundle_threshold(),
-                    )
-                else:
-                    grouping = None
-
-                # Reify "wishlist run" — one shared id stamped on every
-                # sub-batch this invocation produces. The completion
-                # handler uses it to gate the once-per-run cycle toggle
-                # (so it doesn't fire N times for N sub-batches).
                 wishlist_run_id = str(uuid.uuid4())
-
+                _cycle_result = _run_wishlist_cycle(
-                if grouping and grouping.album_groups:
+                    runtime,
-                    for album_idx, group in enumerate(grouping.album_groups):
+                    playlist_id=playlist_id,
-                        album_batch_id = str(uuid.uuid4())
+                    cycle=current_cycle,
-                        album_batch_name = (
+                    tracks=wishlist_tracks,
-                            f"Wishlist (Auto - Album: {group.album_context.get('name', 'Unknown')})"
+                    run_id=wishlist_run_id,
-                        )
+                    auto_initiated=True,
-                        with runtime.tasks_lock:
-                            runtime.download_batches[album_batch_id] = {
-                                # ``queued`` until the master worker
-                                # picks the batch up from the
-                                # ``missing_download_executor`` pool
-                                # (max_workers=3 by default). The worker
-                                # flips phase to ``analysis`` as its
-                                # first action — see
-                                # ``core/downloads/master.py:328``.
-                                # Pre-fix the row was created with
-                                # ``analysis`` directly, so a wishlist
-                                # run with N > 3 sub-batches looked like
-                                # all N were working when really only
-                                # 3 were running.
-                                'phase': 'queued',
-                                'playlist_id': playlist_id,
-                                'playlist_name': album_batch_name,
-                                'queue': [],
-                                'active_count': 0,
-                                'max_concurrent': runtime.get_batch_max_concurrent(),
-                                'queue_index': 0,
-                                'analysis_total': len(group.tracks),
-                                'analysis_processed': 0,
-                                'analysis_results': [],
-                                'permanently_failed_tracks': [],
-                                'cancelled_tracks': set(),
-                                'force_download_all': True,
-                                'auto_initiated': True,
-                                'auto_processing_timestamp': runtime.current_time_fn(),
-                                'current_cycle': current_cycle,
-                                'profile_id': runtime.profile_id,
-                                # Album-bundle dispatch gate reads these
-                                # three. With them set, the master worker
-                                # routes through slskd / torrent / usenet
-                                # album-bundle search instead of per-track.
-                                'is_album_download': True,
-                                'album_context': group.album_context,
-                                'artist_context': group.artist_context,
-                                'wishlist_run_id': wishlist_run_id,
-                            }
-                        logger.info(
-                            f"[Auto-Wishlist] Album sub-batch {album_idx + 1}/{len(grouping.album_groups)}: "
-                            f"'{group.album_context.get('name')}' by '{group.artist_context.get('name')}' "
-                            f"({len(group.tracks)} tracks) → {album_batch_id} [run {wishlist_run_id[:8]}]"
-                        )
-                        _submitted_batches.append(album_batch_id)
-                        # Album bundles block their worker thread for the whole
-                        # search+download, so run them on the dedicated album
-                        # pool — never the shared pool that serves analysis,
-                        # per-track downloads and the manual wishlist (#740).
-                        # Fall back to the shared pool if unset (older callers).
-                        _album_executor = (
-                            runtime.album_bundle_executor or runtime.missing_download_executor
-                        )
-                        _album_executor.submit(
-                            runtime.run_full_missing_tracks_process,
-                            album_batch_id, playlist_id, group.tracks,
-                        )
-
-                # Residual tracks (no album group could be formed, OR
-                # singles cycle): one classic per-track batch as before.
-                residual_tracks = (
-                    grouping.residual_tracks if grouping is not None else wishlist_tracks
                 )
-                if residual_tracks:
-                    batch_id = str(uuid.uuid4())
-                    playlist_name = f"Wishlist (Auto - {current_cycle.capitalize()})"
-                    with runtime.tasks_lock:
-                        runtime.download_batches[batch_id] = {
-                            # See album sub-batch above — ``queued``
-                            # until the master worker picks it up.
-                            'phase': 'queued',
-                            'playlist_id': playlist_id,
-                            'playlist_name': playlist_name,
-                            'queue': [],
-                            'active_count': 0,
-                            'max_concurrent': runtime.get_batch_max_concurrent(),
-                            'queue_index': 0,
-                            'analysis_total': len(residual_tracks),
-                            'analysis_processed': 0,
-                            'analysis_results': [],
-                            'permanently_failed_tracks': [],
-                            'cancelled_tracks': set(),
-                            'force_download_all': True,
-                            'auto_initiated': True,
-                            'auto_processing_timestamp': runtime.current_time_fn(),
-                            'current_cycle': current_cycle,
-                            'profile_id': runtime.profile_id,
-                            'wishlist_run_id': wishlist_run_id,
-                        }
-                    _submitted_batches.append(batch_id)
-                    runtime.missing_download_executor.submit(
-                        runtime.run_full_missing_tracks_process,
-                        batch_id, playlist_id, residual_tracks,
-                    )
-                    logger.info(
-                        f"Starting wishlist residual batch {batch_id} with {len(residual_tracks)} tracks "
-                        f"({'singles' if current_cycle == 'singles' else 'unbucketed albums'}) "
-                        f"[run {wishlist_run_id[:8]}]"
-                    )
 
                 _summary_parts: list[str] = []
-                if grouping and grouping.album_groups:
+                if _cycle_result['album_batches']:
-                    _summary_parts.append(f"{len(grouping.album_groups)} album batch(es)")
+                    _summary_parts.append(f"{_cycle_result['album_batches']} album batch(es)")
-                if residual_tracks:
+                if _cycle_result['residual_count']:
-                    _summary_parts.append(f"{len(residual_tracks)} per-track")
+                    _summary_parts.append(f"{_cycle_result['residual_count']} per-track")
                 _summary_text = ', '.join(_summary_parts) or 'no batches'
                 runtime.update_automation_progress(
                     automation_id, progress=50,

tests/wishlist/test_batch_factory.py

@@ -0,0 +1,102 @@
+"""Tests for make_wishlist_batch_row — the single source of truth for a wishlist
+download_batches row, shared by the auto and manual flows so their batch shapes
+can't drift apart.
+"""
+
+from __future__ import annotations
+
+from core.wishlist.processing import make_wishlist_batch_row
+
+
+_CORE_KEYS = {
+    'phase', 'playlist_id', 'playlist_name', 'queue', 'active_count',
+    'max_concurrent', 'queue_index', 'analysis_total', 'analysis_processed',
+    'analysis_results', 'permanently_failed_tracks', 'cancelled_tracks',
+    'force_download_all', 'profile_id', 'is_album_download', 'album_context',
+    'artist_context', 'wishlist_run_id',
+}
+
+
+def _row(**overrides):
+    base = dict(
+        playlist_id='wishlist', playlist_name='Wishlist', track_count=3,
+        max_concurrent=4, profile_id=1, phase='analysis',
+    )
+    base.update(overrides)
+    return make_wishlist_batch_row(**base)
+
+
+def test_core_fields_always_present_and_consistent():
+    row = _row()
+    assert _CORE_KEYS <= set(row.keys())
+    # Fresh-batch invariants.
+    assert row['queue'] == [] and row['active_count'] == 0 and row['queue_index'] == 0
+    assert row['analysis_processed'] == 0
+    assert row['analysis_results'] == [] and row['permanently_failed_tracks'] == []
+    assert row['cancelled_tracks'] == set()
+    assert row['force_download_all'] is True
+    assert row['analysis_total'] == 3
+    assert row['max_concurrent'] == 4
+    assert row['profile_id'] == 1
+
+
+def test_residual_defaults_are_per_track():
+    row = _row()
+    assert row['is_album_download'] is False
+    assert row['album_context'] is None and row['artist_context'] is None
+    assert row['wishlist_run_id'] is None
+
+
+def test_album_batch_carries_context():
+    row = _row(
+        phase='queued', run_id='run-1', is_album=True,
+        album_context={'name': 'Album One'}, artist_context={'name': 'Artist 1'},
+    )
+    assert row['phase'] == 'queued'
+    assert row['is_album_download'] is True
+    assert row['album_context'] == {'name': 'Album One'}
+    assert row['artist_context'] == {'name': 'Artist 1'}
+    assert row['wishlist_run_id'] == 'run-1'
+
+
+def test_extra_fields_merged_for_auto():
+    row = _row(extra_fields={
+        'auto_initiated': True, 'auto_processing_timestamp': 123.0,
+        'current_cycle': 'albums',
+    })
+    assert row['auto_initiated'] is True
+    assert row['auto_processing_timestamp'] == 123.0
+    assert row['current_cycle'] == 'albums'
+
+
+def test_manual_row_has_no_auto_fields():
+    """Manual rows must not carry the auto-only fields (no extra_fields)."""
+    row = _row(phase='analysis')
+    assert 'auto_initiated' not in row
+    assert 'current_cycle' not in row
+
+
+def test_fresh_rows_do_not_share_mutable_state():
+    """Each row must get its OWN queue/list/set — not a shared reference that
+    one batch's tasks could leak into another's."""
+    a = _row()
+    b = _row()
+    a['queue'].append('task-1')
+    a['cancelled_tracks'].add('x')
+    assert b['queue'] == []
+    assert b['cancelled_tracks'] == set()
+    assert b['analysis_results'] == []
+
+
+def test_auto_and_manual_rows_share_identical_key_shape():
+    """The drift-prevention guarantee: an auto album row and a manual album row
+    expose the same set of keys (modulo the auto-only extras), so the modal /
+    status code sees a consistent shape from both flows."""
+    manual = _row(phase='analysis', run_id='r', is_album=True,
+                  album_context={'name': 'A'}, artist_context={'name': 'B'})
+    auto = _row(phase='queued', run_id='r', is_album=True,
+                album_context={'name': 'A'}, artist_context={'name': 'B'},
+                extra_fields={'auto_initiated': True, 'current_cycle': 'albums'})
+    # Auto is a strict superset (the auto-only extras); the shared core is identical.
+    assert set(manual.keys()) <= set(auto.keys())
+    assert set(auto.keys()) - set(manual.keys()) == {'auto_initiated', 'current_cycle'}

tests/wishlist/test_manual_download.py

@@ -110,6 +110,17 @@ def _run_submitted_bg_job(executor):
     fn(*args, **kwargs)
 
 
+def _dispatched(executor, runtime):
+    """The run_full_missing_tracks_process dispatches the shared engine submitted
+    to the executor (everything after the initial bg-job submission). Manual now
+    parallel-dispatches via the engine instead of running them serially inline,
+    so the master worker is *submitted*, not called directly."""
+    return [
+        s for s in executor.submissions
+        if s[0] is runtime.run_full_missing_tracks_process
+    ]
+
+
 def test_start_manual_wishlist_download_batch_returns_immediately_with_placeholder():
     """Endpoint returns 200 immediately; cleanup runs in the bg job."""
     runtime, service, _db, executor, _logger, activity_calls, batch_map, master_calls = _build_runtime(
@@ -174,11 +185,15 @@ def test_start_manual_wishlist_download_batch_filters_track_ids_and_starts_batch
     _run_submitted_bg_job(executor)
 
     assert activity_calls == [("", "Wishlist Download Started", "1 tracks", "Now")]
-    assert len(master_calls) == 1
+    # One track → no album group (threshold 2) → one residual batch, dispatched
-    master_args, _ = master_calls[0]
+    # via the shared engine (submitted to the executor, not called inline).
-    assert master_args[1] == "wishlist"
+    dispatched = _dispatched(executor, runtime)
-    assert master_args[2][0]["id"] == "track-2"
+    assert len(dispatched) == 1
-    assert master_args[2][0]["_original_index"] == 0
+    args = dispatched[0][1]
+    assert args[1] == "wishlist"
+    assert args[2][0]["id"] == "track-2"
+    assert args[2][0]["_original_index"] == 0
+    assert args[0] == payload["batch_id"]  # placeholder batch_id reused as first sub-batch
     assert batch_map[payload["batch_id"]]["analysis_total"] == 1
     assert batch_map[payload["batch_id"]]["force_download_all"] is True
     assert any("Filtered to 1 specific tracks by ID" in msg for msg in logger.info_messages)
@@ -224,10 +239,12 @@ def test_start_manual_wishlist_download_batch_does_not_run_library_cleanup():
     # The library check is skipped entirely — no per-track DB lookups.
     assert db.track_checks == []
 
-    # All tracks are submitted to the master worker — including the "owned" one.
+    # All tracks are dispatched to the master worker — including the "owned" one.
-    assert len(master_calls) == 1
+    # Two single-track albums → no album groups → one residual batch of both.
-    master_args, _ = master_calls[0]
+    dispatched = _dispatched(executor, runtime)
-    assert [track["id"] for track in master_args[2]] == ["enhance-1", "owned-1"]
+    assert len(dispatched) == 1
+    args = dispatched[0][1]
+    assert [track["id"] for track in args[2]] == ["enhance-1", "owned-1"]
     assert batch_map[payload["batch_id"]]["analysis_total"] == 2
     assert activity_calls == [("", "Wishlist Download Started", "2 tracks", "Now")]
 
@@ -293,28 +310,33 @@ def test_manual_wishlist_splits_into_per_album_sub_batches():
     assert status == 200
     _run_submitted_bg_job(executor)
 
-    # Two album groups → two master-worker calls.
+    # Two album groups → two album sub-batches, PARALLEL-dispatched via the shared
-    assert len(master_calls) == 2
+    # engine (same as auto) — not serial inline calls.
+    dispatched = _dispatched(executor, runtime)
+    assert len(dispatched) == 2
+    assert master_calls == []  # nothing run synchronously inline anymore
 
-    # First sub-batch uses the caller-allocated batch_id.
+    # First sub-batch reuses the caller-allocated placeholder batch_id.
-    first_args, _ = master_calls[0]
+    first_args = dispatched[0][1]
     assert first_args[0] == payload["batch_id"]
     assert batch_map[payload["batch_id"]].get("is_album_download") is True
+    # Both dispatched batches are album bundles.
+    for _fn, args, _kw in dispatched:
+        assert batch_map[args[0]].get("is_album_download") is True
 
     # Second sub-batch gets a fresh uuid; its row exists in batch_map.
-    second_args, _ = master_calls[1]
+    second_args = dispatched[1][1]
     assert second_args[0] != payload["batch_id"]
     assert second_args[0] in batch_map
-    assert batch_map[second_args[0]].get("is_album_download") is True
 
     # Track counts across the two sub-batches: 2 each at threshold=2.
-    counts = sorted(len(args[2]) for args, _ in master_calls)
+    counts = sorted(len(args[2]) for _fn, args, _kw in dispatched)
     assert counts == [2, 2]
 
     # Both sub-batches carry album context populated from spotify_data.
     album_names = {
         batch_map[args[0]]["album_context"]["name"]
-        for args, _ in master_calls
+        for _fn, args, _kw in dispatched
     }
     assert album_names == {"Album One", "Album Two"}
 

web_server.py

@@ -15155,6 +15155,7 @@ def start_wishlist_missing_downloads():
             download_batches=download_batches,
             tasks_lock=tasks_lock,
             missing_download_executor=missing_download_executor,
+            album_bundle_executor=album_bundle_executor,
             run_full_missing_tracks_process=_run_full_missing_tracks_process,
             get_batch_max_concurrent=_get_batch_max_concurrent,
             add_activity_item=add_activity_item,