Extract automation handlers (3/3): maintenance + misc, finishing the lift

Final commit of the automation-handler refactor. With this commit
every closure that used to live in
`web_server._register_automation_handlers` is now a top-level
function in `core/automation/handlers/`.

Handlers extracted in this commit:

- start_database_update + deep_scan_library
    -> core/automation/handlers/database_update.py
    Both share the db_update_state monitoring pattern (poll until
    status flips, stall detection emits warning at 10 min, 2-hour
    outer timeout). Lifted into a shared `_run_with_progress` helper
    inside the module so the per-handler bodies stay tiny.

- run_duplicate_cleaner -> core/automation/handlers/duplicate_cleaner.py
- start_quality_scan    -> core/automation/handlers/quality_scanner.py

- clear_quarantine, cleanup_wishlist, update_discovery_pool,
  backup_database, refresh_beatport_cache
    -> core/automation/handlers/maintenance.py
    Grouped because each body is short (~20-50 lines) and they share
    no state — splitting into per-handler files would just add import
    noise.

- clean_search_history, clean_completed_downloads, full_cleanup
    -> core/automation/handlers/download_cleanup.py
    Grouped because all three reach the download orchestrator,
    tasks_lock, and download_batches/download_tasks accessors. The
    full_cleanup multi-step orchestration shares phase-detection
    logic with clean_completed_downloads.

- run_script         -> core/automation/handlers/run_script.py
- search_and_download -> core/automation/handlers/search_and_download.py

`AutomationDeps` grew with the new dependency surface:
- get_db_update_state + db_update_lock + db_update_executor +
  run_db_update_task + run_deep_scan_task
- get_duplicate_cleaner_state + duplicate_cleaner_lock +
  duplicate_cleaner_executor + run_duplicate_cleaner
- get_quality_scanner_state + quality_scanner_lock +
  quality_scanner_executor + run_quality_scanner
- download_orchestrator + run_async + tasks_lock +
  get_download_batches + get_download_tasks +
  sweep_empty_download_directories + get_staging_path
- docker_resolve_path + get_current_profile_id +
  get_watchlist_scanner + get_app + get_beatport_data_cache
- set_db_update_automation_id (writes the legacy global so the live
  DB-update progress callbacks still living in web_server.py keep
  emitting against the active automation card)

`web_server._register_automation_handlers` is now ~50 lines: build
deps once, call register_all. The 667-line block of remaining
closure definitions and engine register calls is gone.

The final orphan was the `_db_update_automation_id` module global —
the DB-update progress callbacks at line ~14080 still read it
directly, so the extracted database_update handler propagates the
automation id through `deps.set_db_update_automation_id` (a closure
in web_server that writes the global). When the legacy callbacks
get extracted in a future PR the setter goes away.

Tests:
- tests/automation/test_handlers_maintenance.py adds 21 boundary
  tests covering every newly-extracted handler shape: guard
  short-circuits (already-running returns skipped), deps wiring
  (set_db_update_automation_id called with the right id),
  exception swallow contract, status returns, path-traversal
  blocked in run_script, source-mode skip in clean_search_history,
  active-batch skip in clean_completed_downloads, etc.
- 3244 tests pass (was 3223 — 21 new), no regression.

web_server.py: 35,593 -> 34,220 lines (-1,373 net across 3 commits).
Issue #1 from the extraction punch list is now COMPLETE.

core/automation/deps.py

@@ -101,3 +101,35 @@ class AutomationDeps:
     get_deezer_client: Callable[[], Any]
     parse_youtube_playlist: Callable[[str], Any]
     get_sync_states: Callable[[], dict]                  # accessor returns the live dict shared with the sync UI
+
+    # --- Database update + quality scanner (shared state + executors) ---
+    set_db_update_automation_id: Callable[[Optional[str]], None]  # syncs the legacy `_db_update_automation_id` global so the live DB-update progress callbacks (which still read the global directly) keep firing for the active automation
+    get_db_update_state: Callable[[], dict]
+    db_update_lock: Any                                  # threading.Lock
+    db_update_executor: Any                              # ThreadPoolExecutor
+    run_db_update_task: Callable[..., Any]
+    run_deep_scan_task: Callable[..., Any]
+    get_duplicate_cleaner_state: Callable[[], dict]
+    duplicate_cleaner_lock: Any
+    duplicate_cleaner_executor: Any
+    run_duplicate_cleaner: Callable[..., Any]
+    get_quality_scanner_state: Callable[[], dict]
+    quality_scanner_lock: Any
+    quality_scanner_executor: Any
+    run_quality_scanner: Callable[..., Any]
+
+    # --- Download orchestrator + queue accessors ---
+    download_orchestrator: Any
+    run_async: Callable[..., Any]
+    tasks_lock: Any
+    get_download_batches: Callable[[], dict]
+    get_download_tasks: Callable[[], dict]
+    sweep_empty_download_directories: Callable[[], int]
+    get_staging_path: Callable[[], str]
+
+    # --- Maintenance helpers ---
+    docker_resolve_path: Callable[[str], str]
+    get_current_profile_id: Callable[[], int]
+    get_watchlist_scanner: Callable[[Any], Any]
+    get_app: Callable[[], Any]                           # Flask app for test_client (beatport refresh)
+    get_beatport_data_cache: Callable[[], dict]

core/automation/handlers/__init__.py

