Add WebSocket real-time updates with automatic HTTP polling fallback

Migrates 38 HTTP polling loops to WebSocket push events across 6 phases: service status, dashboard stats, enrichment workers, tool progress, sync/discovery progress, and scan status. All original HTTP polling is preserved as automatic fallback — if WebSocket is unavailable or disconnects, the app seamlessly reverts to its previous behavior. Includes 162 tests verifying event delivery, data shape, and HTTP parity. Also fixes a copy-paste bug in Beatport sync error cleanup.

requirements-webui.txt

@@ -46,4 +46,7 @@ pyacoustid>=1.3.0
 websocket-client>=1.7.0
 
 # Tidal download support
-tidalapi>=0.7.6
+tidalapi>=0.7.6
+
+# WebSocket server for real-time UI updates
+flask-socketio>=5.3.0

tests/test_phase1_websocket.py

@@ -0,0 +1,397 @@
+"""Phase 1 WebSocket migration tests.
+
+Verifies that:
+ - WebSocket infrastructure connects and communicates
+ - HTTP endpoints still work (backward compat / fallback)
+ - Socket events deliver identical data to HTTP responses
+ - Download batch room subscriptions work correctly
+
+IMPORTANT: Do NOT use ``from tests.conftest import …`` — pytest's auto-discovered
+conftest is a different module instance. Use the ``shared_state`` fixture instead.
+"""
+
+import pytest
+
+
+# =========================================================================
+# Group A — Infrastructure
+# =========================================================================
+
+class TestInfrastructure:
+    """Socket.IO connects, and HTTP endpoints remain functional."""
+
+    def test_socketio_connects(self, socketio_client):
+        """Client can establish a WebSocket connection."""
+        assert socketio_client.is_connected()
+
+    def test_socketio_disconnect_and_reconnect(self, test_app):
+        """Client can disconnect and reconnect cleanly."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        assert client.is_connected()
+        client.disconnect()
+        assert not client.is_connected()
+        client.connect()
+        assert client.is_connected()
+        client.disconnect()
+
+    def test_http_status_still_works(self, flask_client):
+        """GET /status returns 200 with expected keys."""
+        resp = flask_client.get('/status')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert 'spotify' in data
+        assert 'media_server' in data
+        assert 'soulseek' in data
+        assert 'active_media_server' in data
+
+    def test_http_watchlist_count_still_works(self, flask_client):
+        """GET /api/watchlist/count returns 200 with expected keys."""
+        resp = flask_client.get('/api/watchlist/count')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert data['success'] is True
+        assert 'count' in data
+        assert 'next_run_in_seconds' in data
+
+    def test_http_download_batch_still_works(self, flask_client):
+        """GET /api/download_status/batch returns 200 with expected structure."""
+        resp = flask_client.get('/api/download_status/batch')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert 'batches' in data
+        assert 'metadata' in data
+
+
+# =========================================================================
+# Group B — Service Status Parity
+# =========================================================================
+
+class TestServiceStatus:
+    """status:update socket events match GET /status HTTP responses."""
+
+    def test_status_update_received(self, test_app, shared_state):
+        """Client receives a status:update event."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_status_payload']
+        socketio.emit('status:update', build())
+        received = client.get_received()
+        status_events = [e for e in received if e['name'] == 'status:update']
+        assert len(status_events) >= 1
+
+    def test_status_update_shape(self, test_app, shared_state):
+        """status:update event data has the expected keys."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_status_payload']
+        socketio.emit('status:update', build())
+        received = client.get_received()
+        status_events = [e for e in received if e['name'] == 'status:update']
+        assert len(status_events) >= 1
+        data = status_events[0]['args'][0]
+        assert 'spotify' in data
+        assert 'media_server' in data
+        assert 'soulseek' in data
+        assert 'active_media_server' in data
+
+    def test_status_matches_http(self, test_app, shared_state):
+        """Socket event data matches HTTP endpoint response exactly."""
+        app, socketio = test_app
+        flask_client = app.test_client()
+        ws_client = socketio.test_client(app)
+        build = shared_state['build_status_payload']
+
+        http_data = flask_client.get('/status').get_json()
+
+        socketio.emit('status:update', build())
+        received = ws_client.get_received()
+        status_events = [e for e in received if e['name'] == 'status:update']
+        assert len(status_events) >= 1
+        ws_data = status_events[0]['args'][0]
+
+        assert ws_data['spotify'] == http_data['spotify']
+        assert ws_data['media_server'] == http_data['media_server']
+        assert ws_data['soulseek'] == http_data['soulseek']
+        assert ws_data['active_media_server'] == http_data['active_media_server']
+
+    def test_status_reflects_cache_changes(self, test_app, shared_state):
+        """When _status_cache changes, the next emit reflects it."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        status_cache = shared_state['status_cache']
+        build = shared_state['build_status_payload']
+
+        # Mutate cache
+        status_cache['spotify']['source'] = 'itunes'
+
+        socketio.emit('status:update', build())
+        received = client.get_received()
+        status_events = [e for e in received if e['name'] == 'status:update']
+        data = status_events[-1]['args'][0]
+        assert data['spotify']['source'] == 'itunes'
+
+
+# =========================================================================
+# Group C — Watchlist Count Parity
+# =========================================================================
+
+class TestWatchlistCount:
+    """watchlist:count socket events match GET /api/watchlist/count."""
+
+    def test_watchlist_count_received(self, test_app, shared_state):
+        """Client receives a watchlist:count event."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_watchlist_count_payload']
+        socketio.emit('watchlist:count', build())
+        received = client.get_received()
+        wl_events = [e for e in received if e['name'] == 'watchlist:count']
+        assert len(wl_events) >= 1
+
+    def test_watchlist_count_shape(self, test_app, shared_state):
+        """watchlist:count event data has expected keys."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_watchlist_count_payload']
+        socketio.emit('watchlist:count', build())
+        received = client.get_received()
+        wl_events = [e for e in received if e['name'] == 'watchlist:count']
+        data = wl_events[0]['args'][0]
+        assert data['success'] is True
+        assert isinstance(data['count'], int)
+        assert isinstance(data['next_run_in_seconds'], int)
+
+    def test_watchlist_matches_http(self, test_app, shared_state):
+        """Socket event data matches HTTP endpoint response."""
+        app, socketio = test_app
+        flask_client = app.test_client()
+        ws_client = socketio.test_client(app)
+        build = shared_state['build_watchlist_count_payload']
+
+        http_data = flask_client.get('/api/watchlist/count').get_json()
+
+        socketio.emit('watchlist:count', build())
+        received = ws_client.get_received()
+        wl_events = [e for e in received if e['name'] == 'watchlist:count']
+        ws_data = wl_events[0]['args'][0]
+
+        assert ws_data['success'] == http_data['success']
+        assert ws_data['count'] == http_data['count']
+        assert ws_data['next_run_in_seconds'] == http_data['next_run_in_seconds']
+
+    def test_watchlist_reflects_count_change(self, test_app, shared_state):
+        """When watchlist count changes, the emit reflects it."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        wl_state = shared_state['watchlist_state']
+        build = shared_state['build_watchlist_count_payload']
+
+        wl_state['count'] = 42
+
+        socketio.emit('watchlist:count', build())
+        received = client.get_received()
+        wl_events = [e for e in received if e['name'] == 'watchlist:count']
+        data = wl_events[-1]['args'][0]
+        assert data['count'] == 42
+
+
+# =========================================================================
+# Group D — Download Batch Rooms
+# =========================================================================
+
+class TestDownloadBatch:
+    """Download batch updates are delivered via room subscriptions."""
+
+    def _add_batch(self, shared_state, batch_id, **kwargs):
+        """Helper to add a fake download batch."""
+        defaults = {
+            'phase': 'downloading',
+            'tasks': [
+                {'task_id': 't1', 'status': 'downloading', 'progress': 50},
+                {'task_id': 't2', 'status': 'searching', 'progress': 0},
+            ],
+            'active_count': 2,
+            'max_concurrent': 3,
+            'playlist_id': 'spotify_test',
+            'playlist_name': 'Test Playlist',
+        }
+        defaults.update(kwargs)
+        batches = shared_state['download_batches']
+        lock = shared_state['tasks_lock']
+        with lock:
+            batches[batch_id] = defaults
+
+    def test_download_subscribe(self, test_app):
+        """Client can subscribe to a batch room."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        client.emit('downloads:subscribe', {'batch_ids': ['batch_abc']})
+
+    def test_download_receives_updates(self, test_app, shared_state):
+        """After subscribing, client receives batch_update for that batch."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build_batch = shared_state['build_batch_status_data']
+
+        self._add_batch(shared_state, 'batch_123')
+        client.emit('downloads:subscribe', {'batch_ids': ['batch_123']})
+        client.get_received()  # clear
+
+        batches = shared_state['download_batches']
+        lock = shared_state['tasks_lock']
+        with lock:
+            batch = batches['batch_123']
+            socketio.emit('downloads:batch_update', {
+                'batch_id': 'batch_123',
+                'data': build_batch('batch_123', batch),
+            }, room='batch:batch_123')
+
+        received = client.get_received()
+        dl_events = [e for e in received if e['name'] == 'downloads:batch_update']
+        assert len(dl_events) >= 1
+        payload = dl_events[0]['args'][0]
+        assert payload['batch_id'] == 'batch_123'
+        assert payload['data']['phase'] == 'downloading'
+        assert len(payload['data']['tasks']) == 2
+
+    def test_download_only_subscribed_batches(self, test_app, shared_state):
+        """Client only receives updates for subscribed batches, not others."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build_batch = shared_state['build_batch_status_data']
+
+        self._add_batch(shared_state, 'batch_A')
+        self._add_batch(shared_state, 'batch_B')
+
+        client.emit('downloads:subscribe', {'batch_ids': ['batch_A']})
+        client.get_received()  # clear
+
+        batches = shared_state['download_batches']
+        lock = shared_state['tasks_lock']
+        with lock:
+            for bid in ['batch_A', 'batch_B']:
+                socketio.emit('downloads:batch_update', {
+                    'batch_id': bid,
+                    'data': build_batch(bid, batches[bid]),
+                }, room=f'batch:{bid}')
+
+        received = client.get_received()
+        dl_events = [e for e in received if e['name'] == 'downloads:batch_update']
+        batch_ids_received = {e['args'][0]['batch_id'] for e in dl_events}
+
+        assert 'batch_A' in batch_ids_received
+        assert 'batch_B' not in batch_ids_received
+
+    def test_download_unsubscribe_stops_updates(self, test_app, shared_state):
+        """After unsubscribing, client stops receiving updates for that batch."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build_batch = shared_state['build_batch_status_data']
+
+        self._add_batch(shared_state, 'batch_X')
+        client.emit('downloads:subscribe', {'batch_ids': ['batch_X']})
+        client.get_received()  # clear
+
+        client.emit('downloads:unsubscribe', {'batch_ids': ['batch_X']})
+        client.get_received()  # clear
+
+        batches = shared_state['download_batches']
+        lock = shared_state['tasks_lock']
+        with lock:
+            socketio.emit('downloads:batch_update', {
+                'batch_id': 'batch_X',
+                'data': build_batch('batch_X', batches['batch_X']),
+            }, room='batch:batch_X')
+
+        received = client.get_received()
+        dl_events = [e for e in received if e['name'] == 'downloads:batch_update']
+        assert len(dl_events) == 0
+
+    def test_download_batch_shape(self, test_app, shared_state):
+        """Batch update data has the expected structure."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build_batch = shared_state['build_batch_status_data']
+
+        self._add_batch(shared_state, 'batch_shape')
+        client.emit('downloads:subscribe', {'batch_ids': ['batch_shape']})
+        client.get_received()
+
+        batches = shared_state['download_batches']
+        lock = shared_state['tasks_lock']
+        with lock:
+            socketio.emit('downloads:batch_update', {
+                'batch_id': 'batch_shape',
+                'data': build_batch('batch_shape', batches['batch_shape']),
+            }, room='batch:batch_shape')
+
+        received = client.get_received()
+        dl_events = [e for e in received if e['name'] == 'downloads:batch_update']
+        payload = dl_events[0]['args'][0]
+
+        assert 'batch_id' in payload
+        assert 'data' in payload
+        data = payload['data']
+        assert 'phase' in data
+        assert 'tasks' in data
+        assert 'active_count' in data
+        assert 'max_concurrent' in data
+
+    def test_download_http_batch_still_works(self, test_app, shared_state):
+        """HTTP batch endpoint works alongside WebSocket rooms."""
+        app, socketio = test_app
+        flask_client = app.test_client()
+
+        self._add_batch(shared_state, 'batch_http')
+
+        resp = flask_client.get('/api/download_status/batch?batch_ids=batch_http')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert 'batch_http' in data['batches']
+        assert data['batches']['batch_http']['phase'] == 'downloading'
+
+
+# =========================================================================
+# Group E — Fallback Behavior
+# =========================================================================
+
+class TestFallback:
+    """HTTP endpoints work when no WebSocket is connected."""
+
+    def test_http_works_without_websocket(self, flask_client):
+        """All three HTTP endpoints work without any WebSocket connection."""
+        # Status
+        resp = flask_client.get('/status')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert data['spotify']['connected'] is True
+
+        # Watchlist
+        resp = flask_client.get('/api/watchlist/count')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert data['count'] == 7
+
+        # Download batch (empty — no active batches)
+        resp = flask_client.get('/api/download_status/batch')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert data['batches'] == {}
+
+    def test_multiple_clients_get_updates(self, test_app, shared_state):
+        """Multiple WebSocket clients each receive broadcast events."""
+        app, socketio = test_app
+        client1 = socketio.test_client(app)
+        client2 = socketio.test_client(app)
+        build = shared_state['build_status_payload']
+
+        socketio.emit('status:update', build())
+
+        for client in [client1, client2]:
+            received = client.get_received()
+            status_events = [e for e in received if e['name'] == 'status:update']
+            assert len(status_events) >= 1
+
+        client1.disconnect()
+        client2.disconnect()

