Manual wishlist run: also split into per-album sub-batches

The Phase-1 fix (commit c3b88e69) only extended the per-album
bundle dispatch to ``process_wishlist_automatically``. The manual
"Run Wishlist Now" path goes through
``_prepare_and_run_manual_wishlist_batch`` instead, so the
behavior didn't change for users who triggered downloads from the
Wishlist tab UI — they still saw N per-track Soulseek searches
when N missing tracks all came from one album.

Caught in a real-app test: user added Katy Perry's PRISM (Deluxe)
to the wishlist + clicked "Download Wishlist" → app log shows
``_prepare_and_run_manual_wishlist_batch:421`` running a single
batch with 16 tracks + per-track searches firing one by one
("katy perry prism deluxe legendary lovers", "katy perry prism
deluxe roar", etc.), no album-bundle dispatch.

Fix:

- ``_prepare_and_run_manual_wishlist_batch`` now runs the same
  ``group_wishlist_tracks_by_album`` helper after filtering. For
  each detected album, it builds a sub-batch with
  ``is_album_download=True`` + populated album/artist context.
  Residual tracks (no resolvable album metadata) land in a single
  per-track residual batch.
- The first sub-batch re-uses the caller-allocated ``batch_id``
  so the frontend's existing poll against it keeps working;
  additional sub-batches get fresh ids materialized into
  ``download_batches`` so they show up in the Downloads view.
- Sub-batches dispatch serially — each ``run_full_missing_tracks_process``
  call blocks until the album-bundle staging + per-track tasks
  complete before the next album's bundle search fires.

New test ``test_manual_wishlist_splits_into_per_album_sub_batches``
pins the contract — multi-album wishlist content with
nested-spotify_data shape produces N master-worker calls (one per
album), each batch carries the album_context, first sub-batch
re-uses the original batch_id. 106 wishlist tests + 1099 across
the broader suite green.

Adding 16 Katy Perry PRISM tracks to wishlist + clicking download
should now fire ONE slskd album-bundle search for the release
instead of 16 individual searches.

core/wishlist/processing.py

@@ -469,15 +469,108 @@ def _prepare_and_run_manual_wishlist_batch(
         for i, track in enumerate(wishlist_tracks):
             track['_original_index'] = i
 
-        # Update batch with the real track count now that filtering is done
-        with runtime.tasks_lock:
-            if batch_id in runtime.download_batches:
-                runtime.download_batches[batch_id]['analysis_total'] = len(wishlist_tracks)
-
         runtime.add_activity_item("", "Wishlist Download Started", f"{len(wishlist_tracks)} tracks", "Now")
 
-        logger.info(f"Starting wishlist batch {batch_id} with {len(wishlist_tracks)} tracks")
+        # Try to split into per-album sub-batches so each album fires
-        runtime.run_full_missing_tracks_process(batch_id, "wishlist", wishlist_tracks)
+        # ONE slskd / torrent / usenet album-bundle search (gates on
+        # ``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)
+
+        # 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
+        # fresh batch_ids for the rest.
+        payloads[0]['batch_id'] = batch_id
+        for payload in payloads[1:]:
+            payload['batch_id'] = str(uuid.uuid4())
+
+        # Materialize each sub-batch's row state up-front so the
+        # frontend's polling can see them all under the original
+        # batch's flow.
+        with runtime.tasks_lock:
+            if batch_id in runtime.download_batches:
+                # Re-purpose the existing row for the first payload.
+                first = payloads[0]
+                runtime.download_batches[batch_id]['analysis_total'] = len(first['tracks'])
+                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'],
+                }
+
+        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}")

tests/wishlist/test_manual_download.py

@@ -232,6 +232,80 @@ def test_start_manual_wishlist_download_batch_does_not_run_library_cleanup():
     assert activity_calls == [("", "Wishlist Download Started", "2 tracks", "Now")]
 
 
+def test_manual_wishlist_splits_into_per_album_sub_batches():
+    """Manual wishlist run with multi-album content splits into one
+    sub-batch per album. Each sub-batch flips
+    ``is_album_download=True`` + populates album/artist context so
+    the slskd / Prowlarr album-bundle dispatch engages.
+
+    Pinned to verify the manual path matches the auto path's
+    behavior — the user's first real-world test hit the manual
+    flow, not the auto flow."""
+    runtime, _service, _db, executor, _logger, _activity, batch_map, master_calls = _build_runtime(
+        tracks=[
+            {
+                "id": "trk-a1",
+                "spotify_track_id": "trk-a1",
+                "name": "Song A1",
+                "artists": [{"name": "Artist 1"}],
+                "spotify_data": {
+                    "album": {"id": "alb1", "name": "Album One", "album_type": "album"},
+                    "artists": [{"name": "Artist 1"}],
+                },
+            },
+            {
+                "id": "trk-a2",
+                "spotify_track_id": "trk-a2",
+                "name": "Song A2",
+                "artists": [{"name": "Artist 1"}],
+                "spotify_data": {
+                    "album": {"id": "alb1", "name": "Album One", "album_type": "album"},
+                    "artists": [{"name": "Artist 1"}],
+                },
+            },
+            {
+                "id": "trk-b1",
+                "spotify_track_id": "trk-b1",
+                "name": "Song B1",
+                "artists": [{"name": "Artist 2"}],
+                "spotify_data": {
+                    "album": {"id": "alb2", "name": "Album Two", "album_type": "album"},
+                    "artists": [{"name": "Artist 2"}],
+                },
+            },
+        ]
+    )
+
+    payload, status = processing.start_manual_wishlist_download_batch(runtime)
+    assert status == 200
+    _run_submitted_bg_job(executor)
+
+    # Two album groups → two master-worker calls.
+    assert len(master_calls) == 2
+
+    # First sub-batch uses the caller-allocated batch_id.
+    first_args, _ = master_calls[0]
+    assert first_args[0] == payload["batch_id"]
+    assert batch_map[payload["batch_id"]].get("is_album_download") is True
+
+    # Second sub-batch gets a fresh uuid; its row exists in batch_map.
+    second_args, _ = master_calls[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 sum to the wishlist total.
+    counts = sorted(len(args[2]) for args, _ in master_calls)
+    assert counts == [1, 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
+    }
+    assert album_names == {"Album One", "Album Two"}
+
+
 def test_bg_job_marks_batch_complete_when_wishlist_genuinely_empty():
     """If the wishlist is empty before the manual click, the bg job marks the batch complete."""
     runtime, _service, _db, executor, _logger, _activity, batch_map, master_calls = _build_runtime(