Wishlist: unify batch-row construction into make_wishlist_batch_row

The auto and manual wishlist flows each built the same ~20-field
download_batches row in separate places (auto album, auto residual, manual
placeholder, manual sub-batches) — four near-identical literals that could (and
did) drift apart, producing subtly different batch shapes between the flows.

Extract make_wishlist_batch_row() as the single source of truth: it emits the
consistent core field set, with the genuinely per-flow differences as explicit
arguments — initial phase ('queued' for auto / 'analysis' for manual), the
auto-only auto_initiated/auto_processing_timestamp/current_cycle via
extra_fields, and album-vs-residual contexts. All four sites now go through it,
so every wishlist batch has an IDENTICAL shape (this also removes the field
drift that confused the modal-hydration code).

Deliberately NOT unified — and left explicit in each caller, per the
'don't cargo-cult genuinely-different code' principle: the grouping decision
(auto groups only on the albums cycle), batch-id allocation (manual reuses the
caller's placeholder id for the first sub-batch), and dispatch (auto
parallel-submits album batches to the dedicated pool + residual to the shared
pool; manual runs them serially on one thread). Those are real behavioral
differences, not duplication.

Behavior-preserving: verified safe to normalize the row shape (grep confirmed
every reader uses .get() with defaults, no key-presence checks). The existing
auto (test_automation.py) and manual (test_manual_download.py) characterization
suites stay green = differential proof of identical behavior. Adds
test_batch_factory.py (core fields, album/residual, extra_fields, no shared
mutable state, consistent key shape). 131 wishlist tests pass.

core/wishlist/processing.py

@@ -100,6 +100,58 @@ 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 add_cancelled_tracks_to_failed_tracks(
     batch: Dict[str, Any],
     download_tasks: Dict[str, Dict[str, Any]],
@@ -455,24 +507,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,
@@ -631,26 +675,18 @@ def _prepare_and_run_manual_wishlist_batch(
                     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']] = {
+                runtime.download_batches[payload['batch_id']] = make_wishlist_batch_row(
-                    'phase': 'analysis',
+                    playlist_id='wishlist',
-                    'playlist_id': 'wishlist',
+                    playlist_name=payload['display_name'],
-                    'playlist_name': payload['display_name'],
+                    track_count=len(payload['tracks']),
-                    'queue': [],
+                    max_concurrent=runtime.get_batch_max_concurrent(),
-                    'active_count': 0,
+                    profile_id=runtime.profile_id,
-                    'max_concurrent': runtime.get_batch_max_concurrent(),
+                    phase='analysis',
-                    'queue_index': 0,
+                    run_id=wishlist_run_id,
-                    'analysis_total': len(payload['tracks']),
+                    is_album=bool(payload['is_album']),
-                    'analysis_processed': 0,
+                    album_context=payload['album_context'],
-                    'analysis_results': [],
+                    artist_context=payload['artist_context'],
-                    '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) "
@@ -864,45 +900,29 @@ def process_wishlist_automatically(runtime: WishlistAutoProcessingRuntime, autom
                             f"Wishlist (Auto - Album: {group.album_context.get('name', 'Unknown')})"
                         )
                         with runtime.tasks_lock:
-                            runtime.download_batches[album_batch_id] = {
+                            # ``queued`` (not ``analysis``) so a run with N > pool
-                                # ``queued`` until the master worker
+                            # sub-batches doesn't render all N as "analyzing" while
-                                # picks the batch up from the
+                            # only the pool's worth actually run; the master worker
-                                # ``missing_download_executor`` pool
+                            # flips it to ``analysis`` when it picks the batch up.
-                                # (max_workers=3 by default). The worker
+                            # is_album_download + contexts route it through the
-                                # flips phase to ``analysis`` as its
+                            # album-bundle search instead of per-track.
-                                # first action — see
+                            runtime.download_batches[album_batch_id] = make_wishlist_batch_row(
-                                # ``core/downloads/master.py:328``.
+                                playlist_id=playlist_id,
-                                # Pre-fix the row was created with
+                                playlist_name=album_batch_name,
-                                # ``analysis`` directly, so a wishlist
+                                track_count=len(group.tracks),
-                                # run with N > 3 sub-batches looked like
+                                max_concurrent=runtime.get_batch_max_concurrent(),
-                                # all N were working when really only
+                                profile_id=runtime.profile_id,
-                                # 3 were running.
+                                phase='queued',
-                                'phase': 'queued',
+                                run_id=wishlist_run_id,
-                                'playlist_id': playlist_id,
+                                is_album=True,
-                                'playlist_name': album_batch_name,
+                                album_context=group.album_context,
-                                'queue': [],
+                                artist_context=group.artist_context,
-                                'active_count': 0,
+                                extra_fields={
-                                'max_concurrent': runtime.get_batch_max_concurrent(),
+                                    'auto_initiated': True,
-                                'queue_index': 0,
+                                    'auto_processing_timestamp': runtime.current_time_fn(),
-                                'analysis_total': len(group.tracks),
+                                    'current_cycle': current_cycle,
-                                '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')}' "
@@ -931,28 +951,23 @@ def process_wishlist_automatically(runtime: WishlistAutoProcessingRuntime, autom
                     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
-                            # See album sub-batch above — ``queued``
+                        # worker picks it up. Residual = classic per-track flow
-                            # until the master worker picks it up.
+                        # (is_album_download defaults False).
-                            'phase': 'queued',
+                        runtime.download_batches[batch_id] = make_wishlist_batch_row(
-                            'playlist_id': playlist_id,
+                            playlist_id=playlist_id,
-                            'playlist_name': playlist_name,
+                            playlist_name=playlist_name,
-                            'queue': [],
+                            track_count=len(residual_tracks),
-                            'active_count': 0,
+                            max_concurrent=runtime.get_batch_max_concurrent(),
-                            'max_concurrent': runtime.get_batch_max_concurrent(),
+                            profile_id=runtime.profile_id,
-                            'queue_index': 0,
+                            phase='queued',
-                            'analysis_total': len(residual_tracks),
+                            run_id=wishlist_run_id,
-                            'analysis_processed': 0,
+                            extra_fields={
-                            'analysis_results': [],
+                                'auto_initiated': True,
-                            'permanently_failed_tracks': [],
+                                'auto_processing_timestamp': runtime.current_time_fn(),
-                            'cancelled_tracks': set(),
+                                'current_cycle': current_cycle,
-                            '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,

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'}