Wishlist: reify run id + gate cycle toggle on last-sibling completion

Phase 1c.2.1 splits each wishlist invocation into per-album sub-
batches so the album-bundle dispatch can engage once per album.
Side effect: the completion handler ``finalize_auto_wishlist_completion``
ran end-of-run logic (cycle toggle + state reset + automation
event emit) once per BATCH, so a 2-album run fired the cycle
toggle twice + emitted two ``wishlist_processing_completed``
events. The cycle landed at the right value either way but the
state machine had become per-batch instead of per-run.

Fix: reify "wishlist run" as a first-class concept via a shared
``wishlist_run_id`` UUID. Generated once per wishlist invocation
in both the auto- and manual-wishlist paths, stamped on every
sub-batch row in ``download_batches``.

``finalize_auto_wishlist_completion`` now reads the completing
batch's ``wishlist_run_id`` and, when present, scans
``download_batches`` for siblings still in pre-terminal phases.
If any sibling is still active, the per-batch summary records
but the cycle toggle + state reset + automation emit are
deferred. Only the last completing sibling fires the run-level
finalization. Legacy single-batch runs (no run_id field) keep
their toggle-immediately behavior — back-compat by absence.

The run_id also lays groundwork for frontend grouping (one
logical row in the Downloads view per wishlist run instead of N
sibling rows), but that UX work is deferred.

3 new tests in ``test_processing.py`` pin: defer-when-siblings-
active, toggle-when-last-sibling-done, back-compat-without-run_id.
1 new assertion in ``test_automation.py`` confirms all sub-batches
of one auto-wishlist invocation share the same run_id. 309 tests
across wishlist + automation suites green.

Notes: dispatch concurrency unchanged — sub-batches still run via
the shared download worker pool. Slskd serializes per-uploader at
its own layer (same uploader = automatic queue, different
uploaders = legit parallel), so SoulSync-side serial enforcement
would duplicate work the right layer already handles.

core/wishlist/processing.py

