Merge pull request #614 from Nezreka/feature/personalized-playlist-pipeline

Feature/personalized playlist pipeline

core/artists/map.py

@@ -20,10 +20,14 @@ logger = logging.getLogger(__name__)
 
 
 def get_current_profile_id() -> int:
-    """Mirror of web_server.get_current_profile_id — uses Flask g."""
+    """Mirror of web_server.get_current_profile_id — uses Flask g.
+
+    Catches RuntimeError too because reading `g` outside a request
+    context raises that (not AttributeError) — happens when this is
+    called from background threads (sync, automation, scanners)."""
     try:
         return g.profile_id
-    except AttributeError:
+    except (AttributeError, RuntimeError):
         return 1
 
 

core/automation/blocks.py

@@ -146,6 +146,15 @@ ACTIONS: list[dict] = [
          {"key": "all", "type": "checkbox", "label": "Process all mirrored playlists", "default": False},
          {"key": "skip_wishlist", "type": "checkbox", "label": "Skip wishlist processing", "default": False},
      ]},
+    {"type": "personalized_pipeline", "label": "Personalized Playlist Pipeline", "icon": "sparkles",
+     "description": "Sync personalized / discover-page playlists (Hidden Gems, Time Machine, Fresh Tape, etc.) to your media server + queue missing tracks for download.",
+     "available": True,
+     "config_fields": [
+         {"key": "kinds", "type": "personalized_playlist_select", "label": "Playlists to sync",
+          "description": "Multi-select: Hidden Gems, Discovery Shuffle, Time Machine (per decade), Genre playlists, Fresh Tape, The Archives, Seasonal Mix (per season)"},
+         {"key": "refresh_first", "type": "checkbox", "label": "Refresh playlists before sync (regenerate snapshots)", "default": False},
+         {"key": "skip_wishlist", "type": "checkbox", "label": "Skip wishlist processing", "default": False},
+     ]},
     {"type": "notify_only", "label": "Notify Only", "icon": "bell", "description": "No action — just send notification", "available": True},
     # Phase 3 actions
     {"type": "start_database_update", "label": "Update Database", "icon": "database",

core/automation/deps.py

@@ -138,3 +138,9 @@ class AutomationDeps:
     # the engine's progress callback hooks). ---
     init_automation_progress: Callable[..., Any]
     record_progress_history: Callable[..., Any]
+
+    # --- Personalized playlist pipeline ---
+    # Lazy builder so the pipeline handler can construct a fresh
+    # PersonalizedPlaylistManager per run (cheap accessors inside,
+    # no caching needed yet).
+    build_personalized_manager: Callable[[], Any]

core/automation/handlers/__init__.py

@@ -17,6 +17,7 @@ from core.automation.handlers.refresh_mirrored import auto_refresh_mirrored
 from core.automation.handlers.sync_playlist import auto_sync_playlist
 from core.automation.handlers.discover_playlist import auto_discover_playlist
 from core.automation.handlers.playlist_pipeline import auto_playlist_pipeline
+from core.automation.handlers.personalized_pipeline import auto_personalized_pipeline
 from core.automation.handlers.database_update import auto_start_database_update, auto_deep_scan_library
 from core.automation.handlers.duplicate_cleaner import auto_run_duplicate_cleaner
 from core.automation.handlers.quality_scanner import auto_start_quality_scan
@@ -44,6 +45,7 @@ __all__ = [
     'auto_sync_playlist',
     'auto_discover_playlist',
     'auto_playlist_pipeline',
+    'auto_personalized_pipeline',
     'auto_start_database_update',
     'auto_deep_scan_library',
     'auto_run_duplicate_cleaner',

core/automation/handlers/_pipeline_shared.py

@@ -0,0 +1,201 @@
+"""Shared helpers between mirrored + personalized playlist pipelines.
+
+Both pipelines end in the same shape:
+1. SYNC each playlist to the active media server.
+2. WISHLIST: trigger the wishlist processor for missing tracks.
+
+The differing prefix (mirrored = REFRESH external sources + DISCOVER
+metadata; personalized = SNAPSHOT manager-backed playlists) is owned
+by each pipeline. This module owns the SYNC + WISHLIST tail so both
+pipelines stay consistent + DRY.
+"""
+
+from __future__ import annotations
+
+import time
+from typing import Any, Callable, Dict, List, Optional
+
+from core.automation.deps import AutomationDeps
+
+
+# Per-playlist sync poll cap (mirrored side already used this).
+_SYNC_PER_PLAYLIST_TIMEOUT_SECONDS = 600
+# Sync-status final-state markers.
+_SYNC_TERMINAL_STATUSES = ('finished', 'complete', 'error', 'failed')
+
+
+def run_sync_and_wishlist(
+    deps: AutomationDeps,
+    automation_id: Optional[str],
+    playlists: List[Dict[str, Any]],
+    *,
+    sync_one_fn: Callable[[Dict[str, Any]], Dict[str, Any]],
+    sync_id_for_fn: Callable[[Dict[str, Any]], str],
+    skip_wishlist: bool = False,
+    progress_start: int = 56,
+    progress_end: int = 85,
+    sync_phase_label: str = 'Phase: Syncing to server...',
+    sync_phase_start_log: str = 'Sync',
+    wishlist_phase_label: str = 'Phase: Processing wishlist...',
+    wishlist_phase_start_log: str = 'Wishlist',
+) -> Dict[str, int]:
+    """Run the SYNC + WISHLIST tail of a playlist pipeline.
+
+    The caller supplies:
+    - ``playlists``: list of playlist payload dicts. Each must have at
+      least a ``name`` key (used in progress logs). The shape beyond
+      ``name`` is opaque to the helper — ``sync_one_fn`` receives the
+      payload and returns a sync_result dict.
+    - ``sync_one_fn(payload) -> sync_result``: launches sync for one
+      playlist. Result dict must carry ``status`` ∈ ``('started',
+      'skipped', 'error')`` and may carry ``reason``.
+    - ``sync_id_for_fn(payload) -> str``: returns the sync-state key
+      the helper polls on (so we can wait for the background sync
+      thread to complete + read the matched_tracks count).
+
+    Returns ``{'synced': int, 'skipped': int, 'errors': int,
+    'wishlist_queued': int}`` so the caller can stitch it into its
+    final status.
+    """
+    deps.update_progress(
+        automation_id,
+        progress=progress_start,
+        phase=sync_phase_label,
+        log_line=sync_phase_start_log,
+        log_type='info',
+    )
+
+    total_synced = 0
+    total_skipped = 0
+    sync_errors = 0
+    sync_states = deps.get_sync_states()
+    n_playlists = max(1, len(playlists))
+    progress_span = max(1, progress_end - progress_start - 1)
+
+    for pl_idx, pl in enumerate(playlists):
+        pl_name = pl.get('name', '')
+        sync_result = sync_one_fn(pl)
+        sync_status = sync_result.get('status', '')
+
+        if sync_status == 'started':
+            sync_id = sync_id_for_fn(pl)
+            sync_poll_start = time.time()
+            while time.time() - sync_poll_start < _SYNC_PER_PLAYLIST_TIMEOUT_SECONDS:
+                if (sync_id in sync_states
+                        and sync_states[sync_id].get('status') in _SYNC_TERMINAL_STATUSES):
+                    break
+                time.sleep(2)
+                elapsed = int(time.time() - sync_poll_start)
+                sub_progress = progress_start + 1 + ((pl_idx + 1) / n_playlists) * progress_span
+                deps.update_progress(
+                    automation_id,
+                    progress=min(int(sub_progress), progress_end - 1),
+                    phase=f'{sync_phase_label.rstrip(".")} — "{pl_name}" ({elapsed}s)',
+                )
+
+            ss = sync_states.get(sync_id, {})
+            ss_result = ss.get('result', ss.get('progress', {}))
+            matched = ss_result.get('matched_tracks', 0) if isinstance(ss_result, dict) else 0
+            total_synced += int(matched) if matched else 0
+            deps.update_progress(
+                automation_id,
+                log_line=f'Synced "{pl_name}": {matched} tracks matched',
+                log_type='success',
+            )
+
+        elif sync_status == 'skipped':
+            total_skipped += 1
+            reason = sync_result.get('reason', 'unchanged')
+            deps.update_progress(
+                automation_id,
+                log_line=f'Skipped "{pl_name}": {reason}',
+                log_type='skip',
+            )
+        elif sync_status == 'error':
+            sync_errors += 1
+            deps.update_progress(
+                automation_id,
+                log_line=f'Sync error "{pl_name}": {sync_result.get("reason", "unknown")}',
+                log_type='error',
+            )
+
+    deps.update_progress(
+        automation_id,
+        progress=progress_end,
+        phase=f'{sync_phase_label.rstrip(".")} complete',
+        log_line=f'Sync done: {total_synced} matched, {total_skipped} skipped, {sync_errors} errors',
+        log_type='success' if sync_errors == 0 else 'warning',
+    )
+
+    wishlist_queued = run_wishlist_phase(
+        deps, automation_id,
+        skip=skip_wishlist,
+        progress_pct=progress_end + 1,
+        wishlist_phase_label=wishlist_phase_label,
+        wishlist_phase_start_log=wishlist_phase_start_log,
+    )
+
+    return {
+        'synced': total_synced,
+        'skipped': total_skipped,
+        'errors': sync_errors,
+        'wishlist_queued': wishlist_queued,
+    }
+
+
+def run_wishlist_phase(
+    deps: AutomationDeps,
+    automation_id: Optional[str],
+    *,
+    skip: bool,
+    progress_pct: int,
+    wishlist_phase_label: str = 'Phase: Processing wishlist...',
+    wishlist_phase_start_log: str = 'Wishlist',
+) -> int:
+    """Trigger the wishlist processor unless skipped or already running.
+
+    Returns 1 when the processor was triggered, 0 otherwise. Errors are
+    logged but never raised — wishlist failure should not abort the
+    pipeline."""
+    if skip:
+        deps.update_progress(
+            automation_id,
+            progress=progress_pct,
+            log_line=f'{wishlist_phase_start_log}: skipped (disabled)',
+            log_type='skip',
+        )
+        return 0
+
+    deps.update_progress(
+        automation_id,
+        progress=progress_pct,
+        phase=wishlist_phase_label,
+        log_line=wishlist_phase_start_log,
+        log_type='info',
+    )
+
+    try:
+        if not deps.is_wishlist_actually_processing():
+            deps.process_wishlist_automatically(automation_id=None)
+            deps.update_progress(
+                automation_id,
+                log_line='Wishlist processing triggered',
+                log_type='success',
+            )
+            return 1
+        deps.update_progress(
+            automation_id,
+            log_line='Wishlist already running — skipped',
+            log_type='skip',
+        )
+        return 0
+    except Exception as e:  # noqa: BLE001 — wishlist failure must never abort pipeline
+        deps.update_progress(
+            automation_id,
+            log_line=f'Wishlist error: {e}',
+            log_type='warning',
+        )
+        return 0
+
+
+__all__ = ['run_sync_and_wishlist', 'run_wishlist_phase']

core/automation/handlers/personalized_pipeline.py

@@ -0,0 +1,356 @@
+"""Personalized Playlist Pipeline automation handler.
+
+Sibling to ``auto_playlist_pipeline`` (mirrored). Where the mirrored
+pipeline runs REFRESH external sources → DISCOVER metadata → SYNC →
+WISHLIST, the personalized pipeline is simpler:
+
+    SNAPSHOT → SYNC → WISHLIST
+
+SNAPSHOT reads the persisted track list from
+``PersonalizedPlaylistManager``. When ``refresh_first=True`` (config),
+each playlist is refreshed BEFORE syncing — useful when the user
+wants the cron to capture a fresh-each-run view (e.g. "give me a new
+Hidden Gems set every night"). Default is to sync the existing
+snapshot, on the assumption the user / a separate cron has already
+refreshed when they wanted to.
+
+Config schema:
+    {
+        'kinds': [
+            {'kind': 'hidden_gems'},
+            {'kind': 'time_machine', 'variant': '1980s'},
+            {'kind': 'seasonal_mix', 'variant': 'halloween'},
+            ...
+        ],
+        'refresh_first': bool,    # default false
+        'skip_wishlist': bool,    # default false
+    }
+
+Each kind dict has at minimum ``kind``; ``variant`` is required for
+kinds that need it (time_machine, genre_playlist, daily_mix,
+seasonal_mix). Singleton kinds (hidden_gems, discovery_shuffle,
+popular_picks, fresh_tape, archives) ignore variant.
+
+Pipeline-running flag (``deps.state.pipeline_running``) is shared
+with the mirrored pipeline so the two can't overlap. (One sync
+queue, one wishlist worker — overlapping triggers would step on
+each other.)"""
+
+from __future__ import annotations
+
+import json
+import threading
+import time
+from typing import Any, Dict, List, Optional
+
+from core.automation.deps import AutomationDeps
+from core.automation.handlers._pipeline_shared import run_sync_and_wishlist
+
+
+# Sync state key prefix so personalized syncs don't collide with
+# mirrored ones (`auto_mirror_<id>`).
+_SYNC_ID_PREFIX = 'auto_personalized'
+
+
+def auto_personalized_pipeline(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
+    """Run SNAPSHOT → SYNC → WISHLIST for selected personalized playlists."""
+    deps.state.set_pipeline_running(True)
+    automation_id = config.get('_automation_id')
+    pipeline_start = time.time()
+
+    try:
+        kinds_config = config.get('kinds') or []
+        if not isinstance(kinds_config, list) or not kinds_config:
+            deps.state.set_pipeline_running(False)
+            return {
+                'status': 'error',
+                'error': 'No personalized playlist kinds selected',
+            }
+
+        refresh_first = bool(config.get('refresh_first', False))
+        skip_wishlist = bool(config.get('skip_wishlist', False))
+
+        manager = deps.build_personalized_manager()
+
+        deps.update_progress(
+            automation_id,
+            progress=2,
+            phase=f'Personalized pipeline: {len(kinds_config)} playlist(s)',
+            log_line=f'Starting pipeline for {len(kinds_config)} playlist(s)',
+            log_type='info',
+        )
+
+        # ── PHASE 1: SNAPSHOT (optionally refresh) ──────────────────
+        deps.update_progress(
+            automation_id,
+            progress=3,
+            phase='Phase 1/2: Loading snapshots...' if not refresh_first
+                  else 'Phase 1/2: Refreshing snapshots...',
+            log_line='Phase 1: Snapshot' + (' (with refresh)' if refresh_first else ''),
+            log_type='info',
+        )
+
+        profile_id = deps.get_current_profile_id()
+        playload_payloads = _build_payloads_for_kinds(
+            deps, manager, kinds_config, profile_id,
+            automation_id=automation_id,
+            refresh_first=refresh_first,
+        )
+
+        if not playload_payloads:
+            deps.state.set_pipeline_running(False)
+            deps.update_progress(
+                automation_id,
+                status='finished', progress=100,
+                phase='No playlists to sync',
+                log_line='No personalized playlists had tracks to sync',
+                log_type='warning',
+            )
+            return {
+                'status': 'completed',
+                '_manages_own_progress': True,
+                'playlists_synced': '0',
+                'tracks_synced': '0',
+                'duration_seconds': str(int(time.time() - pipeline_start)),
+            }
+
+        deps.update_progress(
+            automation_id,
+            progress=50,
+            phase='Phase 1/2: Snapshot complete',
+            log_line=f'Phase 1 done: {len(playload_payloads)} playlist(s) ready to sync',
+            log_type='success',
+        )
+
+        # ── PHASE 2: SYNC + WISHLIST (shared helper) ────────────────
+        sync_summary = run_sync_and_wishlist(
+            deps,
+            automation_id,
+            playload_payloads,
+            sync_one_fn=lambda pl: _sync_personalized_playlist(deps, pl),
+            sync_id_for_fn=lambda pl: pl['sync_id'],
+            skip_wishlist=skip_wishlist,
+            progress_start=51,
+            progress_end=90,
+            sync_phase_label='Phase 2/2: Syncing to server...',
+            sync_phase_start_log='Phase 2: Sync',
+            wishlist_phase_label='Phase 2/2: Processing wishlist...',
+            wishlist_phase_start_log='Wishlist',
+        )
+
+        # ── COMPLETE ────────────────────────────────────────────────
+        duration = int(time.time() - pipeline_start)
+        deps.update_progress(
+            automation_id,
+            status='finished', progress=100,
+            phase='Pipeline complete',
+            log_line=f'Personalized pipeline finished in {duration // 60}m {duration % 60}s',
+            log_type='success',
+        )
+
+        deps.state.set_pipeline_running(False)
+        return {
+            'status': 'completed',
+            '_manages_own_progress': True,
+            'playlists_synced': str(len(playload_payloads)),
+            'tracks_synced': str(sync_summary['synced']),
+            'sync_skipped': str(sync_summary['skipped']),
+            'wishlist_queued': str(sync_summary['wishlist_queued']),
+            'duration_seconds': str(duration),
+        }
+
+    except Exception as e:  # noqa: BLE001 — automation handlers must never raise into engine
+        deps.state.set_pipeline_running(False)
+        deps.update_progress(
+            automation_id,
+            status='error', progress=100,
+            phase='Pipeline error',
+            log_line=f'Personalized pipeline failed: {e}',
+            log_type='error',
+        )
+        return {'status': 'error', 'error': str(e), '_manages_own_progress': True}
+
+
+def _build_payloads_for_kinds(
+    deps: AutomationDeps,
+    manager: Any,
+    kinds_config: List[Dict[str, Any]],
+    profile_id: int,
+    *,
+    automation_id: Optional[str],
+    refresh_first: bool,
+) -> List[Dict[str, Any]]:
+    """Resolve each requested kind+variant into a sync-payload dict.
+
+    Each payload has: ``{'name', 'kind', 'variant', 'tracks_json',
+    'image_url', 'sync_id'}``. Playlists with no tracks (e.g. a
+    seasonal mix that hasn't been populated yet) are omitted from
+    the result so the sync loop doesn't waste time on empty pushes.
+    """
+    payloads: List[Dict[str, Any]] = []
+    for entry in kinds_config:
+        if not isinstance(entry, dict):
+            continue
+        kind = entry.get('kind')
+        variant = entry.get('variant') or ''
+        if not kind:
+            continue
+
+        try:
+            # Refresh when ANY of:
+            #   - explicit user flag (cron use case: regenerate each run)
+            #   - snapshot marked stale by upstream data refresher
+            #   - playlist was never generated yet (auto-created by
+            #     ensure_playlist; track_count=0, last_generated_at=NULL).
+            #     Without this branch, a first-run pipeline reads the
+            #     empty snapshot and silently skips — user picks a kind,
+            #     hits run, gets "No tracks to sync" with no clue why.
+            if refresh_first:
+                record = manager.refresh_playlist(kind, variant, profile_id)
+            else:
+                existing = manager.ensure_playlist(kind, variant, profile_id)
+                needs_first_gen = existing.last_generated_at is None
+                if existing.is_stale or needs_first_gen:
+                    record = manager.refresh_playlist(kind, variant, profile_id)
+                else:
+                    record = existing
+        except Exception as exc:  # noqa: BLE001 — log + continue with next kind
+            deps.update_progress(
+                automation_id,
+                log_line=f'Skipping {kind}{("/" + variant) if variant else ""}: {exc}',
+                log_type='warning',
+            )
+            continue
+
+        tracks = manager.get_playlist_tracks(record.id)
+        if not tracks:
+            deps.update_progress(
+                automation_id,
+                log_line=f'No tracks in {record.name} — skipping sync',
+                log_type='skip',
+            )
+            continue
+
+        tracks_json = [_track_to_sync_shape(t) for t in tracks]
+        payloads.append({
+            'name': record.name,
+            'kind': record.kind,
+            'variant': record.variant,
+            'tracks_json': tracks_json,
+            'image_url': '',  # personalized playlists don't have a cover image yet
+            'sync_id': f'{_SYNC_ID_PREFIX}_{record.kind}_{record.variant or "_"}',
+        })
+    return payloads
+
+
+def _track_to_sync_shape(track: Any) -> Dict[str, Any]:
+    """Convert a personalized.types.Track into the dict shape
+    `_run_sync_task` expects. Mirrors what the mirrored pipeline
+    builds from extra_data.matched_data, preserving enriched metadata
+    from personalized snapshots when available."""
+    primary_id = track.spotify_track_id or track.itunes_track_id or track.deezer_track_id or ''
+    rich_data = _coerce_track_data_json(getattr(track, 'track_data_json', None))
+
+    if not rich_data:
+        album = {'name': track.album_name or ''}
+        cover_url = getattr(track, 'album_cover_url', None)
+        if cover_url:
+            album['images'] = [{'url': cover_url}]
+        return {
+            'name': track.track_name,
+            'artists': [{'name': track.artist_name}],
+            'album': album,
+            'duration_ms': int(track.duration_ms or 0),
+            'id': primary_id,
+        }
+
+    payload = dict(rich_data)
+    cover_url = (
+        getattr(track, 'album_cover_url', None)
+        or payload.get('album_cover_url')
+        or payload.get('image_url')
+    )
+    payload['id'] = payload.get('id') or primary_id
+    payload['name'] = payload.get('name') or track.track_name
+    payload['artists'] = _normalize_artists(payload.get('artists'), track.artist_name)
+    payload['album'] = _normalize_album(payload.get('album'), track, cover_url=cover_url)
+    payload['duration_ms'] = int(payload.get('duration_ms') or track.duration_ms or 0)
+    if 'popularity' not in payload and getattr(track, 'popularity', None) is not None:
+        payload['popularity'] = int(track.popularity or 0)
+
+    if cover_url and not payload.get('image_url'):
+        payload['image_url'] = cover_url
+
+    return payload
+
+
+def _coerce_track_data_json(value: Any) -> Dict[str, Any]:
+    if isinstance(value, dict):
+        return value
+    if isinstance(value, str) and value.strip():
+        try:
+            loaded = json.loads(value)
+        except (TypeError, ValueError):
+            return {}
+        return loaded if isinstance(loaded, dict) else {}
+    return {}
+
+
+def _normalize_artists(artists: Any, fallback_artist: str) -> List[Dict[str, Any]]:
+    if not artists:
+        return [{'name': fallback_artist or 'Unknown Artist'}]
+    if isinstance(artists, list):
+        normalized = []
+        for artist in artists:
+            if isinstance(artist, dict):
+                normalized.append(artist if artist.get('name') else {'name': str(artist)})
+            elif isinstance(artist, str):
+                normalized.append({'name': artist})
+            else:
+                normalized.append({'name': str(artist)})
+        return normalized or [{'name': fallback_artist or 'Unknown Artist'}]
+    if isinstance(artists, dict):
+        return [artists if artists.get('name') else {'name': str(artists)}]
+    return [{'name': str(artists)}]
+
+
+def _normalize_album(album: Any, track: Any, cover_url: Optional[str] = None) -> Dict[str, Any]:
+    if isinstance(album, dict):
+        normalized = dict(album)
+    else:
+        normalized = {'name': str(album) if album else (track.album_name or '')}
+
+    normalized['name'] = normalized.get('name') or track.album_name or ''
+    images = normalized.get('images')
+    if not isinstance(images, list):
+        images = []
+    if cover_url and not images:
+        images = [{'url': cover_url}]
+    if images:
+        normalized['images'] = images
+    return normalized
+
+
+def _sync_personalized_playlist(deps: AutomationDeps, payload: Dict[str, Any]) -> Dict[str, Any]:
+    """Launch a personalized playlist sync via _run_sync_task on a
+    daemon thread + return immediately with status='started'.
+
+    Mirrors the mirrored ``auto_sync_playlist`` return contract so the
+    shared helper can poll on ``sync_states[sync_id]`` and aggregate
+    results identically."""
+    sync_id = payload['sync_id']
+    name = payload['name']
+    tracks_json = payload['tracks_json']
+    profile_id = deps.get_current_profile_id()
+
+    threading.Thread(
+        target=deps.run_sync_task,
+        args=(sync_id, name, tracks_json, None, profile_id, payload.get('image_url', '')),
+        daemon=True,
+        name=f'auto-personalized-{sync_id}',
+    ).start()
+    return {
+        'status': 'started',
+        'playlist_name': name,
+        '_manages_own_progress': True,
+    }

core/automation/handlers/playlist_pipeline.py

@@ -28,12 +28,12 @@ import time
 from typing import Any, Dict
 
 from core.automation.deps import AutomationDeps
+from core.automation.handlers._pipeline_shared import run_sync_and_wishlist
 from core.automation.handlers.refresh_mirrored import auto_refresh_mirrored
 from core.automation.handlers.sync_playlist import auto_sync_playlist
 
 
 # Per-playlist sync poll cap inside Phase 3.
-_SYNC_PER_PLAYLIST_TIMEOUT_SECONDS = 600
 # Discovery poll cap inside Phase 2.
 _DISCOVERY_TIMEOUT_SECONDS = 3600
 
@@ -165,124 +165,31 @@ def auto_playlist_pipeline(config: Dict[str, Any], deps: AutomationDeps) -> Dict
             log_type='success',
         )
 
-        # ── PHASE 3: SYNC ─────────────────────────────────────────────
-        deps.update_progress(
+        # ── PHASE 3 + 4: SYNC + WISHLIST (delegated to shared helper) ──
+        # Each mirrored playlist payload only needs `id` + `name` for
+        # the helper; `auto_sync_playlist` reads the rest from the
+        # mirrored DB by id.
+        sync_summary = run_sync_and_wishlist(
+            deps,
             automation_id,
-            progress=56,
-            phase='Phase 3/4: Syncing to server...',
-            log_line='Phase 3: Sync',
-            log_type='info',
+            [pl for pl in playlists if pl.get('id')],
+            sync_one_fn=lambda pl: auto_sync_playlist(
+                {'playlist_id': str(pl['id']), '_automation_id': None},
+                deps,
+            ),
+            sync_id_for_fn=lambda pl: f"auto_mirror_{pl['id']}",
+            skip_wishlist=skip_wishlist,
+            progress_start=56,
+            progress_end=85,
+            sync_phase_label='Phase 3/4: Syncing to server...',
+            sync_phase_start_log='Phase 3: Sync',
+            wishlist_phase_label='Phase 4/4: Processing wishlist...',
+            wishlist_phase_start_log='Phase 4: Wishlist',
         )
-
-        total_synced = 0
-        total_skipped = 0
-        sync_errors = 0
-        sync_states = deps.get_sync_states()
-
-        for pl_idx, pl in enumerate(playlists):
-            pl_id = pl.get('id')
-            if not pl_id:
-                continue
-
-            sync_config = {
-                'playlist_id': str(pl_id),
-                '_automation_id': None,  # Don't let sync handler hijack our progress.
-            }
-            sync_result = auto_sync_playlist(sync_config, deps)
-            sync_status = sync_result.get('status', '')
-
-            if sync_status == 'started':
-                # Sync launched a background thread — wait for it.
-                sync_id = f"auto_mirror_{pl_id}"
-                sync_poll_start = time.time()
-                while time.time() - sync_poll_start < _SYNC_PER_PLAYLIST_TIMEOUT_SECONDS:
-                    if (sync_id in sync_states
-                            and sync_states[sync_id].get('status')
-                            in ('finished', 'complete', 'error', 'failed')):
-                        break
-                    time.sleep(2)
-                    elapsed = int(time.time() - sync_poll_start)
-                    sub_progress = 56 + ((pl_idx + 1) / max(1, len(playlists))) * 29
-                    deps.update_progress(
-                        automation_id,
-                        progress=min(int(sub_progress), 84),
-                        phase=f'Phase 3/4: Syncing "{pl.get("name", "")}" ({elapsed}s)',
-                    )
-
-                # Check result.
-                ss = sync_states.get(sync_id, {})
-                ss_result = ss.get('result', ss.get('progress', {}))
-                matched = ss_result.get('matched_tracks', 0) if isinstance(ss_result, dict) else 0
-                total_synced += int(matched) if matched else 0
-                deps.update_progress(
-                    automation_id,
-                    log_line=f'Synced "{pl.get("name", "")}": {matched} tracks matched',
-                    log_type='success',
-                )
-
-            elif sync_status == 'skipped':
-                total_skipped += 1
-                reason = sync_result.get('reason', 'unchanged')
-                deps.update_progress(
-                    automation_id,
-                    log_line=f'Skipped "{pl.get("name", "")}": {reason}',
-                    log_type='skip',
-                )
-            elif sync_status == 'error':
-                sync_errors += 1
-                deps.update_progress(
-                    automation_id,
-                    log_line=f'Sync error "{pl.get("name", "")}": {sync_result.get("reason", "unknown")}',
-                    log_type='error',
-                )
-
-        deps.update_progress(
-            automation_id,
-            progress=85,
-            phase='Phase 3/4: Sync complete',
-            log_line=f'Phase 3 done: {total_synced} matched, {total_skipped} skipped, {sync_errors} errors',
-            log_type='success' if sync_errors == 0 else 'warning',
-        )
-
-        # ── PHASE 4: WISHLIST ─────────────────────────────────────────
-        wishlist_queued = 0
-        if not skip_wishlist:
-            deps.update_progress(
-                automation_id,
-                progress=86,
-                phase='Phase 4/4: Processing wishlist...',
-                log_line='Phase 4: Wishlist',
-                log_type='info',
-            )
-
-            try:
-                if not deps.is_wishlist_actually_processing():
-                    deps.process_wishlist_automatically(automation_id=None)
-                    deps.update_progress(
-                        automation_id,
-                        log_line='Wishlist processing triggered',
-                        log_type='success',
-                    )
-                    wishlist_queued = 1
-                else:
-                    deps.update_progress(
-                        automation_id,
-                        log_line='Wishlist already running — skipped',
-                        log_type='skip',
-                    )
-            except Exception as e:
-                deps.update_progress(
-                    automation_id,
-                    log_line=f'Wishlist error: {e}',
-                    log_type='warning',
-                )
-        else:
-            deps.update_progress(
-                automation_id,
-                progress=86,
-                log_line='Phase 4: Wishlist skipped (disabled)',
-                log_type='skip',
-            )
+        total_synced = sync_summary['synced']
+        total_skipped = sync_summary['skipped']
+        sync_errors = sync_summary['errors']
+        wishlist_queued = sync_summary['wishlist_queued']
 
         # ── COMPLETE ──────────────────────────────────────────────────
         duration = int(time.time() - pipeline_start)

core/automation/handlers/registration.py

@@ -15,6 +15,7 @@ from core.automation.handlers.refresh_mirrored import auto_refresh_mirrored
 from core.automation.handlers.sync_playlist import auto_sync_playlist
 from core.automation.handlers.discover_playlist import auto_discover_playlist
 from core.automation.handlers.playlist_pipeline import auto_playlist_pipeline
+from core.automation.handlers.personalized_pipeline import auto_personalized_pipeline
 from core.automation.handlers.database_update import (
     auto_start_database_update, auto_deep_scan_library,
 )
@@ -94,6 +95,14 @@ def register_all(deps: AutomationDeps) -> None:
         lambda config: auto_playlist_pipeline(config, deps),
         deps.state.is_pipeline_running,
     )
+    # Personalized pipeline shares the pipeline_running flag with the
+    # mirrored pipeline so the two can't overlap (single sync queue,
+    # single wishlist worker).
+    engine.register_action_handler(
+        'personalized_pipeline',
+        lambda config: auto_personalized_pipeline(config, deps),
+        deps.state.is_pipeline_running,
+    )
 
     # Database update + deep scan share the db_update_state guard —
     # only one operation can mutate that state at a time.

core/discovery/hero.py

@@ -17,10 +17,14 @@ logger = logging.getLogger(__name__)
 
 
 def get_current_profile_id() -> int:
-    """Mirror of web_server.get_current_profile_id — uses Flask g."""
+    """Mirror of web_server.get_current_profile_id — uses Flask g.
+
+    Catches RuntimeError too because reading `g` outside a request
+    context raises that (not AttributeError) — happens when this is
+    called from background threads (sync, automation, scanners)."""
     try:
         return g.profile_id
-    except AttributeError:
+    except (AttributeError, RuntimeError):
         return 1
 
 

core/personalized/api.py

@@ -62,23 +62,35 @@ def _track_to_dict(track: Track) -> Dict[str, Any]:
     }
 
 
-def list_kinds(registry: Optional[PlaylistKindRegistry] = None) -> Dict[str, Any]:
+def list_kinds(
+    registry: Optional[PlaylistKindRegistry] = None,
+    manager: Optional[PersonalizedPlaylistManager] = None,
+) -> Dict[str, Any]:
     """Return every registered playlist kind with metadata.
 
     UI uses this to render the "available playlists" picker. Each
-    kind reports whether it requires a variant and the resolved
-    variant set so the UI can render variant choices when relevant."""
+    kind reports whether it requires a variant; when a manager is
+    supplied AND the kind has a variant_resolver, the resolved
+    variant list is also included so the UI can render variant
+    checkboxes without a second round-trip per kind."""
     reg = registry or get_registry()
     out = []
     for spec in reg.all():
-        out.append({
+        entry = {
             'kind': spec.kind,
             'name_template': spec.name_template,
             'description': spec.description,
             'requires_variant': spec.requires_variant,
             'tags': list(spec.tags),
             'default_config': spec.default_config.to_json_dict(),
-        })
+            'variants': [],
+        }
+        if manager is not None and spec.variant_resolver is not None:
+            try:
+                entry['variants'] = list(spec.variant_resolver(manager.deps) or [])
+            except Exception:
+                entry['variants'] = []
+        out.append(entry)
     return {'success': True, 'kinds': out}
 
 

core/personalized/manager.py

@@ -102,7 +102,8 @@ class PersonalizedPlaylistManager:
                 """
                 SELECT id, profile_id, kind, variant, name, config_json,
                        track_count, last_generated_at, last_synced_at,
-                       last_generation_source, last_generation_error
+                       last_generation_source, last_generation_error,
+                       is_stale
                 FROM personalized_playlists
                 WHERE profile_id = ?
                 ORDER BY COALESCE(last_generated_at, created_at) DESC
@@ -231,6 +232,34 @@ class PersonalizedPlaylistManager:
             rows = cursor.fetchall()
         return [self._row_to_track(r) for r in rows]
 
+    # ─── snapshot freshness vs source data ───────────────────────────
+
+    def mark_kinds_stale(self, kinds: List[str], profile_id: Optional[int] = None) -> int:
+        """Flag every playlist row matching one of ``kinds`` as stale.
+
+        Called by upstream data refreshers (watchlist scan finishing
+        / Spotify enrichment worker re-pulling Release Radar / etc)
+        so pipelines auto-regenerate snapshots before the next sync
+        instead of pushing stale data to the media server.
+
+        Returns the number of rows touched. When ``profile_id`` is
+        None, flags rows across every profile.
+        """
+        if not kinds:
+            return 0
+        placeholders = ','.join('?' * len(kinds))
+        sql = f"UPDATE personalized_playlists SET is_stale = 1, updated_at = CURRENT_TIMESTAMP WHERE kind IN ({placeholders})"
+        params: List[Any] = list(kinds)
+        if profile_id is not None:
+            sql += " AND profile_id = ?"
+            params.append(profile_id)
+        with self.database._get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute(sql, params)
+            count = cursor.rowcount
+            conn.commit()
+        return count
+
     # ─── staleness history ───────────────────────────────────────────
 
     def recent_track_ids(self, profile_id: int, kind: str, days: int) -> List[str]:
@@ -294,6 +323,7 @@ class PersonalizedPlaylistManager:
                     UPDATE personalized_playlists
                     SET track_count = ?, last_generated_at = CURRENT_TIMESTAMP,
                         last_generation_source = ?, last_generation_error = NULL,
+                        is_stale = 0,
                         updated_at = CURRENT_TIMESTAMP
                     WHERE id = ?
                     """,
@@ -346,7 +376,8 @@ class PersonalizedPlaylistManager:
                 """
                 SELECT id, profile_id, kind, variant, name, config_json,
                        track_count, last_generated_at, last_synced_at,
-                       last_generation_source, last_generation_error
+                       last_generation_source, last_generation_error,
+                       is_stale
                 FROM personalized_playlists
                 WHERE profile_id = ? AND kind = ? AND variant = ?
                 """,
@@ -362,7 +393,8 @@ class PersonalizedPlaylistManager:
                 """
                 SELECT id, profile_id, kind, variant, name, config_json,
                        track_count, last_generated_at, last_synced_at,
-                       last_generation_source, last_generation_error
+                       last_generation_source, last_generation_error,
+                       is_stale
                 FROM personalized_playlists
                 WHERE id = ?
                 """,
@@ -386,6 +418,7 @@ class PersonalizedPlaylistManager:
                 last_synced_at=row.get('last_synced_at'),
                 last_generation_source=row.get('last_generation_source'),
                 last_generation_error=row.get('last_generation_error'),
+                is_stale=bool(row.get('is_stale') or 0),
             )
         # Tuple form: positional access matches SELECT order above.
         return PlaylistRecord(
@@ -398,6 +431,7 @@ class PersonalizedPlaylistManager:
             last_synced_at=row[8],
             last_generation_source=row[9],
             last_generation_error=row[10],
+            is_stale=bool(row[11] or 0) if len(row) > 11 else False,
         )
 
     @staticmethod

core/personalized/types.py

@@ -152,7 +152,13 @@ class PlaylistRecord:
     The live track list is fetched separately via
     ``PersonalizedPlaylistManager.get_playlist_tracks(playlist_id)``
     so list / detail responses can stay cheap when the caller only
-    needs metadata."""
+    needs metadata.
+
+    ``is_stale`` flips to True when the underlying source data
+    changes (e.g. watchlist scan updates the discovery pool) and is
+    cleared on the next successful refresh. Pipelines auto-refresh
+    stale snapshots before syncing so the server playlist always
+    reflects the latest source data."""
 
     id: int
     profile_id: int
@@ -165,6 +171,7 @@ class PlaylistRecord:
     last_synced_at: Optional[str]
     last_generation_source: Optional[str]
     last_generation_error: Optional[str]
+    is_stale: bool = False
 
 
 __all__ = [

core/watchlist_scanner.py

@@ -24,6 +24,19 @@ from core.wishlist_service import get_wishlist_service
 from core.matching_engine import MusicMatchingEngine
 from utils.logging_config import get_logger
 
+
+def _mark_personalized_kinds_stale(database, kinds, profile_id=1):
+    """Module-level helper so the inline call sites stay tiny.
+
+    Constructs a PersonalizedPlaylistManager with the minimal deps
+    needed for `mark_kinds_stale` (database access only — no generator
+    dispatch required) and flips the is_stale flag for matching rows.
+    Best-effort: any exception is swallowed by the caller's try/except
+    since stale-flagging is non-critical for the scan itself."""
+    from core.personalized.manager import PersonalizedPlaylistManager
+    mgr = PersonalizedPlaylistManager(database, deps=None)
+    return mgr.mark_kinds_stale(list(kinds), profile_id=profile_id)
+
 logger = get_logger("watchlist_scanner")
 
 # Rate limiting constants for watchlist operations
@@ -2980,6 +2993,22 @@ class WatchlistScanner:
             self.database.update_discovery_pool_timestamp(track_count=final_count, profile_id=profile_id)
             logger.info(f"Discovery pool now contains {final_count} total tracks (built over time)")
 
+            # Mark every personalized-playlist kind that draws from the
+            # discovery pool as stale so the playlist pipeline auto-
+            # regenerates snapshots on its next run. Without this the
+            # server playlists stay frozen even though the source pool
+            # just got fresh tracks. Best-effort — pool refresh succeeds
+            # even if the manager isn't wired (no personalized tables).
+            try:
+                _mark_personalized_kinds_stale(
+                    self.database,
+                    kinds=['hidden_gems', 'discovery_shuffle', 'popular_picks',
+                           'time_machine', 'genre_playlist', 'daily_mix'],
+                    profile_id=profile_id,
+                )
+            except Exception as e:  # noqa: BLE001 — never abort scan for staleness flag
+                logger.debug("Failed to mark personalized kinds stale: %s", e)
+
             # Cache recent albums for discovery page
             logger.info("Caching recent albums for discovery page...")
             if progress_callback:
@@ -3660,6 +3689,14 @@ class WatchlistScanner:
                 playlist_key = f'release_radar_{source}'
                 self.database.save_curated_playlist(playlist_key, release_radar_tracks, profile_id=profile_id)
                 logger.info(f"Release Radar ({source}) curated: {len(release_radar_tracks)} tracks")
+                # Flag personalized Fresh Tape snapshot as stale so the
+                # pipeline auto-regenerates it on the next run.
+                try:
+                    _mark_personalized_kinds_stale(
+                        self.database, kinds=['fresh_tape'], profile_id=profile_id,
+                    )
+                except Exception as e:  # noqa: BLE001
+                    logger.debug("Fresh Tape stale-flag failed: %s", e)
 
                 # 2. Curate Discovery Weekly - 50 tracks from discovery pool
                 logger.info(f"Curating Discovery Weekly for {source}...")
@@ -3735,6 +3772,12 @@ class WatchlistScanner:
                 playlist_key = f'discovery_weekly_{source}'
                 self.database.save_curated_playlist(playlist_key, discovery_weekly_tracks, profile_id=profile_id)
                 logger.info(f"Discovery Weekly ({source}) curated: {len(discovery_weekly_tracks)} tracks")
+                try:
+                    _mark_personalized_kinds_stale(
+                        self.database, kinds=['archives'], profile_id=profile_id,
+                    )
+                except Exception as e:  # noqa: BLE001
+                    logger.debug("Archives stale-flag failed: %s", e)
 
             # 3. "Because You Listen To" — personalized sections based on top played artists
             if profile['has_data']:

database/personalized_schema.py

@@ -66,12 +66,18 @@ CREATE TABLE IF NOT EXISTS personalized_playlists (
     last_synced_at TIMESTAMP,
     last_generation_source TEXT,
     last_generation_error TEXT,
+    is_stale INTEGER NOT NULL DEFAULT 0,
     created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
     updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
     UNIQUE (profile_id, kind, variant)
 )
 """
 
+# Migration for installs that created the table before is_stale existed.
+PERSONALIZED_PLAYLISTS_STALE_MIGRATION = """
+ALTER TABLE personalized_playlists ADD COLUMN is_stale INTEGER NOT NULL DEFAULT 0
+"""
+
 PERSONALIZED_PLAYLIST_TRACKS_DDL = """
 CREATE TABLE IF NOT EXISTS personalized_playlist_tracks (
     id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -126,6 +132,20 @@ def ensure_personalized_schema(connection: Any) -> None:
     cursor.execute(PERSONALIZED_PLAYLIST_TRACKS_INDEX)
     cursor.execute(PERSONALIZED_TRACK_HISTORY_DDL)
     cursor.execute(PERSONALIZED_TRACK_HISTORY_INDEX)
+
+    # Add is_stale column on installs that created the table before
+    # this column existed. SQLite has no `ADD COLUMN IF NOT EXISTS` so
+    # we probe with PRAGMA + tolerate the OperationalError that fires
+    # when the column is already there.
+    cursor.execute("PRAGMA table_info(personalized_playlists)")
+    cols = {row[1] for row in cursor.fetchall()}
+    if 'is_stale' not in cols:
+        try:
+            cursor.execute(PERSONALIZED_PLAYLISTS_STALE_MIGRATION)
+            logger.info("Added is_stale column to personalized_playlists")
+        except Exception as e:
+            logger.debug("is_stale column migration: %s", e)
+
     logger.debug("Personalized-playlist schema ensured")
 
 

tests/automation/test_automation_blocks.py

@@ -34,7 +34,8 @@ def _shape_check(items, allowed_types):
 
 _FIELD_TYPES = {
     'number', 'select', 'time', 'multi_select', 'checkbox', 'text',
-    'mirrored_playlist_select', 'signal_input', 'script_select',
+    'mirrored_playlist_select', 'personalized_playlist_select',
+    'signal_input', 'script_select',
 }
 
 

tests/automation/test_handler_registration.py

@@ -38,6 +38,7 @@ EXPECTED_ACTION_NAMES = frozenset({
     'sync_playlist',
     'discover_playlist',
     'playlist_pipeline',
+    'personalized_pipeline',
     'start_database_update',
     'deep_scan_library',
     'run_duplicate_cleaner',
@@ -60,6 +61,7 @@ EXPECTED_GUARDED_ACTIONS = frozenset({
     'scan_watchlist',
     'scan_library',
     'playlist_pipeline',
+    'personalized_pipeline',
     'start_database_update',
     'deep_scan_library',
     'run_duplicate_cleaner',
@@ -156,6 +158,7 @@ def _build_deps(engine, scan_mgr=None) -> AutomationDeps:
         get_beatport_data_cache=lambda: {'cache_lock': threading.Lock(), 'homepage': {}},
         init_automation_progress=lambda *a, **k: None,
         record_progress_history=lambda *a, **k: None,
+        build_personalized_manager=lambda: None,
     )
 
 

tests/automation/test_handlers_maintenance.py

@@ -105,6 +105,7 @@ def _build_deps(**overrides) -> AutomationDeps:
         get_beatport_data_cache=lambda: {'cache_lock': threading.Lock(), 'homepage': {}},
         init_automation_progress=lambda *a, **k: None,
         record_progress_history=lambda *a, **k: None,
+        build_personalized_manager=lambda: None,
     )
     defaults.update(overrides)
     return AutomationDeps(**defaults)  # type: ignore[arg-type]

tests/automation/test_handlers_personalized_pipeline.py

@@ -0,0 +1,529 @@
+"""Boundary tests for the personalized playlist pipeline handler.
+
+Pin every shape: empty kinds error, refresh_first behaviour, snapshot
+load + sync dispatch, missing-tracks skip, exception swallowing,
+pipeline_running flag cleanup, sync payload shape passed to
+_run_sync_task."""
+
+from __future__ import annotations
+
+import threading
+from types import SimpleNamespace
+from typing import Any, List
+
+import pytest
+
+from core.automation.deps import AutomationDeps, AutomationState
+from core.automation.handlers.personalized_pipeline import (
+    auto_personalized_pipeline,
+    _build_payloads_for_kinds,
+    _track_to_sync_shape,
+    _sync_personalized_playlist,
+)
+
+
+class _StubLogger:
+    def debug(self, *a, **k): pass
+    def info(self, *a, **k): pass
+    def warning(self, *a, **k): pass
+    def error(self, *a, **k): pass
+
+
+def _build_deps(**overrides) -> AutomationDeps:
+    defaults = dict(
+        engine=object(),
+        state=AutomationState(),
+        config_manager=object(),
+        update_progress=lambda *a, **k: None,
+        logger=_StubLogger(),
+        get_database=lambda: object(),
+        spotify_client=None,
+        tidal_client=None,
+        web_scan_manager=None,
+        process_wishlist_automatically=lambda **k: None,
+        process_watchlist_scan_automatically=lambda **k: None,
+        is_wishlist_actually_processing=lambda: False,
+        is_watchlist_actually_scanning=lambda: False,
+        get_watchlist_scan_state=lambda: {},
+        run_playlist_discovery_worker=lambda *a, **k: None,
+        run_sync_task=lambda *a, **k: None,
+        load_sync_status_file=lambda: {},
+        get_deezer_client=lambda: None,
+        parse_youtube_playlist=lambda url: None,
+        get_sync_states=lambda: {},
+        set_db_update_automation_id=lambda v: None,
+        get_db_update_state=lambda: {},
+        db_update_lock=threading.Lock(),
+        db_update_executor=None,
+        run_db_update_task=lambda *a, **k: None,
+        run_deep_scan_task=lambda *a, **k: None,
+        get_duplicate_cleaner_state=lambda: {},
+        duplicate_cleaner_lock=threading.Lock(),
+        duplicate_cleaner_executor=None,
+        run_duplicate_cleaner=lambda: None,
+        get_quality_scanner_state=lambda: {},
+        quality_scanner_lock=threading.Lock(),
+        quality_scanner_executor=None,
+        run_quality_scanner=lambda *a, **k: None,
+        download_orchestrator=None,
+        run_async=lambda coro: None,
+        tasks_lock=threading.Lock(),
+        get_download_batches=lambda: {},
+        get_download_tasks=lambda: {},
+        sweep_empty_download_directories=lambda: 0,
+        get_staging_path=lambda: '/staging',
+        docker_resolve_path=lambda p: p,
+        get_current_profile_id=lambda: 1,
+        get_watchlist_scanner=lambda spc: None,
+        get_app=lambda: None,
+        get_beatport_data_cache=lambda: {'cache_lock': threading.Lock(), 'homepage': {}},
+        init_automation_progress=lambda *a, **k: None,
+        record_progress_history=lambda *a, **k: None,
+        build_personalized_manager=lambda: None,
+    )
+    defaults.update(overrides)
+    return AutomationDeps(**defaults)  # type: ignore[arg-type]
+
+
+# ─── Track shape converter ───────────────────────────────────────────
+
+
+class TestTrackToSyncShape:
+    def test_basic_shape(self):
+        track = SimpleNamespace(
+            track_name='Song', artist_name='Artist', album_name='Album',
+            spotify_track_id='sp-1', itunes_track_id=None, deezer_track_id=None,
+            duration_ms=200000,
+        )
+        out = _track_to_sync_shape(track)
+        assert out == {
+            'name': 'Song',
+            'artists': [{'name': 'Artist'}],
+            'album': {'name': 'Album'},
+            'duration_ms': 200000,
+            'id': 'sp-1',
+        }
+
+    def test_falls_back_through_source_ids(self):
+        t1 = SimpleNamespace(track_name='', artist_name='', album_name='',
+                              spotify_track_id=None, itunes_track_id='it-1',
+                              deezer_track_id=None, duration_ms=0)
+        assert _track_to_sync_shape(t1)['id'] == 'it-1'
+
+        t2 = SimpleNamespace(track_name='', artist_name='', album_name='',
+                              spotify_track_id=None, itunes_track_id=None,
+                              deezer_track_id='dz-1', duration_ms=0)
+        assert _track_to_sync_shape(t2)['id'] == 'dz-1'
+
+    def test_no_id_returns_empty_string(self):
+        t = SimpleNamespace(track_name='X', artist_name='Y', album_name='Z',
+                             spotify_track_id=None, itunes_track_id=None,
+                             deezer_track_id=None, duration_ms=0)
+        assert _track_to_sync_shape(t)['id'] == ''
+
+    def test_preserves_enriched_track_data_for_wishlist_metadata(self):
+        track = SimpleNamespace(
+            track_name='Bare Name', artist_name='Bare Artist', album_name='Bare Album',
+            spotify_track_id='sp-rich', itunes_track_id=None, deezer_track_id=None,
+            album_cover_url=None, duration_ms=200000, popularity=33,
+            track_data_json={
+                'id': 'sp-rich',
+                'name': 'Rich Name',
+                'artists': [{'name': 'Rich Artist', 'id': 'artist-1'}],
+                'album': {
+                    'id': 'album-1',
+                    'name': 'Rich Album',
+                    'images': [{'url': 'https://example.test/cover.jpg'}],
+                },
+                'duration_ms': 201000,
+                'preview_url': 'https://example.test/preview.mp3',
+            },
+        )
+
+        out = _track_to_sync_shape(track)
+
+        assert out['name'] == 'Rich Name'
+        assert out['artists'][0] == {'name': 'Rich Artist', 'id': 'artist-1'}
+        assert out['album']['id'] == 'album-1'
+        assert out['album']['images'][0]['url'] == 'https://example.test/cover.jpg'
+        assert out['preview_url'] == 'https://example.test/preview.mp3'
+
+    def test_album_cover_url_fills_album_images_when_no_rich_blob(self):
+        track = SimpleNamespace(
+            track_name='Song', artist_name='Artist', album_name='Album',
+            spotify_track_id='sp-1', itunes_track_id=None, deezer_track_id=None,
+            album_cover_url='https://example.test/fallback.jpg',
+            duration_ms=200000, track_data_json=None,
+        )
+
+        out = _track_to_sync_shape(track)
+
+        assert out['album'] == {
+            'name': 'Album',
+            'images': [{'url': 'https://example.test/fallback.jpg'}],
+        }
+
+
+# ─── Empty / config validation ──────────────────────────────────────
+
+
+class TestEmptyConfig:
+    def test_no_kinds_returns_error_and_clears_flag(self):
+        deps = _build_deps()
+        deps.state.set_pipeline_running(True)  # simulate already-running
+        result = auto_personalized_pipeline({}, deps)
+        assert result['status'] == 'error'
+        assert 'No personalized playlist' in result['error']
+        assert deps.state.pipeline_running is False
+
+    def test_empty_kinds_list_returns_error(self):
+        deps = _build_deps()
+        result = auto_personalized_pipeline({'kinds': []}, deps)
+        assert result['status'] == 'error'
+        assert deps.state.pipeline_running is False
+
+    def test_non_list_kinds_returns_error(self):
+        deps = _build_deps()
+        result = auto_personalized_pipeline({'kinds': 'not_a_list'}, deps)
+        assert result['status'] == 'error'
+
+
+# ─── Payload building ───────────────────────────────────────────────
+
+
+class _StubManagerNoTracks:
+    def ensure_playlist(self, kind, variant, profile_id):
+        # last_generated_at non-None so pipeline treats the snapshot as
+        # already-generated-but-empty (rather than first-run-needs-gen).
+        return SimpleNamespace(
+            id=1, name=f'{kind}-{variant}', kind=kind, variant=variant,
+            is_stale=False, last_generated_at='2026-05-15T20:00:00',
+        )
+
+    def refresh_playlist(self, kind, variant, profile_id):
+        return self.ensure_playlist(kind, variant, profile_id)
+
+    def get_playlist_tracks(self, playlist_id):
+        return []
+
+
+class _StubManagerWithTracks:
+    def __init__(self, tracks_per_kind=None):
+        self.tracks_per_kind = tracks_per_kind or {}
+        self.refresh_calls: List[tuple] = []
+        self.ensure_calls: List[tuple] = []
+
+    def ensure_playlist(self, kind, variant, profile_id):
+        self.ensure_calls.append((kind, variant, profile_id))
+        return SimpleNamespace(
+            id=hash((kind, variant)) % 10000,
+            name=f'{kind}-{variant or "S"}', kind=kind, variant=variant,
+            is_stale=False, last_generated_at='2026-05-15T20:00:00',
+        )
+
+    def refresh_playlist(self, kind, variant, profile_id):
+        self.refresh_calls.append((kind, variant, profile_id))
+        # Mirror real manager: refresh returns a record without invoking
+        # the public ensure_playlist API path again.
+        return SimpleNamespace(
+            id=hash((kind, variant)) % 10000,
+            name=f'{kind}-{variant or "S"}', kind=kind, variant=variant,
+            is_stale=False, last_generated_at='2026-05-15T20:00:00',
+        )
+
+    def get_playlist_tracks(self, playlist_id):
+        # Return all tracks regardless of id — tests scope to one playlist at a time.
+        for tracks in self.tracks_per_kind.values():
+            if tracks:
+                return [SimpleNamespace(
+                    track_name=t['name'], artist_name=t.get('artist', 'A'),
+                    album_name=t.get('album', 'Al'),
+                    spotify_track_id=t.get('id'),
+                    itunes_track_id=None, deezer_track_id=None,
+                    duration_ms=200000,
+                ) for t in tracks]
+        return []
+
+
+class TestPayloadBuilding:
+    def test_skips_kinds_with_no_tracks(self):
+        deps = _build_deps()
+        manager = _StubManagerNoTracks()
+        payloads = _build_payloads_for_kinds(
+            deps, manager,
+            [{'kind': 'hidden_gems'}, {'kind': 'discovery_shuffle'}],
+            profile_id=1, automation_id=None, refresh_first=False,
+        )
+        assert payloads == []
+
+    def test_skips_invalid_entries(self):
+        deps = _build_deps()
+        manager = _StubManagerNoTracks()
+        payloads = _build_payloads_for_kinds(
+            deps, manager,
+            ['not-a-dict', {}, {'variant': 'no-kind'}],  # all invalid
+            profile_id=1, automation_id=None, refresh_first=False,
+        )
+        assert payloads == []
+
+    def test_refresh_first_calls_refresh(self):
+        deps = _build_deps()
+        manager = _StubManagerWithTracks(
+            tracks_per_kind={'hidden_gems': [{'name': 'T', 'id': 'sp-1'}]},
+        )
+        _build_payloads_for_kinds(
+            deps, manager,
+            [{'kind': 'hidden_gems'}],
+            profile_id=1, automation_id=None, refresh_first=True,
+        )
+        assert manager.refresh_calls == [('hidden_gems', '', 1)]
+        assert manager.ensure_calls == []
+
+    def test_no_refresh_calls_ensure(self):
+        deps = _build_deps()
+        manager = _StubManagerWithTracks(
+            tracks_per_kind={'hidden_gems': [{'name': 'T', 'id': 'sp-1'}]},
+        )
+        _build_payloads_for_kinds(
+            deps, manager,
+            [{'kind': 'hidden_gems'}],
+            profile_id=1, automation_id=None, refresh_first=False,
+        )
+        assert manager.ensure_calls == [('hidden_gems', '', 1)]
+        assert manager.refresh_calls == []
+
+    def test_payload_shape(self):
+        deps = _build_deps()
+        manager = _StubManagerWithTracks(
+            tracks_per_kind={'hidden_gems': [
+                {'name': 'Track1', 'id': 'sp-1'},
+                {'name': 'Track2', 'id': 'sp-2'},
+            ]},
+        )
+        payloads = _build_payloads_for_kinds(
+            deps, manager,
+            [{'kind': 'hidden_gems'}],
+            profile_id=1, automation_id=None, refresh_first=False,
+        )
+        assert len(payloads) == 1
+        p = payloads[0]
+        assert p['kind'] == 'hidden_gems'
+        assert p['variant'] == ''
+        assert p['name'] == 'hidden_gems-S'
+        assert p['sync_id'].startswith('auto_personalized_hidden_gems_')
+        assert len(p['tracks_json']) == 2
+        assert p['tracks_json'][0]['id'] == 'sp-1'
+
+    def test_stale_snapshot_auto_refreshes_even_without_refresh_first(self):
+        """When the manager reports is_stale=True, the pipeline refreshes
+        regardless of the refresh_first config flag — the source data
+        (discovery_pool / curated lists) changed, so the snapshot must
+        be regenerated before syncing or we'd push stale data."""
+        deps = _build_deps()
+        # Stub manager whose ensure_playlist returns a stale record.
+        # refresh_playlist should still get called.
+        refresh_called = []
+
+        class _StaleMgr:
+            def ensure_playlist(self, kind, variant, profile_id):
+                return SimpleNamespace(
+                    id=1, name=kind, kind=kind, variant=variant, is_stale=True,
+                    last_generated_at='2026-05-15T20:00:00',
+                )
+            def refresh_playlist(self, kind, variant, profile_id):
+                refresh_called.append((kind, variant))
+                return SimpleNamespace(
+                    id=1, name=kind, kind=kind, variant=variant, is_stale=False,
+                    last_generated_at='2026-05-15T20:00:00',
+                )
+            def get_playlist_tracks(self, _id):
+                return [SimpleNamespace(
+                    track_name='Refreshed', artist_name='A', album_name='Al',
+                    spotify_track_id='sp-fresh', itunes_track_id=None,
+                    deezer_track_id=None, duration_ms=200000,
+                )]
+
+        payloads = _build_payloads_for_kinds(
+            deps, _StaleMgr(),
+            [{'kind': 'hidden_gems'}],
+            profile_id=1, automation_id=None, refresh_first=False,
+        )
+        assert refresh_called == [('hidden_gems', '')]
+        assert len(payloads) == 1
+        assert payloads[0]['tracks_json'][0]['name'] == 'Refreshed'
+
+    def test_non_stale_snapshot_skips_refresh(self):
+        """When the snapshot is fresh AND refresh_first is False, just
+        read the existing tracks without re-running the generator."""
+        deps = _build_deps()
+        refresh_called = []
+
+        class _FreshMgr:
+            def ensure_playlist(self, kind, variant, profile_id):
+                return SimpleNamespace(
+                    id=1, name=kind, kind=kind, variant=variant, is_stale=False,
+                    last_generated_at='2026-05-15T20:00:00',
+                )
+            def refresh_playlist(self, *_a, **_k):
+                refresh_called.append('called')
+                return SimpleNamespace(
+                    id=1, name='x', kind='x', variant='', is_stale=False,
+                    last_generated_at='2026-05-15T20:00:00',
+                )
+            def get_playlist_tracks(self, _id):
+                return [SimpleNamespace(
+                    track_name='Cached', artist_name='A', album_name='Al',
+                    spotify_track_id='sp-1', itunes_track_id=None,
+                    deezer_track_id=None, duration_ms=200000,
+                )]
+
+        _build_payloads_for_kinds(
+            deps, _FreshMgr(),
+            [{'kind': 'hidden_gems'}],
+            profile_id=1, automation_id=None, refresh_first=False,
+        )
+        assert refresh_called == []
+
+    def test_never_generated_snapshot_triggers_first_refresh(self):
+        """First-run case: pipeline picks a brand-new kind, ensure_playlist
+        auto-creates the row with track_count=0 and last_generated_at=None.
+        Without this branch the pipeline would read the empty snapshot and
+        silently skip — user picked a kind and got nothing. With the branch,
+        last_generated_at=None forces a refresh so the generator actually runs."""
+        deps = _build_deps()
+        refresh_called = []
+
+        class _NeverGenMgr:
+            def ensure_playlist(self, kind, variant, profile_id):
+                return SimpleNamespace(
+                    id=1, name=kind, kind=kind, variant=variant,
+                    is_stale=False, last_generated_at=None,
+                )
+            def refresh_playlist(self, kind, variant, profile_id):
+                refresh_called.append((kind, variant))
+                return SimpleNamespace(
+                    id=1, name=kind, kind=kind, variant=variant,
+                    is_stale=False, last_generated_at='2026-05-15T20:00:00',
+                )
+            def get_playlist_tracks(self, _id):
+                return [SimpleNamespace(
+                    track_name='Generated', artist_name='A', album_name='Al',
+                    spotify_track_id='sp-new', itunes_track_id=None,
+                    deezer_track_id=None, duration_ms=200000,
+                )]
+
+        payloads = _build_payloads_for_kinds(
+            deps, _NeverGenMgr(),
+            [{'kind': 'fresh_tape'}],
+            profile_id=1, automation_id=None, refresh_first=False,
+        )
+        assert refresh_called == [('fresh_tape', '')]
+        assert len(payloads) == 1
+        assert payloads[0]['tracks_json'][0]['name'] == 'Generated'
+
+    def test_manager_exception_swallowed_continues_to_next(self):
+        deps = _build_deps()
+
+        class _ExplodingMgr:
+            def __init__(self):
+                self.calls = []
+            def ensure_playlist(self, kind, variant, profile_id):
+                self.calls.append(kind)
+                if kind == 'broken':
+                    raise RuntimeError('manager boom')
+                return SimpleNamespace(
+                    id=1, name=kind, kind=kind, variant=variant, is_stale=False,
+                    last_generated_at='2026-05-15T20:00:00',
+                )
+            def get_playlist_tracks(self, _id):
+                return []
+
+        mgr = _ExplodingMgr()
+        # broken raises, hidden_gems proceeds (just no tracks).
+        payloads = _build_payloads_for_kinds(
+            deps, mgr,
+            [{'kind': 'broken'}, {'kind': 'hidden_gems'}],
+            profile_id=1, automation_id=None, refresh_first=False,
+        )
+        assert mgr.calls == ['broken', 'hidden_gems']
+        assert payloads == []  # neither produced tracks
+
+
+# ─── Sync launch ────────────────────────────────────────────────────
+
+
+class TestSyncLaunch:
+    def test_sync_one_playlist_starts_thread(self):
+        captured: List[tuple] = []
+
+        def fake_run_sync_task(*args):
+            captured.append(args)
+
+        deps = _build_deps(
+            run_sync_task=fake_run_sync_task,
+            get_current_profile_id=lambda: 7,
+        )
+        payload = {
+            'sync_id': 'auto_personalized_hidden_gems_',
+            'name': 'Hidden Gems',
+            'tracks_json': [{'name': 'X', 'id': 'sp-1'}],
+            'image_url': '',
+        }
+        result = _sync_personalized_playlist(deps, payload)
+        assert result['status'] == 'started'
+        # Wait for thread to invoke fake_run_sync_task.
+        for _ in range(100):
+            if captured:
+                break
+            import time
+            time.sleep(0.01)
+        assert len(captured) == 1
+        # Args: (sync_id, name, tracks_json, automation_id, profile_id, image_url)
+        assert captured[0][0] == 'auto_personalized_hidden_gems_'
+        assert captured[0][1] == 'Hidden Gems'
+        assert captured[0][3] is None  # automation_id muted
+        assert captured[0][4] == 7  # profile_id
+
+
+# ─── Full pipeline (with stubbed manager + sync states) ─────────────
+
+
+class TestPipelineHappyPath:
+    def test_pipeline_completes_with_synced_count(self):
+        # Stub manager returns one playlist with 2 tracks.
+        manager = _StubManagerWithTracks(
+            tracks_per_kind={'hidden_gems': [
+                {'name': 'A', 'id': 'sp-1'},
+                {'name': 'B', 'id': 'sp-2'},
+            ]},
+        )
+
+        # sync_states populated as if the sync background task finished.
+        sync_states_storage = {}
+
+        def fake_run_sync(sync_id, name, tracks, aid, pid, img):
+            sync_states_storage[sync_id] = {
+                'status': 'finished',
+                'result': {'matched_tracks': 2},
+            }
+
+        deps = _build_deps(
+            build_personalized_manager=lambda: manager,
+            run_sync_task=fake_run_sync,
+            get_sync_states=lambda: sync_states_storage,
+        )
+        # Patch time.sleep in shared helper so test doesn't take 2s per iter.
+        import core.automation.handlers._pipeline_shared as shared
+        orig = shared.time.sleep
+        shared.time.sleep = lambda _: None
+        try:
+            result = auto_personalized_pipeline(
+                {'_automation_id': 'auto-1', 'kinds': [{'kind': 'hidden_gems'}]},
+                deps,
+            )
+        finally:
+            shared.time.sleep = orig
+        assert result['status'] == 'completed'
+        assert result['_manages_own_progress'] is True
+        # Pipeline-running flag cleaned up.
+        assert deps.state.pipeline_running is False

tests/automation/test_handlers_playlist.py

@@ -122,6 +122,7 @@ def _build_deps(**overrides) -> AutomationDeps:
         get_beatport_data_cache=lambda: {'cache_lock': threading.Lock(), 'homepage': {}},
         init_automation_progress=lambda *a, **k: None,
         record_progress_history=lambda *a, **k: None,
+        build_personalized_manager=lambda: None,
     )
     defaults.update(overrides)
     return AutomationDeps(**defaults)  # type: ignore[arg-type]