tests/test_phase2_dashboard.py

@@ -0,0 +1,388 @@
+"""Phase 2 WebSocket migration tests — Dashboard pollers.
+
+Verifies that:
+ - System stats, activity feed, toasts, DB stats, and wishlist count
+   are delivered identically via WebSocket events and HTTP endpoints
+ - Instant toast push from add_activity_item() works correctly
+ - HTTP endpoints still work as fallback
+
+IMPORTANT: Do NOT use ``from tests.conftest import …`` — pytest's auto-discovered
+conftest is a different module instance. Use the ``shared_state`` fixture instead.
+"""
+
+import pytest
+import time
+
+
+# =========================================================================
+# Group A — System Stats
+# =========================================================================
+
+class TestSystemStats:
+    """dashboard:stats socket events match GET /api/system/stats."""
+
+    def test_stats_event_received(self, test_app, shared_state):
+        """Client receives a dashboard:stats event."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_system_stats']
+        socketio.emit('dashboard:stats', build())
+        received = client.get_received()
+        stats_events = [e for e in received if e['name'] == 'dashboard:stats']
+        assert len(stats_events) >= 1
+
+    def test_stats_shape(self, test_app, shared_state):
+        """dashboard:stats event data has expected keys."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_system_stats']
+        socketio.emit('dashboard:stats', build())
+        received = client.get_received()
+        stats_events = [e for e in received if e['name'] == 'dashboard:stats']
+        assert len(stats_events) >= 1
+        data = stats_events[0]['args'][0]
+        assert 'active_downloads' in data
+        assert 'finished_downloads' in data
+        assert 'download_speed' in data
+        assert 'active_syncs' in data
+        assert 'uptime' in data
+        assert 'memory_usage' in data
+
+    def test_stats_matches_http(self, test_app, shared_state):
+        """Socket event data matches HTTP endpoint response."""
+        app, socketio = test_app
+        flask_client = app.test_client()
+        ws_client = socketio.test_client(app)
+        build = shared_state['build_system_stats']
+
+        http_data = flask_client.get('/api/system/stats').get_json()
+
+        socketio.emit('dashboard:stats', build())
+        received = ws_client.get_received()
+        stats_events = [e for e in received if e['name'] == 'dashboard:stats']
+        assert len(stats_events) >= 1
+        ws_data = stats_events[0]['args'][0]
+
+        assert ws_data['active_downloads'] == http_data['active_downloads']
+        assert ws_data['finished_downloads'] == http_data['finished_downloads']
+        assert ws_data['download_speed'] == http_data['download_speed']
+        assert ws_data['active_syncs'] == http_data['active_syncs']
+        assert ws_data['uptime'] == http_data['uptime']
+        assert ws_data['memory_usage'] == http_data['memory_usage']
+
+    def test_http_stats_still_works(self, flask_client):
+        """GET /api/system/stats returns 200 with expected keys."""
+        resp = flask_client.get('/api/system/stats')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert 'active_downloads' in data
+        assert 'finished_downloads' in data
+        assert 'download_speed' in data
+
+
+# =========================================================================
+# Group B — Activity Feed
+# =========================================================================
+
+class TestActivityFeed:
+    """dashboard:activity socket events match GET /api/activity/feed."""
+
+    def test_activity_event_received(self, test_app, shared_state):
+        """Client receives a dashboard:activity event."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_activity_feed_payload']
+        socketio.emit('dashboard:activity', build())
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'dashboard:activity']
+        assert len(events) >= 1
+
+    def test_activity_shape(self, test_app, shared_state):
+        """dashboard:activity event data has activities array."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        # Add some activities first
+        add_item = shared_state['add_activity_item']
+        add_item('🎵', 'Download Complete', 'Artist - Song', show_toast=False)
+
+        build = shared_state['build_activity_feed_payload']
+        socketio.emit('dashboard:activity', build())
+        received = client.get_received()
+        # Filter out any toast events, get only activity events
+        events = [e for e in received if e['name'] == 'dashboard:activity']
+        assert len(events) >= 1
+        data = events[0]['args'][0]
+        assert 'activities' in data
+        assert isinstance(data['activities'], list)
+
+    def test_activity_matches_http(self, test_app, shared_state):
+        """Socket event data matches HTTP endpoint response."""
+        app, socketio = test_app
+        flask_client = app.test_client()
+        ws_client = socketio.test_client(app)
+
+        # Add an activity
+        add_item = shared_state['add_activity_item']
+        add_item('🎵', 'Test Activity', 'Test subtitle', show_toast=False)
+
+        http_data = flask_client.get('/api/activity/feed').get_json()
+        build = shared_state['build_activity_feed_payload']
+        socketio.emit('dashboard:activity', build())
+        received = ws_client.get_received()
+        events = [e for e in received if e['name'] == 'dashboard:activity']
+        ws_data = events[0]['args'][0]
+
+        assert len(ws_data['activities']) == len(http_data['activities'])
+        if ws_data['activities']:
+            assert ws_data['activities'][0]['title'] == http_data['activities'][0]['title']
+
+    def test_http_activity_still_works(self, flask_client):
+        """GET /api/activity/feed returns 200 with expected structure."""
+        resp = flask_client.get('/api/activity/feed')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert 'activities' in data
+
+
+# =========================================================================
+# Group C — Toasts (instant push)
+# =========================================================================
+
+class TestToasts:
+    """dashboard:toast events are pushed instantly from add_activity_item()."""
+
+    def test_toast_emitted_on_add(self, test_app, shared_state):
+        """Calling add_activity_item() with show_toast=True emits dashboard:toast."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        client.get_received()  # clear
+
+        add_item = shared_state['add_activity_item']
+        add_item('✅', 'Download Complete', 'Artist - Song', show_toast=True)
+
+        received = client.get_received()
+        toast_events = [e for e in received if e['name'] == 'dashboard:toast']
+        assert len(toast_events) >= 1
+        data = toast_events[0]['args'][0]
+        assert data['title'] == 'Download Complete'
+        assert data['subtitle'] == 'Artist - Song'
+
+    def test_toast_not_emitted_when_disabled(self, test_app, shared_state):
+        """add_activity_item() with show_toast=False does NOT emit dashboard:toast."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        client.get_received()  # clear
+
+        add_item = shared_state['add_activity_item']
+        add_item('📊', 'Background Task', 'Silent update', show_toast=False)
+
+        received = client.get_received()
+        toast_events = [e for e in received if e['name'] == 'dashboard:toast']
+        assert len(toast_events) == 0
+
+    def test_toast_shape(self, test_app, shared_state):
+        """Toast data has expected keys."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        client.get_received()  # clear
+
+        add_item = shared_state['add_activity_item']
+        add_item('✅', 'Test Title', 'Test Subtitle', 'Now', show_toast=True)
+
+        received = client.get_received()
+        toast_events = [e for e in received if e['name'] == 'dashboard:toast']
+        assert len(toast_events) >= 1
+        data = toast_events[0]['args'][0]
+        assert 'icon' in data
+        assert 'title' in data
+        assert 'subtitle' in data
+        assert 'time' in data
+        assert 'timestamp' in data
+        assert 'show_toast' in data
+        assert data['show_toast'] is True
+
+    def test_http_toasts_still_works(self, flask_client, shared_state):
+        """GET /api/activity/toasts returns 200 with expected structure."""
+        # Add a toast-worthy activity first
+        add_item = shared_state['add_activity_item']
+        add_item('✅', 'Test', 'Sub', show_toast=True)
+
+        resp = flask_client.get('/api/activity/toasts')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert 'toasts' in data
+
+
+# =========================================================================
+# Group D — DB Stats
+# =========================================================================
+
+class TestDbStats:
+    """dashboard:db_stats socket events match GET /api/database/stats."""
+
+    def test_db_stats_event_received(self, test_app, shared_state):
+        """Client receives a dashboard:db_stats event."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_db_stats']
+        socketio.emit('dashboard:db_stats', build())
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'dashboard:db_stats']
+        assert len(events) >= 1
+
+    def test_db_stats_shape(self, test_app, shared_state):
+        """dashboard:db_stats event data has expected keys."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_db_stats']
+        socketio.emit('dashboard:db_stats', build())
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'dashboard:db_stats']
+        assert len(events) >= 1
+        data = events[0]['args'][0]
+        assert 'artists' in data
+        assert 'albums' in data
+        assert 'tracks' in data
+        assert 'database_size_mb' in data
+        assert 'server_source' in data
+
+    def test_db_stats_matches_http(self, test_app, shared_state):
+        """Socket event data matches HTTP endpoint response."""
+        app, socketio = test_app
+        flask_client = app.test_client()
+        ws_client = socketio.test_client(app)
+        build = shared_state['build_db_stats']
+
+        http_data = flask_client.get('/api/database/stats').get_json()
+
+        socketio.emit('dashboard:db_stats', build())
+        received = ws_client.get_received()
+        events = [e for e in received if e['name'] == 'dashboard:db_stats']
+        ws_data = events[0]['args'][0]
+
+        assert ws_data['artists'] == http_data['artists']
+        assert ws_data['albums'] == http_data['albums']
+        assert ws_data['tracks'] == http_data['tracks']
+        assert ws_data['database_size_mb'] == http_data['database_size_mb']
+        assert ws_data['server_source'] == http_data['server_source']
+
+    def test_http_db_stats_still_works(self, flask_client):
+        """GET /api/database/stats returns 200 with expected keys."""
+        resp = flask_client.get('/api/database/stats')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert 'artists' in data
+        assert 'albums' in data
+        assert 'tracks' in data
+
+
+# =========================================================================
+# Group E — Wishlist Count
+# =========================================================================
+
+class TestWishlistCount:
+    """dashboard:wishlist_count socket events match GET /api/wishlist/count."""
+
+    def test_wishlist_count_received(self, test_app, shared_state):
+        """Client receives a dashboard:wishlist_count event."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_wishlist_count_payload_ws']
+        socketio.emit('dashboard:wishlist_count', build())
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'dashboard:wishlist_count']
+        assert len(events) >= 1
+
+    def test_wishlist_count_shape(self, test_app, shared_state):
+        """dashboard:wishlist_count event data has count key."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_wishlist_count_payload_ws']
+        socketio.emit('dashboard:wishlist_count', build())
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'dashboard:wishlist_count']
+        assert len(events) >= 1
+        data = events[0]['args'][0]
+        assert 'count' in data
+        assert isinstance(data['count'], int)
+
+    def test_wishlist_count_matches_http(self, test_app, shared_state):
+        """Socket event data matches HTTP endpoint response."""
+        app, socketio = test_app
+        flask_client = app.test_client()
+        ws_client = socketio.test_client(app)
+        build = shared_state['build_wishlist_count_payload_ws']
+
+        http_data = flask_client.get('/api/wishlist/count').get_json()
+
+        socketio.emit('dashboard:wishlist_count', build())
+        received = ws_client.get_received()
+        events = [e for e in received if e['name'] == 'dashboard:wishlist_count']
+        ws_data = events[0]['args'][0]
+
+        assert ws_data['count'] == http_data['count']
+
+    def test_http_wishlist_count_still_works(self, flask_client):
+        """GET /api/wishlist/count returns 200 with count."""
+        resp = flask_client.get('/api/wishlist/count')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert 'count' in data
+        assert isinstance(data['count'], int)
+
+
+# =========================================================================
+# Group F — Backward Compatibility
+# =========================================================================
+
+class TestBackwardCompat:
+    """HTTP endpoints work when no WebSocket is connected."""
+
+    def test_all_http_endpoints_work_without_socket(self, flask_client):
+        """All 5 Phase 2 HTTP endpoints work without any WebSocket connection."""
+        # System stats
+        resp = flask_client.get('/api/system/stats')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert data['active_downloads'] == 2
+
+        # Activity feed
+        resp = flask_client.get('/api/activity/feed')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert 'activities' in data
+
+        # Toasts
+        resp = flask_client.get('/api/activity/toasts')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert 'toasts' in data
+
+        # DB stats
+        resp = flask_client.get('/api/database/stats')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert data['artists'] == 350
+
+        # Wishlist count
+        resp = flask_client.get('/api/wishlist/count')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert data['count'] == 5
+
+    def test_multiple_clients_get_dashboard_updates(self, test_app, shared_state):
+        """Multiple WebSocket clients each receive dashboard broadcast events."""
+        app, socketio = test_app
+        client1 = socketio.test_client(app)
+        client2 = socketio.test_client(app)
+        build = shared_state['build_system_stats']
+
+        socketio.emit('dashboard:stats', build())
+
+        for client in [client1, client2]:
+            received = client.get_received()
+            events = [e for e in received if e['name'] == 'dashboard:stats']
+            assert len(events) >= 1
+
+        client1.disconnect()
+        client2.disconnect()

tests/test_phase3_enrichment.py

@@ -0,0 +1,229 @@
+"""Phase 3 WebSocket migration tests — Enrichment sidebar workers.
+
+Verifies that:
+ - All 7 enrichment worker statuses are delivered identically via
+   WebSocket events and HTTP endpoints
+ - Each worker's data shape is correct
+ - HTTP endpoints still work as fallback
+
+IMPORTANT: Do NOT use ``from tests.conftest import …`` — pytest's auto-discovered
+conftest is a different module instance. Use the ``shared_state`` fixture instead.
+"""
+
+import pytest
+
+
+# All 7 enrichment workers
+WORKERS = [
+    'musicbrainz', 'audiodb', 'deezer',
+    'spotify-enrichment', 'itunes-enrichment',
+    'hydrabase', 'repair',
+]
+
+# Endpoint URLs keyed by worker name
+ENDPOINTS = {
+    'musicbrainz': '/api/musicbrainz/status',
+    'audiodb': '/api/audiodb/status',
+    'deezer': '/api/deezer/status',
+    'spotify-enrichment': '/api/spotify-enrichment/status',
+    'itunes-enrichment': '/api/itunes-enrichment/status',
+    'hydrabase': '/api/hydrabase-worker/status',
+    'repair': '/api/repair/status',
+}
+
+
+# =========================================================================
+# Group A — Event Delivery (parameterized)
+# =========================================================================
+
+class TestEnrichmentEventDelivery:
+    """enrichment:<worker> socket events are received by the client."""
+
+    @pytest.mark.parametrize('worker', WORKERS)
+    def test_enrichment_event_received(self, test_app, shared_state, worker):
+        """Client receives an enrichment:<worker> event."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_enrichment_status']
+
+        socketio.emit(f'enrichment:{worker}', build(worker))
+        received = client.get_received()
+        events = [e for e in received if e['name'] == f'enrichment:{worker}']
+        assert len(events) >= 1
+
+
+# =========================================================================
+# Group B — Data Shape (parameterized)
+# =========================================================================
+
+class TestEnrichmentDataShape:
+    """enrichment:<worker> event data has the expected keys."""
+
+    @pytest.mark.parametrize('worker', [
+        'musicbrainz', 'audiodb', 'deezer',
+        'spotify-enrichment', 'itunes-enrichment',
+    ])
+    def test_standard_enrichment_shape(self, test_app, shared_state, worker):
+        """Standard enrichment worker data has running, paused, idle, progress."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_enrichment_status']
+
+        socketio.emit(f'enrichment:{worker}', build(worker))
+        received = client.get_received()
+        events = [e for e in received if e['name'] == f'enrichment:{worker}']
+        assert len(events) >= 1
+        data = events[0]['args'][0]
+
+        assert 'running' in data
+        assert 'paused' in data
+        assert 'idle' in data
+        assert 'current_item' in data
+        assert 'progress' in data
+        assert isinstance(data['running'], bool)
+        assert isinstance(data['paused'], bool)
+
+    def test_spotify_enrichment_has_authenticated(self, test_app, shared_state):
+        """Spotify enrichment includes the 'authenticated' field."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_enrichment_status']
+
+        socketio.emit('enrichment:spotify-enrichment', build('spotify-enrichment'))
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'enrichment:spotify-enrichment']
+        data = events[0]['args'][0]
+        assert 'authenticated' in data
+
+    def test_hydrabase_shape(self, test_app, shared_state):
+        """Hydrabase worker has running, paused, queue_size (no idle/progress)."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_enrichment_status']
+
+        socketio.emit('enrichment:hydrabase', build('hydrabase'))
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'enrichment:hydrabase']
+        data = events[0]['args'][0]
+
+        assert 'running' in data
+        assert 'paused' in data
+        assert 'queue_size' in data
+        assert 'idle' not in data  # Hydrabase doesn't have idle
+
+    def test_repair_shape(self, test_app, shared_state):
+        """Repair worker has progress.tracks with checked/repaired counters."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_enrichment_status']
+
+        socketio.emit('enrichment:repair', build('repair'))
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'enrichment:repair']
+        data = events[0]['args'][0]
+
+        assert 'running' in data
+        assert 'progress' in data
+        tracks = data['progress']['tracks']
+        assert 'checked' in tracks
+        assert 'total' in tracks
+        assert 'repaired' in tracks
+
+
+# =========================================================================
+# Group C — HTTP Parity (parameterized)
+# =========================================================================
+
+class TestEnrichmentHttpParity:
+    """Socket event data matches HTTP endpoint response."""
+
+    @pytest.mark.parametrize('worker', WORKERS)
+    def test_enrichment_matches_http(self, test_app, shared_state, worker):
+        """Socket event data matches GET /api/<worker>/status."""
+        app, socketio = test_app
+        flask_client = app.test_client()
+        ws_client = socketio.test_client(app)
+        build = shared_state['build_enrichment_status']
+
+        endpoint = ENDPOINTS[worker]
+        http_data = flask_client.get(endpoint).get_json()
+
+        socketio.emit(f'enrichment:{worker}', build(worker))
+        received = ws_client.get_received()
+        events = [e for e in received if e['name'] == f'enrichment:{worker}']
+        assert len(events) >= 1
+        ws_data = events[0]['args'][0]
+
+        # Both should have the same running/paused state
+        assert ws_data['running'] == http_data['running']
+        assert ws_data['paused'] == http_data['paused']
+
+
+# =========================================================================
+# Group D — HTTP Still Works (parameterized)
+# =========================================================================
+
+class TestEnrichmentHttpStillWorks:
+    """HTTP endpoints return 200 with expected structure."""
+
+    @pytest.mark.parametrize('worker', WORKERS)
+    def test_http_enrichment_still_works(self, flask_client, worker):
+        """GET /api/<worker>/status returns 200."""
+        endpoint = ENDPOINTS[worker]
+        resp = flask_client.get(endpoint)
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert 'running' in data
+        assert 'paused' in data
+
+
+# =========================================================================
+# Group E — Backward Compatibility
+# =========================================================================
+
+class TestEnrichmentBackwardCompat:
+    """HTTP endpoints work when no WebSocket is connected."""
+
+    def test_all_http_endpoints_work_without_socket(self, flask_client):
+        """All 7 enrichment HTTP endpoints work without any WebSocket connection."""
+        for worker in WORKERS:
+            endpoint = ENDPOINTS[worker]
+            resp = flask_client.get(endpoint)
+            assert resp.status_code == 200
+            data = resp.get_json()
+            assert data['running'] is True
+
+    def test_multiple_clients_get_enrichment_updates(self, test_app, shared_state):
+        """Multiple WebSocket clients each receive enrichment events."""
+        app, socketio = test_app
+        client1 = socketio.test_client(app)
+        client2 = socketio.test_client(app)
+        build = shared_state['build_enrichment_status']
+
+        socketio.emit('enrichment:musicbrainz', build('musicbrainz'))
+
+        for client in [client1, client2]:
+            received = client.get_received()
+            events = [e for e in received if e['name'] == 'enrichment:musicbrainz']
+            assert len(events) >= 1
+
+        client1.disconnect()
+        client2.disconnect()
+
+    def test_enrichment_reflects_state_change(self, test_app, shared_state):
+        """When enrichment state changes, the next emit reflects it."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        enrich = shared_state['enrichment_status']
+        build = shared_state['build_enrichment_status']
+
+        # Mutate state
+        enrich['musicbrainz']['paused'] = True
+        enrich['musicbrainz']['running'] = False
+
+        socketio.emit('enrichment:musicbrainz', build('musicbrainz'))
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'enrichment:musicbrainz']
+        data = events[-1]['args'][0]
+        assert data['paused'] is True
+        assert data['running'] is False

tests/test_phase4_tools.py

@@ -0,0 +1,329 @@
+"""Phase 4 WebSocket migration tests — Tool progress pollers.
+
+Verifies that:
+ - All 7 tool progress statuses are delivered identically via
+   WebSocket events and HTTP endpoints
+ - Each tool's data shape is correct
+ - HTTP endpoints still work as fallback
+
+IMPORTANT: Do NOT use ``from tests.conftest import …`` — pytest's auto-discovered
+conftest is a different module instance. Use the ``shared_state`` fixture instead.
+"""
+
+import pytest
+
+
+# All 7 tool progress pollers
+TOOLS = [
+    'stream', 'quality-scanner', 'duplicate-cleaner',
+    'retag', 'db-update', 'metadata', 'logs',
+]
+
+# Endpoint URLs keyed by tool name
+ENDPOINTS = {
+    'stream': '/api/stream/status',
+    'quality-scanner': '/api/quality-scanner/status',
+    'duplicate-cleaner': '/api/duplicate-cleaner/status',
+    'retag': '/api/retag/status',
+    'db-update': '/api/database/update/status',
+    'metadata': '/api/metadata/status',
+    'logs': '/api/logs',
+}
+
+
+# =========================================================================
+# Group A — Event Delivery (parameterized)
+# =========================================================================
+
+class TestToolEventDelivery:
+    """tool:<name> socket events are received by the client."""
+
+    @pytest.mark.parametrize('tool', TOOLS)
+    def test_tool_event_received(self, test_app, shared_state, tool):
+        """Client receives a tool:<name> event."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_tool_status']
+
+        socketio.emit(f'tool:{tool}', build(tool))
+        received = client.get_received()
+        events = [e for e in received if e['name'] == f'tool:{tool}']
+        assert len(events) >= 1
+
+
+# =========================================================================
+# Group B — Data Shape (individual per tool)
+# =========================================================================
+
+class TestToolDataShape:
+    """tool:<name> event data has the expected keys."""
+
+    def test_stream_shape(self, test_app, shared_state):
+        """Stream status has status, progress, track_info, error_message."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_stream_status']
+
+        socketio.emit('tool:stream', build())
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'tool:stream']
+        assert len(events) >= 1
+        data = events[0]['args'][0]
+
+        assert 'status' in data
+        assert 'progress' in data
+        assert 'track_info' in data
+        assert 'error_message' in data
+        assert isinstance(data['progress'], (int, float))
+
+    def test_quality_scanner_shape(self, test_app, shared_state):
+        """Quality scanner has status, phase, progress, processed, total, quality_met."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_quality_scanner_status']
+
+        socketio.emit('tool:quality-scanner', build())
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'tool:quality-scanner']
+        assert len(events) >= 1
+        data = events[0]['args'][0]
+
+        assert 'status' in data
+        assert 'phase' in data
+        assert 'progress' in data
+        assert 'processed' in data
+        assert 'total' in data
+        assert 'quality_met' in data
+        assert 'low_quality' in data
+        assert 'matched' in data
+
+    def test_duplicate_cleaner_shape(self, test_app, shared_state):
+        """Duplicate cleaner has status, phase, progress, space_freed_mb."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_duplicate_cleaner_status']
+
+        socketio.emit('tool:duplicate-cleaner', build())
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'tool:duplicate-cleaner']
+        assert len(events) >= 1
+        data = events[0]['args'][0]
+
+        assert 'status' in data
+        assert 'phase' in data
+        assert 'progress' in data
+        assert 'files_scanned' in data
+        assert 'total_files' in data
+        assert 'duplicates_found' in data
+        assert 'deleted' in data
+        assert 'space_freed_mb' in data
+        assert isinstance(data['space_freed_mb'], (int, float))
+
+    def test_retag_shape(self, test_app, shared_state):
+        """Retag has status, phase, progress, current_track, total_tracks."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_retag_status']
+
+        socketio.emit('tool:retag', build())
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'tool:retag']
+        assert len(events) >= 1
+        data = events[0]['args'][0]
+
+        assert 'status' in data
+        assert 'phase' in data
+        assert 'progress' in data
+        assert 'current_track' in data
+        assert 'total_tracks' in data
+        assert 'processed' in data
+
+    def test_db_update_shape(self, test_app, shared_state):
+        """DB update has status, phase, progress, removed_artists/albums/tracks."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_db_update_status']
+
+        socketio.emit('tool:db-update', build())
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'tool:db-update']
+        assert len(events) >= 1
+        data = events[0]['args'][0]
+
+        assert 'status' in data
+        assert 'phase' in data
+        assert 'progress' in data
+        assert 'current_item' in data
+        assert 'processed' in data
+        assert 'total' in data
+        assert 'removed_artists' in data
+        assert 'removed_albums' in data
+        assert 'removed_tracks' in data
+
+    def test_metadata_shape(self, test_app, shared_state):
+        """Metadata has {success, status} wrapper with inner percentage, successful, failed."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_metadata_status']
+
+        socketio.emit('tool:metadata', build())
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'tool:metadata']
+        assert len(events) >= 1
+        data = events[0]['args'][0]
+
+        assert 'success' in data
+        assert data['success'] is True
+        assert 'status' in data
+        status = data['status']
+        assert 'status' in status
+        assert 'current_artist' in status
+        assert 'processed' in status
+        assert 'total' in status
+        assert 'percentage' in status
+        assert 'successful' in status
+        assert 'failed' in status
+
+    def test_logs_shape(self, test_app, shared_state):
+        """Logs has logs array of strings."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_logs']
+
+        socketio.emit('tool:logs', build())
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'tool:logs']
+        assert len(events) >= 1
+        data = events[0]['args'][0]
+
+        assert 'logs' in data
+        assert isinstance(data['logs'], list)
+        assert len(data['logs']) >= 1
+        assert isinstance(data['logs'][0], str)
+
+
+# =========================================================================
+# Group C — HTTP Parity (parameterized)
+# =========================================================================
+
+class TestToolHttpParity:
+    """Socket event data matches HTTP endpoint response."""
+
+    @pytest.mark.parametrize('tool', [t for t in TOOLS if t != 'logs'])
+    def test_tool_matches_http(self, test_app, shared_state, tool):
+        """Socket event data matches GET endpoint for non-logs tools."""
+        app, socketio = test_app
+        flask_client = app.test_client()
+        ws_client = socketio.test_client(app)
+        build = shared_state['build_tool_status']
+
+        endpoint = ENDPOINTS[tool]
+        http_data = flask_client.get(endpoint).get_json()
+
+        socketio.emit(f'tool:{tool}', build(tool))
+        received = ws_client.get_received()
+        events = [e for e in received if e['name'] == f'tool:{tool}']
+        assert len(events) >= 1
+        ws_data = events[0]['args'][0]
+
+        if tool == 'metadata':
+            # Metadata wraps in {success, status}
+            assert ws_data['success'] == http_data['success']
+            assert ws_data['status']['status'] == http_data['status']['status']
+            assert ws_data['status']['processed'] == http_data['status']['processed']
+        else:
+            assert ws_data['status'] == http_data['status']
+
+    def test_logs_matches_http(self, test_app, shared_state):
+        """Logs event data matches GET /api/logs."""
+        app, socketio = test_app
+        flask_client = app.test_client()
+        ws_client = socketio.test_client(app)
+        build = shared_state['build_logs']
+
+        http_data = flask_client.get('/api/logs').get_json()
+
+        socketio.emit('tool:logs', build())
+        received = ws_client.get_received()
+        events = [e for e in received if e['name'] == 'tool:logs']
+        assert len(events) >= 1
+        ws_data = events[0]['args'][0]
+
+        assert len(ws_data['logs']) == len(http_data['logs'])
+        if ws_data['logs']:
+            assert ws_data['logs'][0] == http_data['logs'][0]
+
+
+# =========================================================================
+# Group D — HTTP Still Works (parameterized)
+# =========================================================================
+
+class TestToolHttpStillWorks:
+    """HTTP endpoints return 200 with expected structure."""
+
+    @pytest.mark.parametrize('tool', TOOLS)
+    def test_http_tool_still_works(self, flask_client, tool):
+        """GET /api/<tool>/status returns 200."""
+        endpoint = ENDPOINTS[tool]
+        resp = flask_client.get(endpoint)
+        assert resp.status_code == 200
+        data = resp.get_json()
+        if tool == 'logs':
+            assert 'logs' in data
+        elif tool == 'metadata':
+            assert 'success' in data
+            assert 'status' in data
+        else:
+            assert 'status' in data
+
+
+# =========================================================================
+# Group E — Backward Compatibility
+# =========================================================================
+
+class TestToolBackwardCompat:
+    """HTTP endpoints work when no WebSocket is connected."""
+
+    def test_all_http_endpoints_work_without_socket(self, flask_client):
+        """All 7 tool HTTP endpoints work without any WebSocket connection."""
+        for tool in TOOLS:
+            endpoint = ENDPOINTS[tool]
+            resp = flask_client.get(endpoint)
+            assert resp.status_code == 200
+
+    def test_multiple_clients_get_tool_updates(self, test_app, shared_state):
+        """Multiple WebSocket clients each receive tool events."""
+        app, socketio = test_app
+        client1 = socketio.test_client(app)
+        client2 = socketio.test_client(app)
+        build = shared_state['build_tool_status']
+
+        socketio.emit('tool:quality-scanner', build('quality-scanner'))
+
+        for client in [client1, client2]:
+            received = client.get_received()
+            events = [e for e in received if e['name'] == 'tool:quality-scanner']
+            assert len(events) >= 1
+
+        client1.disconnect()
+        client2.disconnect()
+
+    def test_tool_reflects_state_change(self, test_app, shared_state):
+        """When tool state changes, the next emit reflects it."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_tool_status']
+        qs = shared_state['quality_scanner_state']
+
+        # Mutate state
+        qs['status'] = 'finished'
+        qs['progress'] = 100
+        qs['processed'] = 100
+
+        socketio.emit('tool:quality-scanner', build('quality-scanner'))
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'tool:quality-scanner']
+        data = events[-1]['args'][0]
+        assert data['status'] == 'finished'
+        assert data['progress'] == 100
+        assert data['processed'] == 100

