feat(api): relay services from remote servers with pooled sessions and login prompts

docs/dev/rest-api/endpoints.md

@@ -17,6 +17,7 @@ Start the server with [`unshackle serve`](../../guide/cli-reference.md). By defa
 - **CORS.** Every response carries permissive CORS headers, and the server answers `OPTIONS` preflight requests automatically.
 - **Compression.** JSON responses of 256 bytes or more are gzip-compressed when the client sends `Accept-Encoding: gzip`. A [slow response](#slow-responses) is not compressed.
 - **Service allowlist.** The effective allowlist for your API key filters the service-facing endpoints (the intersection of the global `serve.services` list and your per-key list). The server treats services you cannot use as unknown.
+- **Remote services.** A full-mode server can run a service through one of its own `remote_services` servers (see [Remote sessions](remote-sessions.md)). `GET /api/config` lists their names in `remote_services`, and `GET /api/services?remote=<name>` lists that server's services, each with `remote` set to the name. Add `"remote": "<name>"` to `POST /api/search`, `/api/list-titles`, `/api/list-tracks` or `/api/download`, and the server opens a remote session with its own key for that server: the remote logs in and resolves titles, tracks and licences, and a download still runs, decrypts and muxes here. The tag may be any case or an alias, and your allowlist applies to it. An unknown name is `400 INVALID_PARAMETERS`, a server that does not answer is `SERVICE_UNAVAILABLE`, and a tag it does not offer is `INVALID_SERVICE`. List calls for the same title, profile, proxy and service options share one remote session, which closes after 120 seconds unused. An optional `client_name` names your client on the remote server's dashboard as `serve (<client_name>)`. A `--remote-only` server does not relay. A remote login that waits on you (a pair code or URL to approve on another device) cannot answer through a list call, so poll `GET /api/remote/logins` alongside it: each entry carries the login's `notice`, and its `prompt` when it asks a question (an OTP), which you answer with `POST /api/remote/logins/{id}/input` and `{"response": "..."}`. A remote download job shows the same text as its own `notice`, and asks a question (an OTP) through `input_prompt` as a local job does.
 
 !!! warning "Error responses have two shapes"
     Structured application errors (raised inside a handler) return `{"status": "error", "error_code": "...", "message": "...", "timestamp": "..."}`. The authentication middleware, however, returns a **different** shape: `{"status": 401, "message": "..."}` with an integer `status` and no `error_code`. The same middleware answers `429` in that shape when an API key goes over its rate limit. Clients must accept both shapes. The [Error responses](#error-responses) section at the bottom of this page gives the full error model.

tests/remote/unit/test_remote_relay.py

@@ -0,0 +1,223 @@
+import asyncio
+import json
+import time
+
+import pytest
+
+from unshackle.core import remote_service
+from unshackle.core.api import handlers
+from unshackle.core.api.errors import APIError, APIErrorCode
+from unshackle.core.config import config
+
+SERVICES = [
+    {
+        "tag": "AMZN",
+        "aliases": ["amazon"],
+        "cli_params": [{"kind": "option", "name": "extras"}, {"kind": "argument", "name": "title"}],
+    }
+]
+INFO = SERVICES[0]
+
+
+class FakeRemote:
+    service_tag = "AMZN"
+
+    def __init__(self) -> None:
+        self.closed = False
+
+    def close(self) -> None:
+        self.closed = True
+
+
+@pytest.fixture
+def pool(monkeypatch):
+    logins = []
+
+    def fake_setup(data, server, profile, title_id, request=None):
+        time.sleep(0.05)  # long enough that every request queues on the lease lock
+        logins.append(title_id)
+        return FakeRemote()
+
+    monkeypatch.setattr(handlers, "setup_remote_service", fake_setup)
+    monkeypatch.setattr(handlers, "_remote_pool", {})
+    return logins
+
+
+def lease(data, **extra):
+    return handlers.lease_remote_service({**data, **extra}, "srv", INFO, None, "show")
+
+
+DATA = {"service": "AMZN", "title_id": "show", "remote": "srv"}
+
+
+async def test_concurrent_leases_share_one_session(pool):
+    leases = await asyncio.gather(*(lease(DATA, wanted=f"S01E0{i}") for i in range(1, 7)))
+
+    assert pool == ["show"]
+    assert len({id(x.service) for x in leases}) == 1
+    for x in leases:
+        await handlers.release_remote_service(x)
+    assert not leases[0].service.closed and handlers._remote_pool
+
+
+async def test_a_broken_session_leaves_the_pool_at_once_but_closes_after_its_last_holder(pool):
+    first = await lease(DATA)
+    second = await lease(DATA)
+    first.dead = True
+
+    await handlers.release_remote_service(first)
+    assert not handlers._remote_pool and not first.service.closed
+
+    await handlers.release_remote_service(second)
+    assert first.service.closed
+    assert (await lease(DATA)).service is not first.service
+
+
+async def test_only_options_that_reach_the_remote_split_the_pool(pool):
+    a = await lease(DATA, extras="a")
+    b = await lease(DATA, extras="b")
+    c = await lease(DATA, extras="a", tvdb_id="1", client_name="x")
+    assert a.service is not b.service
+    assert a.service is c.service
+
+
+@pytest.fixture
+def remote_config(monkeypatch):
+    monkeypatch.setattr(config, "remote_services", {"srv": {"url": "https://srv", "api_key": "k"}})
+    monkeypatch.setattr(remote_service, "remote_service_list", lambda name: SERVICES)
+    monkeypatch.setattr(handlers, "get_allowed_services", lambda request=None: None)
+
+
+def test_remote_server_resolves_case_and_aliases(remote_config):
+    assert handlers.remote_server({"remote": "srv", "service": "amzn"}) == ("srv", INFO)
+    assert handlers.remote_server({"remote": "srv", "service": "Amazon"}) == ("srv", INFO)
+    assert handlers.remote_server({"service": "AMZN"}) is None
+
+
+@pytest.mark.parametrize(
+    "data, code",
+    [
+        ({"remote": "nope", "service": "AMZN"}, APIErrorCode.INVALID_PARAMETERS),
+        ({"remote": ["srv"], "service": "AMZN"}, APIErrorCode.INVALID_PARAMETERS),
+        ({"remote": "srv", "service": "NF"}, APIErrorCode.INVALID_SERVICE),
+    ],
+)
+def test_remote_server_rejects(remote_config, data, code):
+    with pytest.raises(APIError) as e:
+        handlers.remote_server(data)
+    assert e.value.error_code == code
+
+
+def test_remote_server_applies_the_allowlist_and_reports_an_unreachable_server(remote_config, monkeypatch):
+    monkeypatch.setattr(handlers, "get_allowed_services", lambda request=None: ["NF"])
+    with pytest.raises(APIError) as e:
+        handlers.remote_server({"remote": "srv", "service": "AMZN"})
+    assert e.value.error_code == APIErrorCode.INVALID_SERVICE
+
+    monkeypatch.setattr(remote_service, "remote_service_list", lambda name: None)
+    with pytest.raises(APIError) as e:
+        handlers.remote_server({"remote": "srv", "service": "AMZN"})
+    assert e.value.error_code == APIErrorCode.SERVICE_UNAVAILABLE
+
+
+def test_make_remote_service_passes_only_cli_options(remote_config, monkeypatch):
+    built = {}
+    monkeypatch.setattr(remote_service, "resolve_server", lambda name: ("https://srv", "k", {"AMZN": {"x": 1}}))
+    monkeypatch.setattr(remote_service, "RemoteService", lambda *a, **kw: built.update(args=a, **kw))
+
+    remote_service.make_remote_service(
+        None, "srv", "AMZN", "id", {"service_params": {"extras": "y", "service": "DSNP", "proxy": "x"}}, False
+    )
+    assert built["service_params"] == {"extras": "y"}
+    assert "AMZN" not in built["args"][5]  # apply_overrides=False keeps the remote's overrides out
+
+    remote_service.make_remote_service(None, "srv", "AMZN", "id", {"extras": "z", "service": "DSNP", "quality": 1})
+    assert built["service_params"] == {"extras": "z"}
+    assert built["args"][5]["AMZN"] == {"x": 1}
+
+
+def test_remote_error_carries_its_reason_and_exits_a_cli_run():
+    err = remote_service.RemoteError("Server error [SESSION_NOT_FOUND]: gone", "SESSION_NOT_FOUND")
+    assert isinstance(err, SystemExit) and err.code == 1 and err.kind == "SESSION_NOT_FOUND"
+    assert err.session_broke
+    assert not remote_service.RemoteError("Server error [NOT_FOUND]: no title", "NOT_FOUND").session_broke
+
+
+def test_a_geofenced_remote_service_gets_its_first_allowed_region_as_proxy(monkeypatch):
+    monkeypatch.setattr(handlers, "server_proxy_allowed", lambda request=None: True)
+    fenced = {"tag": "PCOK", "geofence": ["US", "ca"], "geoblock": ["us"]}
+    assert handlers.geofence_proxy({}, {"tag": "PCOK", "geofence": ["US", "ca"]}) == {"proxy": "us"}
+    assert handlers.geofence_proxy({}, fenced) == {"proxy": "ca"}
+    assert handlers.geofence_proxy({"proxy": "gb"}, fenced) == {}
+    assert handlers.geofence_proxy({"no_proxy": True}, fenced) == {}
+    assert handlers.geofence_proxy({}, INFO) == {}
+
+    monkeypatch.setattr(handlers, "server_proxy_allowed", lambda request=None: False)
+    assert handlers.geofence_proxy({}, fenced) == {}
+
+
+def test_a_quiet_remote_login_shows_its_last_server_lines_as_a_notice(monkeypatch):
+    import logging
+    from threading import Lock
+
+    shown = []
+    monkeypatch.setattr(remote_service, "show_notice", shown.append)
+    monkeypatch.setattr(remote_service.time, "sleep", lambda s: None)
+    logs = iter([["+ IP Region: CA"], ["+ Open URL: https://disneyplus.com/start\n+ Enter pair code: 9835-2503"], []])
+    statuses = iter(["authenticating", "authenticating", "authenticating", "authenticated"])
+
+    class Client:
+        def get(self, endpoint):
+            return {"status": next(statuses)}
+
+        def get_optional(self, endpoint):
+            batch = next(logs, [])
+            remote.seq += len(batch)
+            return {"logs": [{"level": "INFO", "message": m} for m in batch], "last_seq": remote.seq}
+
+    remote = object.__new__(remote_service.RemoteService)
+    remote.seq = 0
+    remote.client, remote._session_id, remote._log_seq = Client(), "sid", 0
+    remote._recent_logs, remote.auth_notice, remote._log_drain_lock = [], None, Lock()
+    remote.log, remote._cdm_sessions = logging.getLogger("test"), {}
+    monkeypatch.setattr(remote_service.RemoteService, "local_cdm", None)
+
+    remote.poll_auth_completion(poll_interval=0)
+    # silent while the remote was still logging; the pair code once it went quiet; cleared when done
+    assert shown == ["+ IP Region: CA\n+ Open URL: https://disneyplus.com/start\n+ Enter pair code: 9835-2503", None]
+    assert remote.auth_notice is None
+
+
+async def test_a_remote_login_prompt_waits_for_its_owners_answer(monkeypatch):
+    import threading as th
+
+    class Remote:
+        auth_notice = None
+
+    login = handlers.RemoteLogin("owner", "srv", "DSNP", "show", Remote())
+    monkeypatch.setattr(handlers, "_remote_logins", {login.id: login})
+    monkeypatch.setattr(handlers, "is_admin", lambda request: False)
+    caller = {"key": "owner"}
+    monkeypatch.setattr(handlers, "caller_key", lambda request: caller["key"])
+    got = []
+    asker = th.Thread(target=lambda: got.append(login.bridge.request_input("Enter the code sent to your email")))
+    asker.start()
+    while login.bridge.get_pending_prompt() is None:
+        await asyncio.sleep(0.01)
+
+    listed = json.loads((await handlers.remote_logins_handler(None)).text)["logins"]
+    assert listed[0]["prompt"] == "Enter the code sent to your email"
+
+    caller["key"] = "someone-else"
+    assert json.loads((await handlers.remote_logins_handler(None)).text)["logins"] == []
+    with pytest.raises(APIError) as e:
+        await handlers.remote_login_input_handler(login.id, {"response": "123456"}, None)
+    assert e.value.error_code == APIErrorCode.NOT_FOUND
+
+    caller["key"] = "owner"
+    await handlers.remote_login_input_handler(login.id, {"response": "123456"}, None)
+    asker.join(2)
+    assert got == ["123456"] and login.bridge.get_pending_prompt() is None
+    with pytest.raises(APIError) as e:
+        await handlers.remote_login_input_handler(login.id, {"response": "again"}, None)
+    assert e.value.error_code == APIErrorCode.CONFLICT

tests/remote/unit/test_remote_service_helpers.py

@@ -600,7 +600,7 @@ def test_the_client_session_downloads_through_the_proxy_it_sends(
     """The client fetches playlists and segments itself, so --proxy must reach its own session too."""
     from unshackle.core import remote_service
 
-    monkeypatch.setattr(remote_service, "resolve_remote_proxy_arg", lambda proxy: proxy)
+    monkeypatch.setattr(remote_service, "resolve_remote_proxy_arg", lambda proxy, providers=None: proxy)
     monkeypatch.setattr("unshackle.core.utils.ip_info.get_ip_info", lambda *args, **kwargs: None)
     svc = _remote_service({"no_proxy": no_proxy, "proxy": "http://proxy.example:8080"}, {})
     with pytest.raises(_SentCreate) as exc:

tests/remote/unit/test_session_log.py

@@ -146,6 +146,7 @@ def test_drain_server_logs_reemits_and_advances_seq() -> None:
     svc._session_id = "abc"
     svc._log_seq = 0
     svc._log_drain_lock = threading.Lock()
+    svc._recent_logs = []
     svc.log = logging.getLogger("test-drain-reemit")
 
     class StubClient:

unshackle/commands/serve.py

@@ -273,6 +273,9 @@ def reload_config(app: web.Application, *, no_key: bool, serve_widevine: bool, s
     old_users, new_users = set(app["config"]["users"]), set(serve_config["users"])
     old_devices, new_devices = _device_stems(config.serve), _device_stems(serve)
     vars(config).update(vars(fresh))
+    from unshackle.core.api.handlers import reset_remote_pool
+
+    reset_remote_pool()
     app["config"]["users"] = serve_config["users"]
     if serve_widevine:
         app["config"]["devices"] = _device_map(serve_config["devices"])

unshackle/core/api/download_manager.py

@@ -454,8 +454,9 @@ def perform_download(
             season_range = SeasonRange()
             params["wanted"] = season_range.parse_tokens(*wanted_raw)
 
-    service_config_path = Services.get_path(service) / config.filenames.config
-    if service_config_path.exists():
+    remote = params.get("remote")
+    service_config_path = None if remote else Services.get_path(service) / config.filenames.config
+    if service_config_path and service_config_path.exists():
         service_config = yaml.safe_load(service_config_path.read_text(encoding="utf8"))
     else:
         service_config = {}
@@ -465,7 +466,7 @@ def perform_download(
 
     ctx = click.Context(dl_command.cli)
     ctx.invoked_subcommand = service
-    from unshackle.core.api.handlers import load_full_cdm
+    from unshackle.core.api.handlers import load_full_cdm, relay_label
 
     cdm = load_full_cdm(service, params.get("profile"), params.get("cdm_type"))
     ctx.obj = ContextData(config=service_config, cdm=cdm, proxy_providers=[], profile=params.get("profile"))
@@ -501,6 +502,10 @@ def perform_download(
         "a_lang": params.get("a_lang", []),
         "forced_subs": params.get("forced_subs", False),
         "exact_lang": params.get("exact_lang", False),
+        # dl skips the local service config and guards the proxy for a remote_services server
+        "remote": bool(remote),
+        "server": remote,
+        "client_argv": f"{relay_label(params)} download job {job_id}",
     }
     # Hand-built context: record parameter sources so service dl overrides
     # apply to defaults but never clobber client-sent values.
@@ -539,47 +544,49 @@ def perform_download(
             config.credentials[service] = svc_creds = {}
         svc_creds[params.get("profile") or "default"] = params["credential"]
 
-    service_module = Services.load(service)
+    service_ctx = click.Context(click.Command(service), parent=ctx, obj=ctx.obj)
+    if remote:
+        from unshackle.core.remote_service import make_remote_service
 
-    try:
-        import inspect
+        service_instance = make_remote_service(service_ctx, remote, service, title_id, params)
+    else:
+        service_module = Services.load(service)
 
-        service_init_params = inspect.signature(service_module.__init__).parameters
+        try:
+            import inspect
 
-        service_ctx = click.Context(click.Command(service))
-        service_ctx.parent = ctx
-        service_ctx.obj = ctx.obj
+            service_init_params = inspect.signature(service_module.__init__).parameters
 
-        service_kwargs = {}
+            service_kwargs = {}
 
-        if "title" in service_init_params:
-            service_kwargs["title"] = title_id
+            if "title" in service_init_params:
+                service_kwargs["title"] = title_id
 
-        for key, value in params.items():
-            if key in service_init_params and key not in ["service", "title_id", "profile", "service_params"]:
-                service_kwargs[key] = value
-        service_params = params.get("service_params")
-        if isinstance(service_params, dict):
-            for key, value in service_params.items():
-                if key in service_init_params and key != "title":
+            for key, value in params.items():
+                if key in service_init_params and key not in ["service", "title_id", "profile", "service_params"]:
                     service_kwargs[key] = value
+            service_params = params.get("service_params")
+            if isinstance(service_params, dict):
+                for key, value in service_params.items():
+                    if key in service_init_params and key != "title":
+                        service_kwargs[key] = value
 
-        for param_name, param_info in service_init_params.items():
-            if param_name not in service_kwargs and param_name not in ["self", "ctx"]:
-                if param_info.default is inspect.Parameter.empty:
-                    if param_name == "movie":
-                        service_kwargs[param_name] = "/movies/" in title_id
-                    elif param_name == "meta_lang":
-                        service_kwargs[param_name] = None
-                    else:
-                        log.warning(f"Unknown required parameter '{param_name}' for service {service}, using None")
-                        service_kwargs[param_name] = None
+            for param_name, param_info in service_init_params.items():
+                if param_name not in service_kwargs and param_name not in ["self", "ctx"]:
+                    if param_info.default is inspect.Parameter.empty:
+                        if param_name == "movie":
+                            service_kwargs[param_name] = "/movies/" in title_id
+                        elif param_name == "meta_lang":
+                            service_kwargs[param_name] = None
+                        else:
+                            log.warning(f"Unknown required parameter '{param_name}' for service {service}, using None")
+                            service_kwargs[param_name] = None
 
-        service_instance = service_module(service_ctx, **service_kwargs)
+            service_instance = service_module(service_ctx, **service_kwargs)
 
-    except Exception as exc:  # noqa: BLE001 - propagate meaningful failure
-        log.error(f"Failed to create service instance: {exc}")
-        raise
+        except Exception as exc:  # noqa: BLE001 - propagate meaningful failure
+            log.error(f"Failed to create service instance: {exc}")
+            raise
 
     original_download_dir = config.directories.downloads
 

unshackle/core/api/handlers.py

@@ -6,11 +6,15 @@ import logging
 import os
 import re
 import tempfile
+import threading
 import time
+import uuid
 import zlib
 from collections import Counter
 from contextlib import suppress
+from dataclasses import dataclass, field
 from datetime import date as date_
+from functools import partial
 from http.cookiejar import CookieJar, MozillaCookieJar
 from pathlib import Path
 from typing import Any, Callable, Dict, Iterable, List, Optional, cast
@@ -406,6 +410,329 @@ class RequestCache:
         SessionStore.cleanup_cache_dir(self.tag)
 
 
+def remote_list(name: Any) -> list[dict]:
+    """The services of a configured ``remote_services`` server; an APIError for an unknown or unreachable one."""
+    from unshackle.core.api.stats import stats
+    from unshackle.core.remote_service import remote_service_list
+
+    if stats.mode == "remote_only":
+        raise APIError(APIErrorCode.INVALID_PARAMETERS, "A remote-only server does not relay to other servers")
+    if not isinstance(name, str) or name not in (config.remote_services or {}):
+        raise APIError(APIErrorCode.INVALID_PARAMETERS, f"Unknown remote server: {sanitize_log(str(name))}")
+    services = remote_service_list(name)
+    if services is None:
+        raise APIError(APIErrorCode.SERVICE_UNAVAILABLE, f"Remote server '{name}' is unreachable")
+    return services
+
+
+def remote_server(data: Dict[str, Any], request: Optional[web.Request] = None) -> Optional[tuple[str, dict]]:
+    """The ``remote_services`` server and service a request names in ``remote``, as the CLI resolves it and the
+    caller's key allows it; None for a local service. Blocks on the remote list."""
+    from unshackle.core.remote_service import find_remote_service
+
+    name = data.get("remote")
+    if not name:
+        return None
+    service = data.get("service")
+    info = find_remote_service(remote_list(name), str(service or ""))
+    allowed = get_allowed_services(request)
+    if info is None or (allowed is not None and info["tag"] not in allowed):
+        raise APIError(
+            APIErrorCode.INVALID_SERVICE,
+            f"Remote server '{name}' does not offer {sanitize_log(str(service))}",
+            details={"service": service, "remote": name},
+        )
+    return name, info
+
+
+def geofence_proxy(data: Dict[str, Any], info: dict, request: Optional[web.Request] = None) -> Dict[str, Any]:
+    """``{"proxy": region}`` for a geofenced remote service the request sends no proxy for; else {}.
+
+    The remote counterpart of a local service's geofence auto-proxy, under the same ``server_proxy`` gate.
+    """
+    if data.get("proxy") or data.get("no_proxy") or not server_proxy_allowed(request):
+        return {}
+    blocked = {str(r).lower() for r in info.get("geoblock") or []}
+    fence = [str(r).lower() for r in info.get("geofence") or [] if r and str(r).lower() not in blocked]
+    return {"proxy": fence[0]} if fence else {}
+
+
+async def relay_target(
+    data: Dict[str, Any], request: Optional[web.Request] = None
+) -> tuple[Optional[str], Optional[dict], Dict[str, Any]]:
+    """``(server, info, data)`` for a request naming ``remote``, ``data`` carrying the resolved tag and any
+    geofence proxy; ``(None, None, data)`` for a local one."""
+    if not data.get("remote"):
+        return None, None, data
+    server, info = await asyncio.to_thread(remote_server, data, request)
+    return server, info, {**data, "service": info["tag"], **geofence_proxy(data, info, request)}
+
+
+def relay_label(data: Dict[str, Any]) -> str:
+    """How the remote server's dashboard names this relay: the REST client's own ``client_name``, if it sent one."""
+    name = data.get("client_name")
+    return f"serve ({name[:64] if isinstance(name, str) and name.strip() else 'REST API'})"
+
+
+def relay_error(exc: Any, service: str) -> APIError:
+    """The APIError for a RemoteError: the remote server's own reason, not a bare exit."""
+    return APIError(APIErrorCode.SERVICE_ERROR, exc.message, details={"service": service, "remote_code": exc.kind})
+
+
+def setup_remote_service(
+    data: Dict[str, Any], server: str, profile: Optional[str], title_id: str, request: Optional[web.Request] = None
+) -> Any:
+    """Open and authenticate a remote session and fetch its titles; the caller must close() it."""
+    from unshackle.core.remote_service import RemoteError, make_remote_service
+
+    # the same server_proxy gate a local service gets: without it the providers stay empty
+    proxy_param, providers = resolve_handler_proxy(data, data["service"], request)
+    parent_ctx = build_parent_ctx(
+        profile,
+        None,
+        proxy_param,
+        data.get("no_proxy", False),
+        providers,
+        {},
+        extra_params={"profile": profile, "client_argv": relay_label(data)},
+    )
+    ctx = click.Context(click.Command(data["service"]), parent=parent_ctx, obj=parent_ctx.obj)
+    # this process is long-lived and serves local services too, so the remote's overrides stay out of config
+    service_instance = make_remote_service(ctx, server, data["service"], title_id, data, apply_overrides=False)
+    login = RemoteLogin(caller_key(request), server, data["service"], title_id, service_instance)
+    service_instance._input_bridge = login.bridge
+    _remote_logins[login.id] = login
+    try:
+        service_instance.authenticate()
+        service_instance.get_titles()
+    except BaseException as exc:
+        service_instance.close()
+        if isinstance(exc, RemoteError):
+            raise relay_error(exc, data["service"]) from None
+        raise
+    finally:
+        _remote_logins.pop(login.id, None)
+    return service_instance
+
+
+@dataclass(eq=False)
+class RemoteLogin:
+    """A remote login a list call is waiting on, so a client can show its pair code or answer its prompt.
+
+    Its bridge is the RemoteService's input bridge: an OTP the remote asks for waits there for the answer a
+    client sends to ``POST /api/remote/logins/{id}/input``, as a download job's prompt waits for its own.
+    """
+
+    owner: str
+    remote: str
+    service: str
+    title_id: str
+    instance: Any
+    id: str = field(default_factory=lambda: uuid.uuid4().hex)
+    bridge: InputBridge = field(default_factory=InputBridge)
+
+
+_remote_logins: Dict[str, RemoteLogin] = {}
+
+
+async def remote_logins_handler(request: web.Request) -> web.Response:
+    """Answer the request for the caller's remote logins in progress, each with its prompt and notice."""
+    owner, admin = caller_key(request), is_admin(request)
+    logins = [
+        {
+            "id": login.id,
+            "remote": login.remote,
+            "service": login.service,
+            "title_id": login.title_id,
+            "prompt": login.bridge.get_pending_prompt(),
+            "notice": login.instance.auth_notice,
+        }
+        for login in list(_remote_logins.values())
+        if admin or login.owner == owner
+    ]
+    return web.json_response({"logins": logins})
+
+
+async def remote_login_input_handler(login_id: str, data: Dict[str, Any], request: web.Request) -> web.Response:
+    """Answer the prompt (an OTP, a PIN) a remote login a list call waits on is asking."""
+    login = _remote_logins.get(login_id)
+    if login is None or not (is_admin(request) or login.owner == caller_key(request)):
+        raise APIError(APIErrorCode.NOT_FOUND, "Remote login not found", details={"login_id": login_id})
+    response = data.get("response")
+    if response is None:
+        raise APIError(APIErrorCode.INVALID_INPUT, "Missing required field: response")
+    if not login.bridge.submit_response(str(response)):
+        raise APIError(APIErrorCode.CONFLICT, "This remote login is not asking anything")
+    return web.json_response({"status": "accepted"})
+
+
+REMOTE_SESSION_IDLE = 120.0
+
+
+@dataclass
+class RemoteLease:
+    """One pooled remote session: a title's episodes all list through it instead of one login each.
+
+    Only the event loop touches the pool, so it needs no thread lock; blocking work runs in threads.
+    """
+
+    key: tuple
+    lock: asyncio.Lock = field(default_factory=asyncio.Lock)
+    service: Any = None
+    error: Optional[Exception] = None
+    busy: int = 0
+    used: float = 0.0
+    dead: bool = False
+
+
+_remote_pool: Dict[tuple, RemoteLease] = {}
+_remote_sweeper: Optional["asyncio.Task[None]"] = None
+
+
+def remote_pool_key(data: Dict[str, Any], server: str, info: dict, profile: Optional[str], title_id: str) -> tuple:
+    """What opens a remote session; requests that differ only in episode selection share one."""
+    from unshackle.core.remote_service import remote_service_params
+
+    options = json.dumps(remote_service_params(info, data), sort_keys=True, default=str)
+    return (server, data["service"], title_id, profile, data.get("proxy"), bool(data.get("no_proxy")), options)
+
+
+def close_remote_services(services: List[Any]) -> None:
+    for service in services:
+        with suppress(Exception, SystemExit):
+            service.close()
+
+
+async def sweep_remote_pool() -> None:
+    """Close pooled remote sessions idle for REMOTE_SESSION_IDLE; their keepalive would hold them open forever."""
+    while True:
+        await asyncio.sleep(REMOTE_SESSION_IDLE / 4)
+        now = time.monotonic()
+        idle = [lease for lease in _remote_pool.values() if not lease.busy and now - lease.used > REMOTE_SESSION_IDLE]
+        for lease in idle:
+            del _remote_pool[lease.key]
+        if idle:
+            await asyncio.to_thread(close_remote_services, [lease.service for lease in idle if lease.service])
+        if not _remote_pool:
+            return
+
+
+def reset_remote_pool() -> None:
+    """Drop the cached remote lists and every pooled session, for a config reload that may change a server.
+
+    Runs on the event loop, so the closes, each a blocking DELETE, go to a thread.
+    """
+    from unshackle.core.remote_service import reset_remote_lists
+
+    reset_remote_lists()
+    leases = list(_remote_pool.values())
+    _remote_pool.clear()
+    for lease in leases:
+        lease.dead = True
+    idle = [lease.service for lease in leases if lease.service is not None and not lease.busy]
+    if idle:
+        threading.Thread(target=close_remote_services, args=(idle,), name="remote-pool-reset", daemon=True).start()
+
+
+async def lease_remote_service(
+    data: Dict[str, Any],
+    server: str,
+    info: dict,
+    profile: Optional[str],
+    title_id: str,
+    request: Optional[web.Request] = None,
+    fresh_titles: bool = False,
+) -> RemoteLease:
+    """A pooled lease on an authenticated RemoteService with its titles fetched; return it with release_remote_service.
+
+    ``fresh_titles`` drops the title list a reused session cached, so a relist sees a newly released episode.
+    """
+    global _remote_sweeper
+    if _remote_sweeper is None or _remote_sweeper.done():
+        _remote_sweeper = asyncio.get_running_loop().create_task(sweep_remote_pool())
+    key = remote_pool_key(data, server, info, profile, title_id)
+    lease = _remote_pool.setdefault(key, RemoteLease(key))
+    lease.busy += 1
+    try:
+        # concurrent first requests for a title wait here, on the loop, for one login instead of each opening one
+        async with lease.lock:
+            if lease.error is not None:
+                raise lease.error
+            if lease.service is None:
+                lease.service = await asyncio.to_thread(setup_remote_service, data, server, profile, title_id, request)
+            elif fresh_titles:
+                lease.service._titles = None
+    except BaseException as exc:
+        if isinstance(exc, Exception):
+            lease.error = exc
+        lease.busy -= 1
+        if _remote_pool.get(key) is lease:
+            del _remote_pool[key]
+        raise
+    return lease
+
+
+async def release_remote_service(lease: RemoteLease) -> None:
+    """Return a lease. A broken session left the pool already and closes when its last holder returns it."""
+    lease.busy -= 1
+    lease.used = time.monotonic()
+    if lease.dead and _remote_pool.get(lease.key) is lease:
+        del _remote_pool[lease.key]
+    if lease.dead and not lease.busy and lease.service is not None:
+        await asyncio.to_thread(close_remote_services, [lease.service])
+
+
+async def remote_call(lease: RemoteLease, method: Callable[..., Any], *args: Any) -> Any:
+    """Run a blocking RemoteService call in a thread, answering a RemoteError with the remote's reason."""
+    from unshackle.core.remote_service import RemoteError
+
+    try:
+        return await asyncio.to_thread(method, *args)
+    except RemoteError as exc:
+        # one episode the remote cannot list must not cost the whole title its session
+        lease.dead = lease.dead or exc.session_broke
+        raise relay_error(exc, lease.service.service_tag) from None
+
+
+async def leased_titles(
+    data: Dict[str, Any],
+    server: str,
+    info: dict,
+    profile: Optional[str],
+    title_id: str,
+    request: Optional[web.Request],
+    fresh_titles: bool,
+) -> tuple[RemoteLease, Any]:
+    """Lease a remote session and get its titles, once more on a new session when the pooled one broke.
+
+    The caller releases the returned lease; on an error this releases it itself.
+    """
+    for attempt in (1, 2):
+        lease = await lease_remote_service(data, server, info, profile, title_id, request, fresh_titles)
+        try:
+            return lease, await remote_call(lease, lease.service.get_titles)
+        except BaseException:
+            await release_remote_service(lease)
+            if attempt == 2 or not lease.dead:
+                raise
+    raise AssertionError("unreachable")
+
+
+def remote_services_info(name: str, allowed: Optional[List[str]]) -> List[Dict[str, Any]]:
+    """The services of one ``remote_services`` server, each marked with that server's name in ``remote``."""
+    return [{**svc, "remote": name} for svc in remote_list(name) if allowed is None or svc.get("tag") in allowed]
+
+
+def remote_search(server: str, service: str, query: str) -> Dict[str, Any]:
+    """Run a search on a ``remote_services`` server with this server's key for it."""
+    from unshackle.core.remote_service import RemoteError, remote_client
+
+    try:
+        return remote_client(server).post("/api/search", {"service": service, "query": query})
+    except RemoteError as exc:
+        raise relay_error(exc, service) from None
+
+
 def setup_list_service(
     data: Dict[str, Any],
     normalized_service: str,
@@ -1529,6 +1856,10 @@ async def search_handler(data: Dict[str, Any], request: Optional[web.Request] =
     if not query:
         raise APIError(APIErrorCode.INVALID_PARAMETERS, "Missing required 'query' field")
 
+    remote = await asyncio.to_thread(remote_server, data, request) if data.get("remote") else None
+    if remote:
+        return web.json_response(await asyncio.to_thread(remote_search, remote[0], remote[1]["tag"], query))
+
     # get_tag echoes an unknown tag back, so it can never be falsy; validate_service
     # is the check that actually resolves the service directory and the allowlist.
     normalized_service = validate_service(service_tag, request)
@@ -1554,7 +1885,8 @@ async def list_titles_handler(data: Dict[str, Any], request: Optional[web.Reques
     title_id = data.get("title_id")
     profile = client_profile(data)
 
-    normalized_service = validate_service(service_tag, request)
+    server, info, data = await relay_target(data, request)
+    normalized_service = info["tag"] if info else validate_service(service_tag, request)
     if not normalized_service:
         raise APIError(
             APIErrorCode.INVALID_SERVICE,
@@ -1563,11 +1895,16 @@ async def list_titles_handler(data: Dict[str, Any], request: Optional[web.Reques
         )
 
     request_cache = RequestCache()
+    lease = None
     try:
-        service_instance = await asyncio.to_thread(
-            setup_list_service, data, normalized_service, profile, title_id, request, request_cache
-        )
-        titles = await asyncio.to_thread(service_instance.get_titles)
+        if server:
+            lease, titles = await leased_titles(data, server, info, profile, title_id, request, True)
+            service_instance = lease.service
+        else:
+            service_instance = await asyncio.to_thread(
+                setup_list_service, data, normalized_service, profile, title_id, request, request_cache
+            )
+            titles = await asyncio.to_thread(service_instance.get_titles)
 
         title_list = [stamp_service_flags(serialize_title(t), service_instance) for t in listed_titles(titles)]
 
@@ -1584,6 +1921,8 @@ async def list_titles_handler(data: Dict[str, Any], request: Optional[web.Reques
             debug_mode=debug_mode,
         )
     finally:
+        if lease is not None:
+            await release_remote_service(lease)
         request_cache.cleanup()
 
 
@@ -1594,7 +1933,8 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques
     title_id = data.get("title_id")
     profile = client_profile(data)
 
-    normalized_service = validate_service(service_tag, request)
+    server, info, data = await relay_target(data, request)
+    normalized_service = info["tag"] if info else validate_service(service_tag, request)
     if not normalized_service:
         raise APIError(
             APIErrorCode.INVALID_SERVICE,
@@ -1603,11 +1943,18 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques
         )
 
     request_cache = RequestCache()
+    lease = None
     try:
-        service_instance = await asyncio.to_thread(
-            setup_list_service, data, normalized_service, profile, title_id, request, request_cache
-        )
-        titles = await asyncio.to_thread(service_instance.get_titles)
+        if server:
+            lease, titles = await leased_titles(data, server, info, profile, title_id, request, False)
+            service_instance = lease.service
+            call: Callable[..., Any] = partial(remote_call, lease)
+        else:
+            service_instance = await asyncio.to_thread(
+                setup_list_service, data, normalized_service, profile, title_id, request, request_cache
+            )
+            call = asyncio.to_thread
+            titles = await call(service_instance.get_titles)
 
         wanted_param = data.get("wanted")
         season = data.get("season")
@@ -1686,7 +2033,7 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques
 
                     for title in sorted_titles:
                         try:
-                            tracks = await asyncio.to_thread(service_instance.get_tracks, title)
+                            tracks = await call(service_instance.get_tracks, title)
                             video_tracks = sorted(tracks.videos, key=lambda t: t.bitrate or 0, reverse=True)
                             audio_tracks = sorted(tracks.audio, key=lambda t: t.bitrate or 0, reverse=True)
 
@@ -1733,7 +2080,7 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques
         else:
             first_title = titles
 
-        tracks = await asyncio.to_thread(service_instance.get_tracks, first_title)
+        tracks = await call(service_instance.get_tracks, first_title)
 
         video_tracks = sorted(tracks.videos, key=lambda t: t.bitrate or 0, reverse=True)
         audio_tracks = sorted(tracks.audio, key=lambda t: t.bitrate or 0, reverse=True)
@@ -1759,6 +2106,8 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques
             debug_mode=debug_mode,
         )
     finally:
+        if lease is not None:
+            await release_remote_service(lease)
         request_cache.cleanup()
 
 
@@ -2044,7 +2393,8 @@ async def download_handler(data: Dict[str, Any], request: Optional[web.Request]
     service_tag = data.get("service")
     title_id = data.get("title_id")
 
-    normalized_service = validate_service(service_tag, request)
+    server, info, data = await relay_target(data, request)
+    normalized_service = info["tag"] if info else validate_service(service_tag, request)
     if not normalized_service:
         raise APIError(
             APIErrorCode.INVALID_SERVICE,
@@ -2061,12 +2411,13 @@ async def download_handler(data: Dict[str, Any], request: Optional[web.Request]
         )
 
     try:
-        service_module = Services.load(normalized_service)
+        service_module = None if server else Services.load(normalized_service)
         service_specific_defaults = {}
 
         # Skip None defaults here: this dict overlays into job params; injecting
         # None for keys like `profile` would clobber serve-config overrides.
         # Missing required __init__ params are handled in download_manager.perform_download.
+        # A remote service applies its own defaults on the remote server.
         if hasattr(service_module, "cli") and hasattr(service_module.cli, "params"):
             for param in service_module.cli.params:
                 if hasattr(param, "name") and param.default is not None and not isinstance(param.default, enum.Enum):
@@ -2827,8 +3178,11 @@ async def retry_download_job_handler(job_id: str, request: Optional[web.Request]
             )
 
         # Re-apply creation-time gates so retry cannot bypass the caller's service allowlist
-        # or currently-disabled cdm_overrides / allow_job_credentials config.
-        if not validate_service(job.service, request):
+        # or currently-disabled cdm_overrides / allow_job_credentials config. A remote job is checked
+        # against its remote server, which must still be configured and offer the service.
+        if job.parameters.get("remote"):
+            await asyncio.to_thread(remote_server, {**job.parameters, "service": job.service}, request)
+        elif not validate_service(job.service, request):
             raise APIError(
                 APIErrorCode.INVALID_SERVICE,
                 f"Invalid or unavailable service: {job.service}",
@@ -3000,6 +3354,8 @@ async def server_config_handler(request: Optional[web.Request] = None) -> web.Re
                 or None,
             },
             "services": service_tags,
+            # names only: url and api_key stay on the server
+            "remote_services": list(config.remote_services or {}),
         }
         # Server-local absolute paths are useless to a remote client and name the host layout.
         if is_admin(request):

unshackle/core/api/routes.py

@@ -45,6 +45,9 @@ from unshackle.core.api.handlers import (
     prioritize_download_job_handler,
     profiles_handler,
     refresh_services_handler,
+    remote_login_input_handler,
+    remote_logins_handler,
+    remote_services_info,
     retry_download_job_handler,
     search_handler,
     server_account_regions,
@@ -212,6 +215,16 @@ async def services(request: web.Request) -> web.Response:
     ---
     summary: Show services
     description: Get all available streaming services with their details
+    parameters:
+      - name: remote
+        in: query
+        required: false
+        schema:
+          type: string
+        description: >
+          A remote_services server name. Lists that server's services instead of this server's, each
+          with `remote` set to the name. Pass the name as `remote` to search, list-titles, list-tracks and
+          download to run the service through that server; the download itself still runs here.
     responses:
       '200':
         description: List of services
@@ -302,6 +315,9 @@ async def services(request: web.Request) -> web.Response:
         allowed = get_allowed_services(request)
         if allowed is not None:
             service_tags = [t for t in service_tags if t in allowed]
+        if remote := request.query.get("remote"):
+            remote_info = await asyncio.to_thread(remote_services_info, remote, allowed)
+            return web.json_response({"services": remote_info, "load_errors": []})
         services_info = []
 
         for tag in service_tags:
@@ -449,6 +465,14 @@ async def search(request: web.Request) -> web.Response:
               - service
               - query
             properties:
+              remote:
+                type: string
+                description: >
+                  A remote_services server name from GET /api/config. The service runs through that
+                  server (login, titles, tracks, licences); a download still runs on this server.
+              client_name:
+                type: string
+                description: Names this client on the remote server's dashboard, as serve (client_name)
               service:
                 type: string
                 description: Service tag
@@ -558,6 +582,14 @@ async def list_titles(request: web.Request) -> web.Response:
               - service
               - title_id
             properties:
+              remote:
+                type: string
+                description: >
+                  A remote_services server name from GET /api/config. The service runs through that
+                  server (login, titles, tracks, licences); a download still runs on this server.
+              client_name:
+                type: string
+                description: Names this client on the remote server's dashboard, as serve (client_name)
               service:
                 type: string
                 description: Service tag
@@ -713,6 +745,14 @@ async def list_tracks(request: web.Request) -> web.Response:
               - service
               - title_id
             properties:
+              remote:
+                type: string
+                description: >
+                  A remote_services server name from GET /api/config. The service runs through that
+                  server (login, titles, tracks, licences); a download still runs on this server.
+              client_name:
+                type: string
+                description: Names this client on the remote server's dashboard, as serve (client_name)
               service:
                 type: string
                 description: Service tag
@@ -800,6 +840,14 @@ async def download(request: web.Request) -> web.Response:
               - service
               - title_id
             properties:
+              remote:
+                type: string
+                description: >
+                  A remote_services server name from GET /api/config. The service runs through that
+                  server (login, titles, tracks, licences); a download still runs on this server.
+              client_name:
+                type: string
+                description: Names this client on the remote server's dashboard, as serve (client_name)
               service:
                 type: string
                 description: Service tag
@@ -1507,6 +1555,11 @@ async def server_config(request: web.Request) -> web.Response:
                           description: List of permitted CDM device names, true, or null
                         allow_job_credentials:
                           type: boolean
+                    remote_services:
+                      type: array
+                      items:
+                        type: string
+                      description: remote_services server names; their url and api_key stay on the server
                     directories:
                       type: object
                       properties:
@@ -3090,6 +3143,88 @@ async def dashboard_events(request: web.Request) -> web.StreamResponse:
     return await dashboard_events_handler(request)
 
 
+@api_handler
+async def remote_logins(request: web.Request) -> web.Response:
+    """
+    Remote logins in progress.
+    ---
+    summary: Remote logins in progress
+    description: >
+      The caller's list calls that wait on a remote_services server's login, each with the remote
+      server, service and title, and `notice`: what the login shows the user while it waits on them,
+      such as a URL and pair code to enter on another device, or null. Poll it alongside a remote
+      list-titles, list-tracks or search call; a download job carries the same text as its `notice`.
+    responses:
+      '200':
+        description: Logins in progress
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                logins:
+                  type: array
+                  items:
+                    type: object
+                    properties:
+                      remote:
+                        type: string
+                      service:
+                        type: string
+                      id:
+                        type: string
+                        description: Answer its prompt at POST /api/remote/logins/{id}/input
+                      title_id:
+                        type: string
+                      prompt:
+                        type: string
+                        nullable: true
+                        description: What the login asks the user (an OTP, a PIN), or null
+                      notice:
+                        type: string
+                        nullable: true
+    """
+    return await remote_logins_handler(request)
+
+
+@api_handler
+async def remote_login_input(request: web.Request) -> web.Response:
+    """
+    Answer a remote login's prompt.
+    ---
+    summary: Answer a remote login's prompt
+    description: Send the answer to the prompt (an OTP, a PIN) a remote login a list call waits on is asking.
+    parameters:
+      - name: login_id
+        in: path
+        required: true
+        schema:
+          type: string
+    requestBody:
+      required: true
+      content:
+        application/json:
+          schema:
+            type: object
+            required: [response]
+            properties:
+              response:
+                type: string
+    responses:
+      '200':
+        description: Answer accepted
+      '404':
+        description: No such remote login for this API key
+      '409':
+        description: The login is not asking anything
+    """
+    try:
+        data = await request.json()
+    except Exception as e:
+        raise APIError(APIErrorCode.INVALID_INPUT, "Invalid JSON request body", details={"error": str(e)}) from e
+    return await remote_login_input_handler(request.match_info["login_id"], data, request)
+
+
 # Single source of truth for all API routes. `remote` marks endpoints exposed in
 # --remote-only server mode; full mode registers every route. Both setup_routes and
 # setup_swagger derive from this table so the live routes and the docs cannot drift.
@@ -3116,6 +3251,8 @@ ROUTES: list[tuple[str, str, Handler, bool]] = [
     ("POST", "/api/maintenance/clear-temp", maintenance_clear_temp, False),
     ("POST", "/api/maintenance/refresh-services", maintenance_refresh_services, False),
     ("GET", "/api/env/check", env_check, False),
+    ("GET", "/api/remote/logins", remote_logins, False),
+    ("POST", "/api/remote/logins/{login_id}/input", remote_login_input, False),
     ("POST", "/api/session/create", session_create, True),
     ("GET", "/api/session/{session_id}/titles", session_titles, True),
     ("POST", "/api/session/{session_id}/tracks", session_tracks, True),

unshackle/core/remote_service.py

@@ -17,6 +17,7 @@ import re
 import shlex
 import sys
 import time
+from collections import deque
 from datetime import date as date_
 from enum import Enum
 from http.cookiejar import CookieJar
@@ -35,7 +36,7 @@ from rich.padding import Padding
 from rich.rule import Rule
 
 from unshackle.core.config import config
-from unshackle.core.console import console, prompt_user
+from unshackle.core.console import console, prompt_user, show_notice
 from unshackle.core.constants import AnyTrack
 from unshackle.core.credential import Credential
 from unshackle.core.drm import cbcs, real_kids
@@ -66,6 +67,26 @@ def redact_secrets(text: str, data: Optional[Dict[str, Any]] = None) -> str:
     return redact_text(text, secrets) or ""
 
 
+class RemoteError(SystemExit):
+    """A remote call that failed: logged, then exits a CLI run like ``SystemExit(1)``.
+
+    A serve relaying for a REST client catches it instead and answers with ``message``; ``kind`` is the
+    remote server's error code, or CONNECT / TIMEOUT / AUTH when the session could not be reached or opened.
+    (Not ``code``: SystemExit's ``code`` is the exit status.)
+    """
+
+    def __init__(self, message: str, kind: str) -> None:
+        super().__init__(1)
+        self.message = message
+        self.kind = kind
+        log.error(message)
+
+    @property
+    def session_broke(self) -> bool:
+        """Whether the remote session itself is gone, not just this one call."""
+        return self.kind in ("SESSION_NOT_FOUND", "CONNECT", "TIMEOUT")
+
+
 class RemoteClient:
     """HTTP client for the unshackle serve API."""
 
@@ -129,20 +150,13 @@ class RemoteClient:
                     resp = getattr(self.session, method)(url, timeout=self.timeout, **payload)
             except requests.ConnectionError as e:
                 if isinstance(e.args[0] if e.args else None, urllib3.exceptions.ReadTimeoutError):
-                    log.error(
-                        f"Request to remote server timed out after {self.timeout}s: {endpoint}. "
-                        "Raise 'timeout' for this server in remote_services."
-                    )
-                    raise SystemExit(1)
+                    raise self.timed_out(endpoint) from None
                 server_url = safe_display_url(self.server_url)
-                log.error(f"Could not connect to remote server at {server_url}. Is it running? (unshackle serve)")
-                raise SystemExit(1)
+                raise RemoteError(
+                    f"Could not connect to remote server at {server_url}. Is it running? (unshackle serve)", "CONNECT"
+                ) from None
             except requests.Timeout:
-                log.error(
-                    f"Request to remote server timed out after {self.timeout}s: {endpoint}. "
-                    "Raise 'timeout' for this server in remote_services."
-                )
-                raise SystemExit(1)
+                raise self.timed_out(endpoint) from None
             # servers differ on which header carries the key, so retry the rest before giving up
             if resp.status_code == 401 and self.next_auth_header():
                 continue
@@ -166,10 +180,16 @@ class RemoteClient:
             if optional:
                 log.debug(f"Optional endpoint {endpoint} unavailable [{error_code}]: {error_msg}")
                 return {}
-            log.error(f"Server error [{error_code}]: {error_msg}")
-            raise SystemExit(1)
+            raise RemoteError(f"Server error [{error_code}]: {error_msg}", error_code)
         return detail
 
+    def timed_out(self, endpoint: str) -> RemoteError:
+        return RemoteError(
+            f"Request to remote server timed out after {self.timeout}s: {endpoint}. "
+            "Raise 'timeout' for this server in remote_services.",
+            "TIMEOUT",
+        )
+
     def post(self, endpoint: str, data: Dict[str, Any], expect: tuple[str, ...] = ()) -> Dict[str, Any]:
         return self.request("post", endpoint, data, expect=expect)
 
@@ -732,6 +752,94 @@ def resolve_server(server_name: Optional[str]) -> tuple[str, str, dict]:
     raise click.ClickException(f"Multiple remote services configured. Use --server to select one: {available}")
 
 
+REMOTE_LIST_TTL = 300.0
+REMOTE_LIST_FAILURE_TTL = 30.0
+_remote_lists: Dict[Optional[str], tuple[float, Optional[list[dict]]]] = {}
+_remote_list_locks: Dict[Optional[str], Lock] = {}
+
+
+def remote_client(server_name: Optional[str]) -> RemoteClient:
+    """A RemoteClient for one ``remote_services`` server, with its auth headers and timeout."""
+    url, api_key, services_config = resolve_server(server_name)
+    return RemoteClient(url, api_key, services_config.get("_auth_headers"), services_config.get("_timeout"))
+
+
+def remote_service_list(server_name: Optional[str]) -> Optional[list[dict]]:
+    """The services a ``remote_services`` server offers, cached for ``REMOTE_LIST_TTL``; None when it is unreachable.
+
+    A failure is cached for ``REMOTE_LIST_FAILURE_TTL``, so a server that is down does not stall every caller
+    for its timeout, and a blip does not hide its services for long. One caller fetches; the rest wait for it.
+    """
+
+    def cached() -> Optional[tuple[float, Optional[list[dict]]]]:
+        hit = _remote_lists.get(server_name)
+        ttl = REMOTE_LIST_FAILURE_TTL if hit and hit[1] is None else REMOTE_LIST_TTL
+        return hit if hit and time.monotonic() - hit[0] < ttl else None
+
+    if hit := cached():
+        return hit[1]
+    with _remote_list_locks.setdefault(server_name, Lock()):
+        if hit := cached():
+            return hit[1]
+        services: Optional[list[dict]] = None
+        try:
+            services = remote_client(server_name).get("/api/services").get("services") or []
+        except (Exception, SystemExit) as e:
+            log.warning(f"Could not list the services of remote server '{server_name}': {e}")
+        _remote_lists[server_name] = (time.monotonic(), services)
+        return services
+
+
+def reset_remote_lists() -> None:
+    """Forget every cached remote service list, for a config reload."""
+    _remote_lists.clear()
+
+
+def find_remote_service(services: list[dict], name: str) -> Optional[dict]:
+    """The remote service ``name`` means: a tag in any case first, then an alias, as local get_tag resolves one."""
+    lowered = name.lower()
+    return next((svc for svc in services if str(svc.get("tag", "")).lower() == lowered), None) or next(
+        (svc for svc in services if lowered in (str(a).lower() for a in svc.get("aliases") or [])), None
+    )
+
+
+def remote_service_params(info: dict, data: Dict[str, Any]) -> Dict[str, Any]:
+    """The service options to send: ``data["service_params"]``, else the ``data`` keys naming a cli option.
+
+    Only the remote service's cli option names pass, either way: RemoteService merges them into the
+    session-create body, where a key such as ``service`` or ``proxy`` would override the checked value.
+    """
+    names = {p.get("name") for p in info.get("cli_params") or [] if p.get("kind") == "option"}
+    names -= {"title", "service", "title_id", "profile", "proxy", "no_proxy"}
+    source = data.get("service_params") if isinstance(data.get("service_params"), dict) else data
+    return {k: v for k, v in source.items() if k in names and v is not None and v is not False}
+
+
+def make_remote_service(
+    ctx: click.Context,
+    server_name: Optional[str],
+    service_tag: str,
+    title_id: str,
+    data: Dict[str, Any],
+    apply_overrides: bool = True,
+) -> "RemoteService":
+    """A RemoteService for one ``remote_services`` server, set up the way ``dl --remote`` sets one up.
+
+    ``apply_overrides`` False keeps the per-service ``remote_services.<name>.services`` overrides out of this
+    process's global config.
+    """
+    url, api_key, services_config = resolve_server(server_name)
+    # resolve_server hands back the live config dict; concurrent requests for other tags must not share it
+    services_config = dict(services_config)
+    if not apply_overrides:
+        services_config.pop(service_tag, None)
+    info = next((s for s in remote_service_list(server_name) or [] if s.get("tag") == service_tag), {})
+    services_config["_server_accounts"] = info.get("server_accounts")
+    services_config["_client_config"] = info.get("client_config")
+    params = remote_service_params(info, data)
+    return RemoteService(ctx, service_tag, title_id, url, api_key, services_config, service_params=params)
+
+
 def load_credentials_for_transport(service_tag: str, profile: Optional[str]) -> Optional[Dict[str, str]]:
     from unshackle.commands.dl import dl
 
@@ -768,7 +876,7 @@ def resolve_proxy_arg(proxy_arg: Optional[str]) -> Optional[str]:
         raise click.ClickException(str(e))
 
 
-def resolve_remote_proxy_arg(proxy_arg: str) -> Optional[str]:
+def resolve_remote_proxy_arg(proxy_arg: str, providers: Optional[list] = None) -> Optional[str]:
     """
     Get the proxy URI to send to a server for a --proxy value.
 
@@ -779,7 +887,8 @@ def resolve_remote_proxy_arg(proxy_arg: str) -> Optional[str]:
     from unshackle.core.proxies.resolve import find_provider, initialize_proxy_providers, resolve_proxy
 
     try:
-        providers = initialize_proxy_providers()
+        # the providers dl or a serve already loaded, which a serve leaves empty for a key without server_proxy
+        providers = initialize_proxy_providers() if providers is None else providers
         provider, _, query = proxy_arg.partition(":")
         if provider.lower() == "controld" and query:
             controld = find_provider(providers, "controld")
@@ -944,6 +1053,10 @@ class RemoteService:
         self._segment_filters: Dict[str, Callable[[Any], bool]] = {}
         self._sample_decrypt_tracks: set[str] = set()
         self._log_seq = 0
+        self._recent_logs: deque[str] = deque(maxlen=3)
+        self._input_bridge: Any = None  # a serve relaying for a REST client sets one, as it does for a Service
+        # what a login waiting on the user (a pair code, an approval link) last showed; None otherwise
+        self.auth_notice: Optional[str] = None
         self._log_drain_lock = Lock()
         self._keepalive_stop = Event()
 
@@ -1048,11 +1161,12 @@ class RemoteService:
             except Exception:
                 pass
 
+        providers = getattr(self.ctx.obj, "proxy_providers", None)
         if not no_proxy and not proxy and self._server_accounts and not self._server_accounts.get("global"):
             regions = [str(r).lower() for r in self._server_accounts.get("regions") or []]
             if regions and client_region and client_region not in regions:
                 try:
-                    create_data["proxy"] = resolve_remote_proxy_arg(regions[0])
+                    create_data["proxy"] = resolve_remote_proxy_arg(regions[0], providers)
                     create_data["proxy_region"] = regions[0]
                     self.log.info(f"Using a '{regions[0]}' proxy to match a server account ({', '.join(regions)})")
                 except click.ClickException:
@@ -1062,7 +1176,7 @@ class RemoteService:
                         "rejects the remote session"
                     )
         elif not no_proxy and proxy:
-            resolved_proxy = resolve_remote_proxy_arg(proxy)
+            resolved_proxy = resolve_remote_proxy_arg(proxy, providers)
             if resolved_proxy:
                 create_data["proxy"] = resolved_proxy
                 query = (self.ctx.parent.params.get("proxy_query") if self.ctx.parent else None) or proxy
@@ -1157,7 +1271,9 @@ class RemoteService:
         from unshackle.core import __code_hash__, __version__
 
         secrets = [v for v in (create_data.get("credentials") or {}).values() if isinstance(v, str) and len(v) >= 4]
-        argv = (redact_path(redact_text(shlex.join(sys.argv[1:]), secrets)) or "")[:3000]
+        # a serve relaying for a REST client names itself; its own argv is just the serve command
+        client_argv = self.ctx.parent.params.get("client_argv") if self.ctx.parent else None
+        argv = (redact_path(redact_text(client_argv or shlex.join(sys.argv[1:]), secrets)) or "")[:3000]
         create_data["client"] = {
             "version": __version__,
             "code_hash": __code_hash__ or None,
@@ -1274,6 +1390,7 @@ class RemoteService:
             for record in resp.get("logs", []):
                 level = level_names.get(str(record.get("level", "INFO")), logging.INFO)
                 self.log.log(level, str(record.get("message", "")))
+                self._recent_logs.append(str(record.get("message", "")).strip())
             self._log_seq = int(resp.get("last_seq", self._log_seq))
 
     def poll_auth_completion(self, poll_interval: float = 2.0, timeout: float = 600.0) -> None:
@@ -1288,20 +1405,28 @@ class RemoteService:
             self.poll_auth(deadline, poll_interval)
         finally:
             self.close_cdm_sessions()
+            if self.auth_notice is not None:
+                self.auth_notice = None
+                show_notice(None)
 
     def poll_auth(self, deadline: float, poll_interval: float) -> None:
         while time.monotonic() < deadline:
             resp = self.client.get(f"/api/session/{self._session_id}/prompt")
             status = resp.get("status")
+            seen = self._log_seq
             self.drain_server_logs()
 
             if status == "authenticated":
                 return
 
+            if status == "authenticating" and self._log_seq == seen and self._recent_logs:
+                notice = "\n".join(self._recent_logs)
+                if notice != self.auth_notice:
+                    self.auth_notice = notice
+                    show_notice(notice)
+
             if status == "failed":
-                error = resp.get("error", "Authentication failed on server")
-                log.error(f"Remote auth failed: {error}")
-                raise SystemExit(1)
+                raise RemoteError(f"Remote auth failed: {resp.get('error', 'Authentication failed on server')}", "AUTH")
 
             if status == "pending_input" and isinstance(resp.get("cdm_call"), dict):
                 self.answer_cdm_call(resp["cdm_call"], self._cdm_sessions)
@@ -1309,11 +1434,11 @@ class RemoteService:
 
             if status == "pending_input":
                 prompt = resp.get("prompt", "Enter input: ")
+                bridge = self._input_bridge
                 try:
-                    user_response = prompt_user(prompt)
+                    user_response = bridge.request_input(prompt) if bridge else prompt_user(prompt)
                 except EOFError:
-                    log.error("Remote auth needs interactive input but stdin is closed")
-                    raise SystemExit(1) from None
+                    raise RemoteError("Remote auth needs interactive input but stdin is closed", "AUTH") from None
                 self.client.post(
                     f"/api/session/{self._session_id}/prompt",
                     {"response": user_response},
@@ -1322,8 +1447,7 @@ class RemoteService:
 
             time.sleep(poll_interval)
 
-        log.error("Remote authentication timed out")
-        raise SystemExit(1)
+        raise RemoteError("Remote authentication timed out", "AUTH")
 
     def answer_cdm_call(self, call: Dict[str, Any], sessions: Dict[str, bytes]) -> None:
         """Run one of the server's CDM calls on this machine's device and post the answer."""
@@ -1785,6 +1909,8 @@ class RemoteService:
         pass
 
     def close(self) -> None:
+        # a long-lived serve opens many of these; the atexit hook would keep every one alive
+        atexit.unregister(self.close)
         self._keepalive_stop.set()
         session_id, self._session_id = self._session_id, None
         if session_id:

unshackle/core/services.py

@@ -351,8 +351,6 @@ def check_load_errors() -> None:
 class Services(click.Group):
     """Lazy-loaded command group of project services."""
 
-    remote_services_cache: list[dict] | None = None
-
     def format_help(self, ctx: click.Context, formatter: click.HelpFormatter) -> None:
         """Name the commands panel ``Services``, because this group's commands are services."""
         config = getattr(formatter, "config", None)
@@ -446,24 +444,12 @@ class Services(click.Group):
 
     @staticmethod
     def fetch_remote_services(ctx: click.Context) -> list[dict] | None:
-        """Fetch the service list from the remote server (cached per process)."""
-        if Services.remote_services_cache is not None:
-            return Services.remote_services_cache
-        try:
-            from unshackle.core.remote_service import RemoteClient, resolve_server
+        """Fetch the service list from the remote server (cached); None when it is unreachable."""
+        from unshackle.core.remote_service import remote_service_list, resolve_server
 
-            server_name = ctx.params.get("server")
-            server_url, api_key, services_config = resolve_server(server_name)
-            client = RemoteClient(
-                server_url, api_key, services_config.get("_auth_headers"), services_config.get("_timeout")
-            )
-            result = client.get("/api/services")
-            Services.remote_services_cache = result.get("services", [])
-            return Services.remote_services_cache
-        except click.ClickException:
-            raise
-        except Exception:
-            return None
+        server_name = ctx.params.get("server")
+        resolve_server(server_name)  # a bad --server raises its ClickException, not "unreachable"
+        return remote_service_list(server_name)
 
     @staticmethod
     def make_remote_command(name: str, ctx: click.Context) -> click.Command:
@@ -478,12 +464,9 @@ class Services(click.Group):
             # Resolve against the remote server's tags and aliases, not the local tables: the
             # two drift apart as soon as either side's services change. Tags win over aliases,
             # as in local get_tag.
-            lowered = name.lower()
-            svc_info = next((svc for svc in services if str(svc.get("tag", "")).lower() == lowered), None)
-            if svc_info is None:
-                svc_info = next(
-                    (svc for svc in services if lowered in (str(a).lower() for a in svc.get("aliases") or [])), None
-                )
+            from unshackle.core.remote_service import find_remote_service
+
+            svc_info = find_remote_service(services, name)
             if svc_info is None:
                 if services:
                     available = ", ".join(sorted(svc["tag"] for svc in services))
@@ -507,14 +490,10 @@ class Services(click.Group):
         @click.argument("title", type=str, required=bool(title_arg.get("required", True)) if title_arg else True)
         @click.pass_context
         def remote_cli(ctx: click.Context, title: str, **kwargs: object) -> object:
-            from unshackle.core.remote_service import RemoteService, resolve_server
+            from unshackle.core.remote_service import make_remote_service
 
             server_name = ctx.parent.params.get("server") if ctx.parent else None
-            server_url, api_key, services_config = resolve_server(server_name)
-            services_config["_server_accounts"] = svc_info.get("server_accounts") if svc_info else None
-            services_config["_client_config"] = svc_info.get("client_config") if svc_info else None
-            service_params = {k: v for k, v in kwargs.items() if v is not None and v is not False}
-            return RemoteService(ctx, tag, title, server_url, api_key, services_config, service_params=service_params)
+            return make_remote_service(ctx, server_name, tag, title, {"service_params": kwargs})
 
         if cli_params:
             for param in cli_params: