diff --git a/core/personalized/api.py b/core/personalized/api.py
new file mode 100644
index 00000000..1fc6fef8
--- /dev/null
+++ b/core/personalized/api.py
@@ -0,0 +1,151 @@
+"""HTTP endpoint handlers for the personalized-playlists subsystem.
+
+Wired into the Flask app from web_server.py. Each handler is a thin
+wrapper that:
+1. Pulls profile id + manager from request context.
+2. Calls one PersonalizedPlaylistManager method.
+3. Returns a JSON-serializable shape.
+
+Live routes (registered against the main Flask app):
+- GET    /api/personalized/playlists                       — list
+- GET    /api/personalized/kinds                           — registry
+- GET    /api/personalized/playlist/<kind>                  — singleton
+- GET    /api/personalized/playlist/<kind>/<variant>        — variant
+- POST   /api/personalized/playlist/<kind>/refresh          — singleton
+- POST   /api/personalized/playlist/<kind>/<variant>/refresh — variant
+- PUT    /api/personalized/playlist/<kind>/config           — singleton
+- PUT    /api/personalized/playlist/<kind>/<variant>/config  — variant
+
+The handlers themselves are pure functions returning Python dicts so
+they're testable without spinning up Flask. The wiring step in
+web_server.py wraps them in `jsonify` + URL routing.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional
+
+from core.personalized.manager import PersonalizedPlaylistManager
+from core.personalized.specs import PlaylistKindRegistry, get_registry
+from core.personalized.types import PlaylistRecord, Track
+
+
+def _record_to_dict(record: PlaylistRecord) -> Dict[str, Any]:
+    return {
+        'id': record.id,
+        'profile_id': record.profile_id,
+        'kind': record.kind,
+        'variant': record.variant,
+        'name': record.name,
+        'config': record.config.to_json_dict(),
+        'track_count': record.track_count,
+        'last_generated_at': record.last_generated_at,
+        'last_synced_at': record.last_synced_at,
+        'last_generation_source': record.last_generation_source,
+        'last_generation_error': record.last_generation_error,
+    }
+
+
+def _track_to_dict(track: Track) -> Dict[str, Any]:
+    return {
+        'spotify_track_id': track.spotify_track_id,
+        'itunes_track_id': track.itunes_track_id,
+        'deezer_track_id': track.deezer_track_id,
+        'track_name': track.track_name,
+        'artist_name': track.artist_name,
+        'album_name': track.album_name,
+        'album_cover_url': track.album_cover_url,
+        'duration_ms': track.duration_ms,
+        'popularity': track.popularity,
+        'track_data_json': track.track_data_json,
+        'source': track.source,
+    }
+
+
+def list_kinds(registry: Optional[PlaylistKindRegistry] = None) -> Dict[str, Any]:
+    """Return every registered playlist kind with metadata.
+
+    UI uses this to render the "available playlists" picker. Each
+    kind reports whether it requires a variant and the resolved
+    variant set so the UI can render variant choices when relevant."""
+    reg = registry or get_registry()
+    out = []
+    for spec in reg.all():
+        out.append({
+            'kind': spec.kind,
+            'name_template': spec.name_template,
+            'description': spec.description,
+            'requires_variant': spec.requires_variant,
+            'tags': list(spec.tags),
+            'default_config': spec.default_config.to_json_dict(),
+        })
+    return {'success': True, 'kinds': out}
+
+
+def list_playlists(manager: PersonalizedPlaylistManager, profile_id: int) -> Dict[str, Any]:
+    """List every persisted playlist for a profile."""
+    records = manager.list_playlists(profile_id)
+    return {
+        'success': True,
+        'playlists': [_record_to_dict(r) for r in records],
+    }
+
+
+def get_playlist_with_tracks(
+    manager: PersonalizedPlaylistManager,
+    kind: str,
+    variant: str,
+    profile_id: int,
+) -> Dict[str, Any]:
+    """Get the playlist row + its current track snapshot. Auto-creates
+    the row from default config if it doesn't exist (so the UI's first-
+    paint of an unseen kind works without a separate ensure call)."""
+    record = manager.ensure_playlist(kind, variant, profile_id)
+    tracks = manager.get_playlist_tracks(record.id)
+    return {
+        'success': True,
+        'playlist': _record_to_dict(record),
+        'tracks': [_track_to_dict(t) for t in tracks],
+    }
+
+
+def refresh_playlist(
+    manager: PersonalizedPlaylistManager,
+    kind: str,
+    variant: str,
+    profile_id: int,
+    config_overrides: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    """Run the kind's generator and persist the snapshot. Returns the
+    fresh row + tracks."""
+    record = manager.refresh_playlist(kind, variant, profile_id, config_overrides=config_overrides)
+    tracks = manager.get_playlist_tracks(record.id)
+    return {
+        'success': True,
+        'playlist': _record_to_dict(record),
+        'tracks': [_track_to_dict(t) for t in tracks],
+    }
+
+
+def update_config(
+    manager: PersonalizedPlaylistManager,
+    kind: str,
+    variant: str,
+    profile_id: int,
+    overrides: Dict[str, Any],
+) -> Dict[str, Any]:
+    """Patch the playlist's config with the provided fields."""
+    record = manager.update_config(kind, variant, profile_id, overrides)
+    return {
+        'success': True,
+        'playlist': _record_to_dict(record),
+    }
+
+
+__all__ = [
+    'list_kinds',
+    'list_playlists',
+    'get_playlist_with_tracks',
+    'refresh_playlist',
+    'update_config',
+]
diff --git a/tests/test_personalized_api.py b/tests/test_personalized_api.py
new file mode 100644
index 00000000..7b7ef613
--- /dev/null
+++ b/tests/test_personalized_api.py
@@ -0,0 +1,181 @@
+"""Boundary tests for `core.personalized.api` handler functions.
+
+These are pure-function dispatchers — they take a manager + ids,
+return a JSON-serializable dict. No Flask required, no real DB.
+The Flask wiring in `web_server.py` adds `jsonify` + URL routing
+on top.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+from types import SimpleNamespace
+from typing import Any, List
+
+import pytest
+
+from core.personalized import api as _api
+from core.personalized.manager import PersonalizedPlaylistManager
+from core.personalized.specs import PlaylistKindRegistry, PlaylistKindSpec
+from core.personalized.types import PlaylistConfig, Track
+from database.personalized_schema import ensure_personalized_schema
+
+
+class _FakeDB:
+    def __init__(self, path):
+        self.path = path
+
+    def _get_connection(self):
+        c = sqlite3.connect(self.path)
+        c.row_factory = sqlite3.Row
+        return c
+
+
+@pytest.fixture
+def db(tmp_path):
+    p = str(tmp_path / 't.db')
+    conn = sqlite3.connect(p)
+    ensure_personalized_schema(conn)
+    conn.commit()
+    conn.close()
+    return _FakeDB(p)
+
+
+@pytest.fixture
+def registry():
+    return PlaylistKindRegistry()
+
+
+@pytest.fixture
+def manager(db, registry):
+    return PersonalizedPlaylistManager(db, deps=None, registry=registry)
+
+
+def _register(reg, kind='hidden_gems', requires_variant=False, generator=None):
+    spec = PlaylistKindSpec(
+        kind=kind, name_template=kind.replace('_', ' ').title(),
+        description=f'Description for {kind}',
+        default_config=PlaylistConfig(limit=20),
+        generator=generator or (lambda *a, **k: []),
+        requires_variant=requires_variant,
+        tags=['test'],
+    )
+    reg.register(spec)
+    return spec
+
+
+# ─── list_kinds ──────────────────────────────────────────────────────
+
+
+class TestListKinds:
+    def test_lists_every_registered_kind(self, registry):
+        _register(registry, kind='hidden_gems')
+        _register(registry, kind='time_machine', requires_variant=True)
+        out = _api.list_kinds(registry)
+        assert out['success'] is True
+        kinds = {k['kind'] for k in out['kinds']}
+        assert kinds == {'hidden_gems', 'time_machine'}
+
+    def test_kind_metadata_shape(self, registry):
+        _register(registry, kind='hidden_gems')
+        out = _api.list_kinds(registry)
+        kind = out['kinds'][0]
+        assert kind['kind'] == 'hidden_gems'
+        assert kind['requires_variant'] is False
+        assert kind['tags'] == ['test']
+        assert kind['default_config']['limit'] == 20
+
+    def test_empty_registry(self):
+        out = _api.list_kinds(PlaylistKindRegistry())
+        assert out == {'success': True, 'kinds': []}
+
+
+# ─── list_playlists ─────────────────────────────────────────────────
+
+
+class TestListPlaylists:
+    def test_returns_empty_when_no_playlists(self, manager, registry):
+        _register(registry)
+        out = _api.list_playlists(manager, profile_id=1)
+        assert out == {'success': True, 'playlists': []}
+
+    def test_serializes_playlist_record(self, manager, registry):
+        _register(registry)
+        manager.ensure_playlist('hidden_gems', '', 1)
+        out = _api.list_playlists(manager, profile_id=1)
+        assert out['success'] is True
+        assert len(out['playlists']) == 1
+        pl = out['playlists'][0]
+        assert pl['kind'] == 'hidden_gems'
+        assert pl['variant'] == ''
+        assert pl['name'] == 'Hidden Gems'
+        assert pl['track_count'] == 0
+        assert pl['config']['limit'] == 20
+
+
+# ─── get_playlist_with_tracks ───────────────────────────────────────
+
+
+class TestGetPlaylistWithTracks:
+    def test_auto_creates_on_first_get(self, manager, registry):
+        _register(registry)
+        out = _api.get_playlist_with_tracks(manager, 'hidden_gems', '', 1)
+        assert out['success'] is True
+        assert out['playlist']['kind'] == 'hidden_gems'
+        assert out['tracks'] == []
+
+    def test_returns_persisted_tracks(self, manager, registry):
+        gen_calls = []
+
+        def gen(deps, variant, config):
+            gen_calls.append(1)
+            return [Track(track_name='X', artist_name='Y', spotify_track_id='sp-1')]
+
+        _register(registry, generator=gen)
+        manager.refresh_playlist('hidden_gems', '', 1)
+        out = _api.get_playlist_with_tracks(manager, 'hidden_gems', '', 1)
+        assert len(out['tracks']) == 1
+        assert out['tracks'][0]['track_name'] == 'X'
+        assert out['tracks'][0]['spotify_track_id'] == 'sp-1'
+
+    def test_unknown_kind_raises_value_error(self, manager):
+        with pytest.raises(ValueError):
+            _api.get_playlist_with_tracks(manager, 'nope', '', 1)
+
+
+# ─── refresh_playlist ───────────────────────────────────────────────
+
+
+class TestRefreshPlaylist:
+    def test_refresh_runs_generator_and_returns_tracks(self, manager, registry):
+        _register(registry, generator=lambda *a, **k: [
+            Track(track_name='T1', artist_name='A', spotify_track_id='1'),
+            Track(track_name='T2', artist_name='B', spotify_track_id='2'),
+        ])
+        out = _api.refresh_playlist(manager, 'hidden_gems', '', 1)
+        assert out['success'] is True
+        assert len(out['tracks']) == 2
+        assert out['playlist']['track_count'] == 2
+        assert out['playlist']['last_generated_at'] is not None
+
+    def test_config_overrides_passed_through(self, manager, registry):
+        captured = {}
+
+        def gen(deps, variant, config):
+            captured['limit'] = config.limit
+            return []
+
+        _register(registry, generator=gen)
+        _api.refresh_playlist(manager, 'hidden_gems', '', 1, config_overrides={'limit': 99})
+        assert captured['limit'] == 99
+
+
+# ─── update_config ──────────────────────────────────────────────────
+
+
+class TestUpdateConfig:
+    def test_patches_config(self, manager, registry):
+        _register(registry)
+        out = _api.update_config(manager, 'hidden_gems', '', 1, {'limit': 75})
+        assert out['success'] is True
+        assert out['playlist']['config']['limit'] == 75
diff --git a/web_server.py b/web_server.py
index 208bf49a..7bc3ec5c 100644
--- a/web_server.py
+++ b/web_server.py
@@ -26818,6 +26818,115 @@ def get_discovery_shuffle():
         logger.error(f"Error getting discovery shuffle playlist: {e}")
         return jsonify({"success": False, "error": str(e)}), 500
 
+
+# ========================================================================
+# Personalized Playlists v2 — unified storage + manager-backed routes.
+# Wraps every personalized playlist (Group A + Group B) behind one API
+# surface. Generators in `core/personalized/generators/` register at
+# import time; this set of routes exposes the manager for the UI.
+# Legacy `/api/discover/personalized/...` endpoints stay alive for
+# backward compat during the UI migration window.
+# ========================================================================
+
+# Trigger registration of every generator (side-effect import).
+from core.personalized import generators as _personalized_generators  # noqa: F401
+from core.personalized import api as _personalized_api
+from core.personalized.manager import PersonalizedPlaylistManager as _PersonalizedManager
+
+
+def _build_personalized_manager():
+    """Construct a manager wired with whatever each generator needs.
+
+    Per-request construction: the underlying services are cheap
+    accessors, so we don't bother caching. If profiling shows
+    overhead, this becomes a module-level lazy singleton."""
+    from core.personalized_playlists import get_personalized_playlists_service
+    from core.seasonal_discovery import get_seasonal_discovery_service
+    database = get_database()
+    deps = types.SimpleNamespace(
+        database=database,
+        service=get_personalized_playlists_service(database, spotify_client),
+        seasonal_service=get_seasonal_discovery_service(spotify_client, database),
+        get_current_profile_id=get_current_profile_id,
+        get_active_discovery_source=_get_active_discovery_source,
+    )
+    return _PersonalizedManager(database=database, deps=deps)
+
+
+@app.route('/api/personalized/kinds', methods=['GET'])
+def personalized_list_kinds():
+    """List every registered personalized-playlist kind."""
+    try:
+        return jsonify(_personalized_api.list_kinds())
+    except Exception as e:
+        logger.error(f"Personalized kinds list error: {e}")
+        return jsonify({"success": False, "error": str(e)}), 500
+
+
+@app.route('/api/personalized/playlists', methods=['GET'])
+def personalized_list_playlists():
+    """List every persisted personalized playlist for the active profile."""
+    try:
+        manager = _build_personalized_manager()
+        return jsonify(_personalized_api.list_playlists(manager, get_current_profile_id()))
+    except Exception as e:
+        logger.error(f"Personalized playlists list error: {e}")
+        return jsonify({"success": False, "error": str(e)}), 500
+
+
+@app.route('/api/personalized/playlist/<kind>', methods=['GET'])
+@app.route('/api/personalized/playlist/<kind>/<variant>', methods=['GET'])
+def personalized_get_playlist(kind, variant=''):
+    """Get one personalized playlist + its current track snapshot.
+
+    Auto-creates the row from default config if it doesn't exist."""
+    try:
+        manager = _build_personalized_manager()
+        return jsonify(_personalized_api.get_playlist_with_tracks(
+            manager, kind, variant, get_current_profile_id(),
+        ))
+    except ValueError as e:
+        return jsonify({"success": False, "error": str(e)}), 400
+    except Exception as e:
+        logger.error(f"Personalized playlist get error ({kind}/{variant}): {e}")
+        return jsonify({"success": False, "error": str(e)}), 500
+
+
+@app.route('/api/personalized/playlist/<kind>/refresh', methods=['POST'])
+@app.route('/api/personalized/playlist/<kind>/<variant>/refresh', methods=['POST'])
+def personalized_refresh_playlist(kind, variant=''):
+    """Run the kind's generator and persist the snapshot."""
+    try:
+        manager = _build_personalized_manager()
+        body = request.get_json(silent=True) or {}
+        overrides = body.get('config_overrides') if isinstance(body.get('config_overrides'), dict) else None
+        return jsonify(_personalized_api.refresh_playlist(
+            manager, kind, variant, get_current_profile_id(), config_overrides=overrides,
+        ))
+    except ValueError as e:
+        return jsonify({"success": False, "error": str(e)}), 400
+    except Exception as e:
+        logger.error(f"Personalized playlist refresh error ({kind}/{variant}): {e}")
+        return jsonify({"success": False, "error": str(e)}), 500
+
+
+@app.route('/api/personalized/playlist/<kind>/config', methods=['PUT'])
+@app.route('/api/personalized/playlist/<kind>/<variant>/config', methods=['PUT'])
+def personalized_update_config(kind, variant=''):
+    """Patch the playlist's per-instance config."""
+    try:
+        manager = _build_personalized_manager()
+        body = request.get_json(silent=True) or {}
+        return jsonify(_personalized_api.update_config(
+            manager, kind, variant, get_current_profile_id(), body,
+        ))
+    except ValueError as e:
+        return jsonify({"success": False, "error": str(e)}), 400
+    except Exception as e:
+        logger.error(f"Personalized playlist config error ({kind}/{variant}): {e}")
+        return jsonify({"success": False, "error": str(e)}), 500
+
+
 @app.route('/api/discover/artist-blacklist', methods=['GET'])
 def get_discovery_artist_blacklist():
     """Get all blacklisted discovery artists."""