@@ -17,6 +17,23 @@ 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.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
+from core.automation.handlers.maintenance import (
+    auto_clear_quarantine,
+    auto_cleanup_wishlist,
+    auto_update_discovery_pool,
+    auto_backup_database,
+    auto_refresh_beatport_cache,
+)
+from core.automation.handlers.download_cleanup import (
+    auto_clean_search_history,
+    auto_clean_completed_downloads,
+    auto_full_cleanup,
+)
+from core.automation.handlers.run_script import auto_run_script
+from core.automation.handlers.search_and_download import auto_search_and_download
 from core.automation.handlers.registration import register_all
 
 __all__ = [
@@ -27,5 +44,19 @@ __all__ = [
     'auto_sync_playlist',
     'auto_discover_playlist',
     'auto_playlist_pipeline',
+    'auto_start_database_update',
+    'auto_deep_scan_library',
+    'auto_run_duplicate_cleaner',
+    'auto_start_quality_scan',
+    'auto_clear_quarantine',
+    'auto_cleanup_wishlist',
+    'auto_update_discovery_pool',
+    'auto_backup_database',
+    'auto_refresh_beatport_cache',
+    'auto_clean_search_history',
+    'auto_clean_completed_downloads',
+    'auto_full_cleanup',
+    'auto_run_script',
+    'auto_search_and_download',
     'register_all',
 ]

core/automation/handlers/database_update.py

@@ -0,0 +1,136 @@
+"""Automation handlers: ``start_database_update`` and
+``deep_scan_library`` actions.
+
+Lifted from ``web_server._register_automation_handlers`` (the
+``_auto_start_database_update`` and ``_auto_deep_scan_library``
+closures). Both share the same ``db_update_state`` / executor / lock
+infrastructure -- the only difference is which task they submit
+(``run_db_update_task`` vs ``run_deep_scan_task``).
+
+Pattern: pre-set state to running, submit task to executor, then
+poll the state dict until it transitions away from ``running``.
+Stall-detection emits a warning every 10 minutes when progress
+hasn't budged. 2-hour outer timeout caps the worst case.
+"""
+
+from __future__ import annotations
+
+import time
+from typing import Any, Dict
+
+from core.automation.deps import AutomationDeps
+
+
+_TIMEOUT_SECONDS = 7200          # 2 hours — covers the worst large-library case
+_STALL_WARNING_SECONDS = 600     # 10 minutes without progress = stall
+_POLL_INTERVAL_SECONDS = 3
+_INITIAL_DELAY_SECONDS = 1
+
+
+def auto_start_database_update(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
+    """Run a full or incremental DB update via ``run_db_update_task``."""
+    return _run_with_progress(
+        config, deps,
+        task=deps.run_db_update_task,
+        task_args=(config.get('full_refresh', False), deps.config_manager.get_active_media_server()),
+        initial_phase='Initializing...',
+        stall_label='Database update',
+        finished_extras=lambda: {'full_refresh': str(config.get('full_refresh', False))},
+        timeout_label='Database update timed out after 2 hours',
+    )
+
+
+def auto_deep_scan_library(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
+    """Run a deep library scan via ``run_deep_scan_task``."""
+    return _run_with_progress(
+        config, deps,
+        task=deps.run_deep_scan_task,
+        task_args=(deps.config_manager.get_active_media_server(),),
+        initial_phase='Deep scan: Initializing...',
+        stall_label='Deep scan',
+        finished_extras=lambda: {},
+        timeout_label='Deep scan timed out after 2 hours',
+    )
+
+
+def _run_with_progress(
+    config: Dict[str, Any],
+    deps: AutomationDeps,
+    *,
+    task,
+    task_args: tuple,
+    initial_phase: str,
+    stall_label: str,
+    finished_extras,
+    timeout_label: str,
+) -> Dict[str, Any]:
+    """Shared poll-and-wait body for both DB-update handlers."""
+    automation_id = config.get('_automation_id')
+    state = deps.get_db_update_state()
+    if state.get('status') == 'running':
+        return {'status': 'skipped', 'reason': 'Database update already running'}
+    deps.state.db_update_automation_id = automation_id
+    # Sync legacy module global so the DB-update progress callbacks
+    # (still living in web_server.py) emit against this automation.
+    deps.set_db_update_automation_id(automation_id)
+
+    with deps.db_update_lock:
+        state.update({
+            'status': 'running', 'phase': initial_phase,
+            'progress': 0, 'current_item': '', 'processed': 0, 'total': 0,
+            'error_message': '',
+        })
+    deps.db_update_executor.submit(task, *task_args)
+
+    # Monitor progress (callbacks handle card updates, we just block until done).
+    time.sleep(_INITIAL_DELAY_SECONDS)
+    poll_start = time.time()
+    last_progress_time = time.time()
+    last_progress_val = 0
+    while time.time() - poll_start < _TIMEOUT_SECONDS:
+        time.sleep(_POLL_INTERVAL_SECONDS)
+        with deps.db_update_lock:
+            current_status = state.get('status', 'idle')
+            current_progress = state.get('progress', 0)
+        if current_status != 'running':
+            break
+        # Stall detection — if no progress change in 10 minutes, warn.
+        if current_progress != last_progress_val:
+            last_progress_val = current_progress
+            last_progress_time = time.time()
+        elif time.time() - last_progress_time > _STALL_WARNING_SECONDS:
+            deps.update_progress(
+                automation_id,
+                log_line=f'{stall_label} appears stalled — waiting...',
+                log_type='warning',
+            )
+            last_progress_time = time.time()  # Reset so warning repeats every 10 min.
+    else:
+        # 2-hour timeout reached.
+        deps.update_progress(
+            automation_id, status='error',
+            phase='Timed out', log_line=timeout_label, log_type='error',
+        )
+        return {'status': 'error', 'reason': 'Timed out', '_manages_own_progress': True}
+
+    # Finished/error callback already updated the card — return matching status.
+    with deps.db_update_lock:
+        final_status = state.get('status', 'unknown')
+    if final_status == 'error':
+        return {
+            'status': 'error',
+            'reason': state.get('error_message', 'Unknown error'),
+            '_manages_own_progress': True,
+        }
+    with deps.db_update_lock:
+        stats = {
+            'status': 'completed', '_manages_own_progress': True,
+            'artists': state.get('total', 0),
+            'albums': state.get('total_albums', 0),
+            'tracks': state.get('total_tracks', 0),
+            'removed_artists': state.get('removed_artists', 0),
+            'removed_albums': state.get('removed_albums', 0),
+            'removed_tracks': state.get('removed_tracks', 0),
+        }
+    stats.update(finished_extras())
+    return stats

core/automation/handlers/download_cleanup.py

@@ -0,0 +1,267 @@
+"""Automation handlers: download-queue cleanup actions.
+
+Lifted from ``web_server._register_automation_handlers``:
+- ``clean_search_history``       → :func:`auto_clean_search_history`
+- ``clean_completed_downloads``  → :func:`auto_clean_completed_downloads`
+- ``full_cleanup``               → :func:`auto_full_cleanup`
+
+All three share the download-orchestrator + tasks_lock /
+download_batches / download_tasks accessors. ``full_cleanup`` is a
+multi-step orchestration that pulls in quarantine purge + staging
+sweep on top of the queue cleanup -- kept as one big handler since
+its phases share state-detection logic.
+"""
+
+from __future__ import annotations
+
+import os
+import shutil as _shutil
+from typing import Any, Dict
+
+from core.automation.deps import AutomationDeps
+
+
+# ─── clean_search_history ────────────────────────────────────────────
+
+
+def auto_clean_search_history(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
+    """Remove old searches from Soulseek when configured."""
+    automation_id = config.get('_automation_id')
+    # Skip if soulseek is not the active download source or in hybrid order.
+    dl_mode = deps.config_manager.get('download_source.mode', 'hybrid')
+    hybrid_order = deps.config_manager.get(
+        'download_source.hybrid_order', ['hifi', 'youtube', 'soulseek'],
+    )
+    soulseek_active = (
+        dl_mode == 'soulseek'
+        or (dl_mode == 'hybrid' and 'soulseek' in hybrid_order)
+    )
+    # Reach the underlying SoulseekClient via the orchestrator's
+    # generic accessor.
+    slskd = deps.download_orchestrator.client('soulseek') if deps.download_orchestrator else None
+    if not soulseek_active or not slskd or not slskd.base_url:
+        deps.update_progress(automation_id, log_line='Soulseek not active — skipped', log_type='skip')
+        return {'status': 'skipped'}
+    if not deps.config_manager.get('soulseek.auto_clear_searches', True):
+        deps.update_progress(
+            automation_id, log_line='Auto-clear disabled in settings', log_type='skip',
+        )
+        return {'status': 'skipped'}
+    try:
+        success = deps.run_async(deps.download_orchestrator.maintain_search_history_with_buffer(
+            keep_searches=50, trigger_threshold=200,
+        ))
+        if success:
+            deps.update_progress(
+                automation_id,
+                log_line='Search history maintenance completed',
+                log_type='success',
+            )
+            return {'status': 'completed'}
+        else:
+            deps.update_progress(automation_id, log_line='No cleanup needed', log_type='skip')
+            return {'status': 'completed'}
+    except Exception as e:  # noqa: BLE001 — automation handlers must never raise
+        return {'status': 'error', 'error': str(e)}
+
+
+# ─── clean_completed_downloads ───────────────────────────────────────
+
+
+def auto_clean_completed_downloads(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
+    """Clear completed downloads + sweep empty download directories.
+    Skips when active batches or post-processing is in flight."""
+    automation_id = config.get('_automation_id')
+    try:
+        has_active_batches = False
+        has_post_processing = False
+        with deps.tasks_lock:
+            batches = deps.get_download_batches()
+            for batch_data in batches.values():
+                if batch_data.get('phase') not in ['complete', 'error', 'cancelled', None]:
+                    has_active_batches = True
+                    break
+            if not has_active_batches:
+                tasks = deps.get_download_tasks()
+                for task_data in tasks.values():
+                    if task_data.get('status') == 'post_processing':
+                        has_post_processing = True
+                        break
+
+        if has_active_batches:
+            deps.update_progress(
+                automation_id, log_line='Skipped — downloads active', log_type='skip',
+            )
+            return {'status': 'completed'}
+
+        deps.run_async(deps.download_orchestrator.clear_all_completed_downloads())
+        if not has_post_processing:
+            deps.sweep_empty_download_directories()
+        deps.update_progress(
+            automation_id, log_line='Download cleanup completed', log_type='success',
+        )
+        return {'status': 'completed'}
+    except Exception as e:  # noqa: BLE001 — automation handlers must never raise
+        return {'status': 'error', 'reason': str(e)}
+
+
+# ─── full_cleanup ────────────────────────────────────────────────────
+
+
+def auto_full_cleanup(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
+    """Run all cleanup tasks: quarantine purge → download queue clear
+    → empty-dir sweep → staging sweep → search history."""
+    automation_id = config.get('_automation_id')
+    steps = []
+
+    # --- 1. Clear quarantine ---
+    deps.update_progress(automation_id, phase='Clearing quarantine...', progress=0)
+    quarantine_path = os.path.join(
+        deps.docker_resolve_path(deps.config_manager.get('soulseek.download_path', './downloads')),
+        'ss_quarantine',
+    )
+    q_removed = 0
+    if os.path.exists(quarantine_path):
+        for f in os.listdir(quarantine_path):
+            fp = os.path.join(quarantine_path, f)
+            try:
+                if os.path.isfile(fp):
+                    os.remove(fp)
+                    q_removed += 1
+                elif os.path.isdir(fp):
+                    _shutil.rmtree(fp)
+                    q_removed += 1
+            except Exception as e:  # noqa: BLE001 — best-effort purge
+                deps.logger.debug("quarantine entry purge failed: %s", e)
+    steps.append(f'Quarantine: removed {q_removed} items')
+    deps.update_progress(
+        automation_id,
+        log_line=f'Quarantine: removed {q_removed} items',
+        log_type='success' if q_removed else 'info',
+    )
+
+    # --- 2. Clear completed/errored/cancelled downloads from Soulseek queue ---
+    deps.update_progress(automation_id, phase='Clearing download queue...', progress=20)
+    has_active_batches = False
+    has_post_processing = False
+    with deps.tasks_lock:
+        batches = deps.get_download_batches()
+        for batch_data in batches.values():
+            if batch_data.get('phase') not in ['complete', 'error', 'cancelled', None]:
+                has_active_batches = True
+                break
+        if not has_active_batches:
+            tasks = deps.get_download_tasks()
+            for task_data in tasks.values():
+                if task_data.get('status') == 'post_processing':
+                    has_post_processing = True
+                    break
+    if has_active_batches:
+        steps.append('Download queue: skipped (active batches)')
+        deps.update_progress(
+            automation_id,
+            log_line='Download queue: skipped (active batches)',
+            log_type='skip',
+        )
+    else:
+        try:
+            deps.run_async(deps.download_orchestrator.clear_all_completed_downloads())
+            steps.append('Download queue: cleared')
+            deps.update_progress(
+                automation_id, log_line='Download queue: cleared', log_type='success',
+            )
+        except Exception as e:  # noqa: BLE001 — per-step best-effort
+            steps.append(f'Download queue: error ({e})')
+            deps.update_progress(
+                automation_id,
+                log_line=f'Download queue: error ({e})',
+                log_type='error',
+            )
+
+    # --- 3. Sweep empty download directories ---
+    deps.update_progress(automation_id, phase='Sweeping empty directories...', progress=40)
+    if has_active_batches or has_post_processing:
+        reason = 'active batches' if has_active_batches else 'post-processing active'
+        steps.append(f'Empty directories: skipped ({reason})')
+        deps.update_progress(
+            automation_id,
+            log_line=f'Empty directories: skipped ({reason})',
+            log_type='skip',
+        )
+    else:
+        dirs_removed = deps.sweep_empty_download_directories()
+        steps.append(f'Empty directories: removed {dirs_removed}')
+        deps.update_progress(
+            automation_id,
+            log_line=f'Empty directories: removed {dirs_removed}',
+            log_type='success' if dirs_removed else 'info',
+        )
+
+    # --- 4. Sweep empty staging directories ---
+    deps.update_progress(automation_id, phase='Sweeping import folder...', progress=60)
+    staging_path = deps.get_staging_path()
+    s_removed = 0
+    if os.path.isdir(staging_path):
+        for dirpath, _dirnames, _filenames in os.walk(staging_path, topdown=False):
+            if os.path.normpath(dirpath) == os.path.normpath(staging_path):
+                continue
+            try:
+                entries = os.listdir(dirpath)
+            except OSError:
+                continue
+            visible = [e for e in entries if not e.startswith('.')]
+            if not visible:
+                for hidden in entries:
+                    try:
+                        os.remove(os.path.join(dirpath, hidden))
+                    except Exception as e:  # noqa: BLE001 — best-effort
+                        deps.logger.debug("hidden file cleanup failed: %s", e)
+                try:
+                    os.rmdir(dirpath)
+                    s_removed += 1
+                except OSError:
+                    pass
+    steps.append(f'Staging: removed {s_removed} empty directories')
+    deps.update_progress(
+        automation_id,
+        log_line=f'Staging: removed {s_removed} empty directories',
+        log_type='success' if s_removed else 'info',
+    )
+
+    # --- 5. Clean search history (if enabled) ---
+    deps.update_progress(automation_id, phase='Cleaning search history...', progress=80)
+    try:
+        if not deps.config_manager.get('soulseek.auto_clear_searches', True):
+            steps.append('Search cleanup: disabled in settings')
+            deps.update_progress(
+                automation_id, log_line='Search cleanup: disabled in settings', log_type='skip',
+            )
+        else:
+            deps.run_async(deps.download_orchestrator.maintain_search_history_with_buffer(
+                keep_searches=50, trigger_threshold=200,
+            ))
+            steps.append('Search history: cleaned')
+            deps.update_progress(
+                automation_id, log_line='Search history: cleaned', log_type='success',
+            )
+    except Exception as e:  # noqa: BLE001 — per-step best-effort
+        steps.append(f'Search history: error ({e})')
+        deps.update_progress(
+            automation_id, log_line=f'Search history: error ({e})', log_type='error',
+        )
+
+    total_removed = q_removed + s_removed
+    deps.update_progress(
+        automation_id, status='finished', progress=100,
+        phase='Complete',
+        log_line=f'Full cleanup complete — {total_removed} items removed',
+        log_type='success',
+    )
+    return {
+        'status': 'completed',
+        'quarantine_removed': str(q_removed),
+        'staging_removed': str(s_removed),
+        'total_removed': str(total_removed),
+        'steps': steps,
+        '_manages_own_progress': True,
+    }

core/automation/handlers/duplicate_cleaner.py

@@ -0,0 +1,87 @@
+"""Automation handler: ``run_duplicate_cleaner`` action.
+
+Lifted from ``web_server._register_automation_handlers`` (the
+``_auto_run_duplicate_cleaner`` closure). Submits the duplicate
+cleaner to its executor, then polls the shared state dict until
+the worker transitions away from ``running``.
+"""
+
+from __future__ import annotations
+
+import time
+from typing import Any, Dict
+
+from core.automation.deps import AutomationDeps
+
+
+_TIMEOUT_SECONDS = 7200          # 2 hours
+_POLL_INTERVAL_SECONDS = 3
+_INITIAL_DELAY_SECONDS = 1
+
+
+def auto_run_duplicate_cleaner(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
+    """Kick off the duplicate cleaner and report final stats."""
+    automation_id = config.get('_automation_id')
+    state = deps.get_duplicate_cleaner_state()
+    if state.get('status') == 'running':
+        return {'status': 'skipped', 'reason': 'Duplicate cleaner already running'}
+
+    # Pre-set status before submit so the polling loop doesn't see a
+    # stale 'finished' from a previous run.
+    with deps.duplicate_cleaner_lock:
+        state['status'] = 'running'
+    deps.duplicate_cleaner_executor.submit(deps.run_duplicate_cleaner)
+    deps.update_progress(automation_id, log_line='Duplicate cleaner started', log_type='info')
+
+    # Monitor progress (max 2 hours).
+    time.sleep(_INITIAL_DELAY_SECONDS)
+    poll_start = time.time()
+    while time.time() - poll_start < _TIMEOUT_SECONDS:
+        time.sleep(_POLL_INTERVAL_SECONDS)
+        current_status = state.get('status', 'idle')
+        if current_status not in ('running',):
+            break
+        deps.update_progress(
+            automation_id,
+            phase=state.get('phase', 'Scanning...'),
+            progress=state.get('progress', 0),
+            processed=state.get('files_scanned', 0),
+            total=state.get('total_files', 0),
+        )
+    else:
+        # 2-hour timeout reached.
+        deps.update_progress(
+            automation_id, status='error',
+            phase='Timed out',
+            log_line='Duplicate cleaner timed out after 2 hours',
+            log_type='error',
+        )
+        return {'status': 'error', 'reason': 'Timed out', '_manages_own_progress': True}
+
+    # Check actual exit status (could be 'finished' or 'error').
+    final_status = state.get('status', 'idle')
+    if final_status == 'error':
+        err = state.get('error_message', 'Unknown error')
+        deps.update_progress(
+            automation_id, status='error', progress=100,
+            phase='Error', log_line=err, log_type='error',
+        )
+        return {'status': 'error', 'reason': err, '_manages_own_progress': True}
+
+    dupes = state.get('duplicates_found', 0)
+    removed = state.get('deleted', 0)
+    space_freed = state.get('space_freed', 0)
+    scanned = state.get('files_scanned', 0)
+    deps.update_progress(
+        automation_id, status='finished', progress=100,
+        phase='Complete',
+        log_line=f'Found {dupes} duplicates, removed {removed} files',
+        log_type='success',
+    )
+    return {
+        'status': 'completed', '_manages_own_progress': True,
+        'files_scanned': scanned,
+        'duplicates_found': dupes,
+        'files_deleted': removed,
+        'space_freed_mb': round(space_freed / (1024 * 1024), 1),
+    }

core/automation/handlers/maintenance.py

@@ -0,0 +1,213 @@
+"""Automation handlers: short maintenance actions.
+
+Lifted from ``web_server._register_automation_handlers``:
+- ``clear_quarantine``       → :func:`auto_clear_quarantine`
+- ``cleanup_wishlist``       → :func:`auto_cleanup_wishlist`
+- ``update_discovery_pool``  → :func:`auto_update_discovery_pool`
+- ``backup_database``        → :func:`auto_backup_database`
+- ``refresh_beatport_cache`` → :func:`auto_refresh_beatport_cache`
+
+Each is a thin wrapper around an existing service / helper. Grouped
+in one module because every body is short and they share no state
+between them — splitting into per-handler files would just add
+import noise.
+"""
+
+from __future__ import annotations
+
+import glob as _glob
+import os
+import shutil as _shutil
+import sqlite3
+import time
+from datetime import datetime
+from typing import Any, Dict
+
+from core.automation.deps import AutomationDeps
+
+
+# ─── clear_quarantine ────────────────────────────────────────────────
+
+
+def auto_clear_quarantine(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
+    """Purge every file/folder under the configured ss_quarantine path."""
+    automation_id = config.get('_automation_id')
+    quarantine_path = os.path.join(
+        deps.docker_resolve_path(deps.config_manager.get('soulseek.download_path', './downloads')),
+        'ss_quarantine',
+    )
+    if not os.path.exists(quarantine_path):
+        deps.update_progress(automation_id, log_line='No quarantine folder found', log_type='info')
+        return {'status': 'completed', 'removed': '0'}
+    removed = 0
+    for f in os.listdir(quarantine_path):
+        fp = os.path.join(quarantine_path, f)
+        try:
+            if os.path.isfile(fp):
+                os.remove(fp)
+                removed += 1
+            elif os.path.isdir(fp):
+                _shutil.rmtree(fp)
+                removed += 1
+        except Exception as e:  # noqa: BLE001 — best-effort purge
+            deps.logger.debug("quarantine entry purge failed: %s", e)
+    deps.update_progress(
+        automation_id,
+        log_line=f'Removed {removed} quarantined items',
+        log_type='success' if removed > 0 else 'info',
+    )
+    return {'status': 'completed', 'removed': str(removed)}
+
+
+# ─── cleanup_wishlist ────────────────────────────────────────────────
+
+
+def auto_cleanup_wishlist(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
+    """Drop duplicate entries from the wishlist for the active profile."""
+    automation_id = config.get('_automation_id')
+    db = deps.get_database()
+    removed = db.remove_wishlist_duplicates(deps.get_current_profile_id())
+    deps.update_progress(
+        automation_id,
+        log_line=f'Removed {removed or 0} duplicate wishlist entries',
+        log_type='success' if removed else 'info',
+    )
+    return {'status': 'completed', 'removed': str(removed or 0)}
+
+
+# ─── update_discovery_pool ───────────────────────────────────────────
+
+
+def auto_update_discovery_pool(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
+    """Run an incremental refresh of the discovery pool via the
+    watchlist scanner."""
+    automation_id = config.get('_automation_id')
+    try:
+        scanner = deps.get_watchlist_scanner(deps.spotify_client)
+        deps.update_progress(automation_id, log_line='Updating discovery pool...', log_type='info')
+        scanner.update_discovery_pool_incremental(deps.get_current_profile_id())
+        deps.update_progress(
+            automation_id, status='finished', progress=100,
+            phase='Complete', log_line='Discovery pool updated', log_type='success',
+        )
+        return {'status': 'completed', '_manages_own_progress': True}
+    except Exception as e:  # noqa: BLE001 — automation handlers must never raise
+        deps.update_progress(
+            automation_id, status='error',
+            phase='Error', log_line=str(e), log_type='error',
+        )
+        return {'status': 'error', 'reason': str(e), '_manages_own_progress': True}
+
+
+# ─── backup_database ─────────────────────────────────────────────────
+
+
+_MAX_BACKUPS = 5
+
+
+def auto_backup_database(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
+    """Create a hot SQLite backup, then prune old backups so only the
+    newest ``_MAX_BACKUPS`` remain."""
+    automation_id = config.get('_automation_id')
+    db_path = os.environ.get('DATABASE_PATH', 'database/music_library.db')
+    if not os.path.exists(db_path):
+        return {'status': 'error', 'reason': 'Database file not found'}
+
+    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
+    backup_path = f"{db_path}.backup_{timestamp}"
+    # Use SQLite backup API for a safe hot-copy of an active database.
+    src = sqlite3.connect(db_path)
+    dst = sqlite3.connect(backup_path)
+    src.backup(dst)
+    dst.close()
+    src.close()
+    size_mb = round(os.path.getsize(backup_path) / (1024 * 1024), 1)
+
+    # Rolling cleanup — keep only the newest N backups.
+    existing = sorted(_glob.glob(f"{db_path}.backup_*"), key=os.path.getmtime)
+    while len(existing) > _MAX_BACKUPS:
+        try:
+            os.remove(existing.pop(0))
+        except Exception as e:  # noqa: BLE001 — best-effort cleanup
+            deps.logger.debug("rolling backup cleanup failed: %s", e)
+    deps.update_progress(
+        automation_id,
+        log_line=f'Backup created: {size_mb}MB ({os.path.basename(backup_path)})',
+        log_type='success',
+    )
+    return {'status': 'completed', 'backup_path': backup_path, 'size_mb': str(size_mb)}
+
+
+# ─── refresh_beatport_cache ──────────────────────────────────────────
+
+
+_BEATPORT_SECTIONS = (
+    ('hero_tracks', '/api/beatport/hero-tracks', 'Hero Tracks'),
+    ('new_releases', '/api/beatport/new-releases', 'New Releases'),
+    ('featured_charts', '/api/beatport/featured-charts', 'Featured Charts'),
+    ('dj_charts', '/api/beatport/dj-charts', 'DJ Charts'),
+    ('top_10_lists', '/api/beatport/homepage/top-10-lists', 'Top 10 Lists'),
+    ('top_10_releases', '/api/beatport/homepage/top-10-releases-cards', 'Top 10 Releases'),
+    ('hype_picks', '/api/beatport/hype-picks', 'Hype Picks'),
+)
+
+
+def auto_refresh_beatport_cache(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
+    """Refresh Beatport homepage cache by calling each endpoint internally
+    via Flask's ``test_client``. Invalidates the homepage cache first
+    so endpoints re-scrape rather than returning stale data."""
+    automation_id = config.get('_automation_id')
+    cache = deps.get_beatport_data_cache()
+    # Invalidate all homepage cache timestamps so endpoints re-scrape.
+    with cache['cache_lock']:
+        for key in cache['homepage']:
+            cache['homepage'][key]['timestamp'] = 0
+            cache['homepage'][key]['data'] = None
+
+    refreshed = 0
+    errors = []
+    app = deps.get_app()
+    with app.test_client() as client:
+        for idx, (_, endpoint, label) in enumerate(_BEATPORT_SECTIONS):
+            deps.update_progress(
+                automation_id,
+                progress=(idx / len(_BEATPORT_SECTIONS)) * 100,
+                phase=f'Scraping: {label}',
+                current_item=label,
+            )
+            try:
+                resp = client.get(endpoint)
+                if resp.status_code == 200:
+                    refreshed += 1
+                    deps.update_progress(
+                        automation_id, log_line=f'{label}: cached', log_type='success',
+                    )
+                else:
+                    errors.append(label)
+                    deps.update_progress(
+                        automation_id,
+                        log_line=f'{label}: HTTP {resp.status_code}',
+                        log_type='error',
+                    )
+            except Exception as e:  # noqa: BLE001 — per-section best-effort
+                errors.append(label)
+                deps.update_progress(
+                    automation_id,
+                    log_line=f'{label}: {str(e)}',
+                    log_type='error',
+                )
+            if idx < len(_BEATPORT_SECTIONS) - 1:
+                time.sleep(2)
+
+    deps.update_progress(
+        automation_id, status='finished', progress=100,
+        phase='Complete',
+        log_line=f'Refreshed {refreshed}/{len(_BEATPORT_SECTIONS)} sections',
+        log_type='success',
+    )
+    return {
+        'status': 'completed',
+        'refreshed': str(refreshed),
+        'errors': str(len(errors)),
+        '_manages_own_progress': True,
+    }

core/automation/handlers/quality_scanner.py

@@ -0,0 +1,83 @@
+"""Automation handler: ``start_quality_scan`` action.
+
+Lifted from ``web_server._register_automation_handlers`` (the
+``_auto_start_quality_scan`` closure). Submits the quality scanner
+to its executor with the configured scope (default: ``watchlist``)
+then polls the shared state dict.
+"""
+
+from __future__ import annotations
+
+import time
+from typing import Any, Dict
+
+from core.automation.deps import AutomationDeps
+
+
+_TIMEOUT_SECONDS = 7200          # 2 hours
+_POLL_INTERVAL_SECONDS = 3
+_INITIAL_DELAY_SECONDS = 1
+
+
+def auto_start_quality_scan(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
+    automation_id = config.get('_automation_id')
+    state = deps.get_quality_scanner_state()
+    if state.get('status') == 'running':
+        return {'status': 'skipped', 'reason': 'Quality scan already running'}
+
+    scope = config.get('scope', 'watchlist')
+    # Pre-set status before submit so the polling loop doesn't see a
+    # stale 'finished' from a previous run.
+    with deps.quality_scanner_lock:
+        state['status'] = 'running'
+    deps.quality_scanner_executor.submit(deps.run_quality_scanner, scope, deps.get_current_profile_id())
+    deps.update_progress(
+        automation_id, log_line=f'Quality scan started (scope: {scope})', log_type='info',
+    )
+
+    # Monitor progress (max 2 hours).
+    time.sleep(_INITIAL_DELAY_SECONDS)
+    poll_start = time.time()
+    while time.time() - poll_start < _TIMEOUT_SECONDS:
+        time.sleep(_POLL_INTERVAL_SECONDS)
+        current_status = state.get('status', 'idle')
+        if current_status not in ('running',):
+            break
+        deps.update_progress(
+            automation_id,
+            phase=state.get('phase', 'Scanning...'),
+            progress=state.get('progress', 0),
+            processed=state.get('processed', 0),
+            total=state.get('total', 0),
+        )
+    else:
+        deps.update_progress(
+            automation_id, status='error',
+            phase='Timed out', log_line='Quality scan timed out after 2 hours',
+            log_type='error',
+        )
+        return {'status': 'error', 'reason': 'Timed out', '_manages_own_progress': True}
+
+    final_status = state.get('status', 'idle')
+    if final_status == 'error':
+        err = state.get('error_message', 'Unknown error')
+        deps.update_progress(
+            automation_id, status='error', progress=100,
+            phase='Error', log_line=err, log_type='error',
+        )
+        return {'status': 'error', 'reason': err, '_manages_own_progress': True}
+
+    issues = state.get('low_quality', 0)
+    deps.update_progress(
+        automation_id, status='finished', progress=100,
+        phase='Complete',
+        log_line=f'Quality scan complete — {issues} issues found',
+        log_type='success',
+    )
+    return {
+        'status': 'completed', 'scope': scope, '_manages_own_progress': True,
+        'tracks_scanned': state.get('processed', 0),
+        'quality_met': state.get('quality_met', 0),
+        'low_quality': issues,
+        'matched': state.get('matched', 0),
+    }

core/automation/handlers/registration.py

@@ -15,6 +15,25 @@ 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.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
+from core.automation.handlers.maintenance import (
+    auto_clear_quarantine,
+    auto_cleanup_wishlist,
+    auto_update_discovery_pool,
+    auto_backup_database,
+    auto_refresh_beatport_cache,
+)
+from core.automation.handlers.download_cleanup import (
+    auto_clean_search_history,
+    auto_clean_completed_downloads,
+    auto_full_cleanup,
+)
+from core.automation.handlers.run_script import auto_run_script
+from core.automation.handlers.search_and_download import auto_search_and_download
 
 
 def register_all(deps: AutomationDeps) -> None:
@@ -69,3 +88,66 @@ def register_all(deps: AutomationDeps) -> None:
         lambda config: auto_playlist_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.
+    engine.register_action_handler(
+        'start_database_update',
+        lambda config: auto_start_database_update(config, deps),
+        lambda: deps.get_db_update_state().get('status') == 'running',
+    )
+    engine.register_action_handler(
+        'deep_scan_library',
+        lambda config: auto_deep_scan_library(config, deps),
+        lambda: deps.get_db_update_state().get('status') == 'running',
+    )
+    engine.register_action_handler(
+        'run_duplicate_cleaner',
+        lambda config: auto_run_duplicate_cleaner(config, deps),
+        lambda: deps.get_duplicate_cleaner_state().get('status') == 'running',
+    )
+    engine.register_action_handler(
+        'clear_quarantine',
+        lambda config: auto_clear_quarantine(config, deps),
+    )
+    engine.register_action_handler(
+        'cleanup_wishlist',
+        lambda config: auto_cleanup_wishlist(config, deps),
+    )
+    engine.register_action_handler(
+        'update_discovery_pool',
+        lambda config: auto_update_discovery_pool(config, deps),
+    )
+    engine.register_action_handler(
+        'start_quality_scan',
+        lambda config: auto_start_quality_scan(config, deps),
+        lambda: deps.get_quality_scanner_state().get('status') == 'running',
+    )
+    engine.register_action_handler(
+        'backup_database',
+        lambda config: auto_backup_database(config, deps),
+    )
+    engine.register_action_handler(
+        'refresh_beatport_cache',
+        lambda config: auto_refresh_beatport_cache(config, deps),
+    )
+    engine.register_action_handler(
+        'clean_search_history',
+        lambda config: auto_clean_search_history(config, deps),
+    )
+    engine.register_action_handler(
+        'clean_completed_downloads',
+        lambda config: auto_clean_completed_downloads(config, deps),
+    )
+    engine.register_action_handler(
+        'full_cleanup',
+        lambda config: auto_full_cleanup(config, deps),
+    )
+    engine.register_action_handler(
+        'run_script',
+        lambda config: auto_run_script(config, deps),
+    )
+    engine.register_action_handler(
+        'search_and_download',
+        lambda config: auto_search_and_download(config, deps),
+    )

core/automation/handlers/run_script.py

@@ -0,0 +1,103 @@
+"""Automation handler: ``run_script`` action.
+
+Lifted from ``web_server._register_automation_handlers`` (the
+``_auto_run_script`` closure). Runs a user-provided shell or Python
+script from the configured scripts directory with bounded timeout +
+captured stdout/stderr. Path-traversal guard ensures users can't
+escape the scripts directory.
+
+Environment variables exposed to the script:
+- ``SOULSYNC_EVENT``: triggering event type (when fired by an event)
+- ``SOULSYNC_AUTOMATION``: automation name
+- ``SOULSYNC_SCRIPTS_DIR``: absolute path to the scripts dir
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess as _sp
+from typing import Any, Dict
+
+from core.automation.deps import AutomationDeps
+
+
+_MAX_TIMEOUT_SECONDS = 300  # Hard cap on user-supplied timeout config.
+
+
+def auto_run_script(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
+    script_name = config.get('script_name', '')
+    timeout = min(int(config.get('timeout', 60)), _MAX_TIMEOUT_SECONDS)
+    automation_id = config.get('_automation_id')
+
+    if not script_name:
+        return {'status': 'error', 'error': 'No script selected'}
+
+    scripts_dir = deps.docker_resolve_path(deps.config_manager.get('scripts.path', './scripts'))
+    if not scripts_dir or not os.path.isdir(scripts_dir):
+        os.makedirs(scripts_dir, exist_ok=True)
+        return {
+            'status': 'error',
+            'error': 'Scripts directory is empty. Add scripts to the scripts/ folder.',
+        }
+
+    script_path = os.path.join(scripts_dir, script_name)
+    script_path = os.path.realpath(script_path)
+
+    # Security: block path traversal — script must resolve under
+    # the scripts dir, no symlinks/.. tricks allowed out.
+    if not script_path.startswith(os.path.realpath(scripts_dir)):
+        return {'status': 'error', 'error': 'Script path traversal blocked'}
+
+    if not os.path.isfile(script_path):
+        return {'status': 'error', 'error': f'Script not found: {script_name}'}
+
+    deps.update_progress(automation_id, phase=f'Running {script_name}...', progress=10)
+
+    # Build environment with SoulSync context.
+    env = os.environ.copy()
+    event_data = config.get('_event_data') or {}
+    env['SOULSYNC_EVENT'] = str(event_data.get('type', ''))
+    env['SOULSYNC_AUTOMATION'] = config.get('_automation_name', '')
+    env['SOULSYNC_SCRIPTS_DIR'] = scripts_dir
+
+    try:
+        # Determine how to run the script.
+        if script_path.endswith('.py'):
+            cmd = ['python', script_path]
+        elif script_path.endswith('.sh'):
+            cmd = ['bash', script_path]
+        else:
+            cmd = [script_path]
+
+        result = _sp.run(
+            cmd,
+            capture_output=True, text=True, timeout=timeout,
+            cwd=scripts_dir, env=env,
+        )
+
+        deps.update_progress(automation_id, phase='Script completed', progress=100)
+
+        stdout = result.stdout[:2000] if result.stdout else ''
+        stderr = result.stderr[:1000] if result.stderr else ''
+
+        if result.returncode == 0:
+            deps.logger.info(f"Script '{script_name}' completed (exit 0)")
+        else:
+            deps.logger.warning(f"Script '{script_name}' exited with code {result.returncode}")
+
+        return {
+            'status': 'completed' if result.returncode == 0 else 'error',
+            'exit_code': str(result.returncode),
+            'stdout': stdout,
+            'stderr': stderr,
+            'script': script_name,
+        }
+    except _sp.TimeoutExpired:
+        deps.update_progress(automation_id, phase='Script timed out', progress=100)
+        return {
+            'status': 'error',
+            'error': f'Script timed out after {timeout}s',
+            'script': script_name,
+        }
+    except Exception as e:  # noqa: BLE001 — automation handlers must never raise
+        return {'status': 'error', 'error': str(e), 'script': script_name}

core/automation/handlers/search_and_download.py

@@ -0,0 +1,57 @@
+"""Automation handler: ``search_and_download`` action.
+
+Lifted from ``web_server._register_automation_handlers`` (the
+``_auto_search_and_download`` closure). Searches for a track by
+name/artist string and dispatches the best match through the
+download orchestrator. Query can come from the trigger config
+(direct value) or from event data (e.g. webhook payload).
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict
+
+from core.automation.deps import AutomationDeps
+
+
+def auto_search_and_download(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
+    automation_id = config.get('_automation_id')
+    query = config.get('query', '').strip()
+    # Event-triggered: pull query from event data (e.g. webhook_received).
+    if not query:
+        event_data = config.get('_event_data', {})
+        query = (event_data.get('query', '') or '').strip()
+    if not query:
+        if automation_id:
+            deps.update_progress(
+                automation_id, log_line='No search query provided', log_type='error',
+            )
+        return {'status': 'error', 'error': 'No search query provided'}
+    try:
+        if automation_id:
+            deps.update_progress(
+                automation_id, phase='Searching',
+                log_line=f'Searching: {query}', log_type='info',
+            )
+        result = deps.run_async(deps.download_orchestrator.search_and_download_best(query))
+        if result:
+            if automation_id:
+                deps.update_progress(
+                    automation_id,
+                    log_line=f'Download started for: {query}',
+                    log_type='success',
+                )
+            return {'status': 'completed', 'query': query, 'download_id': result}
+        if automation_id:
+            deps.update_progress(
+                automation_id,
+                log_line=f'No match found for: {query}',
+                log_type='warning',
+            )
+        return {'status': 'not_found', 'query': query, 'error': 'No match found'}
+    except Exception as e:  # noqa: BLE001 — automation handlers must never raise
+        if automation_id:
+            deps.update_progress(
+                automation_id, log_line=f'Error: {e}', log_type='error',
+            )
+        return {'status': 'error', 'query': query, 'error': str(e)}

tests/automation/test_handlers_maintenance.py

@@ -0,0 +1,390 @@
+"""Boundary tests for the maintenance + misc automation handlers
+(database_update / deep_scan_library / duplicate_cleaner /
+quality_scanner / clear_quarantine / cleanup_wishlist /
+update_discovery_pool / backup_database / refresh_beatport_cache /
+clean_search_history / clean_completed_downloads / full_cleanup /
+run_script / search_and_download).
+
+Each handler is tested as a pure function via stub deps. The bodies
+are mechanical lifts of the closures that used to live in
+``web_server._register_automation_handlers`` — these tests pin the
+seam (deps wiring, exception swallow contract, return shapes,
+guard short-circuits) so future drift fails here, not at runtime
+against real executors / clients."""
+
+from __future__ import annotations
+
+import os
+import threading
+from typing import Any, Dict, List
+
+import pytest
+
+from core.automation.deps import AutomationDeps, AutomationState
+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
+from core.automation.handlers.maintenance import (
+    auto_clear_quarantine, auto_cleanup_wishlist,
+    auto_update_discovery_pool, auto_backup_database,
+)
+from core.automation.handlers.download_cleanup import (
+    auto_clean_search_history, auto_clean_completed_downloads,
+)
+from core.automation.handlers.run_script import auto_run_script
+from core.automation.handlers.search_and_download import auto_search_and_download
+
+
+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
+
+
+class _StubConfig:
+    def __init__(self, values=None):
+        self._values = values or {}
+
+    def get(self, key, default=None):
+        return self._values.get(key, default)
+
+    def get_active_media_server(self):
+        return 'plex'
+
+
+def _build_deps(**overrides) -> AutomationDeps:
+    defaults = dict(
+        engine=object(),
+        state=AutomationState(),
+        config_manager=_StubConfig(),
+        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': {}},
+    )
+    defaults.update(overrides)
+    return AutomationDeps(**defaults)  # type: ignore[arg-type]
+
+
+# ─── database_update / deep_scan ──────────────────────────────────────
+
+
+class _StubExecutor:
+    def __init__(self):
+        self.submits: List[tuple] = []
+
+    def submit(self, fn, *args, **kwargs):
+        self.submits.append((fn, args, kwargs))
+
+
+class TestDatabaseUpdate:
+    def test_already_running_returns_skipped(self):
+        state = {'status': 'running'}
+        deps = _build_deps(get_db_update_state=lambda: state)
+        result = auto_start_database_update({}, deps)
+        assert result == {'status': 'skipped', 'reason': 'Database update already running'}
+
+    def test_set_db_update_automation_id_called(self):
+        # Handler must propagate the automation id through the deps
+        # setter so the legacy global stays in sync.
+        captured: List[Any] = []
+        state = {'status': 'idle'}
+        # Make the polling loop terminate immediately by flipping
+        # the status as soon as it's set to 'running'.
+        executor = _StubExecutor()
+
+        def fake_task(*_a, **_k):
+            state['status'] = 'finished'
+
+        executor.submit = lambda fn, *a, **k: fake_task()
+        deps = _build_deps(
+            get_db_update_state=lambda: state,
+            db_update_executor=executor,
+            set_db_update_automation_id=lambda v: captured.append(v),
+            run_db_update_task=fake_task,
+        )
+        # Replace time.sleep so we don't actually wait
+        import core.automation.handlers.database_update as module
+        original = module.time.sleep
+        module.time.sleep = lambda _: None
+        try:
+            result = auto_start_database_update({'_automation_id': 'auto-1'}, deps)
+        finally:
+            module.time.sleep = original
+        assert captured == ['auto-1']
+        assert result['status'] == 'completed'
+
+
+class TestDeepScan:
+    def test_already_running_returns_skipped(self):
+        state = {'status': 'running'}
+        deps = _build_deps(get_db_update_state=lambda: state)
+        result = auto_deep_scan_library({}, deps)
+        assert result == {'status': 'skipped', 'reason': 'Database update already running'}
+
+
+# ─── duplicate_cleaner ────────────────────────────────────────────────
+
+
+class TestDuplicateCleaner:
+    def test_already_running_returns_skipped(self):
+        state = {'status': 'running'}
+        deps = _build_deps(get_duplicate_cleaner_state=lambda: state)
+        result = auto_run_duplicate_cleaner({}, deps)
+        assert result == {'status': 'skipped', 'reason': 'Duplicate cleaner already running'}
+
+
+# ─── quality_scanner ──────────────────────────────────────────────────
+
+
+class TestQualityScanner:
+    def test_already_running_returns_skipped(self):
+        state = {'status': 'running'}
+        deps = _build_deps(get_quality_scanner_state=lambda: state)
+        result = auto_start_quality_scan({}, deps)
+        assert result == {'status': 'skipped', 'reason': 'Quality scan already running'}
+
+
+# ─── clear_quarantine ────────────────────────────────────────────────
+
+
+class TestClearQuarantine:
+    def test_no_quarantine_folder_returns_zero(self, tmp_path):
+        # Point at a non-existent path.
+        deps = _build_deps(
+            config_manager=_StubConfig({'soulseek.download_path': str(tmp_path / 'nonexistent')}),
+            docker_resolve_path=lambda p: p,
+        )
+        result = auto_clear_quarantine({}, deps)
+        assert result == {'status': 'completed', 'removed': '0'}
+
+    def test_clears_files_from_quarantine(self, tmp_path):
+        download_path = tmp_path
+        quarantine_path = download_path / 'ss_quarantine'
+        quarantine_path.mkdir()
+        (quarantine_path / 'a.flac').write_bytes(b'')
+        (quarantine_path / 'b.flac').write_bytes(b'')
+        deps = _build_deps(
+            config_manager=_StubConfig({'soulseek.download_path': str(download_path)}),
+            docker_resolve_path=lambda p: p,
+        )
+        result = auto_clear_quarantine({}, deps)
+        assert result == {'status': 'completed', 'removed': '2'}
+
+
+# ─── cleanup_wishlist ────────────────────────────────────────────────
+
+
+class TestCleanupWishlist:
+    def test_returns_count_from_db(self):
+        class _DB:
+            def remove_wishlist_duplicates(self, profile_id):
+                assert profile_id == 1
+                return 7
+
+        deps = _build_deps(get_database=lambda: _DB())
+        result = auto_cleanup_wishlist({}, deps)
+        assert result == {'status': 'completed', 'removed': '7'}
+
+    def test_returns_zero_when_db_returns_falsey(self):
+        class _DB:
+            def remove_wishlist_duplicates(self, profile_id):
+                return None
+
+        deps = _build_deps(get_database=lambda: _DB())
+        result = auto_cleanup_wishlist({}, deps)
+        assert result == {'status': 'completed', 'removed': '0'}
+
+
+# ─── update_discovery_pool ───────────────────────────────────────────
+
+
+class TestUpdateDiscoveryPool:
+    def test_success(self):
+        called: List[Any] = []
+
+        class _Scanner:
+            def update_discovery_pool_incremental(self, profile_id):
+                called.append(profile_id)
+
+        deps = _build_deps(
+            get_watchlist_scanner=lambda spc: _Scanner(),
+            get_current_profile_id=lambda: 7,
+        )
+        result = auto_update_discovery_pool({}, deps)
+        assert result['status'] == 'completed'
+        assert result['_manages_own_progress'] is True
+        assert called == [7]
+
+    def test_exception_swallowed(self):
+        def boom(_):
+            raise RuntimeError('scanner down')
+
+        deps = _build_deps(get_watchlist_scanner=boom)
+        result = auto_update_discovery_pool({}, deps)
+        assert result['status'] == 'error'
+        assert 'scanner down' in result['reason']
+
+
+# ─── backup_database ─────────────────────────────────────────────────
+
+
+class TestBackupDatabase:
+    def test_missing_database_returns_error(self, tmp_path, monkeypatch):
+        monkeypatch.setenv('DATABASE_PATH', str(tmp_path / 'no.db'))
+        deps = _build_deps()
+        result = auto_backup_database({}, deps)
+        assert result == {'status': 'error', 'reason': 'Database file not found'}
+
+
+# ─── clean_search_history ────────────────────────────────────────────
+
+
+class TestCleanSearchHistory:
+    def test_soulseek_inactive_skips(self):
+        deps = _build_deps(
+            config_manager=_StubConfig({'download_source.mode': 'tidal'}),
+        )
+        result = auto_clean_search_history({}, deps)
+        assert result == {'status': 'skipped'}
+
+    def test_no_orchestrator_skips(self):
+        deps = _build_deps(
+            config_manager=_StubConfig({'download_source.mode': 'soulseek'}),
+            download_orchestrator=None,
+        )
+        result = auto_clean_search_history({}, deps)
+        assert result == {'status': 'skipped'}
+
+
+# ─── clean_completed_downloads ───────────────────────────────────────
+
+
+class TestCleanCompletedDownloads:
+    def test_active_batches_skip(self):
+        # Active batch present → handler returns 'completed' without doing anything.
+        deps = _build_deps(
+            get_download_batches=lambda: {'b1': {'phase': 'downloading'}},
+            get_download_tasks=lambda: {},
+        )
+        result = auto_clean_completed_downloads({}, deps)
+        assert result == {'status': 'completed'}
+
+
+# ─── run_script ──────────────────────────────────────────────────────
+
+
+class TestRunScript:
+    def test_no_script_name_returns_error(self):
+        deps = _build_deps()
+        result = auto_run_script({}, deps)
+        assert result == {'status': 'error', 'error': 'No script selected'}
+
+    def test_path_traversal_blocked(self, tmp_path):
+        scripts_dir = tmp_path / 'scripts'
+        scripts_dir.mkdir()
+        # Place a script OUTSIDE the scripts dir + try to reach it
+        # via ../ traversal.
+        evil = tmp_path / 'evil.sh'
+        evil.write_text('#!/bin/bash\necho evil')
+        deps = _build_deps(
+            config_manager=_StubConfig({'scripts.path': str(scripts_dir)}),
+            docker_resolve_path=lambda p: p,
+        )
+        result = auto_run_script({'script_name': '../evil.sh'}, deps)
+        assert result['status'] == 'error'
+        assert 'path traversal' in result['error'].lower()
+
+    def test_missing_script_returns_error(self, tmp_path):
+        scripts_dir = tmp_path / 'scripts'
+        scripts_dir.mkdir()
+        deps = _build_deps(
+            config_manager=_StubConfig({'scripts.path': str(scripts_dir)}),
+        )
+        result = auto_run_script({'script_name': 'no_such_script.sh'}, deps)
+        assert result['status'] == 'error'
+        assert 'not found' in result['error'].lower()
+
+
+# ─── search_and_download ─────────────────────────────────────────────
+
+
+class TestSearchAndDownload:
+    def test_no_query_returns_error(self):
+        deps = _build_deps()
+        result = auto_search_and_download({}, deps)
+        assert result == {'status': 'error', 'error': 'No search query provided'}
+
+    def test_query_from_event_data_used(self):
+        captured_queries: List[str] = []
+
+        class _Orchestrator:
+            async def search_and_download_best(self, q):
+                captured_queries.append(q)
+                return 'dl-id-123'
+
+        deps = _build_deps(
+            download_orchestrator=_Orchestrator(),
+            run_async=lambda coro: 'dl-id-123',
+        )
+        result = auto_search_and_download(
+            {'_event_data': {'query': 'Adele Hello'}}, deps,
+        )
+        assert result['status'] == 'completed'
+        assert result['query'] == 'Adele Hello'
+        assert result['download_id'] == 'dl-id-123'
+
+    def test_no_match_returns_not_found(self):
+        class _Orchestrator:
+            async def search_and_download_best(self, q):
+                return None
+
+        deps = _build_deps(
+            download_orchestrator=_Orchestrator(),
+            run_async=lambda coro: None,
+        )
+        result = auto_search_and_download({'query': 'xyz'}, deps)
+        assert result['status'] == 'not_found'
+        assert result['query'] == 'xyz'

tests/automation/test_handlers_playlist.py

@@ -94,6 +94,32 @@ def _build_deps(**overrides) -> AutomationDeps:
         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': {}},
     )
     defaults.update(overrides)
     return AutomationDeps(**defaults)  # type: ignore[arg-type]

tests/automation/test_handlers_simple.py

@@ -64,6 +64,32 @@ def _build_deps(**overrides: Any) -> AutomationDeps:
         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': {}},
     )
     defaults.update(overrides)
     return AutomationDeps(**defaults)  # type: ignore[arg-type]

web_server.py

@@ -767,6 +767,15 @@ db_update_state = {
 _db_update_automation_id = None  # Set when automation triggers DB update, used by callbacks
 db_update_lock = threading.Lock()
 
+
+def _set_db_update_automation_id(value):
+    """Setter exposed to extracted automation handlers — keeps the
+    legacy `_db_update_automation_id` global in sync so the live
+    DB-update progress callbacks below (which still read the global
+    directly) emit against the right automation card."""
+    global _db_update_automation_id
+    _db_update_automation_id = value
+
 # Quality Scanner state
 quality_scanner_state = {
     "status": "idle",  # idle, running, finished, error
@@ -932,6 +941,8 @@ def _register_automation_handlers():
     # work; they'll be removed once the rest of the lift is done.
     _automation_state = AutomationState()
 
+    from core.watchlist_scanner import get_watchlist_scanner as _get_watchlist_scanner_fn
+
     _automation_deps = AutomationDeps(
         engine=automation_engine,
         state=_automation_state,
@@ -953,679 +964,38 @@ def _register_automation_handlers():
         get_deezer_client=_get_deezer_client,
         parse_youtube_playlist=parse_youtube_playlist,
         get_sync_states=lambda: sync_states,
+        set_db_update_automation_id=_set_db_update_automation_id,
+        get_db_update_state=lambda: db_update_state,
+        db_update_lock=db_update_lock,
+        db_update_executor=db_update_executor,
+        run_db_update_task=_run_db_update_task,
+        run_deep_scan_task=_run_deep_scan_task,
+        get_duplicate_cleaner_state=lambda: duplicate_cleaner_state,
+        duplicate_cleaner_lock=duplicate_cleaner_lock,
+        duplicate_cleaner_executor=duplicate_cleaner_executor,
+        run_duplicate_cleaner=_run_duplicate_cleaner,
+        get_quality_scanner_state=lambda: quality_scanner_state,
+        quality_scanner_lock=quality_scanner_lock,
+        quality_scanner_executor=quality_scanner_executor,
+        run_quality_scanner=_run_quality_scanner,
+        download_orchestrator=download_orchestrator,
+        run_async=run_async,
+        tasks_lock=tasks_lock,
+        get_download_batches=lambda: download_batches,
+        get_download_tasks=lambda: download_tasks,
+        sweep_empty_download_directories=_sweep_empty_download_directories,
+        get_staging_path=get_staging_path,
+        docker_resolve_path=docker_resolve_path,
+        get_current_profile_id=get_current_profile_id,
+        get_watchlist_scanner=_get_watchlist_scanner_fn,
+        get_app=lambda: app,
+        get_beatport_data_cache=lambda: beatport_data_cache,
     )
     _register_extracted_handlers(_automation_deps)
 
 
     # --- Phase 3 action handlers ---
 
-    def _auto_start_database_update(config):
-        global _db_update_automation_id
-        automation_id = config.get('_automation_id')
-        if db_update_state.get('status') == 'running':
-            return {'status': 'skipped', 'reason': 'Database update already running'}
-        _db_update_automation_id = automation_id
-        full = config.get('full_refresh', False)
-        active_server = config_manager.get_active_media_server()
-        with db_update_lock:
-            db_update_state.update({
-                "status": "running", "phase": "Initializing...",
-                "progress": 0, "current_item": "", "processed": 0, "total": 0, "error_message": ""
-            })
-        db_update_executor.submit(_run_db_update_task, full, active_server)
-
-        # Monitor DB update progress (callbacks handle card updates, we just block until done)
-        time.sleep(1)
-        poll_start = time.time()
-        last_progress_time = time.time()
-        last_progress_val = 0
-        while time.time() - poll_start < 7200:  # Max 2 hours
-            time.sleep(3)
-            with db_update_lock:
-                status = db_update_state.get('status', 'idle')
-                current_progress = db_update_state.get('progress', 0)
-            if status != 'running':
-                break
-            # Track stall detection — if no progress change in 10 minutes, warn
-            if current_progress != last_progress_val:
-                last_progress_val = current_progress
-                last_progress_time = time.time()
-            elif time.time() - last_progress_time > 600:
-                _update_automation_progress(automation_id,
-                    log_line='Database update appears stalled — waiting...', log_type='warning')
-                last_progress_time = time.time()  # Reset so warning repeats every 10 min
-        else:
-            # 2-hour timeout reached
-            _update_automation_progress(automation_id, status='error',
-                phase='Timed out', log_line='Database update timed out after 2 hours', log_type='error')
-            return {'status': 'error', 'reason': 'Timed out', '_manages_own_progress': True}
-
-        # Finished/error callback already updated the card — return matching status
-        with db_update_lock:
-            final_status = db_update_state.get('status', 'unknown')
-        if final_status == 'error':
-            return {'status': 'error', 'reason': db_update_state.get('error_message', 'Unknown error'), '_manages_own_progress': True}
-        with db_update_lock:
-            stats = {
-                'status': 'completed', 'full_refresh': str(full), '_manages_own_progress': True,
-                'artists': db_update_state.get('total', 0),
-                'albums': db_update_state.get('total_albums', 0),
-                'tracks': db_update_state.get('total_tracks', 0),
-                'removed_artists': db_update_state.get('removed_artists', 0),
-                'removed_albums': db_update_state.get('removed_albums', 0),
-                'removed_tracks': db_update_state.get('removed_tracks', 0),
-            }
-        return stats
-
-    def _auto_deep_scan_library(config):
-        global _db_update_automation_id
-        automation_id = config.get('_automation_id')
-        if db_update_state.get('status') == 'running':
-            return {'status': 'skipped', 'reason': 'Database update already running'}
-        _db_update_automation_id = automation_id
-        active_server = config_manager.get_active_media_server()
-        with db_update_lock:
-            db_update_state.update({
-                "status": "running", "phase": "Deep scan: Initializing...",
-                "progress": 0, "current_item": "", "processed": 0, "total": 0, "error_message": ""
-            })
-        db_update_executor.submit(_run_deep_scan_task, active_server)
-
-        # Monitor progress (callbacks handle card updates, we just block until done)
-        time.sleep(1)
-        poll_start = time.time()
-        last_progress_time = time.time()
-        last_progress_val = 0
-        while time.time() - poll_start < 7200:  # Max 2 hours
-            time.sleep(3)
-            with db_update_lock:
-                status = db_update_state.get('status', 'idle')
-                current_progress = db_update_state.get('progress', 0)
-            if status != 'running':
-                break
-            if current_progress != last_progress_val:
-                last_progress_val = current_progress
-                last_progress_time = time.time()
-            elif time.time() - last_progress_time > 600:
-                _update_automation_progress(automation_id,
-                    log_line='Deep scan appears stalled — waiting...', log_type='warning')
-                last_progress_time = time.time()
-        else:
-            _update_automation_progress(automation_id, status='error',
-                phase='Timed out', log_line='Deep scan timed out after 2 hours', log_type='error')
-            return {'status': 'error', 'reason': 'Timed out', '_manages_own_progress': True}
-
-        with db_update_lock:
-            final_status = db_update_state.get('status', 'unknown')
-        if final_status == 'error':
-            return {'status': 'error', 'reason': db_update_state.get('error_message', 'Unknown error'), '_manages_own_progress': True}
-        with db_update_lock:
-            stats = {
-                'status': 'completed', '_manages_own_progress': True,
-                'artists': db_update_state.get('total', 0),
-                'albums': db_update_state.get('total_albums', 0),
-                'tracks': db_update_state.get('total_tracks', 0),
-                'removed_artists': db_update_state.get('removed_artists', 0),
-                'removed_albums': db_update_state.get('removed_albums', 0),
-                'removed_tracks': db_update_state.get('removed_tracks', 0),
-            }
-        return stats
-
-    def _auto_run_duplicate_cleaner(config):
-        automation_id = config.get('_automation_id')
-        if duplicate_cleaner_state.get('status') == 'running':
-            return {'status': 'skipped', 'reason': 'Duplicate cleaner already running'}
-
-        # Pre-set status before submit so polling loop doesn't see stale 'finished' from last run
-        with duplicate_cleaner_lock:
-            duplicate_cleaner_state["status"] = "running"
-        duplicate_cleaner_executor.submit(_run_duplicate_cleaner)
-        _update_automation_progress(automation_id,
-            log_line='Duplicate cleaner started', log_type='info')
-
-        # Monitor duplicate cleaner progress (max 2 hours)
-        time.sleep(1)  # Brief pause for executor to start
-        poll_start = time.time()
-        while time.time() - poll_start < 7200:
-            time.sleep(3)
-            status = duplicate_cleaner_state.get('status', 'idle')
-            if status not in ('running',):
-                break
-            phase = duplicate_cleaner_state.get('phase', 'Scanning...')
-            progress = duplicate_cleaner_state.get('progress', 0)
-            scanned = duplicate_cleaner_state.get('files_scanned', 0)
-            total = duplicate_cleaner_state.get('total_files', 0)
-            _update_automation_progress(automation_id,
-                phase=phase, progress=progress,
-                processed=scanned, total=total)
-        else:
-            # 2-hour timeout reached
-            _update_automation_progress(automation_id, status='error',
-                phase='Timed out', log_line='Duplicate cleaner timed out after 2 hours', log_type='error')
-            return {'status': 'error', 'reason': 'Timed out', '_manages_own_progress': True}
-
-        # Check actual exit status (could be 'finished' or 'error')
-        final_status = duplicate_cleaner_state.get('status', 'idle')
-        if final_status == 'error':
-            err = duplicate_cleaner_state.get('error_message', 'Unknown error')
-            _update_automation_progress(automation_id, status='error', progress=100,
-                phase='Error', log_line=err, log_type='error')
-            return {'status': 'error', 'reason': err, '_manages_own_progress': True}
-
-        dupes = duplicate_cleaner_state.get('duplicates_found', 0)
-        removed = duplicate_cleaner_state.get('deleted', 0)
-        space_freed = duplicate_cleaner_state.get('space_freed', 0)
-        scanned = duplicate_cleaner_state.get('files_scanned', 0)
-        _update_automation_progress(automation_id, status='finished', progress=100,
-            phase='Complete',
-            log_line=f'Found {dupes} duplicates, removed {removed} files', log_type='success')
-        return {
-            'status': 'completed', '_manages_own_progress': True,
-            'files_scanned': scanned,
-            'duplicates_found': dupes,
-            'files_deleted': removed,
-            'space_freed_mb': round(space_freed / (1024 * 1024), 1),
-        }
-
-    def _auto_clear_quarantine(config):
-        import shutil as _shutil
-        automation_id = config.get('_automation_id')
-        quarantine_path = os.path.join(docker_resolve_path(config_manager.get('soulseek.download_path', './downloads')), 'ss_quarantine')
-        if not os.path.exists(quarantine_path):
-            _update_automation_progress(automation_id,
-                log_line='No quarantine folder found', log_type='info')
-            return {'status': 'completed', 'removed': '0'}
-        removed = 0
-        for f in os.listdir(quarantine_path):
-            fp = os.path.join(quarantine_path, f)
-            try:
-                if os.path.isfile(fp):
-                    os.remove(fp)
-                    removed += 1
-                elif os.path.isdir(fp):
-                    _shutil.rmtree(fp)
-                    removed += 1
-            except Exception as e:
-                logger.debug("quarantine entry purge failed: %s", e)
-        _update_automation_progress(automation_id,
-            log_line=f'Removed {removed} quarantined items', log_type='success' if removed > 0 else 'info')
-        return {'status': 'completed', 'removed': str(removed)}
-
-    def _auto_cleanup_wishlist(config):
-        automation_id = config.get('_automation_id')
-        db = get_database()
-        removed = db.remove_wishlist_duplicates(get_current_profile_id())
-        _update_automation_progress(automation_id,
-            log_line=f'Removed {removed or 0} duplicate wishlist entries', log_type='success' if removed else 'info')
-        return {'status': 'completed', 'removed': str(removed or 0)}
-
-    def _auto_update_discovery_pool(config):
-        automation_id = config.get('_automation_id')
-        try:
-            from core.watchlist_scanner import get_watchlist_scanner
-            scanner = get_watchlist_scanner(spotify_client)
-            _update_automation_progress(automation_id,
-                log_line='Updating discovery pool...', log_type='info')
-            scanner.update_discovery_pool_incremental(get_current_profile_id())
-            _update_automation_progress(automation_id, status='finished', progress=100,
-                phase='Complete',
-                log_line='Discovery pool updated', log_type='success')
-            return {'status': 'completed', '_manages_own_progress': True}
-        except Exception as e:
-            _update_automation_progress(automation_id, status='error',
-                phase='Error', log_line=str(e), log_type='error')
-            return {'status': 'error', 'reason': str(e), '_manages_own_progress': True}
-
-    def _auto_start_quality_scan(config):
-        automation_id = config.get('_automation_id')
-        if quality_scanner_state.get('status') == 'running':
-            return {'status': 'skipped', 'reason': 'Quality scan already running'}
-
-        scope = config.get('scope', 'watchlist')
-        # Pre-set status before submit so polling loop doesn't see stale 'finished' from last run
-        with quality_scanner_lock:
-            quality_scanner_state["status"] = "running"
-        quality_scanner_executor.submit(_run_quality_scanner, scope, get_current_profile_id())
-        _update_automation_progress(automation_id,
-            log_line=f'Quality scan started (scope: {scope})', log_type='info')
-
-        # Monitor quality scanner progress (max 2 hours)
-        time.sleep(1)  # Brief pause for executor to start
-        poll_start = time.time()
-        while time.time() - poll_start < 7200:
-            time.sleep(3)
-            status = quality_scanner_state.get('status', 'idle')
-            if status not in ('running',):
-                break
-            phase = quality_scanner_state.get('phase', 'Scanning...')
-            progress = quality_scanner_state.get('progress', 0)
-            processed = quality_scanner_state.get('processed', 0)
-            total = quality_scanner_state.get('total', 0)
-            _update_automation_progress(automation_id,
-                phase=phase, progress=progress,
-                processed=processed, total=total)
-        else:
-            # 2-hour timeout reached
-            _update_automation_progress(automation_id, status='error',
-                phase='Timed out', log_line='Quality scan timed out after 2 hours', log_type='error')
-            return {'status': 'error', 'reason': 'Timed out', '_manages_own_progress': True}
-
-        # Check actual exit status (could be 'finished' or 'error')
-        final_status = quality_scanner_state.get('status', 'idle')
-        if final_status == 'error':
-            err = quality_scanner_state.get('error_message', 'Unknown error')
-            _update_automation_progress(automation_id, status='error', progress=100,
-                phase='Error', log_line=err, log_type='error')
-            return {'status': 'error', 'reason': err, '_manages_own_progress': True}
-
-        issues = quality_scanner_state.get('low_quality', 0)
-        _update_automation_progress(automation_id, status='finished', progress=100,
-            phase='Complete',
-            log_line=f'Quality scan complete — {issues} issues found', log_type='success')
-        return {
-            'status': 'completed', 'scope': scope, '_manages_own_progress': True,
-            'tracks_scanned': quality_scanner_state.get('processed', 0),
-            'quality_met': quality_scanner_state.get('quality_met', 0),
-            'low_quality': issues,
-            'matched': quality_scanner_state.get('matched', 0),
-        }
-
-    def _auto_backup_database(config):
-        import sqlite3, glob as _glob
-        automation_id = config.get('_automation_id')
-        db_path = os.environ.get('DATABASE_PATH', 'database/music_library.db')
-        if not os.path.exists(db_path):
-            return {'status': 'error', 'reason': 'Database file not found'}
-        max_backups = 5
-        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
-        backup_path = f"{db_path}.backup_{timestamp}"
-        # Use SQLite backup API for safe hot-copy of active database
-        src = sqlite3.connect(db_path)
-        dst = sqlite3.connect(backup_path)
-        src.backup(dst)
-        dst.close()
-        src.close()
-        size_mb = round(os.path.getsize(backup_path) / (1024 * 1024), 1)
-        # Rolling cleanup — keep only the newest N backups
-        existing = sorted(_glob.glob(f"{db_path}.backup_*"), key=os.path.getmtime)
-        while len(existing) > max_backups:
-            try:
-                os.remove(existing.pop(0))
-            except Exception as e:
-                logger.debug("rolling backup cleanup failed: %s", e)
-        _update_automation_progress(automation_id,
-            log_line=f'Backup created: {size_mb}MB ({os.path.basename(backup_path)})', log_type='success')
-        return {'status': 'completed', 'backup_path': backup_path, 'size_mb': str(size_mb)}
-
-    def _auto_refresh_beatport_cache(config):
-        """Refresh Beatport homepage cache by calling each endpoint internally."""
-        automation_id = config.get('_automation_id')
-        sections = [
-            ('hero_tracks', '/api/beatport/hero-tracks', 'Hero Tracks'),
-            ('new_releases', '/api/beatport/new-releases', 'New Releases'),
-            ('featured_charts', '/api/beatport/featured-charts', 'Featured Charts'),
-            ('dj_charts', '/api/beatport/dj-charts', 'DJ Charts'),
-            ('top_10_lists', '/api/beatport/homepage/top-10-lists', 'Top 10 Lists'),
-            ('top_10_releases', '/api/beatport/homepage/top-10-releases-cards', 'Top 10 Releases'),
-            ('hype_picks', '/api/beatport/hype-picks', 'Hype Picks'),
-        ]
-        # Invalidate all homepage cache timestamps so endpoints re-scrape
-        with beatport_data_cache['cache_lock']:
-            for key in beatport_data_cache['homepage']:
-                beatport_data_cache['homepage'][key]['timestamp'] = 0
-                beatport_data_cache['homepage'][key]['data'] = None
-
-        refreshed = 0
-        errors = []
-        with app.test_client() as client:
-            for idx, (_, endpoint, label) in enumerate(sections):
-                _update_automation_progress(automation_id,
-                    progress=(idx / len(sections)) * 100,
-                    phase=f'Scraping: {label}',
-                    current_item=label)
-                try:
-                    resp = client.get(endpoint)
-                    if resp.status_code == 200:
-                        refreshed += 1
-                        _update_automation_progress(automation_id,
-                            log_line=f'{label}: cached', log_type='success')
-                    else:
-                        errors.append(label)
-                        _update_automation_progress(automation_id,
-                            log_line=f'{label}: HTTP {resp.status_code}', log_type='error')
-                except Exception as e:
-                    errors.append(label)
-                    _update_automation_progress(automation_id,
-                        log_line=f'{label}: {str(e)}', log_type='error')
-                if idx < len(sections) - 1:
-                    time.sleep(2)
-
-        _update_automation_progress(automation_id, status='finished', progress=100,
-            phase='Complete',
-            log_line=f'Refreshed {refreshed}/{len(sections)} sections', log_type='success')
-        return {'status': 'completed', 'refreshed': str(refreshed), 'errors': str(len(errors)),
-                '_manages_own_progress': True}
-
-    def _auto_clean_search_history(config):
-        """Remove old searches from Soulseek."""
-        automation_id = config.get('_automation_id')
-        # Skip if soulseek is not the active download source or in hybrid order
-        dl_mode = config_manager.get('download_source.mode', 'hybrid')
-        hybrid_order = config_manager.get('download_source.hybrid_order', ['hifi', 'youtube', 'soulseek'])
-        soulseek_active = (dl_mode == 'soulseek' or
-                          (dl_mode == 'hybrid' and 'soulseek' in hybrid_order))
-        # Reach the underlying SoulseekClient via the orchestrator's
-        # generic accessor.
-        slskd = download_orchestrator.client('soulseek') if download_orchestrator else None
-        if not soulseek_active or not slskd or not slskd.base_url:
-            _update_automation_progress(automation_id,
-                log_line='Soulseek not active — skipped', log_type='skip')
-            return {'status': 'skipped'}
-        if not config_manager.get('soulseek.auto_clear_searches', True):
-            _update_automation_progress(automation_id,
-                log_line='Auto-clear disabled in settings', log_type='skip')
-            return {'status': 'skipped'}
-        try:
-            success = run_async(download_orchestrator.maintain_search_history_with_buffer(
-                keep_searches=50, trigger_threshold=200
-            ))
-            if success:
-                _update_automation_progress(automation_id,
-                    log_line='Search history maintenance completed', log_type='success')
-                return {'status': 'completed'}
-            else:
-                _update_automation_progress(automation_id,
-                    log_line='No cleanup needed', log_type='skip')
-                return {'status': 'completed'}
-        except Exception as e:
-            return {'status': 'error', 'error': str(e)}
-
-    def _auto_clean_completed_downloads(config):
-        """Clear completed downloads and empty directories."""
-        automation_id = config.get('_automation_id')
-        try:
-            has_active_batches = False
-            has_post_processing = False
-            with tasks_lock:
-                for batch_data in download_batches.values():
-                    if batch_data.get('phase') not in ['complete', 'error', 'cancelled', None]:
-                        has_active_batches = True
-                        break
-                if not has_active_batches:
-                    for task_data in download_tasks.values():
-                        if task_data.get('status') == 'post_processing':
-                            has_post_processing = True
-                            break
-
-            if has_active_batches:
-                _update_automation_progress(automation_id,
-                    log_line='Skipped — downloads active', log_type='skip')
-                return {'status': 'completed'}
-
-            run_async(download_orchestrator.clear_all_completed_downloads())
-            if not has_post_processing:
-                _sweep_empty_download_directories()
-            _update_automation_progress(automation_id,
-                log_line='Download cleanup completed', log_type='success')
-            return {'status': 'completed'}
-        except Exception as e:
-            return {'status': 'error', 'reason': str(e)}
-
-    def _auto_full_cleanup(config):
-        """Run all cleanup tasks: quarantine, download queue, empty dirs, staging, search history."""
-        import shutil as _shutil
-        automation_id = config.get('_automation_id')
-        steps = []
-
-        # --- 1. Clear quarantine ---
-        _update_automation_progress(automation_id, phase='Clearing quarantine...', progress=0)
-        quarantine_path = os.path.join(docker_resolve_path(config_manager.get('soulseek.download_path', './downloads')), 'ss_quarantine')
-        q_removed = 0
-        if os.path.exists(quarantine_path):
-            for f in os.listdir(quarantine_path):
-                fp = os.path.join(quarantine_path, f)
-                try:
-                    if os.path.isfile(fp):
-                        os.remove(fp)
-                        q_removed += 1
-                    elif os.path.isdir(fp):
-                        _shutil.rmtree(fp)
-                        q_removed += 1
-                except Exception as e:
-                    logger.debug("quarantine entry purge failed: %s", e)
-        steps.append(f'Quarantine: removed {q_removed} items')
-        _update_automation_progress(automation_id,
-            log_line=f'Quarantine: removed {q_removed} items', log_type='success' if q_removed else 'info')
-
-        # --- 2. Clear completed/errored/cancelled downloads from Soulseek queue ---
-        _update_automation_progress(automation_id, phase='Clearing download queue...', progress=20)
-        has_active_batches = False
-        has_post_processing = False
-        with tasks_lock:
-            for batch_data in download_batches.values():
-                if batch_data.get('phase') not in ['complete', 'error', 'cancelled', None]:
-                    has_active_batches = True
-                    break
-            if not has_active_batches:
-                for task_data in download_tasks.values():
-                    if task_data.get('status') == 'post_processing':
-                        has_post_processing = True
-                        break
-        if has_active_batches:
-            steps.append('Download queue: skipped (active batches)')
-            _update_automation_progress(automation_id,
-                log_line='Download queue: skipped (active batches)', log_type='skip')
-        else:
-            try:
-                run_async(download_orchestrator.clear_all_completed_downloads())
-                steps.append('Download queue: cleared')
-                _update_automation_progress(automation_id,
-                    log_line='Download queue: cleared', log_type='success')
-            except Exception as e:
-                steps.append(f'Download queue: error ({e})')
-                _update_automation_progress(automation_id,
-                    log_line=f'Download queue: error ({e})', log_type='error')
-
-        # --- 3. Sweep empty download directories ---
-        _update_automation_progress(automation_id, phase='Sweeping empty directories...', progress=40)
-        if has_active_batches or has_post_processing:
-            reason = 'active batches' if has_active_batches else 'post-processing active'
-            steps.append(f'Empty directories: skipped ({reason})')
-            _update_automation_progress(automation_id,
-                log_line=f'Empty directories: skipped ({reason})', log_type='skip')
-        else:
-            dirs_removed = _sweep_empty_download_directories()
-            steps.append(f'Empty directories: removed {dirs_removed}')
-            _update_automation_progress(automation_id,
-                log_line=f'Empty directories: removed {dirs_removed}', log_type='success' if dirs_removed else 'info')
-
-        # --- 4. Sweep empty staging directories ---
-        _update_automation_progress(automation_id, phase='Sweeping import folder...', progress=60)
-        staging_path = get_staging_path()
-        s_removed = 0
-        if os.path.isdir(staging_path):
-            for dirpath, _dirnames, _filenames in os.walk(staging_path, topdown=False):
-                if os.path.normpath(dirpath) == os.path.normpath(staging_path):
-                    continue
-                try:
-                    entries = os.listdir(dirpath)
-                except OSError:
-                    continue
-                visible = [e for e in entries if not e.startswith('.')]
-                if not visible:
-                    for hidden in entries:
-                        try:
-                            os.remove(os.path.join(dirpath, hidden))
-                        except Exception as e:
-                            logger.debug("hidden file cleanup failed: %s", e)
-                    try:
-                        os.rmdir(dirpath)
-                        s_removed += 1
-                    except OSError:
-                        pass
-        steps.append(f'Staging: removed {s_removed} empty directories')
-        _update_automation_progress(automation_id,
-            log_line=f'Staging: removed {s_removed} empty directories', log_type='success' if s_removed else 'info')
-
-        # --- 5. Clean search history (if enabled) ---
-        _update_automation_progress(automation_id, phase='Cleaning search history...', progress=80)
-        try:
-            if not config_manager.get('soulseek.auto_clear_searches', True):
-                steps.append('Search cleanup: disabled in settings')
-                _update_automation_progress(automation_id,
-                    log_line='Search cleanup: disabled in settings', log_type='skip')
-            else:
-                run_async(download_orchestrator.maintain_search_history_with_buffer(
-                    keep_searches=50, trigger_threshold=200
-                ))
-                steps.append('Search history: cleaned')
-                _update_automation_progress(automation_id,
-                    log_line='Search history: cleaned', log_type='success')
-        except Exception as e:
-            steps.append(f'Search history: error ({e})')
-            _update_automation_progress(automation_id,
-                log_line=f'Search history: error ({e})', log_type='error')
-
-        total_removed = q_removed + s_removed
-        _update_automation_progress(automation_id, status='finished', progress=100,
-            phase='Complete',
-            log_line=f'Full cleanup complete — {total_removed} items removed', log_type='success')
-        return {
-            'status': 'completed',
-            'quarantine_removed': str(q_removed),
-            'staging_removed': str(s_removed),
-            'total_removed': str(total_removed),
-            'steps': steps,
-            '_manages_own_progress': True,
-        }
-
-    def _auto_run_script(config):
-        """Execute a user script from the scripts directory."""
-        import subprocess as _sp
-        script_name = config.get('script_name', '')
-        timeout = min(int(config.get('timeout', 60)), 300)
-        automation_id = config.get('_automation_id')
-
-        if not script_name:
-            return {'status': 'error', 'error': 'No script selected'}
-
-        scripts_dir = docker_resolve_path(config_manager.get('scripts.path', './scripts'))
-        if not scripts_dir or not os.path.isdir(scripts_dir):
-            os.makedirs(scripts_dir, exist_ok=True)
-            return {'status': 'error', 'error': 'Scripts directory is empty. Add scripts to the scripts/ folder.'}
-
-        script_path = os.path.join(scripts_dir, script_name)
-        script_path = os.path.realpath(script_path)
-
-        # Security: block path traversal
-        if not script_path.startswith(os.path.realpath(scripts_dir)):
-            return {'status': 'error', 'error': 'Script path traversal blocked'}
-
-        if not os.path.isfile(script_path):
-            return {'status': 'error', 'error': f'Script not found: {script_name}'}
-
-        _update_automation_progress(automation_id, phase=f'Running {script_name}...', progress=10)
-
-        # Build environment with SoulSync context
-        env = os.environ.copy()
-        event_data = config.get('_event_data') or {}
-        env['SOULSYNC_EVENT'] = str(event_data.get('type', ''))
-        env['SOULSYNC_AUTOMATION'] = config.get('_automation_name', '')
-        env['SOULSYNC_SCRIPTS_DIR'] = scripts_dir
-
-        try:
-            # Determine how to run the script
-            if script_path.endswith('.py'):
-                cmd = ['python', script_path]
-            elif script_path.endswith('.sh'):
-                cmd = ['bash', script_path]
-            else:
-                cmd = [script_path]
-
-            result = _sp.run(
-                cmd,
-                capture_output=True, text=True, timeout=timeout,
-                cwd=scripts_dir, env=env
-            )
-
-            _update_automation_progress(automation_id, phase='Script completed', progress=100)
-
-            stdout = result.stdout[:2000] if result.stdout else ''
-            stderr = result.stderr[:1000] if result.stderr else ''
-
-            if result.returncode == 0:
-                logger.info(f"Script '{script_name}' completed (exit 0)")
-            else:
-                logger.warning(f"Script '{script_name}' exited with code {result.returncode}")
-
-            return {
-                'status': 'completed' if result.returncode == 0 else 'error',
-                'exit_code': str(result.returncode),
-                'stdout': stdout,
-                'stderr': stderr,
-                'script': script_name,
-            }
-        except _sp.TimeoutExpired:
-            _update_automation_progress(automation_id, phase='Script timed out', progress=100)
-            return {'status': 'error', 'error': f'Script timed out after {timeout}s', 'script': script_name}
-        except Exception as e:
-            return {'status': 'error', 'error': str(e), 'script': script_name}
-
-    automation_engine.register_action_handler('run_script', _auto_run_script)
-    automation_engine.register_action_handler('full_cleanup', _auto_full_cleanup)
-
-    automation_engine.register_action_handler('start_database_update', _auto_start_database_update,
-                                              lambda: db_update_state.get('status') == 'running')
-    automation_engine.register_action_handler('deep_scan_library', _auto_deep_scan_library,
-                                              lambda: db_update_state.get('status') == 'running')
-    automation_engine.register_action_handler('run_duplicate_cleaner', _auto_run_duplicate_cleaner,
-                                              lambda: duplicate_cleaner_state.get('status') == 'running')
-    automation_engine.register_action_handler('clear_quarantine', _auto_clear_quarantine)
-    automation_engine.register_action_handler('cleanup_wishlist', _auto_cleanup_wishlist)
-    automation_engine.register_action_handler('update_discovery_pool', _auto_update_discovery_pool)
-    automation_engine.register_action_handler('start_quality_scan', _auto_start_quality_scan,
-                                              lambda: quality_scanner_state.get('status') == 'running')
-    automation_engine.register_action_handler('backup_database', _auto_backup_database)
-    automation_engine.register_action_handler('refresh_beatport_cache', _auto_refresh_beatport_cache)
-    automation_engine.register_action_handler('clean_search_history', _auto_clean_search_history)
-    automation_engine.register_action_handler('clean_completed_downloads', _auto_clean_completed_downloads)
-
-    def _auto_search_and_download(config):
-        """Search for a track and download the best match."""
-        automation_id = config.get('_automation_id')
-        query = config.get('query', '').strip()
-        # Event-triggered: pull query from event data (e.g. webhook_received)
-        if not query:
-            event_data = config.get('_event_data', {})
-            query = (event_data.get('query', '') or '').strip()
-        if not query:
-            if automation_id:
-                _update_automation_progress(automation_id,
-                    log_line='No search query provided', log_type='error')
-            return {'status': 'error', 'error': 'No search query provided'}
-        try:
-            if automation_id:
-                _update_automation_progress(automation_id,
-                    phase='Searching', log_line=f'Searching: {query}', log_type='info')
-            result = run_async(download_orchestrator.search_and_download_best(query))
-            if result:
-                if automation_id:
-                    _update_automation_progress(automation_id,
-                        log_line=f'Download started for: {query}', log_type='success')
-                return {'status': 'completed', 'query': query, 'download_id': result}
-            else:
-                if automation_id:
-                    _update_automation_progress(automation_id,
-                        log_line=f'No match found for: {query}', log_type='warning')
-                return {'status': 'not_found', 'query': query, 'error': 'No match found'}
-        except Exception as e:
-            if automation_id:
-                _update_automation_progress(automation_id,
-                    log_line=f'Error: {e}', log_type='error')
-            return {'status': 'error', 'query': query, 'error': str(e)}
-
-    automation_engine.register_action_handler('search_and_download', _auto_search_and_download)
-
     # Register progress tracking callbacks
     def _progress_init(aid, name, action_type):
         _init_automation_progress(aid, name, action_type)