@@ -183,6 +183,34 @@ def build_wishlist_source_context(batch: Dict[str, Any], current_time: datetime
     return context
 
 
+def _wishlist_run_has_siblings_still_active(
+    download_batches: Dict[str, Dict[str, Any]],
+    run_id: str,
+    completing_batch_id: str,
+) -> bool:
+    """Return True if any sibling batch sharing ``run_id`` is still
+    pre-terminal.
+
+    Used by ``finalize_auto_wishlist_completion`` to gate the run-
+    level cycle toggle. The caller already holds ``tasks_lock``.
+
+    The completing batch may or may not have its phase flipped to
+    'complete' yet by the time we land here; either way we skip it
+    in the sibling scan since we're handling its completion now."""
+    terminal_phases = {'complete', 'error', 'cancelled'}
+    for sibling_id, sibling in download_batches.items():
+        if sibling_id == completing_batch_id:
+            continue
+        if not isinstance(sibling, dict):
+            continue
+        if sibling.get('wishlist_run_id') != run_id:
+            continue
+        if sibling.get('phase') in terminal_phases:
+            continue
+        return True
+    return False
+
+
 def finalize_auto_wishlist_completion(
     batch_id: str,
     completion_summary: Dict[str, Any],
@@ -195,7 +223,17 @@ def finalize_auto_wishlist_completion(
     db_factory: Callable[[], Any],
     logger=logger,
 ) -> Dict[str, Any]:
-    """Finalize auto wishlist processing after a batch finishes."""
+    """Finalize auto wishlist processing after a batch finishes.
+
+    For wishlist runs that split into multiple sub-batches (Phase
+    1c.2.1: per-album bundle dispatch), the cycle toggle + state
+    reset only fire when the LAST sibling sub-batch of the same
+    ``wishlist_run_id`` completes. Earlier completions just record
+    their per-batch summary and return without toggling.
+
+    Back-compat: legacy single-batch runs (no ``wishlist_run_id``
+    field on the batch) keep the original toggle-immediately
+    behavior — the gate treats a missing run_id as "lone batch"."""
     tracks_added = completion_summary.get('tracks_added', 0)
     total_failed = completion_summary.get('total_failed', 0)
     logger.error(
@@ -205,6 +243,22 @@ def finalize_auto_wishlist_completion(
     if tracks_added > 0:
         add_activity_item("", "Wishlist Updated", f"{tracks_added} failed tracks added to wishlist", "Now")
 
+    # Run-level gate: if siblings of the same wishlist run are still
+    # active, defer cycle toggle + state reset until they finish.
+    with tasks_lock:
+        run_id = ''
+        if batch_id in download_batches:
+            run_id = download_batches[batch_id].get('wishlist_run_id') or ''
+        siblings_active = bool(run_id) and _wishlist_run_has_siblings_still_active(
+            download_batches, run_id, batch_id,
+        )
+    if siblings_active:
+        logger.info(
+            f"[Auto-Wishlist] Sub-batch {batch_id[:8]} done; waiting on sibling sub-batches "
+            f"of run {run_id[:8]} before toggling cycle"
+        )
+        return completion_summary
+
     try:
         with tasks_lock:
             if batch_id in download_batches:
@@ -517,6 +571,13 @@ def _prepare_and_run_manual_wishlist_batch(
         for payload in payloads[1:]:
             payload['batch_id'] = str(uuid.uuid4())
 
+        # Reify "wishlist run" — one shared id stamped on every sub-
+        # batch this manual invocation produces. Mirrors the auto
+        # path. Note manual wishlist completion currently doesn't
+        # toggle the cycle (only auto does), but the id is set anyway
+        # so future code + UI grouping have a consistent hook.
+        wishlist_run_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.
@@ -525,6 +586,7 @@ def _prepare_and_run_manual_wishlist_batch(
                 # Re-purpose the existing row for the first payload.
                 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']
@@ -549,6 +611,7 @@ def _prepare_and_run_manual_wishlist_batch(
                     '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(
@@ -747,6 +810,12 @@ def process_wishlist_automatically(runtime: WishlistAutoProcessingRuntime, autom
                 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())
+
                 if grouping and grouping.album_groups:
                     for album_idx, group in enumerate(grouping.album_groups):
                         album_batch_id = str(uuid.uuid4())
@@ -779,11 +848,12 @@ def process_wishlist_automatically(runtime: WishlistAutoProcessingRuntime, autom
                                 '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}"
+                            f"({len(group.tracks)} tracks) → {album_batch_id} [run {wishlist_run_id[:8]}]"
                         )
                         _submitted_batches.append(album_batch_id)
                         runtime.missing_download_executor.submit(
@@ -818,6 +888,7 @@ def process_wishlist_automatically(runtime: WishlistAutoProcessingRuntime, autom
                             '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(
@@ -826,7 +897,8 @@ def process_wishlist_automatically(runtime: WishlistAutoProcessingRuntime, autom
                     )
                     logger.info(
                         f"Starting wishlist residual batch {batch_id} with {len(residual_tracks)} tracks "
-                        f"({'singles' if current_cycle == 'singles' else 'unbucketed albums'})"
+                        f"({'singles' if current_cycle == 'singles' else 'unbucketed albums'}) "
+                        f"[run {wishlist_run_id[:8]}]"
                     )
 
                 _summary_parts: list[str] = []

tests/wishlist/test_automation.py

@@ -292,6 +292,13 @@ def test_wishlist_albums_cycle_splits_into_per_album_batches():
     track_counts = sorted(len(tracks) for tracks in submitted_track_lists)
     assert track_counts == [1, 2]
 
+    # All sub-batches of one wishlist invocation share a single
+    # ``wishlist_run_id`` so the completion handler can gate the
+    # cycle toggle on "all siblings done".
+    run_ids = {batch.get("wishlist_run_id") for batch in batch_map.values()}
+    assert len(run_ids) == 1
+    assert next(iter(run_ids))  # non-empty string
+
 
 def test_wishlist_albums_cycle_residual_for_orphan_tracks():
     """Tracks without resolvable album metadata fall to the classic

tests/wishlist/test_processing.py

@@ -182,6 +182,111 @@ def test_recover_uncaptured_failed_tracks_builds_entries():
     assert failed[0]["failure_reason"] == "boom"
 
 
+def test_finalize_auto_wishlist_completion_defers_toggle_when_siblings_active():
+    """When the completing batch shares a ``wishlist_run_id`` with
+    siblings still in pre-terminal phases, finalize must NOT toggle
+    the cycle yet — that only happens when the LAST sibling done.
+    Pinned to prevent the regression where every sub-batch's
+    completion fired its own cycle toggle (Phase 1c.2.1 split path)."""
+    db = _FakeDB()
+    automation_engine = _FakeAutomationEngine()
+    resets = []
+    activities = []
+    summary = {"tracks_added": 1, "total_failed": 1, "errors": 0}
+
+    # Two sub-batches share the same run id. The first finishes,
+    # the second is still 'analysis'.
+    download_batches = {
+        "batch-A": {"current_cycle": "albums", "wishlist_run_id": "run-1", "phase": "complete"},
+        "batch-B": {"current_cycle": "albums", "wishlist_run_id": "run-1", "phase": "analysis"},
+    }
+
+    processing.finalize_auto_wishlist_completion(
+        "batch-A",
+        summary,
+        download_batches=download_batches,
+        tasks_lock=_FakeLock(),
+        reset_processing_state=lambda: resets.append(True),
+        add_activity_item=lambda *args: activities.append(args),
+        automation_engine=automation_engine,
+        db_factory=lambda: db,
+        logger=_FakeLogger(),
+    )
+
+    # Activity log still fires (it's a per-batch record), but cycle
+    # toggle + state reset + automation emit are deferred.
+    assert activities == [("", "Wishlist Updated", "1 failed tracks added to wishlist", "Now")]
+    assert resets == []  # NOT reset yet — siblings still active
+    assert automation_engine.events == []  # NOT emitted yet
+    assert db.connection.cursor_obj.calls == []  # DB cycle-toggle NOT written
+
+
+def test_finalize_auto_wishlist_completion_toggles_when_last_sibling_done():
+    """When all siblings of the same run are in terminal phases (or
+    don't exist), the completing batch IS the last → cycle toggles
+    + state resets + automation event fires."""
+    db = _FakeDB()
+    automation_engine = _FakeAutomationEngine()
+    resets = []
+    summary = {"tracks_added": 1, "total_failed": 1, "errors": 0}
+
+    download_batches = {
+        # Both siblings already terminal — current batch is the last.
+        "batch-A": {"current_cycle": "albums", "wishlist_run_id": "run-1", "phase": "complete"},
+        "batch-B": {"current_cycle": "albums", "wishlist_run_id": "run-1", "phase": "complete"},
+        "batch-C": {"current_cycle": "albums", "wishlist_run_id": "run-1", "phase": "analysis"},  # the completing one
+    }
+
+    processing.finalize_auto_wishlist_completion(
+        "batch-C",
+        summary,
+        download_batches=download_batches,
+        tasks_lock=_FakeLock(),
+        reset_processing_state=lambda: resets.append(True),
+        add_activity_item=lambda *_a: None,
+        automation_engine=automation_engine,
+        db_factory=lambda: db,
+        logger=_FakeLogger(),
+    )
+
+    assert resets == [True]
+    assert db.connection.committed is True
+    assert db.connection.cursor_obj.calls[0][1] == ("singles",)
+    assert automation_engine.events  # event emitted
+
+
+def test_finalize_auto_wishlist_completion_legacy_no_run_id_toggles_immediately():
+    """Back-compat: a batch with NO ``wishlist_run_id`` (legacy
+    single-batch run from before Phase 1c.2.1) should keep firing
+    the toggle on its own completion regardless of any unrelated
+    batches in the dict."""
+    db = _FakeDB()
+    automation_engine = _FakeAutomationEngine()
+    resets = []
+    summary = {"tracks_added": 0, "total_failed": 0, "errors": 0}
+
+    download_batches = {
+        "batch-legacy": {"current_cycle": "albums"},  # no wishlist_run_id
+        # Even with another unrelated batch active, legacy should toggle.
+        "unrelated": {"current_cycle": "singles", "phase": "analysis"},
+    }
+
+    processing.finalize_auto_wishlist_completion(
+        "batch-legacy",
+        summary,
+        download_batches=download_batches,
+        tasks_lock=_FakeLock(),
+        reset_processing_state=lambda: resets.append(True),
+        add_activity_item=lambda *_a: None,
+        automation_engine=automation_engine,
+        db_factory=lambda: db,
+        logger=_FakeLogger(),
+    )
+
+    assert resets == [True]
+    assert db.connection.cursor_obj.calls[0][1] == ("singles",)
+
+
 def test_finalize_auto_wishlist_completion_toggles_cycle_and_resets_state():
     db = _FakeDB()
     automation_engine = _FakeAutomationEngine()

webui/static/helper.js

@@ -3423,6 +3423,7 @@ const WHATS_NEW = {
         { title: 'SoulSync Discovery Sync tab', desc: 'last of the unified-tab trio. surfaces your personalized SoulSync Discovery playlists (decade mixes, hidden gems, popular picks, daily mixes, discovery shuffle, etc.) on the Sync page. clicking a card regenerates the playlist + mirrors it under a stable synthetic id, so the same mirror updates in place every Auto-Sync refresh. tracks come out already matched against Spotify / iTunes / Deezer so there\'s no discovery hop — straight to download / sync.', page: 'sync' },
         { title: 'Fix: album-bundle downloads all landing as track 1', desc: 'soulseek album-bundle downloads (and any other untagged release-staging path) were importing every track with track_number=1. the staging-file reader was using the auto-import\'s filename extractor that defaults to 1 when no NN- prefix is present in the filename — so for albums like Ryoto\'s "Cha-La Head-Cha-La" where slskd hands you bare titles, every file got "track 1" stamped on it. now the staging path uses a strict extractor that returns 0 when it can\'t see an explicit prefix, so the downstream resolver correctly falls through to the authoritative Spotify metadata and the right track numbers land in the library.' },
         { title: 'Wishlist albums-cycle: one album-bundle search per album', desc: 'auto-wishlist runs in two cycles — albums + singles. previously the albums cycle dumped every missing track from every album into one big batch and ran a separate per-track Soulseek / Prowlarr search for each (~50 searches for a typical scan). now the albums cycle splits the wishlist into per-album sub-batches at submission time, and each one engages the existing slskd / torrent / usenet album-bundle release-first flow with that album\'s context. tracks without resolvable album metadata stay on the classic per-track residual batch so nothing falls off. singles cycle is unchanged.' },
+        { title: 'Wishlist cycle toggles once per run, not per sub-batch', desc: 'follow-up to the per-album split above. with N sub-batches per wishlist run, the cycle toggle (albums → singles or vice versa) was firing once per sub-batch as each one completed — so the cycle could flip mid-run and the automation completion event got emitted multiple times. each wishlist invocation now stamps every sub-batch with a shared run id; the completion handler waits until the LAST sibling finishes before toggling the cycle + emitting the run-complete event. cleaner state machine, future-proofs frontend grouping of sub-batches under one logical run.' },
     ],
     '2.6.2': [
         { date: 'May 24, 2026 — 2.6.2 release' },