tests/automation/test_handlers_simple.py

@@ -92,6 +92,7 @@ def _build_deps(**overrides: Any) -> AutomationDeps:
         get_beatport_data_cache=lambda: {'cache_lock': threading.Lock(), 'homepage': {}},
         init_automation_progress=lambda *a, **k: None,
         record_progress_history=lambda *a, **k: None,
+        build_personalized_manager=lambda: None,
     )
     defaults.update(overrides)
     return AutomationDeps(**defaults)  # type: ignore[arg-type]

tests/automation/test_progress_callbacks.py

@@ -78,6 +78,7 @@ def _build_deps(**overrides) -> AutomationDeps:
         get_beatport_data_cache=lambda: {'cache_lock': threading.Lock(), 'homepage': {}},
         init_automation_progress=lambda *a, **k: None,
         record_progress_history=lambda *a, **k: None,
+        build_personalized_manager=lambda: None,
     )
     defaults.update(overrides)
     return AutomationDeps(**defaults)  # type: ignore[arg-type]

tests/test_personalized_manager.py

@@ -449,6 +449,62 @@ class TestStalenessFilter:
         assert r2.track_count == 1
 
 
+class TestStaleFlag:
+    """`is_stale` flips when upstream data changes; refresh clears it."""
+
+    def test_default_is_false(self, db, registry):
+        _register_simple_kind(registry, lambda *a, **k: [])
+        mgr = PersonalizedPlaylistManager(db, deps=None, registry=registry)
+        record = mgr.ensure_playlist('hidden_gems', '', 1)
+        assert record.is_stale is False
+
+    def test_mark_kinds_stale_flips_flag(self, db, registry):
+        _register_simple_kind(registry, lambda *a, **k: [], kind='hidden_gems')
+        _register_simple_kind(registry, lambda *a, **k: [], kind='discovery_shuffle')
+        mgr = PersonalizedPlaylistManager(db, deps=None, registry=registry)
+        mgr.ensure_playlist('hidden_gems', '', 1)
+        mgr.ensure_playlist('discovery_shuffle', '', 1)
+        n = mgr.mark_kinds_stale(['hidden_gems', 'discovery_shuffle'])
+        assert n == 2
+        assert mgr.ensure_playlist('hidden_gems', '', 1).is_stale is True
+        assert mgr.ensure_playlist('discovery_shuffle', '', 1).is_stale is True
+
+    def test_mark_kinds_stale_only_matching_kinds(self, db, registry):
+        _register_simple_kind(registry, lambda *a, **k: [], kind='hidden_gems')
+        _register_simple_kind(registry, lambda *a, **k: [], kind='popular_picks')
+        mgr = PersonalizedPlaylistManager(db, deps=None, registry=registry)
+        mgr.ensure_playlist('hidden_gems', '', 1)
+        mgr.ensure_playlist('popular_picks', '', 1)
+        mgr.mark_kinds_stale(['hidden_gems'])
+        assert mgr.ensure_playlist('hidden_gems', '', 1).is_stale is True
+        assert mgr.ensure_playlist('popular_picks', '', 1).is_stale is False
+
+    def test_mark_kinds_stale_scopes_to_profile(self, db, registry):
+        _register_simple_kind(registry, lambda *a, **k: [])
+        mgr = PersonalizedPlaylistManager(db, deps=None, registry=registry)
+        mgr.ensure_playlist('hidden_gems', '', 1)
+        mgr.ensure_playlist('hidden_gems', '', 2)
+        mgr.mark_kinds_stale(['hidden_gems'], profile_id=1)
+        assert mgr.ensure_playlist('hidden_gems', '', 1).is_stale is True
+        assert mgr.ensure_playlist('hidden_gems', '', 2).is_stale is False
+
+    def test_refresh_clears_stale_flag(self, db, registry):
+        _register_simple_kind(registry, lambda *a, **k: [_make_track()])
+        mgr = PersonalizedPlaylistManager(db, deps=None, registry=registry)
+        mgr.ensure_playlist('hidden_gems', '', 1)
+        mgr.mark_kinds_stale(['hidden_gems'])
+        assert mgr.ensure_playlist('hidden_gems', '', 1).is_stale is True
+        record = mgr.refresh_playlist('hidden_gems', '', 1)
+        assert record.is_stale is False
+
+    def test_mark_kinds_stale_empty_list_noop(self, db, registry):
+        _register_simple_kind(registry, lambda *a, **k: [])
+        mgr = PersonalizedPlaylistManager(db, deps=None, registry=registry)
+        mgr.ensure_playlist('hidden_gems', '', 1)
+        n = mgr.mark_kinds_stale([])
+        assert n == 0
+
+
 class TestStalenessHistory:
     def test_recent_track_ids_returns_zero_when_days_zero(self, db, registry):
         _register_simple_kind(registry, lambda *a, **k: [_make_track(sid='spot-1')])