tests/test_phase5_sync.py

@@ -0,0 +1,920 @@
+"""Phase 5 WebSocket migration tests — Sync/Discovery Progress + Scans.
+
+Verifies that:
+ - Room-based sync:progress events are delivered only to subscribed clients
+ - Room-based discovery:progress events are delivered only to subscribed clients
+ - Broadcast scan:watchlist and scan:media events reach all clients
+ - Data shapes are correct for each event type
+ - HTTP endpoints still work as fallback
+
+IMPORTANT: Do NOT use ``from tests.conftest import …`` — pytest's auto-discovered
+conftest is a different module instance. Use the ``shared_state`` fixture instead.
+"""
+
+import pytest
+
+
+# =========================================================================
+# Constants
+# =========================================================================
+
+DISCOVERY_PLATFORMS = ['tidal', 'youtube']
+
+SYNC_PLAYLIST_IDS = ['test-playlist-1']
+
+
+# =========================================================================
+# Group A — Sync Event Delivery (room-based)
+# =========================================================================
+
+class TestSyncEventDelivery:
+    """sync:progress socket events are received by subscribed clients."""
+
+    def test_sync_event_received(self, test_app, shared_state):
+        """Client subscribes and receives sync:progress event."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_sync_status']
+
+        # Subscribe to sync room
+        client.emit('sync:subscribe', {'playlist_ids': ['test-playlist-1']})
+
+        # Server emits to the room
+        socketio.emit('sync:progress', {
+            'playlist_id': 'test-playlist-1',
+            **build('test-playlist-1')
+        }, room='sync:test-playlist-1')
+
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'sync:progress']
+        assert len(events) >= 1
+
+        client.disconnect()
+
+    def test_sync_only_subscribed(self, test_app, shared_state):
+        """Unsubscribed client does NOT receive sync:progress events."""
+        app, socketio = test_app
+        subscribed = socketio.test_client(app)
+        unsubscribed = socketio.test_client(app)
+        build = shared_state['build_sync_status']
+
+        # Only one client subscribes
+        subscribed.emit('sync:subscribe', {'playlist_ids': ['test-playlist-1']})
+
+        socketio.emit('sync:progress', {
+            'playlist_id': 'test-playlist-1',
+            **build('test-playlist-1')
+        }, room='sync:test-playlist-1')
+
+        sub_events = [e for e in subscribed.get_received()
+                      if e['name'] == 'sync:progress']
+        unsub_events = [e for e in unsubscribed.get_received()
+                        if e['name'] == 'sync:progress']
+
+        assert len(sub_events) >= 1
+        assert len(unsub_events) == 0
+
+        subscribed.disconnect()
+        unsubscribed.disconnect()
+
+    def test_sync_subscribe_unsubscribe(self, test_app, shared_state):
+        """Client stops receiving events after unsubscribing."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_sync_status']
+
+        # Subscribe
+        client.emit('sync:subscribe', {'playlist_ids': ['test-playlist-1']})
+
+        # First emit — should receive
+        socketio.emit('sync:progress', {
+            'playlist_id': 'test-playlist-1',
+            **build('test-playlist-1')
+        }, room='sync:test-playlist-1')
+
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'sync:progress']
+        assert len(events) >= 1
+
+        # Unsubscribe
+        client.emit('sync:unsubscribe', {'playlist_ids': ['test-playlist-1']})
+
+        # Second emit — should NOT receive
+        socketio.emit('sync:progress', {
+            'playlist_id': 'test-playlist-1',
+            **build('test-playlist-1')
+        }, room='sync:test-playlist-1')
+
+        received2 = client.get_received()
+        events2 = [e for e in received2 if e['name'] == 'sync:progress']
+        assert len(events2) == 0
+
+        client.disconnect()
+
+
+# =========================================================================
+# Group B — Sync Data Shape
+# =========================================================================
+
+class TestSyncDataShape:
+    """sync:progress event data has the expected keys."""
+
+    def test_sync_progress_shape(self, test_app, shared_state):
+        """Sync progress has playlist_id, status, progress dict."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_sync_status']
+
+        client.emit('sync:subscribe', {'playlist_ids': ['test-playlist-1']})
+
+        payload = {'playlist_id': 'test-playlist-1', **build('test-playlist-1')}
+        socketio.emit('sync:progress', payload, room='sync:test-playlist-1')
+
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'sync:progress']
+        assert len(events) >= 1
+        data = events[0]['args'][0]
+
+        assert 'playlist_id' in data
+        assert data['playlist_id'] == 'test-playlist-1'
+        assert 'status' in data
+        assert data['status'] == 'syncing'
+        assert 'progress' in data
+        progress = data['progress']
+        assert 'total_tracks' in progress
+        assert 'matched_tracks' in progress
+        assert 'progress' in progress
+        assert isinstance(progress['progress'], (int, float))
+
+        client.disconnect()
+
+
+# =========================================================================
+# Group C — Discovery Event Delivery (room-based)
+# =========================================================================
+
+class TestDiscoveryEventDelivery:
+    """discovery:progress socket events are received by subscribed clients."""
+
+    @pytest.mark.parametrize('platform,pid', [
+        ('tidal', 'test-tidal-1'),
+        ('youtube', 'test-yt-hash'),
+    ])
+    def test_discovery_event_received(self, test_app, shared_state, platform, pid):
+        """Client subscribes and receives discovery:progress event."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_discovery_status']
+
+        client.emit('discovery:subscribe', {'ids': [pid]})
+
+        payload = build(platform, pid)
+        payload['platform'] = platform
+        payload['id'] = pid
+        socketio.emit('discovery:progress', payload, room=f'discovery:{pid}')
+
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'discovery:progress']
+        assert len(events) >= 1
+
+        client.disconnect()
+
+    def test_discovery_only_subscribed(self, test_app, shared_state):
+        """Unsubscribed client does NOT receive discovery:progress events."""
+        app, socketio = test_app
+        subscribed = socketio.test_client(app)
+        unsubscribed = socketio.test_client(app)
+        build = shared_state['build_discovery_status']
+
+        subscribed.emit('discovery:subscribe', {'ids': ['test-tidal-1']})
+
+        payload = build('tidal', 'test-tidal-1')
+        payload['platform'] = 'tidal'
+        payload['id'] = 'test-tidal-1'
+        socketio.emit('discovery:progress', payload, room='discovery:test-tidal-1')
+
+        sub_events = [e for e in subscribed.get_received()
+                      if e['name'] == 'discovery:progress']
+        unsub_events = [e for e in unsubscribed.get_received()
+                        if e['name'] == 'discovery:progress']
+
+        assert len(sub_events) >= 1
+        assert len(unsub_events) == 0
+
+        subscribed.disconnect()
+        unsubscribed.disconnect()
+
+    def test_discovery_subscribe_unsubscribe(self, test_app, shared_state):
+        """Client stops receiving discovery events after unsubscribing."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_discovery_status']
+
+        client.emit('discovery:subscribe', {'ids': ['test-yt-hash']})
+
+        payload = build('youtube', 'test-yt-hash')
+        payload['platform'] = 'youtube'
+        payload['id'] = 'test-yt-hash'
+        socketio.emit('discovery:progress', payload, room='discovery:test-yt-hash')
+
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'discovery:progress']
+        assert len(events) >= 1
+
+        client.emit('discovery:unsubscribe', {'ids': ['test-yt-hash']})
+
+        socketio.emit('discovery:progress', payload, room='discovery:test-yt-hash')
+        received2 = client.get_received()
+        events2 = [e for e in received2 if e['name'] == 'discovery:progress']
+        assert len(events2) == 0
+
+        client.disconnect()
+
+
+# =========================================================================
+# Group D — Discovery Data Shape
+# =========================================================================
+
+class TestDiscoveryDataShape:
+    """discovery:progress event data has the expected keys."""
+
+    @pytest.mark.parametrize('platform,pid', [
+        ('tidal', 'test-tidal-1'),
+        ('youtube', 'test-yt-hash'),
+    ])
+    def test_discovery_progress_shape(self, test_app, shared_state, platform, pid):
+        """Discovery progress has platform, id, phase, status, complete, results."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_discovery_status']
+
+        client.emit('discovery:subscribe', {'ids': [pid]})
+
+        payload = build(platform, pid)
+        payload['platform'] = platform
+        payload['id'] = pid
+        socketio.emit('discovery:progress', payload, room=f'discovery:{pid}')
+
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'discovery:progress']
+        assert len(events) >= 1
+        data = events[0]['args'][0]
+
+        assert data['platform'] == platform
+        assert data['id'] == pid
+        assert 'phase' in data
+        assert 'status' in data
+        assert 'progress' in data
+        assert 'spotify_matches' in data
+        assert 'spotify_total' in data
+        assert 'results' in data
+        assert 'complete' in data
+        assert isinstance(data['results'], list)
+        assert isinstance(data['complete'], bool)
+
+        client.disconnect()
+
+
+# =========================================================================
+# Group E — Scan Events (broadcast)
+# =========================================================================
+
+class TestScanEventDelivery:
+    """Broadcast scan events are received by all connected clients."""
+
+    def test_watchlist_scan_received(self, test_app, shared_state):
+        """All clients receive scan:watchlist event."""
+        app, socketio = test_app
+        client1 = socketio.test_client(app)
+        client2 = socketio.test_client(app)
+        build = shared_state['build_watchlist_scan_status']
+
+        socketio.emit('scan:watchlist', build())
+
+        for client in [client1, client2]:
+            received = client.get_received()
+            events = [e for e in received if e['name'] == 'scan:watchlist']
+            assert len(events) >= 1
+
+        client1.disconnect()
+        client2.disconnect()
+
+    def test_watchlist_scan_shape(self, test_app, shared_state):
+        """scan:watchlist data has success, status, current_artist_name."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_watchlist_scan_status']
+
+        socketio.emit('scan:watchlist', build())
+
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'scan:watchlist']
+        assert len(events) >= 1
+        data = events[0]['args'][0]
+
+        assert data['success'] is True
+        assert 'status' in data
+        assert data['status'] == 'scanning'
+        assert 'current_artist_name' in data
+        assert 'current_album' in data
+        assert 'current_track_name' in data
+
+        client.disconnect()
+
+    def test_media_scan_received(self, test_app, shared_state):
+        """All clients receive scan:media event."""
+        app, socketio = test_app
+        client1 = socketio.test_client(app)
+        client2 = socketio.test_client(app)
+        build = shared_state['build_media_scan_status']
+
+        socketio.emit('scan:media', build())
+
+        for client in [client1, client2]:
+            received = client.get_received()
+            events = [e for e in received if e['name'] == 'scan:media']
+            assert len(events) >= 1
+
+        client1.disconnect()
+        client2.disconnect()
+
+    def test_media_scan_shape(self, test_app, shared_state):
+        """scan:media data has success and status with is_scanning."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_media_scan_status']
+
+        socketio.emit('scan:media', build())
+
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'scan:media']
+        assert len(events) >= 1
+        data = events[0]['args'][0]
+
+        assert data['success'] is True
+        assert 'status' in data
+        status = data['status']
+        assert 'is_scanning' in status
+        assert 'status' in status
+        assert 'progress_message' in status
+
+        client.disconnect()
+
+
+# =========================================================================
+# Group F — HTTP Parity
+# =========================================================================
+
+class TestSyncHttpParity:
+    """Socket event data matches HTTP endpoint response for sync."""
+
+    def test_sync_matches_http(self, test_app, shared_state):
+        """Sync socket data matches GET /api/sync/status/:id."""
+        app, socketio = test_app
+        flask_client = app.test_client()
+        ws_client = socketio.test_client(app)
+        build = shared_state['build_sync_status']
+
+        endpoint = shared_state['sync_endpoints']['sync']
+        http_data = flask_client.get(endpoint).get_json()
+
+        ws_client.emit('sync:subscribe', {'playlist_ids': ['test-playlist-1']})
+
+        payload = {'playlist_id': 'test-playlist-1', **build('test-playlist-1')}
+        socketio.emit('sync:progress', payload, room='sync:test-playlist-1')
+
+        received = ws_client.get_received()
+        events = [e for e in received if e['name'] == 'sync:progress']
+        assert len(events) >= 1
+        ws_data = events[0]['args'][0]
+
+        assert ws_data['status'] == http_data['status']
+        assert ws_data['playlist_id'] == http_data['playlist_id']
+        assert ws_data['progress']['total_tracks'] == http_data['progress']['total_tracks']
+        assert ws_data['progress']['matched_tracks'] == http_data['progress']['matched_tracks']
+
+        ws_client.disconnect()
+
+
+class TestDiscoveryHttpParity:
+    """Socket event data matches HTTP endpoint response for discovery."""
+
+    @pytest.mark.parametrize('platform,pid,endpoint_key', [
+        ('tidal', 'test-tidal-1', 'tidal'),
+        ('youtube', 'test-yt-hash', 'youtube'),
+    ])
+    def test_discovery_matches_http(self, test_app, shared_state,
+                                    platform, pid, endpoint_key):
+        """Discovery socket data matches GET /api/<platform>/discovery/status/:id."""
+        app, socketio = test_app
+        flask_client = app.test_client()
+        ws_client = socketio.test_client(app)
+        build = shared_state['build_discovery_status']
+
+        endpoint = shared_state['discovery_endpoints'][endpoint_key]
+        http_data = flask_client.get(endpoint).get_json()
+
+        ws_client.emit('discovery:subscribe', {'ids': [pid]})
+
+        payload = build(platform, pid)
+        payload['platform'] = platform
+        payload['id'] = pid
+        socketio.emit('discovery:progress', payload, room=f'discovery:{pid}')
+
+        received = ws_client.get_received()
+        events = [e for e in received if e['name'] == 'discovery:progress']
+        assert len(events) >= 1
+        ws_data = events[0]['args'][0]
+
+        assert ws_data['phase'] == http_data['phase']
+        assert ws_data['status'] == http_data['status']
+        assert ws_data['spotify_matches'] == http_data['spotify_matches']
+        assert ws_data['spotify_total'] == http_data['spotify_total']
+        assert ws_data['complete'] == http_data['complete']
+
+        ws_client.disconnect()
+
+
+class TestScanHttpParity:
+    """Socket event data matches HTTP endpoint response for scans."""
+
+    def test_watchlist_scan_matches_http(self, test_app, shared_state):
+        """scan:watchlist socket data matches GET /api/watchlist/scan/status."""
+        app, socketio = test_app
+        flask_client = app.test_client()
+        ws_client = socketio.test_client(app)
+        build = shared_state['build_watchlist_scan_status']
+
+        endpoint = shared_state['scan_endpoints']['watchlist']
+        http_data = flask_client.get(endpoint).get_json()
+
+        socketio.emit('scan:watchlist', build())
+
+        received = ws_client.get_received()
+        events = [e for e in received if e['name'] == 'scan:watchlist']
+        assert len(events) >= 1
+        ws_data = events[0]['args'][0]
+
+        assert ws_data['success'] == http_data['success']
+        assert ws_data['status'] == http_data['status']
+        assert ws_data['current_artist_name'] == http_data['current_artist_name']
+
+        ws_client.disconnect()
+
+    def test_media_scan_matches_http(self, test_app, shared_state):
+        """scan:media socket data matches GET /api/scan/status."""
+        app, socketio = test_app
+        flask_client = app.test_client()
+        ws_client = socketio.test_client(app)
+        build = shared_state['build_media_scan_status']
+
+        endpoint = shared_state['scan_endpoints']['media']
+        http_data = flask_client.get(endpoint).get_json()
+
+        socketio.emit('scan:media', build())
+
+        received = ws_client.get_received()
+        events = [e for e in received if e['name'] == 'scan:media']
+        assert len(events) >= 1
+        ws_data = events[0]['args'][0]
+
+        assert ws_data['success'] == http_data['success']
+        assert ws_data['status']['is_scanning'] == http_data['status']['is_scanning']
+        assert ws_data['status']['status'] == http_data['status']['status']
+
+        ws_client.disconnect()
+
+
+# =========================================================================
+# Group G — HTTP Still Works
+# =========================================================================
+
+class TestHttpStillWorks:
+    """HTTP endpoints return 200 with expected structure."""
+
+    def test_sync_http_works(self, flask_client):
+        """GET /api/sync/status/:id returns 200."""
+        resp = flask_client.get('/api/sync/status/test-playlist-1')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert 'status' in data
+        assert 'progress' in data
+
+    def test_sync_http_404_unknown(self, flask_client):
+        """GET /api/sync/status/:id returns 404 for unknown playlist."""
+        resp = flask_client.get('/api/sync/status/nonexistent')
+        assert resp.status_code == 404
+
+    @pytest.mark.parametrize('platform,endpoint', [
+        ('tidal', '/api/tidal/discovery/status/test-tidal-1'),
+        ('youtube', '/api/youtube/discovery/status/test-yt-hash'),
+    ])
+    def test_discovery_http_works(self, flask_client, platform, endpoint):
+        """GET /api/<platform>/discovery/status/:id returns 200."""
+        resp = flask_client.get(endpoint)
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert 'phase' in data
+        assert 'status' in data
+        assert 'results' in data
+
+    def test_discovery_http_not_found(self, flask_client):
+        """GET /api/tidal/discovery/status/:id returns error for unknown ID."""
+        resp = flask_client.get('/api/tidal/discovery/status/nonexistent')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert 'error' in data
+
+    def test_watchlist_scan_http_works(self, flask_client):
+        """GET /api/watchlist/scan/status returns 200."""
+        resp = flask_client.get('/api/watchlist/scan/status')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert data['success'] is True
+        assert 'status' in data
+
+    def test_media_scan_http_works(self, flask_client):
+        """GET /api/scan/status returns 200."""
+        resp = flask_client.get('/api/scan/status')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert data['success'] is True
+        assert 'status' in data
+
+
+# =========================================================================
+# Group H — Backward Compatibility
+# =========================================================================
+
+class TestBackwardCompat:
+    """HTTP endpoints work when no WebSocket is connected."""
+
+    def test_all_http_endpoints_work_without_socket(self, flask_client):
+        """All Phase 5 HTTP endpoints work without any WebSocket connection."""
+        endpoints = [
+            '/api/sync/status/test-playlist-1',
+            '/api/tidal/discovery/status/test-tidal-1',
+            '/api/youtube/discovery/status/test-yt-hash',
+            '/api/watchlist/scan/status',
+            '/api/scan/status',
+        ]
+        for endpoint in endpoints:
+            resp = flask_client.get(endpoint)
+            assert resp.status_code == 200
+
+    def test_multiple_clients_get_scan_updates(self, test_app, shared_state):
+        """Multiple WebSocket clients each receive scan events."""
+        app, socketio = test_app
+        client1 = socketio.test_client(app)
+        client2 = socketio.test_client(app)
+        build = shared_state['build_watchlist_scan_status']
+
+        socketio.emit('scan:watchlist', build())
+
+        for client in [client1, client2]:
+            received = client.get_received()
+            events = [e for e in received if e['name'] == 'scan:watchlist']
+            assert len(events) >= 1
+
+        client1.disconnect()
+        client2.disconnect()
+
+    def test_sync_reflects_state_change(self, test_app, shared_state):
+        """When sync state changes, the next emit reflects it."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        ss = shared_state['sync_states']
+        build = shared_state['build_sync_status']
+
+        client.emit('sync:subscribe', {'playlist_ids': ['test-playlist-1']})
+
+        # Mutate state
+        ss['test-playlist-1']['status'] = 'completed'
+        ss['test-playlist-1']['progress']['progress'] = 100
+        ss['test-playlist-1']['progress']['matched_tracks'] = 11
+
+        payload = {'playlist_id': 'test-playlist-1', **build('test-playlist-1')}
+        socketio.emit('sync:progress', payload, room='sync:test-playlist-1')
+
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'sync:progress']
+        data = events[-1]['args'][0]
+        assert data['status'] == 'completed'
+        assert data['progress']['progress'] == 100
+        assert data['progress']['matched_tracks'] == 11
+
+        client.disconnect()
+
+    def test_discovery_reflects_state_change(self, test_app, shared_state):
+        """When discovery state changes, the next emit reflects it."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        ds = shared_state['discovery_states']
+        build = shared_state['build_discovery_status']
+
+        client.emit('discovery:subscribe', {'ids': ['test-tidal-1']})
+
+        # Mutate state
+        ds['tidal']['test-tidal-1']['phase'] = 'discovered'
+        ds['tidal']['test-tidal-1']['spotify_matches'] = 10
+
+        payload = build('tidal', 'test-tidal-1')
+        payload['platform'] = 'tidal'
+        payload['id'] = 'test-tidal-1'
+        socketio.emit('discovery:progress', payload, room='discovery:test-tidal-1')
+
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'discovery:progress']
+        data = events[-1]['args'][0]
+        assert data['phase'] == 'discovered'
+        assert data['complete'] is True
+        assert data['spotify_matches'] == 10
+
+        client.disconnect()
+
+    def test_scan_reflects_state_change(self, test_app, shared_state):
+        """When scan state changes, the next emit reflects it."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        wss = shared_state['watchlist_scan_state']
+        build = shared_state['build_watchlist_scan_status']
+
+        # Mutate state
+        wss['status'] = 'completed'
+        wss['current_artist_name'] = 'Led Zeppelin'
+
+        socketio.emit('scan:watchlist', build())
+
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'scan:watchlist']
+        data = events[-1]['args'][0]
+        assert data['status'] == 'completed'
+        assert data['current_artist_name'] == 'Led Zeppelin'
+
+        client.disconnect()
+
+
+# =========================================================================
+# Group I — Phase 6: Platform Sync via WebSocket Rooms
+# =========================================================================
+
+PLATFORM_SYNC_IDS = [
+    ('tidal', 'tidal_test-tidal-1'),
+    ('youtube', 'youtube_test-yt-hash'),
+    ('beatport', 'beatport_sync_test-bp-hash_1234'),
+    ('listenbrainz', 'listenbrainz_test-lb-mbid'),
+]
+
+
+class TestPlatformSyncEventDelivery:
+    """Platform sync pollers receive sync:progress via WS rooms."""
+
+    @pytest.mark.parametrize('platform,sync_id', PLATFORM_SYNC_IDS)
+    def test_platform_sync_event_received(self, test_app, shared_state, platform, sync_id):
+        """Client subscribes to platform sync_playlist_id and receives sync:progress."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_sync_status']
+
+        client.emit('sync:subscribe', {'playlist_ids': [sync_id]})
+
+        socketio.emit('sync:progress', {
+            'playlist_id': sync_id, **build(sync_id)
+        }, room=f'sync:{sync_id}')
+
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'sync:progress']
+        assert len(events) >= 1
+        data = events[0]['args'][0]
+        assert data['playlist_id'] == sync_id
+        assert data['status'] == 'syncing'
+
+        client.disconnect()
+
+    @pytest.mark.parametrize('platform,sync_id', PLATFORM_SYNC_IDS)
+    def test_platform_sync_not_received_when_unsubscribed(
+            self, test_app, shared_state, platform, sync_id):
+        """Unsubscribed client does NOT receive platform sync events."""
+        app, socketio = test_app
+        subscribed = socketio.test_client(app)
+        unsubscribed = socketio.test_client(app)
+        build = shared_state['build_sync_status']
+
+        subscribed.emit('sync:subscribe', {'playlist_ids': [sync_id]})
+
+        socketio.emit('sync:progress', {
+            'playlist_id': sync_id, **build(sync_id)
+        }, room=f'sync:{sync_id}')
+
+        sub_events = [e for e in subscribed.get_received()
+                      if e['name'] == 'sync:progress']
+        unsub_events = [e for e in unsubscribed.get_received()
+                        if e['name'] == 'sync:progress']
+
+        assert len(sub_events) >= 1
+        assert len(unsub_events) == 0
+
+        subscribed.disconnect()
+        unsubscribed.disconnect()
+
+    @pytest.mark.parametrize('platform,sync_id', PLATFORM_SYNC_IDS)
+    def test_platform_sync_progress_shape(self, test_app, shared_state, platform, sync_id):
+        """Platform sync progress data has expected keys."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_sync_status']
+
+        client.emit('sync:subscribe', {'playlist_ids': [sync_id]})
+
+        payload = {'playlist_id': sync_id, **build(sync_id)}
+        socketio.emit('sync:progress', payload, room=f'sync:{sync_id}')
+
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'sync:progress']
+        assert len(events) >= 1
+        data = events[0]['args'][0]
+
+        assert 'playlist_id' in data
+        assert 'status' in data
+        assert 'progress' in data
+        progress = data['progress']
+        assert 'total_tracks' in progress
+        assert 'matched_tracks' in progress
+        assert 'failed_tracks' in progress
+        assert 'progress' in progress
+
+        client.disconnect()
+
+    def test_platform_sync_completion_detection(self, test_app, shared_state):
+        """When status changes to 'finished', client detects completion."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        ss = shared_state['sync_states']
+
+        sync_id = 'tidal_test-tidal-1'
+        client.emit('sync:subscribe', {'playlist_ids': [sync_id]})
+
+        # Mutate to finished
+        ss[sync_id]['status'] = 'finished'
+        ss[sync_id]['progress']['progress'] = 100
+        ss[sync_id]['progress']['matched_tracks'] = 8
+
+        build = shared_state['build_sync_status']
+        socketio.emit('sync:progress', {
+            'playlist_id': sync_id, **build(sync_id)
+        }, room=f'sync:{sync_id}')
+
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'sync:progress']
+        data = events[-1]['args'][0]
+        assert data['status'] == 'finished'
+        assert data['progress']['progress'] == 100
+
+        client.disconnect()
+
+    def test_platform_sync_error_detection(self, test_app, shared_state):
+        """When status changes to 'error', client detects the error."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        ss = shared_state['sync_states']
+
+        sync_id = 'youtube_test-yt-hash'
+        client.emit('sync:subscribe', {'playlist_ids': [sync_id]})
+
+        ss[sync_id]['status'] = 'error'
+        ss[sync_id]['error'] = 'Connection lost'
+
+        build = shared_state['build_sync_status']
+        socketio.emit('sync:progress', {
+            'playlist_id': sync_id, **build(sync_id)
+        }, room=f'sync:{sync_id}')
+
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'sync:progress']
+        data = events[-1]['args'][0]
+        assert data['status'] == 'error'
+
+        client.disconnect()
+
+    @pytest.mark.parametrize('platform,sync_id', PLATFORM_SYNC_IDS)
+    def test_platform_sync_http_still_works(self, flask_client, platform, sync_id):
+        """GET /api/sync/status/<platform_sync_id> returns 200."""
+        resp = flask_client.get(f'/api/sync/status/{sync_id}')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert 'status' in data
+        assert 'progress' in data
+
+
+# =========================================================================
+# Group H — Wishlist Stats (broadcast)
+# =========================================================================
+
+class TestWishlistStatsEventDelivery:
+    """wishlist:stats broadcast events are received by the client."""
+
+    def test_wishlist_stats_event_received(self, test_app, shared_state):
+        """Client receives a wishlist:stats event."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_wishlist_stats']
+
+        socketio.emit('wishlist:stats', build())
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'wishlist:stats']
+        assert len(events) >= 1
+        client.disconnect()
+
+    def test_wishlist_stats_data_shape(self, test_app, shared_state):
+        """wishlist:stats has is_auto_processing and next_run_in_seconds."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        build = shared_state['build_wishlist_stats']
+
+        socketio.emit('wishlist:stats', build())
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'wishlist:stats']
+        assert len(events) >= 1
+        data = events[0]['args'][0]
+
+        assert 'is_auto_processing' in data
+        assert 'next_run_in_seconds' in data
+        assert isinstance(data['is_auto_processing'], bool)
+        assert isinstance(data['next_run_in_seconds'], (int, float))
+        client.disconnect()
+
+    def test_wishlist_stats_http_still_works(self, flask_client):
+        """GET /api/wishlist/stats returns 200 with expected keys."""
+        resp = flask_client.get('/api/wishlist/stats')
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert 'is_auto_processing' in data
+        assert 'next_run_in_seconds' in data
+
+    def test_wishlist_stats_auto_processing_detection(self, test_app, shared_state):
+        """When is_auto_processing is True, client detects it."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        ws = shared_state['wishlist_stats_state']
+
+        ws['is_auto_processing'] = True
+        ws['next_run_in_seconds'] = 0
+        build = shared_state['build_wishlist_stats']
+
+        socketio.emit('wishlist:stats', build())
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'wishlist:stats']
+        data = events[-1]['args'][0]
+        assert data['is_auto_processing'] is True
+        client.disconnect()
+
+    def test_wishlist_stats_countdown_value(self, test_app, shared_state):
+        """next_run_in_seconds reflects the mutable state."""
+        app, socketio = test_app
+        client = socketio.test_client(app)
+        ws = shared_state['wishlist_stats_state']
+
+        ws['next_run_in_seconds'] = 42
+        build = shared_state['build_wishlist_stats']
+
+        socketio.emit('wishlist:stats', build())
+        received = client.get_received()
+        events = [e for e in received if e['name'] == 'wishlist:stats']
+        data = events[-1]['args'][0]
+        assert data['next_run_in_seconds'] == 42
+        client.disconnect()
+
+    def test_wishlist_stats_multiple_clients(self, test_app, shared_state):
+        """Multiple clients each receive wishlist:stats broadcast."""
+        app, socketio = test_app
+        client1 = socketio.test_client(app)
+        client2 = socketio.test_client(app)
+        build = shared_state['build_wishlist_stats']
+
+        socketio.emit('wishlist:stats', build())
+
+        for client in [client1, client2]:
+            received = client.get_received()
+            events = [e for e in received if e['name'] == 'wishlist:stats']
+            assert len(events) >= 1
+
+        client1.disconnect()
+        client2.disconnect()
+
+    def test_wishlist_stats_matches_http(self, test_app, shared_state):
+        """Socket event data matches HTTP endpoint response."""
+        app, socketio = test_app
+        flask_client = app.test_client()
+        ws_client = socketio.test_client(app)
+        build = shared_state['build_wishlist_stats']
+
+        http_data = flask_client.get('/api/wishlist/stats').get_json()
+
+        socketio.emit('wishlist:stats', build())
+        received = ws_client.get_received()
+        events = [e for e in received if e['name'] == 'wishlist:stats']
+        ws_data = events[0]['args'][0]
+
+        assert ws_data['is_auto_processing'] == http_data['is_auto_processing']
+        assert ws_data['next_run_in_seconds'] == http_data['next_run_in_seconds']
+        ws_client.disconnect()

web_server.py

@@ -18,6 +18,7 @@ from urllib.parse import urljoin
 
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from flask import Flask, render_template, request, jsonify, redirect, send_file, Response
+from flask_socketio import SocketIO, emit, join_room, leave_room
 from utils.logging_config import get_logger
 from utils.async_helpers import run_async
 
@@ -119,6 +120,9 @@ app = Flask(
     static_folder=os.path.join(base_dir, 'webui', 'static')
 )
 
+# --- WebSocket (Socket.IO) Setup ---
+socketio = SocketIO(app, async_mode='threading', cors_allowed_origins='*')
+
 # --- Docker Helper Functions ---
 def docker_resolve_path(path_str):
     """
@@ -2366,85 +2370,88 @@ def save_playlist_m3u():
         logger.error(f"❌ Error saving M3U file: {e}")
         return jsonify({"status": "error", "message": str(e)}), 500
 
+def _build_system_stats():
+    """Build system statistics dict — shared by HTTP handler and WebSocket emitter."""
+    import psutil
+    import time
+    from datetime import timedelta
+
+    # Calculate uptime
+    start_time = getattr(app, 'start_time', time.time())
+    uptime_seconds = time.time() - start_time
+    uptime = str(timedelta(seconds=int(uptime_seconds)))
+
+    # Get memory usage
+    memory = psutil.virtual_memory()
+    memory_usage = f"{memory.percent}%"
+
+    # Count active downloads from download_batches (batches that are currently downloading)
+    active_downloads = len([batch_id for batch_id, batch_data in download_batches.items()
+                           if batch_data.get('phase') == 'downloading'])
+
+    # Count finished downloads (completed this session) - use session counter like dashboard.py
+    with session_stats_lock:
+        finished_downloads = session_completed_downloads
+
+    # Calculate total download speed from active soulseek transfers
+    total_download_speed = 0.0
+    try:
+        transfers_data = run_async(soulseek_client._make_request('GET', 'transfers/downloads'))
+        if transfers_data:
+            for user_data in transfers_data:
+                if 'directories' in user_data:
+                    for directory in user_data['directories']:
+                        if 'files' in directory:
+                            for file_info in directory['files']:
+                                state = file_info.get('state', '').lower()
+                                # Only count actively downloading files
+                                if 'inprogress' in state or 'downloading' in state or 'transferring' in state:
+                                    speed = file_info.get('averageSpeed', 0)
+                                    if isinstance(speed, (int, float)) and speed > 0:
+                                        total_download_speed += float(speed)
+    except Exception as e:
+        print(f"Warning: Could not fetch download speeds: {e}")
+
+    # Convert bytes/sec to KB/s and format
+    if total_download_speed > 0:
+        speed_kb_s = total_download_speed / 1024
+        if speed_kb_s >= 1024:
+            speed_mb_s = speed_kb_s / 1024
+            download_speed_str = f"{speed_mb_s:.1f} MB/s"
+        else:
+            download_speed_str = f"{speed_kb_s:.1f} KB/s"
+    else:
+        download_speed_str = "0 KB/s"
+
+    # Count active syncs (playlists currently syncing)
+    active_syncs = 0
+    # Count Spotify playlist syncs
+    for playlist_id, sync_state in sync_states.items():
+        if sync_state.get('status') == 'syncing':
+            active_syncs += 1
+    # Count YouTube playlist syncs
+    for url_hash, state in youtube_playlist_states.items():
+        if state.get('phase') == 'syncing':
+            active_syncs += 1
+    # Count Tidal playlist syncs
+    for playlist_id, state in tidal_discovery_states.items():
+        if state.get('phase') == 'syncing':
+            active_syncs += 1
+
+    return {
+        'active_downloads': active_downloads,
+        'finished_downloads': finished_downloads,
+        'download_speed': download_speed_str,
+        'active_syncs': active_syncs,
+        'uptime': uptime,
+        'memory_usage': memory_usage
+    }
+
 @app.route('/api/system/stats')
 def get_system_stats():
     """Get system statistics for dashboard"""
     try:
-        import psutil
-        import time
-        from datetime import timedelta
-        
-        # Calculate uptime
-        start_time = getattr(app, 'start_time', time.time())
-        uptime_seconds = time.time() - start_time
-        uptime = str(timedelta(seconds=int(uptime_seconds)))
-        
-        # Get memory usage
-        memory = psutil.virtual_memory()
-        memory_usage = f"{memory.percent}%"
-        
-        # Count active downloads from download_batches (batches that are currently downloading)
-        active_downloads = len([batch_id for batch_id, batch_data in download_batches.items() 
-                               if batch_data.get('phase') == 'downloading'])
-        
-        # Count finished downloads (completed this session) - use session counter like dashboard.py
-        with session_stats_lock:
-            finished_downloads = session_completed_downloads
-        
-        # Calculate total download speed from active soulseek transfers
-        total_download_speed = 0.0
-        try:
-            transfers_data = run_async(soulseek_client._make_request('GET', 'transfers/downloads'))
-            if transfers_data:
-                for user_data in transfers_data:
-                    if 'directories' in user_data:
-                        for directory in user_data['directories']:
-                            if 'files' in directory:
-                                for file_info in directory['files']:
-                                    state = file_info.get('state', '').lower()
-                                    # Only count actively downloading files
-                                    if 'inprogress' in state or 'downloading' in state or 'transferring' in state:
-                                        speed = file_info.get('averageSpeed', 0)
-                                        if isinstance(speed, (int, float)) and speed > 0:
-                                            total_download_speed += float(speed)
-        except Exception as e:
-            print(f"Warning: Could not fetch download speeds: {e}")
-        
-        # Convert bytes/sec to KB/s and format
-        if total_download_speed > 0:
-            speed_kb_s = total_download_speed / 1024
-            if speed_kb_s >= 1024:
-                speed_mb_s = speed_kb_s / 1024
-                download_speed_str = f"{speed_mb_s:.1f} MB/s"
-            else:
-                download_speed_str = f"{speed_kb_s:.1f} KB/s"
-        else:
-            download_speed_str = "0 KB/s"
-        
-        # Count active syncs (playlists currently syncing)
-        active_syncs = 0
-        # Count Spotify playlist syncs
-        for playlist_id, sync_state in sync_states.items():
-            if sync_state.get('status') == 'syncing':
-                active_syncs += 1
-        # Count YouTube playlist syncs
-        for url_hash, state in youtube_playlist_states.items():
-            if state.get('phase') == 'syncing':
-                active_syncs += 1
-        # Count Tidal playlist syncs
-        for playlist_id, state in tidal_discovery_states.items():
-            if state.get('phase') == 'syncing':
-                active_syncs += 1
-        
-        stats_data = {
-            'active_downloads': active_downloads,
-            'finished_downloads': finished_downloads,
-            'download_speed': download_speed_str,
-            'active_syncs': active_syncs,
-            'uptime': uptime,
-            'memory_usage': memory_usage
-        }
-        return jsonify(stats_data)
+        return jsonify(_build_system_stats())
     except Exception as e:
         return jsonify({'error': str(e)}), 500
 
@@ -2532,13 +2539,20 @@ def add_activity_item(icon: str, title: str, subtitle: str, time_ago: str = "Now
             'timestamp': time.time(),
             'show_toast': show_toast
         }
-        
+
         with activity_feed_lock:
             activity_feed.append(activity_item)
             # Keep only last 20 items to prevent memory growth
             if len(activity_feed) > 20:
                 activity_feed.pop(0)
-        
+
+        # Instant toast push via WebSocket (replaces 3-second polling)
+        if show_toast:
+            try:
+                socketio.emit('dashboard:toast', activity_item)
+            except Exception:
+                pass
+
         print(f"📝 Activity: {icon} {title} - {subtitle}")
     except Exception as e:
         print(f"Error adding activity item: {e}")
@@ -21595,6 +21609,11 @@ def add_to_watchlist():
                 # Don't fail the add operation if image fetch fails
                 print(f"⚠️ Could not fetch artist image for {artist_name}: {img_error}")
 
+            # Push updated count to all WebSocket clients immediately
+            try:
+                socketio.emit('watchlist:count', _build_watchlist_count_payload())
+            except Exception:
+                pass
             return jsonify({"success": True, "message": f"Added {artist_name} to watchlist"})
         else:
             return jsonify({"success": False, "error": "Failed to add artist to watchlist"}), 500
@@ -21609,14 +21628,19 @@ def remove_from_watchlist():
     try:
         data = request.get_json()
         artist_id = data.get('artist_id')
-        
+
         if not artist_id:
             return jsonify({"success": False, "error": "Missing artist_id"}), 400
-        
+
         database = get_database()
         success = database.remove_artist_from_watchlist(artist_id)
-        
+
         if success:
+            # Push updated count to all WebSocket clients immediately
+            try:
+                socketio.emit('watchlist:count', _build_watchlist_count_payload())
+            except Exception:
+                pass
             return jsonify({"success": True, "message": "Removed artist from watchlist"})
         else:
             return jsonify({"success": False, "error": "Failed to remove artist from watchlist"}), 500
@@ -29643,6 +29667,355 @@ def import_staging_suggestions():
 # ================================================================================================
 
 
+# ================================================================================================
+# WEBSOCKET (SOCKET.IO) EVENT HANDLERS AND BACKGROUND EMITTERS
+# ================================================================================================
+
+def _build_status_payload():
+    """Build the same status payload used by GET /status, reading from the cache."""
+    download_mode = config_manager.get('download_source.mode', 'soulseek')
+    soulseek_data = dict(_status_cache.get('soulseek', {}))
+    soulseek_data['source'] = download_mode
+    return {
+        'spotify': _status_cache.get('spotify', {}),
+        'media_server': _status_cache.get('media_server', {}),
+        'soulseek': soulseek_data,
+        'active_media_server': config_manager.get_active_media_server()
+    }
+
+def _build_watchlist_count_payload():
+    """Build the same payload used by GET /api/watchlist/count."""
+    try:
+        database = get_database()
+        count = database.get_watchlist_count()
+    except Exception:
+        count = 0
+    next_run_in_seconds = 0
+    with watchlist_timer_lock:
+        if watchlist_next_run_time > 0:
+            next_run_in_seconds = max(0, int(watchlist_next_run_time - time.time()))
+    return {
+        'success': True,
+        'count': count,
+        'next_run_in_seconds': next_run_in_seconds
+    }
+
+def _emit_service_status_loop():
+    """Background thread that pushes service status every 10 seconds."""
+    while True:
+        socketio.sleep(10)
+        try:
+            socketio.emit('status:update', _build_status_payload())
+        except Exception as e:
+            logger.debug(f"Error emitting service status: {e}")
+
+def _emit_watchlist_count_loop():
+    """Background thread that pushes watchlist count every 30 seconds."""
+    while True:
+        socketio.sleep(30)
+        try:
+            socketio.emit('watchlist:count', _build_watchlist_count_payload())
+        except Exception as e:
+            logger.debug(f"Error emitting watchlist count: {e}")
+
+def _emit_download_status_loop():
+    """Background thread that pushes download batch status every 2 seconds to subscribed rooms."""
+    while True:
+        socketio.sleep(2)
+        try:
+            live_transfers_lookup = get_cached_transfer_data()
+            with tasks_lock:
+                for batch_id, batch in download_batches.items():
+                    try:
+                        status_data = _build_batch_status_data(
+                            batch_id, batch, live_transfers_lookup
+                        )
+                        socketio.emit('downloads:batch_update', {
+                            'batch_id': batch_id,
+                            'data': status_data
+                        }, room=f'batch:{batch_id}')
+                    except Exception as e:
+                        logger.debug(f"Error building batch status for {batch_id}: {e}")
+        except Exception as e:
+            logger.debug(f"Error in download status emit loop: {e}")
+
+# --- Socket.IO event handlers ---
+
+@socketio.on('connect')
+def handle_connect():
+    logger.info("WebSocket client connected")
+
+@socketio.on('disconnect')
+def handle_disconnect():
+    logger.info("WebSocket client disconnected")
+
+@socketio.on('downloads:subscribe')
+def handle_download_subscribe(data):
+    """Client subscribes to download batch updates by joining rooms."""
+    batch_ids = data.get('batch_ids', [])
+    for bid in batch_ids:
+        join_room(f'batch:{bid}')
+    logger.debug(f"Client subscribed to batches: {batch_ids}")
+
+@socketio.on('downloads:unsubscribe')
+def handle_download_unsubscribe(data):
+    """Client unsubscribes from download batch updates by leaving rooms."""
+    batch_ids = data.get('batch_ids', [])
+    for bid in batch_ids:
+        leave_room(f'batch:{bid}')
+    logger.debug(f"Client unsubscribed from batches: {batch_ids}")
+
+# --- Phase 2: Dashboard emitters ---
+
+def _emit_system_stats_loop():
+    """Background thread that pushes system stats every 10 seconds."""
+    while True:
+        socketio.sleep(10)
+        try:
+            socketio.emit('dashboard:stats', _build_system_stats())
+        except Exception as e:
+            logger.debug(f"Error emitting system stats: {e}")
+
+def _emit_activity_feed_loop():
+    """Background thread that pushes activity feed every 5 seconds."""
+    while True:
+        socketio.sleep(5)
+        try:
+            with activity_feed_lock:
+                activities = activity_feed[-10:][::-1]
+            socketio.emit('dashboard:activity', {'activities': activities})
+        except Exception as e:
+            logger.debug(f"Error emitting activity feed: {e}")
+
+def _emit_db_stats_loop():
+    """Background thread that pushes database stats every 30 seconds."""
+    while True:
+        socketio.sleep(30)
+        try:
+            db = get_database()
+            stats = db.get_database_info_for_server()
+            socketio.emit('dashboard:db_stats', stats)
+        except Exception as e:
+            logger.debug(f"Error emitting db stats: {e}")
+
+def _emit_wishlist_count_loop():
+    """Background thread that pushes wishlist count every 30 seconds."""
+    while True:
+        socketio.sleep(30)
+        try:
+            from core.wishlist_service import get_wishlist_service
+            count = get_wishlist_service().get_wishlist_count()
+            socketio.emit('dashboard:wishlist_count', {'count': count})
+        except Exception as e:
+            logger.debug(f"Error emitting wishlist count: {e}")
+
+# Note: Toasts are NOT on a timer — they emit instantly from add_activity_item()
+
+# --- Phase 3: Enrichment sidebar worker emitters ---
+
+def _emit_enrichment_status_loop():
+    """Background thread that pushes all enrichment worker statuses every 2 seconds."""
+    workers = {
+        'musicbrainz': lambda: mb_worker,
+        'audiodb': lambda: audiodb_worker,
+        'deezer': lambda: deezer_worker,
+        'spotify-enrichment': lambda: spotify_enrichment_worker,
+        'itunes-enrichment': lambda: itunes_enrichment_worker,
+        'hydrabase': lambda: hydrabase_worker,
+        'repair': lambda: repair_worker,
+    }
+    while True:
+        socketio.sleep(2)
+        for name, get_worker in workers.items():
+            try:
+                worker = get_worker()
+                if worker is None:
+                    continue
+                status = worker.get_stats()
+                socketio.emit(f'enrichment:{name}', status)
+            except Exception as e:
+                logger.debug(f"Error emitting {name} status: {e}")
+
+def _emit_tool_progress_loop():
+    """Background thread that pushes all tool progress statuses every 1 second."""
+    while True:
+        socketio.sleep(1)
+        # Stream status
+        try:
+            with stream_lock:
+                socketio.emit('tool:stream', {
+                    "status": stream_state["status"],
+                    "progress": stream_state["progress"],
+                    "track_info": stream_state["track_info"],
+                    "error_message": stream_state["error_message"]
+                })
+        except Exception as e:
+            logger.debug(f"Error emitting stream status: {e}")
+        # Quality Scanner
+        try:
+            with quality_scanner_lock:
+                socketio.emit('tool:quality-scanner', dict(quality_scanner_state))
+        except Exception as e:
+            logger.debug(f"Error emitting quality scanner status: {e}")
+        # Duplicate Cleaner (add computed space_freed_mb)
+        try:
+            with duplicate_cleaner_lock:
+                state_copy = duplicate_cleaner_state.copy()
+                state_copy["space_freed_mb"] = duplicate_cleaner_state["space_freed"] / (1024 * 1024)
+                socketio.emit('tool:duplicate-cleaner', state_copy)
+        except Exception as e:
+            logger.debug(f"Error emitting duplicate cleaner status: {e}")
+        # Retag
+        try:
+            with retag_lock:
+                socketio.emit('tool:retag', dict(retag_state))
+        except Exception as e:
+            logger.debug(f"Error emitting retag status: {e}")
+        # DB Update
+        try:
+            with db_update_lock:
+                socketio.emit('tool:db-update', dict(db_update_state))
+        except Exception as e:
+            logger.debug(f"Error emitting db update status: {e}")
+        # Metadata Update (match HTTP wrapper: {success, status})
+        try:
+            state_copy = metadata_update_state.copy()
+            if state_copy.get('started_at'):
+                state_copy['started_at'] = state_copy['started_at'].isoformat()
+            if state_copy.get('completed_at'):
+                state_copy['completed_at'] = state_copy['completed_at'].isoformat()
+            socketio.emit('tool:metadata', {"success": True, "status": state_copy})
+        except Exception as e:
+            logger.debug(f"Error emitting metadata status: {e}")
+        # Logs (format activity_feed same as HTTP endpoint)
+        try:
+            with activity_feed_lock:
+                recent = activity_feed[-50:][::-1]
+                formatted = []
+                for a in recent:
+                    ts = a.get('time', 'Unknown')
+                    icon = a.get('icon', '•')
+                    title = a.get('title', 'Activity')
+                    sub = a.get('subtitle', '')
+                    formatted.append(f"[{ts}] {icon} {title} - {sub}" if sub else f"[{ts}] {icon} {title}")
+                if not formatted:
+                    formatted = ["No recent activity.", "Sync and download operations..."]
+                socketio.emit('tool:logs', {'logs': formatted})
+        except Exception as e:
+            logger.debug(f"Error emitting logs: {e}")
+
+@socketio.on('sync:subscribe')
+def handle_sync_subscribe(data):
+    for pid in data.get('playlist_ids', []):
+        join_room(f'sync:{pid}')
+
+@socketio.on('sync:unsubscribe')
+def handle_sync_unsubscribe(data):
+    for pid in data.get('playlist_ids', []):
+        leave_room(f'sync:{pid}')
+
+@socketio.on('discovery:subscribe')
+def handle_discovery_subscribe(data):
+    for pid in data.get('ids', []):
+        join_room(f'discovery:{pid}')
+
+@socketio.on('discovery:unsubscribe')
+def handle_discovery_unsubscribe(data):
+    for pid in data.get('ids', []):
+        leave_room(f'discovery:{pid}')
+
+def _emit_sync_progress_loop():
+    """Push sync progress to subscribed rooms every 1 second."""
+    while True:
+        socketio.sleep(1)
+        try:
+            with sync_lock:
+                for pid, state in list(sync_states.items()):
+                    try:
+                        socketio.emit('sync:progress', {
+                            'playlist_id': pid, **state
+                        }, room=f'sync:{pid}')
+                    except Exception:
+                        pass
+        except Exception as e:
+            logger.debug(f"Error in sync progress loop: {e}")
+
+def _emit_discovery_progress_loop():
+    """Push discovery progress to subscribed rooms every 1 second."""
+    platform_states = {
+        'tidal': lambda: tidal_discovery_states,
+        'youtube': lambda: youtube_playlist_states,
+        'beatport': lambda: beatport_chart_states,
+        'listenbrainz': lambda: listenbrainz_playlist_states,
+    }
+    while True:
+        socketio.sleep(1)
+        for platform, get_states in platform_states.items():
+            try:
+                states_dict = get_states()
+                for pid, state in list(states_dict.items()):
+                    try:
+                        phase = state.get('phase', '')
+                        if phase in ('', 'idle'):
+                            continue
+                        payload = {
+                            'platform': platform,
+                            'id': pid,
+                            'phase': state.get('phase'),
+                            'status': state.get('status', 'unknown'),
+                            'progress': state.get('discovery_progress', 0),
+                            'discovery_progress': state.get('discovery_progress', {}),
+                            'spotify_matches': state.get('spotify_matches', 0),
+                            'spotify_total': state.get('spotify_total', 0),
+                            'results': state.get('discovery_results', state.get('results', [])),
+                            'complete': state.get('phase') == 'discovered',
+                        }
+                        socketio.emit('discovery:progress', payload, room=f'discovery:{pid}')
+                    except Exception:
+                        pass
+            except Exception as e:
+                logger.debug(f"Error in {platform} discovery loop: {e}")
+
+def _emit_scan_status_loop():
+    """Push watchlist and media scan status every 2 seconds."""
+    while True:
+        socketio.sleep(2)
+        # Watchlist scan
+        try:
+            state = watchlist_scan_state.copy()
+            if state.get('started_at'):
+                state['started_at'] = state['started_at'].isoformat()
+            if state.get('completed_at'):
+                state['completed_at'] = state['completed_at'].isoformat()
+            state.pop('results', None)
+            socketio.emit('scan:watchlist', {"success": True, **state})
+        except Exception as e:
+            logger.debug(f"Error emitting watchlist scan: {e}")
+        # Media scan
+        try:
+            if web_scan_manager:
+                scan_status = web_scan_manager.get_scan_status()
+                socketio.emit('scan:media', {"success": True, "status": scan_status})
+        except Exception as e:
+            logger.debug(f"Error emitting media scan: {e}")
+        # Wishlist stats (auto-processing detection + countdown refresh)
+        try:
+            next_run = 0
+            with wishlist_timer_lock:
+                if wishlist_next_run_time > 0:
+                    next_run = max(0, int(wishlist_next_run_time - time.time()))
+            socketio.emit('wishlist:stats', {
+                "is_auto_processing": is_wishlist_actually_processing(),
+                "next_run_in_seconds": next_run,
+            })
+        except Exception as e:
+            logger.debug(f"Error emitting wishlist stats: {e}")
+
+# ================================================================================================
+# END WEBSOCKET HANDLERS
+# ================================================================================================
+
+
 if __name__ == '__main__':
     # Initialize logging for web server
     from utils.logging_config import setup_logging
@@ -29694,4 +30067,25 @@ if __name__ == '__main__':
     # Add a test activity to verify the system is working
     add_activity_item("🔧", "Debug Test", "Activity feed system test", "Now")
 
-    app.run(host='0.0.0.0', port=8008, debug=False)
+    # Start WebSocket background emitters
+    print("🔧 Starting WebSocket background emitters...")
+    # Phase 1: Global pollers
+    socketio.start_background_task(_emit_service_status_loop)
+    socketio.start_background_task(_emit_watchlist_count_loop)
+    socketio.start_background_task(_emit_download_status_loop)
+    # Phase 2: Dashboard pollers
+    socketio.start_background_task(_emit_system_stats_loop)
+    socketio.start_background_task(_emit_activity_feed_loop)
+    socketio.start_background_task(_emit_db_stats_loop)
+    socketio.start_background_task(_emit_wishlist_count_loop)
+    # Phase 3: Enrichment sidebar workers
+    socketio.start_background_task(_emit_enrichment_status_loop)
+    # Phase 4: Tool progress pollers
+    socketio.start_background_task(_emit_tool_progress_loop)
+    # Phase 5: Sync/discovery progress + scans
+    socketio.start_background_task(_emit_sync_progress_loop)
+    socketio.start_background_task(_emit_discovery_progress_loop)
+    socketio.start_background_task(_emit_scan_status_loop)
+    print("✅ WebSocket emitters started (Phase 1-5: global/dashboard/enrichment/tools/sync)")
+
+    socketio.run(app, host='0.0.0.0', port=8008, debug=False, allow_unsafe_werkzeug=True)

webui/index.html

@@ -4358,6 +4358,7 @@
         </div>
     </div>
 
+    <script src="{{ url_for('static', filename='vendor/socket.io.min.js') }}"></script>
     <script src="{{ url_for('static', filename='script.js') }}"></script>
 </body>
 

webui/static/style.css

@@ -5460,7 +5460,7 @@ body {
 
     .header-button {
         flex: 1;
-        min-width: 120px;
+        min-width: 100%;
     }
 
     .service-status-grid,