diff --git a/core/stats/__init__.py b/core/stats/__init__.py
new file mode 100644
index 00000000..c531efc7
--- /dev/null
+++ b/core/stats/__init__.py
@@ -0,0 +1 @@
+"""Stats API helpers package."""
diff --git a/core/stats/queries.py b/core/stats/queries.py
new file mode 100644
index 00000000..df7cfa68
--- /dev/null
+++ b/core/stats/queries.py
@@ -0,0 +1,276 @@
+"""Stats API query helpers.
+
+Lifted from web_server.py /api/stats/* and /api/listening-stats/* routes.
+Pure-ish functions: take dependencies as args, return data dicts/lists. Route
+handlers stay in web_server.py and are responsible for request parsing,
+jsonify, and error responses.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import threading
+import time
+import traceback
+from typing import Any, Callable, Optional
+
+logger = logging.getLogger(__name__)
+
+ImageUrlFixer = Callable[[Optional[str]], Optional[str]]
+
+
+def get_cached_stats(database, image_url_fixer: ImageUrlFixer, time_range: str) -> dict:
+    """Read pre-computed stats cache for a time range. Instant response."""
+    conn = database._get_connection()
+    try:
+        cursor = conn.cursor()
+
+        cursor.execute("SELECT value FROM metadata WHERE key = ?", (f'stats_cache_{time_range}',))
+        row = cursor.fetchone()
+        data = json.loads(row[0]) if row and row[0] else {}
+
+        cursor.execute("SELECT value FROM metadata WHERE key = 'stats_cache_recent'")
+        row = cursor.fetchone()
+        recent = json.loads(row[0]) if row and row[0] else []
+
+        cursor.execute("SELECT value FROM metadata WHERE key = 'stats_cache_health'")
+        row = cursor.fetchone()
+        health = json.loads(row[0]) if row and row[0] else {}
+    finally:
+        conn.close()
+
+    for item in (data.get('top_artists') or []) + (data.get('top_albums') or []) + (data.get('top_tracks') or []):
+        if item.get('image_url'):
+            item['image_url'] = image_url_fixer(item['image_url'])
+
+    return {
+        'cached': True,
+        **data,
+        'recent': recent,
+        'health': health,
+    }
+
+
+def get_overview(database, time_range: str) -> dict:
+    """Aggregate listening stats for a time range."""
+    return database.get_listening_stats(time_range)
+
+
+def get_top_artists(database, image_url_fixer: ImageUrlFixer, time_range: str, limit: int) -> list[dict]:
+    """Top artists by play count, enriched with image / Last.fm stats / soul_id."""
+    artists = database.get_top_artists(time_range, limit)
+
+    for artist in artists:
+        try:
+            conn = database._get_connection()
+            try:
+                cursor = conn.cursor()
+                cursor.execute(
+                    """
+                    SELECT thumb_url, id, lastfm_listeners, lastfm_playcount, soul_id
+                    FROM artists
+                    WHERE LOWER(name) = LOWER(?)
+                    LIMIT 1
+                    """,
+                    (artist['name'],),
+                )
+                row = cursor.fetchone()
+                if row:
+                    artist['image_url'] = image_url_fixer(row[0]) if row[0] else None
+                    artist['id'] = row[1]
+                    artist['global_listeners'] = row[2]
+                    artist['global_playcount'] = row[3]
+                    artist['soul_id'] = row[4]
+            finally:
+                conn.close()
+        except Exception:
+            pass
+
+    return artists
+
+
+def get_top_albums(database, image_url_fixer: ImageUrlFixer, time_range: str, limit: int) -> list[dict]:
+    """Top albums by play count, enriched with album thumb."""
+    albums = database.get_top_albums(time_range, limit)
+
+    for album in albums:
+        try:
+            conn = database._get_connection()
+            try:
+                cursor = conn.cursor()
+                cursor.execute(
+                    """
+                    SELECT al.thumb_url, al.id, al.artist_id FROM albums al
+                    WHERE LOWER(al.title) = LOWER(?) AND al.thumb_url IS NOT NULL AND al.thumb_url != ''
+                    LIMIT 1
+                    """,
+                    (album['name'],),
+                )
+                row = cursor.fetchone()
+                if row:
+                    album['image_url'] = image_url_fixer(row[0]) if row[0] else None
+                    album['id'] = row[1]
+                    album['artist_id'] = row[2]
+            finally:
+                conn.close()
+        except Exception:
+            pass
+
+    return albums
+
+
+def get_top_tracks(database, image_url_fixer: ImageUrlFixer, time_range: str, limit: int) -> list[dict]:
+    """Top tracks by play count, enriched with album thumb."""
+    tracks = database.get_top_tracks(time_range, limit)
+
+    for track in tracks:
+        try:
+            conn = database._get_connection()
+            try:
+                cursor = conn.cursor()
+                cursor.execute(
+                    """
+                    SELECT al.thumb_url, t.id, t.artist_id FROM tracks t
+                    JOIN albums al ON al.id = t.album_id
+                    JOIN artists ar ON ar.id = t.artist_id
+                    WHERE LOWER(t.title) = LOWER(?) AND LOWER(ar.name) = LOWER(?)
+                    LIMIT 1
+                    """,
+                    (track['name'], track['artist']),
+                )
+                row = cursor.fetchone()
+                if row:
+                    track['image_url'] = image_url_fixer(row[0]) if row[0] else None
+                    track['id'] = row[1]
+                    track['artist_id'] = row[2]
+            finally:
+                conn.close()
+        except Exception:
+            pass
+
+    return tracks
+
+
+def get_timeline(database, time_range: str, granularity: str) -> Any:
+    """Play count per time period for chart rendering."""
+    return database.get_listening_timeline(time_range, granularity)
+
+
+def get_genres(database, time_range: str) -> Any:
+    """Genre distribution by play count."""
+    return database.get_genre_breakdown(time_range)
+
+
+def get_library_health(database) -> dict:
+    """Library health metrics."""
+    return database.get_library_health()
+
+
+def get_db_storage(database) -> dict:
+    """Database storage breakdown by table."""
+    return database.get_db_storage_stats()
+
+
+def get_recent_tracks(database, limit: int) -> list[dict]:
+    """Recently played tracks from listening_history."""
+    conn = database._get_connection()
+    try:
+        cursor = conn.cursor()
+        cursor.execute(
+            """
+            SELECT title, artist, album, played_at, duration_ms
+            FROM listening_history
+            ORDER BY played_at DESC
+            LIMIT ?
+            """,
+            (limit,),
+        )
+        rows = cursor.fetchall()
+    finally:
+        conn.close()
+
+    return [
+        {
+            'title': row[0],
+            'artist': row[1],
+            'album': row[2],
+            'played_at': row[3],
+            'duration_ms': row[4],
+        }
+        for row in rows
+    ]
+
+
+def resolve_track(database, image_url_fixer: ImageUrlFixer, title: str, artist: str) -> Optional[dict]:
+    """Resolve a track by title+artist to its file_path / metadata. Returns None if not found."""
+    conn = database._get_connection()
+    try:
+        cursor = conn.cursor()
+        cursor.execute(
+            """
+            SELECT t.id, t.title, t.file_path, t.bitrate, t.duration,
+                   ar.name as artist_name, al.title as album_title,
+                   al.thumb_url, t.artist_id, t.album_id
+            FROM tracks t
+            JOIN artists ar ON ar.id = t.artist_id
+            LEFT JOIN albums al ON al.id = t.album_id
+            WHERE LOWER(t.title) = LOWER(?) AND LOWER(ar.name) = LOWER(?)
+              AND t.file_path IS NOT NULL AND t.file_path != ''
+            LIMIT 1
+            """,
+            (title.strip(), artist.strip()),
+        )
+        row = cursor.fetchone()
+    finally:
+        conn.close()
+
+    if not row:
+        return None
+
+    return {
+        'id': row[0],
+        'title': row[1],
+        'file_path': row[2],
+        'bitrate': row[3],
+        'duration': row[4],
+        'artist_name': row[5],
+        'album_title': row[6],
+        'image_url': image_url_fixer(row[7]) if row[7] else None,
+        'artist_id': row[8],
+        'album_id': row[9],
+    }
+
+
+def trigger_listening_sync(worker) -> None:
+    """Spawn a daemon thread that runs the worker's poll loop once.
+
+    Caller is responsible for verifying worker is not None before calling.
+    """
+    def _do_sync():
+        try:
+            logger.info("[Stats Sync] Starting manual poll...")
+            worker._poll()
+            worker.stats['polls_completed'] += 1
+            worker.stats['last_poll'] = time.strftime('%Y-%m-%d %H:%M:%S')
+            logger.info("[Stats Sync] Manual poll completed")
+        except Exception as e:
+            logger.error(f"[Stats Sync] Manual poll failed: {e}")
+            traceback.print_exc()
+            logger.error(f"Manual stats sync failed: {e}")
+
+    threading.Thread(target=_do_sync, daemon=True).start()
+
+
+def get_listening_status(worker) -> dict:
+    """Worker status dict. Returns disabled-state shape if worker is None."""
+    if worker is None:
+        return {
+            'enabled': False,
+            'running': False,
+            'paused': False,
+            'idle': False,
+            'current_item': None,
+            'stats': {},
+        }
+    return worker.get_stats()
diff --git a/tests/stats/__init__.py b/tests/stats/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/stats/test_stats_queries.py b/tests/stats/test_stats_queries.py
new file mode 100644
index 00000000..288c2672
--- /dev/null
+++ b/tests/stats/test_stats_queries.py
@@ -0,0 +1,433 @@
+"""Tests for core/stats/queries.py — lifted from web_server.py /api/stats/* routes."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from core.stats import queries
+from database.music_database import MusicDatabase
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def db(tmp_path):
+    return MusicDatabase(str(tmp_path / "music.db"))
+
+
+@pytest.fixture
+def fix_url():
+    """Image-url fixer stub: prefixes inputs to make calls observable."""
+    return lambda u: f"FIXED::{u}" if u else None
+
+
+_id_counter = {'n': 0}
+
+
+def _next_id(prefix):
+    _id_counter['n'] += 1
+    return f"{prefix}-{_id_counter['n']}"
+
+
+def _seed_artist(db, name, thumb=None, lastfm_listeners=None, lastfm_playcount=None, soul_id=None):
+    aid = _next_id('art')
+    conn = db._get_connection()
+    try:
+        cursor = conn.cursor()
+        cursor.execute(
+            "INSERT INTO artists (id, name, thumb_url, lastfm_listeners, lastfm_playcount, soul_id) "
+            "VALUES (?, ?, ?, ?, ?, ?)",
+            (aid, name, thumb, lastfm_listeners, lastfm_playcount, soul_id),
+        )
+        conn.commit()
+        return aid
+    finally:
+        conn.close()
+
+
+def _seed_album(db, artist_id, title, thumb=None):
+    alb = _next_id('alb')
+    conn = db._get_connection()
+    try:
+        cursor = conn.cursor()
+        cursor.execute(
+            "INSERT INTO albums (id, artist_id, title, thumb_url) VALUES (?, ?, ?, ?)",
+            (alb, artist_id, title, thumb),
+        )
+        conn.commit()
+        return alb
+    finally:
+        conn.close()
+
+
+def _seed_track(db, album_id, artist_id, title, file_path=None, bitrate=None, duration=None):
+    tid = _next_id('trk')
+    conn = db._get_connection()
+    try:
+        cursor = conn.cursor()
+        cursor.execute(
+            "INSERT INTO tracks (id, album_id, artist_id, title, file_path, bitrate, duration) "
+            "VALUES (?, ?, ?, ?, ?, ?, ?)",
+            (tid, album_id, artist_id, title, file_path, bitrate, duration),
+        )
+        conn.commit()
+        return tid
+    finally:
+        conn.close()
+
+
+def _seed_history(db, title, artist, album, played_at, duration_ms=180000):
+    conn = db._get_connection()
+    try:
+        cursor = conn.cursor()
+        cursor.execute(
+            "INSERT INTO listening_history (title, artist, album, played_at, duration_ms) "
+            "VALUES (?, ?, ?, ?, ?)",
+            (title, artist, album, played_at, duration_ms),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _seed_metadata(db, key, value):
+    conn = db._get_connection()
+    try:
+        cursor = conn.cursor()
+        cursor.execute(
+            "INSERT OR REPLACE INTO metadata (key, value) VALUES (?, ?)",
+            (key, json.dumps(value)),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+# ---------------------------------------------------------------------------
+# get_recent_tracks
+# ---------------------------------------------------------------------------
+
+def test_get_recent_tracks_orders_by_played_at_desc(db):
+    _seed_history(db, "Old", "A", "Album", "2026-01-01 00:00:00")
+    _seed_history(db, "Newest", "A", "Album", "2026-04-01 00:00:00")
+    _seed_history(db, "Mid", "A", "Album", "2026-02-15 00:00:00")
+
+    rows = queries.get_recent_tracks(db, limit=10)
+    titles = [r['title'] for r in rows]
+
+    assert titles == ["Newest", "Mid", "Old"]
+
+
+def test_get_recent_tracks_respects_limit(db):
+    for i in range(5):
+        _seed_history(db, f"T{i}", "A", "Album", f"2026-04-0{i + 1} 00:00:00")
+    rows = queries.get_recent_tracks(db, limit=2)
+    assert len(rows) == 2
+
+
+def test_get_recent_tracks_empty_returns_empty(db):
+    rows = queries.get_recent_tracks(db, limit=10)
+    assert rows == []
+
+
+def test_get_recent_tracks_returns_full_shape(db):
+    _seed_history(db, "Money", "Pink Floyd", "DSOTM", "2026-04-01 00:00:00", duration_ms=383000)
+    rows = queries.get_recent_tracks(db, limit=1)
+    assert rows == [{
+        'title': "Money",
+        'artist': "Pink Floyd",
+        'album': "DSOTM",
+        'played_at': "2026-04-01 00:00:00",
+        'duration_ms': 383000,
+    }]
+
+
+# ---------------------------------------------------------------------------
+# resolve_track
+# ---------------------------------------------------------------------------
+
+def test_resolve_track_returns_full_metadata(db, fix_url):
+    aid = _seed_artist(db, "Pink Floyd")
+    alb = _seed_album(db, aid, "DSOTM", thumb="local://thumb.jpg")
+    _seed_track(db, alb, aid, "Money", file_path="/music/money.flac", bitrate=1411, duration=383000)
+
+    result = queries.resolve_track(db, fix_url, "Money", "Pink Floyd")
+    assert result['title'] == "Money"
+    assert result['file_path'] == "/music/money.flac"
+    assert result['bitrate'] == 1411
+    assert result['duration'] == 383000
+    assert result['artist_name'] == "Pink Floyd"
+    assert result['album_title'] == "DSOTM"
+    assert result['image_url'] == "FIXED::local://thumb.jpg"
+    assert result['album_id'] == alb
+    assert result['artist_id'] == aid
+
+
+def test_resolve_track_case_insensitive_match(db, fix_url):
+    aid = _seed_artist(db, "Pink Floyd")
+    alb = _seed_album(db, aid, "DSOTM")
+    _seed_track(db, alb, aid, "Money", file_path="/music/x.flac")
+
+    result = queries.resolve_track(db, fix_url, "money", "pink floyd")
+    assert result is not None
+    assert result['title'] == "Money"
+
+
+def test_resolve_track_returns_none_when_no_file_path(db, fix_url):
+    aid = _seed_artist(db, "Pink Floyd")
+    alb = _seed_album(db, aid, "DSOTM")
+    _seed_track(db, alb, aid, "Money", file_path=None)
+
+    result = queries.resolve_track(db, fix_url, "Money", "Pink Floyd")
+    assert result is None
+
+
+def test_resolve_track_returns_none_when_file_path_empty(db, fix_url):
+    aid = _seed_artist(db, "Pink Floyd")
+    alb = _seed_album(db, aid, "DSOTM")
+    _seed_track(db, alb, aid, "Money", file_path="")
+
+    result = queries.resolve_track(db, fix_url, "Money", "Pink Floyd")
+    assert result is None
+
+
+def test_resolve_track_strips_whitespace(db, fix_url):
+    aid = _seed_artist(db, "Pink Floyd")
+    alb = _seed_album(db, aid, "DSOTM")
+    _seed_track(db, alb, aid, "Money", file_path="/x.flac")
+
+    result = queries.resolve_track(db, fix_url, "  Money  ", "  Pink Floyd  ")
+    assert result is not None
+
+
+# ---------------------------------------------------------------------------
+# get_top_artists / get_top_albums / get_top_tracks — enrichment
+# ---------------------------------------------------------------------------
+
+def test_get_top_artists_enriches_with_artist_table_columns(db, fix_url, monkeypatch):
+    aid = _seed_artist(
+        db, "Pink Floyd", thumb="local://pf.jpg",
+        lastfm_listeners=5000000, lastfm_playcount=100000000, soul_id="soul-pf",
+    )
+
+    monkeypatch.setattr(db, "get_top_artists", lambda tr, lim: [{'name': 'Pink Floyd', 'play_count': 42}])
+
+    result = queries.get_top_artists(db, fix_url, time_range='all', limit=10)
+    assert result[0]['name'] == 'Pink Floyd'
+    assert result[0]['image_url'] == 'FIXED::local://pf.jpg'
+    assert result[0]['id'] == aid
+    assert result[0]['global_listeners'] == 5000000
+    assert result[0]['global_playcount'] == 100000000
+    assert result[0]['soul_id'] == 'soul-pf'
+
+
+def test_get_top_artists_no_match_leaves_record_unenriched(db, fix_url, monkeypatch):
+    monkeypatch.setattr(db, "get_top_artists", lambda tr, lim: [{'name': 'Unknown', 'play_count': 1}])
+    result = queries.get_top_artists(db, fix_url, time_range='all', limit=10)
+    assert result == [{'name': 'Unknown', 'play_count': 1}]
+
+
+def test_get_top_albums_enriches_with_album_thumb(db, fix_url, monkeypatch):
+    aid = _seed_artist(db, "Pink Floyd")
+    alb = _seed_album(db, aid, "DSOTM", thumb="local://album.jpg")
+
+    monkeypatch.setattr(db, "get_top_albums", lambda tr, lim: [{'name': 'DSOTM', 'play_count': 5}])
+
+    result = queries.get_top_albums(db, fix_url, time_range='all', limit=10)
+    assert result[0]['image_url'] == 'FIXED::local://album.jpg'
+    assert result[0]['id'] == alb
+    assert result[0]['artist_id'] == aid
+
+
+def test_get_top_albums_skips_empty_thumb(db, fix_url, monkeypatch):
+    aid = _seed_artist(db, "X")
+    _seed_album(db, aid, "Empty", thumb="")
+    monkeypatch.setattr(db, "get_top_albums", lambda tr, lim: [{'name': 'Empty', 'play_count': 1}])
+
+    result = queries.get_top_albums(db, fix_url, time_range='all', limit=10)
+    assert 'image_url' not in result[0]
+
+
+def test_get_top_tracks_enriches_with_album_thumb(db, fix_url, monkeypatch):
+    aid = _seed_artist(db, "Pink Floyd")
+    alb = _seed_album(db, aid, "DSOTM", thumb="local://thumb.jpg")
+    tid = _seed_track(db, alb, aid, "Money")
+
+    monkeypatch.setattr(db, "get_top_tracks", lambda tr, lim: [{'name': 'Money', 'artist': 'Pink Floyd'}])
+
+    result = queries.get_top_tracks(db, fix_url, time_range='all', limit=10)
+    assert result[0]['image_url'] == 'FIXED::local://thumb.jpg'
+    assert result[0]['id'] == tid
+    assert result[0]['artist_id'] == aid
+
+
+def test_get_top_tracks_unmatched_record_passed_through(db, fix_url, monkeypatch):
+    monkeypatch.setattr(db, "get_top_tracks", lambda tr, lim: [{'name': 'Phantom', 'artist': 'Nobody'}])
+    result = queries.get_top_tracks(db, fix_url, time_range='all', limit=10)
+    assert result == [{'name': 'Phantom', 'artist': 'Nobody'}]
+
+
+# ---------------------------------------------------------------------------
+# get_cached_stats
+# ---------------------------------------------------------------------------
+
+def test_get_cached_stats_reads_three_metadata_keys(db, fix_url):
+    _seed_metadata(db, 'stats_cache_7d', {
+        'top_artists': [{'name': 'PF', 'image_url': 'local://a.jpg'}],
+        'top_albums': [{'name': 'DSOTM'}],
+        'top_tracks': [{'name': 'Money', 'image_url': 'local://t.jpg'}],
+        'overview': {'plays': 100},
+    })
+    _seed_metadata(db, 'stats_cache_recent', [{'title': 'Money'}])
+    _seed_metadata(db, 'stats_cache_health', {'orphan_tracks': 0})
+
+    result = queries.get_cached_stats(db, fix_url, '7d')
+
+    assert result['cached'] is True
+    assert result['top_artists'][0]['image_url'] == 'FIXED::local://a.jpg'
+    assert result['top_tracks'][0]['image_url'] == 'FIXED::local://t.jpg'
+    assert result['overview'] == {'plays': 100}
+    assert result['recent'] == [{'title': 'Money'}]
+    assert result['health'] == {'orphan_tracks': 0}
+
+
+def test_get_cached_stats_missing_keys_return_empty_defaults(db, fix_url):
+    result = queries.get_cached_stats(db, fix_url, '30d')
+    assert result['cached'] is True
+    assert result['recent'] == []
+    assert result['health'] == {}
+
+
+def test_get_cached_stats_skips_image_fix_when_no_url(db, fix_url):
+    _seed_metadata(db, 'stats_cache_7d', {
+        'top_artists': [{'name': 'PF'}],
+    })
+    result = queries.get_cached_stats(db, fix_url, '7d')
+    assert 'image_url' not in result['top_artists'][0]
+
+
+# ---------------------------------------------------------------------------
+# Pass-through helpers — verify they delegate to the right DB method
+# ---------------------------------------------------------------------------
+
+def test_get_overview_delegates_to_db(monkeypatch):
+    sentinel = object()
+    called = {}
+
+    class _DB:
+        def get_listening_stats(self, time_range):
+            called['arg'] = time_range
+            return sentinel
+
+    assert queries.get_overview(_DB(), '7d') is sentinel
+    assert called['arg'] == '7d'
+
+
+def test_get_timeline_delegates_to_db():
+    called = {}
+
+    class _DB:
+        def get_listening_timeline(self, time_range, granularity):
+            called['args'] = (time_range, granularity)
+            return ['data']
+
+    assert queries.get_timeline(_DB(), '30d', 'week') == ['data']
+    assert called['args'] == ('30d', 'week')
+
+
+def test_get_genres_delegates_to_db():
+    called = {}
+
+    class _DB:
+        def get_genre_breakdown(self, time_range):
+            called['arg'] = time_range
+            return [{'genre': 'rock'}]
+
+    assert queries.get_genres(_DB(), 'all') == [{'genre': 'rock'}]
+    assert called['arg'] == 'all'
+
+
+def test_get_library_health_delegates_to_db():
+    class _DB:
+        def get_library_health(self):
+            return {'orphan_tracks': 5}
+
+    assert queries.get_library_health(_DB()) == {'orphan_tracks': 5}
+
+
+def test_get_db_storage_delegates_to_db():
+    class _DB:
+        def get_db_storage_stats(self):
+            return {'total_mb': 42}
+
+    assert queries.get_db_storage(_DB()) == {'total_mb': 42}
+
+
+# ---------------------------------------------------------------------------
+# Listening worker glue
+# ---------------------------------------------------------------------------
+
+def test_get_listening_status_handles_none_worker():
+    result = queries.get_listening_status(None)
+    assert result == {
+        'enabled': False,
+        'running': False,
+        'paused': False,
+        'idle': False,
+        'current_item': None,
+        'stats': {},
+    }
+
+
+def test_get_listening_status_delegates_to_worker():
+    class _Worker:
+        def get_stats(self):
+            return {'enabled': True, 'running': True, 'stats': {'polls_completed': 42}}
+
+    result = queries.get_listening_status(_Worker())
+    assert result['enabled'] is True
+    assert result['stats']['polls_completed'] == 42
+
+
+def test_trigger_listening_sync_runs_worker_poll_in_thread():
+    poll_called = []
+    stats_dict = {'polls_completed': 0, 'last_poll': None}
+
+    class _Worker:
+        stats = stats_dict
+
+        def _poll(self):
+            poll_called.append(True)
+
+    queries.trigger_listening_sync(_Worker())
+
+    # Wait briefly for thread to run
+    import time as _time
+    for _ in range(50):
+        if poll_called:
+            break
+        _time.sleep(0.01)
+
+    assert poll_called == [True]
+    assert stats_dict['polls_completed'] == 1
+    assert stats_dict['last_poll'] is not None
+
+
+def test_trigger_listening_sync_swallows_worker_errors():
+    class _BrokenWorker:
+        stats = {'polls_completed': 0, 'last_poll': None}
+
+        def _poll(self):
+            raise RuntimeError("boom")
+
+    # Should NOT raise — error is caught + logged inside the thread
+    queries.trigger_listening_sync(_BrokenWorker())
+
+    import time as _time
+    _time.sleep(0.1)  # give thread time to crash
+    # Counter not incremented because exception was raised before increment
+    assert _BrokenWorker.stats['polls_completed'] == 0
diff --git a/web_server.py b/web_server.py
index c8e1bbfe..d119a05c 100644
--- a/web_server.py
+++ b/web_server.py
@@ -46801,46 +46801,18 @@ except Exception as e:
     listening_stats_worker = None
 
 # --- Stats API Endpoints ---
+# Logic lives in core/stats/queries.py — these routes are thin handlers.
+
+from core.stats import queries as _stats_queries
+
 
 @app.route('/api/stats/cached', methods=['GET'])
 def stats_cached():
     """Get all pre-computed stats for a time range from cache. Instant response."""
     try:
-        import json as _json
         time_range = request.args.get('range', '7d')
-        database = get_database()
-        conn = database._get_connection()
-        cursor = conn.cursor()
-
-        # Get time-range-specific cache
-        cursor.execute("SELECT value FROM metadata WHERE key = ?", (f'stats_cache_{time_range}',))
-        row = cursor.fetchone()
-        data = _json.loads(row[0]) if row and row[0] else {}
-
-        # Get recent plays cache
-        cursor.execute("SELECT value FROM metadata WHERE key = 'stats_cache_recent'")
-        row = cursor.fetchone()
-        recent = _json.loads(row[0]) if row and row[0] else []
-
-        # Get library health cache
-        cursor.execute("SELECT value FROM metadata WHERE key = 'stats_cache_health'")
-        row = cursor.fetchone()
-        health = _json.loads(row[0]) if row and row[0] else {}
-
-        conn.close()
-
-        # Fix image URLs (stored as relative paths, need server prefix)
-        for item in (data.get('top_artists') or []) + (data.get('top_albums') or []) + (data.get('top_tracks') or []):
-            if item.get('image_url'):
-                item['image_url'] = fix_artist_image_url(item['image_url'])
-
-        return jsonify({
-            'success': True,
-            'cached': True,
-            **data,
-            'recent': recent,
-            'health': health,
-        })
+        data = _stats_queries.get_cached_stats(get_database(), fix_artist_image_url, time_range)
+        return jsonify({'success': True, **data})
     except Exception as e:
         return jsonify({'success': False, 'error': str(e)}), 500
 
@@ -46849,8 +46821,7 @@ def stats_overview():
     """Get aggregate listening stats for a time range."""
     try:
         time_range = request.args.get('range', 'all')
-        database = get_database()
-        data = database.get_listening_stats(time_range)
+        data = _stats_queries.get_overview(get_database(), time_range)
         return jsonify({'success': True, **data})
     except Exception as e:
         return jsonify({'success': False, 'error': str(e)}), 500
@@ -46861,31 +46832,7 @@ def stats_top_artists():
     try:
         time_range = request.args.get('range', 'all')
         limit = int(request.args.get('limit', 10))
-        database = get_database()
-        artists = database.get_top_artists(time_range, limit)
-
-        # Enrich with images, Last.fm stats, and soul_id from the library
-        for artist in artists:
-            try:
-                conn = database._get_connection()
-                cursor = conn.cursor()
-                cursor.execute("""
-                    SELECT thumb_url, id, lastfm_listeners, lastfm_playcount, soul_id
-                    FROM artists
-                    WHERE LOWER(name) = LOWER(?)
-                    LIMIT 1
-                """, (artist['name'],))
-                row = cursor.fetchone()
-                if row:
-                    artist['image_url'] = fix_artist_image_url(row[0]) if row[0] else None
-                    artist['id'] = row[1]
-                    artist['global_listeners'] = row[2]
-                    artist['global_playcount'] = row[3]
-                    artist['soul_id'] = row[4]
-                conn.close()
-            except Exception:
-                pass
-
+        artists = _stats_queries.get_top_artists(get_database(), fix_artist_image_url, time_range, limit)
         return jsonify({'success': True, 'artists': artists})
     except Exception as e:
         return jsonify({'success': False, 'error': str(e)}), 500
@@ -46896,28 +46843,7 @@ def stats_top_albums():
     try:
         time_range = request.args.get('range', 'all')
         limit = int(request.args.get('limit', 10))
-        database = get_database()
-        albums = database.get_top_albums(time_range, limit)
-
-        # Enrich with images
-        for album in albums:
-            try:
-                conn = database._get_connection()
-                cursor = conn.cursor()
-                cursor.execute("""
-                    SELECT al.thumb_url, al.id, al.artist_id FROM albums al
-                    WHERE LOWER(al.title) = LOWER(?) AND al.thumb_url IS NOT NULL AND al.thumb_url != ''
-                    LIMIT 1
-                """, (album['name'],))
-                row = cursor.fetchone()
-                if row:
-                    album['image_url'] = fix_artist_image_url(row[0]) if row[0] else None
-                    album['id'] = row[1]
-                    album['artist_id'] = row[2]
-                conn.close()
-            except Exception:
-                pass
-
+        albums = _stats_queries.get_top_albums(get_database(), fix_artist_image_url, time_range, limit)
         return jsonify({'success': True, 'albums': albums})
     except Exception as e:
         return jsonify({'success': False, 'error': str(e)}), 500
@@ -46928,30 +46854,7 @@ def stats_top_tracks():
     try:
         time_range = request.args.get('range', 'all')
         limit = int(request.args.get('limit', 10))
-        database = get_database()
-        tracks = database.get_top_tracks(time_range, limit)
-
-        # Enrich with album images
-        for track in tracks:
-            try:
-                conn = database._get_connection()
-                cursor = conn.cursor()
-                cursor.execute("""
-                    SELECT al.thumb_url, t.id, t.artist_id FROM tracks t
-                    JOIN albums al ON al.id = t.album_id
-                    JOIN artists ar ON ar.id = t.artist_id
-                    WHERE LOWER(t.title) = LOWER(?) AND LOWER(ar.name) = LOWER(?)
-                    LIMIT 1
-                """, (track['name'], track['artist']))
-                row = cursor.fetchone()
-                if row:
-                    track['image_url'] = fix_artist_image_url(row[0]) if row[0] else None
-                    track['id'] = row[1]
-                    track['artist_id'] = row[2]
-                conn.close()
-            except Exception:
-                pass
-
+        tracks = _stats_queries.get_top_tracks(get_database(), fix_artist_image_url, time_range, limit)
         return jsonify({'success': True, 'tracks': tracks})
     except Exception as e:
         return jsonify({'success': False, 'error': str(e)}), 500
@@ -46962,8 +46865,7 @@ def stats_timeline():
     try:
         time_range = request.args.get('range', '30d')
         granularity = request.args.get('granularity', 'day')
-        database = get_database()
-        data = database.get_listening_timeline(time_range, granularity)
+        data = _stats_queries.get_timeline(get_database(), time_range, granularity)
         return jsonify({'success': True, 'timeline': data})
     except Exception as e:
         return jsonify({'success': False, 'error': str(e)}), 500
@@ -46973,8 +46875,7 @@ def stats_genres():
     """Get genre distribution by play count."""
     try:
         time_range = request.args.get('range', 'all')
-        database = get_database()
-        data = database.get_genre_breakdown(time_range)
+        data = _stats_queries.get_genres(get_database(), time_range)
         return jsonify({'success': True, 'genres': data})
     except Exception as e:
         return jsonify({'success': False, 'error': str(e)}), 500
@@ -46983,8 +46884,7 @@ def stats_genres():
 def stats_library_health():
     """Get library health metrics."""
     try:
-        database = get_database()
-        data = database.get_library_health()
+        data = _stats_queries.get_library_health(get_database())
         return jsonify({'success': True, **data})
     except Exception as e:
         return jsonify({'success': False, 'error': str(e)}), 500
@@ -46993,8 +46893,7 @@ def stats_library_health():
 def stats_db_storage():
     """Get database storage breakdown by table."""
     try:
-        database = get_database()
-        data = database.get_db_storage_stats()
+        data = _stats_queries.get_db_storage(get_database())
         return jsonify({'success': True, **data})
     except Exception as e:
         return jsonify({'success': False, 'error': str(e)}), 500
@@ -47004,28 +46903,7 @@ def stats_recent():
     """Get recently played tracks."""
     try:
         limit = int(request.args.get('limit', 20))
-        database = get_database()
-        conn = database._get_connection()
-        cursor = conn.cursor()
-        cursor.execute("""
-            SELECT title, artist, album, played_at, duration_ms
-            FROM listening_history
-            ORDER BY played_at DESC
-            LIMIT ?
-        """, (limit,))
-        rows = cursor.fetchall()
-        conn.close()
-
-        tracks = []
-        for row in rows:
-            tracks.append({
-                'title': row[0],
-                'artist': row[1],
-                'album': row[2],
-                'played_at': row[3],
-                'duration_ms': row[4],
-            })
-
+        tracks = _stats_queries.get_recent_tracks(get_database(), limit)
         return jsonify({'success': True, 'tracks': tracks})
     except Exception as e:
         return jsonify({'success': False, 'error': str(e)}), 500
@@ -47040,41 +46918,10 @@ def stats_resolve_track():
         if not title:
             return jsonify({'success': False, 'error': 'Title required'}), 400
 
-        database = get_database()
-        conn = database._get_connection()
-        cursor = conn.cursor()
-        cursor.execute("""
-            SELECT t.id, t.title, t.file_path, t.bitrate, t.duration,
-                   ar.name as artist_name, al.title as album_title,
-                   al.thumb_url, t.artist_id, t.album_id
-            FROM tracks t
-            JOIN artists ar ON ar.id = t.artist_id
-            LEFT JOIN albums al ON al.id = t.album_id
-            WHERE LOWER(t.title) = LOWER(?) AND LOWER(ar.name) = LOWER(?)
-              AND t.file_path IS NOT NULL AND t.file_path != ''
-            LIMIT 1
-        """, (title.strip(), artist.strip()))
-        row = cursor.fetchone()
-        conn.close()
-
-        if not row:
+        track = _stats_queries.resolve_track(get_database(), fix_artist_image_url, title, artist)
+        if track is None:
             return jsonify({'success': False, 'error': 'Track not found in library'})
-
-        return jsonify({
-            'success': True,
-            'track': {
-                'id': row[0],
-                'title': row[1],
-                'file_path': row[2],
-                'bitrate': row[3],
-                'duration': row[4],
-                'artist_name': row[5],
-                'album_title': row[6],
-                'image_url': fix_artist_image_url(row[7]) if row[7] else None,
-                'artist_id': row[8],
-                'album_id': row[9],
-            }
-        })
+        return jsonify({'success': True, 'track': track})
     except Exception as e:
         return jsonify({'success': False, 'error': str(e)}), 500
 
@@ -47084,22 +46931,7 @@ def listening_stats_sync():
     try:
         if not listening_stats_worker:
             return jsonify({'success': False, 'error': 'Listening stats worker not initialized'}), 400
-
-        import threading
-        def _do_sync():
-            try:
-                logger.info("[Stats Sync] Starting manual poll...")
-                listening_stats_worker._poll()
-                listening_stats_worker.stats['polls_completed'] += 1
-                listening_stats_worker.stats['last_poll'] = time.strftime('%Y-%m-%d %H:%M:%S')
-                logger.info("[Stats Sync] Manual poll completed")
-            except Exception as e:
-                logger.error(f"[Stats Sync] Manual poll failed: {e}")
-                import traceback
-                traceback.print_exc()
-                logger.error(f"Manual stats sync failed: {e}")
-
-        threading.Thread(target=_do_sync, daemon=True).start()
+        _stats_queries.trigger_listening_sync(listening_stats_worker)
         return jsonify({'success': True, 'message': 'Sync started'})
     except Exception as e:
         return jsonify({'success': False, 'error': str(e)}), 500
@@ -47108,10 +46940,7 @@ def listening_stats_sync():
 def listening_stats_status():
     """Get listening stats worker status."""
     try:
-        if listening_stats_worker is None:
-            return jsonify({'enabled': False, 'running': False, 'paused': False,
-                            'idle': False, 'current_item': None, 'stats': {}})
-        return jsonify(listening_stats_worker.get_stats())
+        return jsonify(_stats_queries.get_listening_status(listening_stats_worker))
     except Exception as e:
         return jsonify({'error': str(e)}), 500
 
diff --git a/webui/static/helper.js b/webui/static/helper.js
index 4b8a260a..d69563b8 100644
--- a/webui/static/helper.js
+++ b/webui/static/helper.js
@@ -3449,6 +3449,7 @@ const WHATS_NEW = {
         { title: 'Settings Endpoints: Admin-Only', desc: 'the /api/settings endpoints (read, write, log-level, config-status, verify) had no auth gate — any logged-in profile could read or change service tokens, oauth secrets, api keys. now admin-only. single-admin setups (no multi-profile config) work transparently as before.', page: 'settings' },
         { title: 'Browser Caching for Static Assets + Discover Pages', desc: 'static assets (js/css/icons) now get a 1-year browser cache instead of revalidating on every page load. safe because the existing ?v=static_v cache-bust query changes every server restart, so deploys still ship live. discover pages (hero, similar artists, recent releases, deep cuts, etc.) now cache 5 minutes browser-side so toggling between sections doesn\'t re-fetch everything. faster repeat loads, fewer round-trips.', page: 'discover' },
         { title: 'Service Worker for Cover Art + Installable PWA', desc: 'cover art used to re-fetch from the cdn on every library / discover page visit. now a service worker caches images locally — second visit serves art instantly from disk, no network hit. also added a pwa manifest so soulsync can be installed to home screen / desktop as a standalone app (chrome / edge / safari → install soulsync). cache versioned so future strategy changes invalidate cleanly.' },
+        { title: 'Stats Endpoints Lifted to core/stats', desc: 'internal — moved /api/stats/* and /api/listening-stats/* logic out of web_server.py into core/stats/queries.py with full test coverage. no behavior change. step toward breaking up the web_server.py monolith.' },
     ],
     '2.4.0': [
         // --- April 26, 2026 — Search & Artists unification + reorganize queue ---