web_server.py

@@ -474,10 +474,16 @@ def _add_discover_cache_headers(response):
 
 
 def get_current_profile_id() -> int:
-    """Get the current profile ID from Flask g context or default to 1"""
+    """Get the current profile ID from Flask g context or default to 1.
+
+    Background callers (automation engine, sync threads, watchlist
+    scanner) have no request context, so `g.profile_id` raises
+    `RuntimeError("Working outside of application context")` rather
+    than `AttributeError`. Catch both so non-request callers degrade
+    to the admin profile instead of crashing the handler."""
     try:
         return g.profile_id
-    except AttributeError:
+    except (AttributeError, RuntimeError):
         return 1
 
 
@@ -992,6 +998,7 @@ def _register_automation_handlers():
         get_beatport_data_cache=lambda: beatport_data_cache,
         init_automation_progress=_init_automation_progress,
         record_progress_history=_auto_progress.record_history,
+        build_personalized_manager=_build_personalized_manager,
     )
     _register_extracted_handlers(_automation_deps)
 
@@ -26855,9 +26862,12 @@ def _build_personalized_manager():
 
 @app.route('/api/personalized/kinds', methods=['GET'])
 def personalized_list_kinds():
-    """List every registered personalized-playlist kind."""
+    """List every registered personalized-playlist kind. Includes the
+    resolved variant list per kind that supports variants so the UI
+    can render kind+variant checkboxes without per-kind round-trips."""
     try:
-        return jsonify(_personalized_api.list_kinds())
+        manager = _build_personalized_manager()
+        return jsonify(_personalized_api.list_kinds(manager=manager))
     except Exception as e:
         logger.error(f"Personalized kinds list error: {e}")
         return jsonify({"success": False, "error": str(e)}), 500

webui/static/helper.js

@@ -3416,6 +3416,8 @@ const WHATS_NEW = {
     '2.5.2': [
         // --- May 13, 2026 — 2.5.2 release ---
         { date: 'May 13, 2026 — 2.5.2 release' },
+        { title: 'Personalized Pipeline: Auto-Refresh Stale Snapshots', desc: 'follow-up polish on the personalized playlist pipeline. snapshots now know when their source data went stale. when watchlist scan finishes — refreshes the discovery pool OR re-curates Release Radar / Discovery Weekly — every playlist that draws from that data gets flagged `is_stale = 1`. next pipeline run sees the flag and auto-refreshes BEFORE syncing, so the server playlist always reflects the latest pool. no more pushing day-old Hidden Gems to plex right after the scan dropped 200 new tracks into the pool. pool-fed kinds (Hidden Gems / Discovery Shuffle / Popular Picks / Time Machine / Genre / Daily Mix) flagged after the discovery pool refresh. Fresh Tape flagged after Release Radar curates. Archives flagged after Discovery Weekly curates. flag clears on the next successful refresh. independent of the existing `refresh_first` config — that flag is now for "ALWAYS refresh, even when nothing changed" (cron use case: nightly Hidden Gems regen). idempotent schema migration adds the `is_stale` column to installs created before this PR. 12 new boundary tests pin: stale flag flips, refresh clears it, profile scoping, multi-kind batching, pipeline auto-refreshes on stale even without refresh_first, pipeline skips refresh when fresh. 3391 tests pass.', page: 'discover' },
+        { title: 'Personalized Playlist Pipeline: Auto-Sync Discover-Page Playlists', desc: 'follow-up to the personalized-playlists standardization PR. new automation action `personalized_pipeline` syncs your selected discover-page playlists (Hidden Gems, Time Machine per-decade, Fresh Tape, The Archives, Seasonal Mix per-season, etc.) to your active media server + queues missing tracks for download — same pattern as the existing mirrored playlist pipeline but two phases instead of four (no REFRESH or DISCOVER needed since manager-backed snapshots are already metadata-matched). config: pick which kinds+variants to include, optional `refresh_first` to regenerate snapshots before syncing, optional `skip_wishlist`. shares the pipeline_running guard with the mirrored pipeline so the two can\'t overlap (one sync queue, one wishlist worker). lifted PHASE 3 (SYNC loop) + PHASE 4 (WISHLIST tail) of the mirrored pipeline into shared `core/automation/handlers/_pipeline_shared.run_sync_and_wishlist` so both pipelines reuse the same sync-state polling / progress emission / wishlist trigger logic — 0 duplication. trigger UI block declared at `core/automation/blocks.py` (full multi-select picker UI is its own follow-up). 14 new boundary tests pin: track→sync_shape conversion + source ID fallback, empty-kinds error, payload building skips no-tracks playlists, refresh_first vs ensure dispatch, manager exception swallowed continues to next kind, full pipeline happy-path with stubbed sync_states. 3383 tests pass total. now usable via API today, polished UI dropping next.', page: 'discover' },
         { title: 'Personalized Playlists Standardization', desc: 'all 8 personalized / discover-page playlists (Hidden Gems, Discovery Shuffle, Popular Picks, Time Machine per-decade, Genre playlists per-genre, Daily Mixes, Fresh Tape, The Archives, Seasonal Mix per-season) now share one unified storage layer. pre-overhaul: Group A (Fresh Tape / Archives / Seasonal Mix) lived in one shape, Group B (everything else) was computed-on-demand with no persistence — every page-load re-rolled the dice and tracks rotated under your feet. post-overhaul: every playlist has a stable identity, persistent track snapshot, explicit refresh button, and per-playlist tweakable config (limit, diversity caps, popularity bounds, recency window, exclude-recent-days staleness window). prerequisite for the playlist pipeline integration coming in the next PR (sync these to your media server + send missing tracks to wishlist on a timer). fixed a stub: Daily Mixes used to promise 50% library + 50% discovery but the library half always returned [] (tracks table has no source IDs to sync) — now honestly discovery-only so it actually works. also: each kind\'s body lifted into its own module under `core/personalized/generators/`, behavior preserved verbatim from the legacy `PersonalizedPlaylistsService` and `SeasonalDiscoveryService`. new REST endpoints under `/api/personalized/*`. 134 boundary tests cover every kind + the manager + the API + staleness filter; full suite at 3369 tests.', page: 'discover' },
         { title: 'Dashboard Activity Feed: Stop Showing "NaNmo ago"', desc: 'recent activity items on the dashboard all rendered "NaNmo ago" because the formatter was parsing `activity.time` (a human label like "Now") as a date. backend has always emitted `activity.timestamp` (Unix epoch seconds) alongside the label — frontend now uses that for relative-time formatting. falls back to the literal label only when no timestamp present (legacy items / future shapes).', page: 'home' },
         { title: 'Token Leak Round 2: URL-Encoded Form In Artist Endpoint + Playlist Sync', desc: 'security follow-up to the prior token-leak fix. found three sites in `web_server.py` (artist endpoint) that logged the full `image_url` and the entire artist_info dict at INFO on every artist-page render — the dict contained the `image_url` field routed through the image proxy (`/api/image-proxy?url=<encoded>`), URL-encoding the X-Plex-Token / X-Emby-Token / Subsonic auth straight into the log line. also one site in `core/discovery/sync.py` logged the playlist poster URL during sync. fixes: dropped the three artist-endpoint dev-time debug log lines entirely (before-fix, after-fix, "Final artist data being sent"). playlist-image log now logs `has_image=True/False`, not the URL. strengthened `_redact_url_secrets` with a second regex pattern that matches the URL-encoded form (`%3FX-Plex-Token%3D...`) so any future log-through-redactor catches both plain and encoded shapes. wipe your existing app.log if it captured tokens in either form, and rotate Plex / Jellyfin / Navidrome credentials.', page: 'settings' },

webui/static/stats-automations.js

@@ -5754,6 +5754,18 @@ function _renderBuilderCanvas() {
     canvas.innerHTML = html;
     // Load mirrored playlist selects if any are present
     _autoLoadMirroredSelects();
+    // Personalized-pipeline kind pickers (one per slot that uses it)
+    ['when', 'do'].forEach(sk => {
+        if (document.getElementById('cfg-' + sk + '-kinds-picker')) {
+            _autoLoadPersonalizedKinds(sk);
+        }
+    });
+    _autoBuilder.then.forEach((item, i) => {
+        const sk = 'then-' + i;
+        if (document.getElementById('cfg-' + sk + '-kinds-picker')) {
+            _autoLoadPersonalizedKinds(sk);
+        }
+    });
     // Set up checkbox state for refresh_mirrored
     ['when', 'do'].forEach(sk => {
         const allCb = document.getElementById('cfg-' + sk + '-all');
@@ -5961,6 +5973,33 @@ function _renderBlockConfigFields(slotKey, blockType, config) {
         </div>
         <div class="config-row" style="color:rgba(255,255,255,0.35);font-size:11px;">Runs 4 phases: Refresh → Discover → Sync → Download Missing</div>`;
     }
+    if (blockType === 'personalized_pipeline') {
+        const refreshFirstChecked = config.refresh_first ? ' checked' : '';
+        const skipWishlistChecked = config.skip_wishlist ? ' checked' : '';
+        // Stash existing selections on a hidden input as JSON so the
+        // async populator can mark them checked once kinds load. The
+        // visible picker container starts as a loading message.
+        const initialKinds = JSON.stringify(Array.isArray(config.kinds) ? config.kinds : []);
+        // The default `.config-row` flex layout puts label on the left
+        // and input on the right — wrong for a tall multi-select picker.
+        // Use a dedicated column-layout wrapper so the picker spans
+        // the full card width.
+        return `<div class="config-row" style="flex-direction:column;align-items:stretch;gap:6px;">
+            <label style="min-width:0;">Personalized playlists to sync</label>
+            <input type="hidden" id="cfg-${slotKey}-kinds" data-initial='${_escAttr(initialKinds)}' value='${_escAttr(initialKinds)}'>
+            <div id="cfg-${slotKey}-kinds-picker" class="personalized-kinds-picker" style="max-height:280px;overflow-y:auto;border:1px solid rgba(255,255,255,0.08);border-radius:6px;padding:10px 12px;background:rgba(0,0,0,0.25);width:100%;box-sizing:border-box;">
+                <div style="color:rgba(255,255,255,0.4);font-size:12px;">Loading personalized playlists…</div>
+            </div>
+            <div style="color:rgba(255,255,255,0.35);font-size:11px;text-transform:none;letter-spacing:0;">Pick which Discover-page playlists this automation will sync. Variant kinds (Time Machine, Genre, Daily Mix, Seasonal Mix) expose individual instances below their kind row.</div>
+        </div>
+        <div class="config-row">
+            <label><input type="checkbox" id="cfg-${slotKey}-refresh_first"${refreshFirstChecked}> Refresh playlists before sync (regenerate snapshots)</label>
+        </div>
+        <div class="config-row">
+            <label><input type="checkbox" id="cfg-${slotKey}-skip_wishlist"${skipWishlistChecked}> Skip wishlist processing</label>
+        </div>
+        <div class="config-row" style="color:rgba(255,255,255,0.35);font-size:11px;">Runs 2 phases: Snapshot → Sync → Download Missing</div>`;
+    }
     // Shared variable tags builder for notification types
     function _notifyVarHtml(slotKey) {
         let allVars = ['time', 'name', 'run_count', 'status'];
@@ -6156,6 +6195,86 @@ function _autoTogglePlaylistSelect(slotKey) {
     if (sel) sel.disabled = allCb && allCb.checked;
 }
 
+// --- Personalized Playlist Picker (multi-select kind+variant) ---
+
+// Cached so the second action card renders instantly without re-fetching.
+let _autoPersonalizedKinds = null;
+
+async function _autoLoadPersonalizedKinds(slotKey) {
+    const picker = document.getElementById('cfg-' + slotKey + '-kinds-picker');
+    const hidden = document.getElementById('cfg-' + slotKey + '-kinds');
+    if (!picker || !hidden) return;
+
+    // Read existing selection so we can pre-check matching boxes after render.
+    let selected = [];
+    try {
+        const initial = hidden.dataset.initial || hidden.value || '[]';
+        selected = JSON.parse(initial);
+        if (!Array.isArray(selected)) selected = [];
+    } catch (_) { selected = []; }
+    const selectedKey = (kind, variant) => `${kind}::${variant || ''}`;
+    const selectedSet = new Set(selected.map(s => selectedKey(s.kind, s.variant)));
+
+    // Fetch + cache the kinds catalog.
+    if (!_autoPersonalizedKinds) {
+        try {
+            const res = await fetch('/api/personalized/kinds');
+            const data = await res.json();
+            _autoPersonalizedKinds = (data && data.kinds) || [];
+        } catch (e) {
+            _autoPersonalizedKinds = [];
+        }
+    }
+
+    if (!_autoPersonalizedKinds.length) {
+        picker.innerHTML = '<div style="color:rgba(255,255,255,0.4);font-size:12px;">No personalized playlists registered. Restart the server if you just upgraded.</div>';
+        return;
+    }
+
+    // Render: one row per (kind, variant). Singletons get one row;
+    // variant kinds get one row per resolved variant + an "all" toggle.
+    // Inline styles override the parent `.placed-block-config label`
+    // rule (uppercase / min-width / letter-spacing) — without these
+    // overrides every row would render as a tiny narrow chip.
+    const rowStyle = 'display:flex;align-items:center;gap:8px;padding:4px 0;cursor:pointer;font-size:13px;text-transform:none;letter-spacing:0;color:rgba(255,255,255,0.85);min-width:0;';
+    const variantRowStyle = rowStyle + 'padding-left:20px;font-size:12px;color:rgba(255,255,255,0.7);';
+    const sectionHeader = 'font-weight:600;margin:8px 0 2px;color:rgba(255,255,255,0.85);font-size:13px;text-transform:none;letter-spacing:0;';
+    let html = '';
+    _autoPersonalizedKinds.forEach(spec => {
+        const baseLabel = spec.name_template
+            ? spec.name_template.replace('{variant}', '').replace(/\s*[—-]\s*$/,'').trim()
+            : spec.kind;
+
+        if (!spec.requires_variant) {
+            // Singleton: one checkbox.
+            const key = selectedKey(spec.kind, '');
+            const checked = selectedSet.has(key) ? ' checked' : '';
+            html += `<label style="${rowStyle}font-weight:600;">
+                <input type="checkbox" data-kind="${_escAttr(spec.kind)}"${checked}>
+                <span>${_escAttr(baseLabel)}</span>
+            </label>`;
+        } else {
+            const variants = Array.isArray(spec.variants) ? spec.variants : [];
+            html += `<div style="margin:6px 0 4px;border-top:1px solid rgba(255,255,255,0.05);padding-top:6px;">
+                <div style="${sectionHeader}">${_escAttr(baseLabel)}</div>`;
+            if (variants.length === 0) {
+                html += `<div style="font-size:11px;color:rgba(255,255,255,0.35);padding-left:20px;text-transform:none;letter-spacing:0;">(no variants available — populate via the discover page first)</div>`;
+            } else {
+                variants.forEach(variant => {
+                    const key = selectedKey(spec.kind, variant);
+                    const checked = selectedSet.has(key) ? ' checked' : '';
+                    html += `<label style="${variantRowStyle}">
+                        <input type="checkbox" data-kind="${_escAttr(spec.kind)}" data-variant="${_escAttr(variant)}"${checked}>
+                        <span>${_escAttr(variant)}</span>
+                    </label>`;
+                });
+            }
+            html += '</div>';
+        }
+    });
+    picker.innerHTML = html;
+}
+
 async function _autoLoadMirroredSelects() {
     const selects = document.querySelectorAll('.mirrored-playlist-select');
     const nameSelects = document.querySelectorAll('.mirrored-playlist-name-select');
@@ -6265,6 +6384,26 @@ function _readPlacedConfig(slotKey) {
             skip_wishlist: skipWl ? skipWl.checked : false,
         };
     }
+    if (type === 'personalized_pipeline') {
+        // Each checked box has data-kind / data-variant attributes;
+        // walk them to assemble the kinds list.
+        const picker = document.getElementById('cfg-' + slotKey + '-kinds-picker');
+        const kinds = [];
+        if (picker) {
+            picker.querySelectorAll('input[type="checkbox"][data-kind]:checked').forEach(cb => {
+                const entry = { kind: cb.dataset.kind };
+                if (cb.dataset.variant) entry.variant = cb.dataset.variant;
+                kinds.push(entry);
+            });
+        }
+        const refreshFirstCb = document.getElementById('cfg-' + slotKey + '-refresh_first');
+        const skipWl = document.getElementById('cfg-' + slotKey + '-skip_wishlist');
+        return {
+            kinds,
+            refresh_first: refreshFirstCb ? refreshFirstCb.checked : false,
+            skip_wishlist: skipWl ? skipWl.checked : false,
+        };
+    }
     if (type === 'signal_received' || type === 'fire_signal') {
         return { signal_name: document.getElementById('cfg-' + slotKey + '-signal_name')?.value?.trim() || '' };
     }