fix(remote): forward dl track selection and profile to remote sessions

Remote sessions now carry -l/-vl/-al/-a/-fs under a nested dl_params, so services fetch the requested audio versions. Server-account tokens for non-default accounts move from default to the account name, so each account logs in once after upgrade.

docs/dev/rest-api/endpoints.md

@@ -143,6 +143,7 @@ Find titles in a service by query string. The service must have a `search()` met
 | `profile` | string | no | `null` | Credential/cookie profile to use. |
 | `proxy` | string | no | `null` | Full proxy URI, or a country code when the API key has `server_proxy`. |
 | `no_proxy` | boolean | no | `false` | Force-disable all proxy use. |
+| `credentials`, `cookies`, `cache` | - | no | `null` | Your own login material, on a `--remote-only` server only. See [Client cache on a remote-only server](#client-cache-on-a-remote-only-server). |
 
 === "Request"
 
@@ -192,9 +193,34 @@ Get the list of titles behind a title ID, for example episodes or a movie, witho
 | `proxy` | string | no | Full proxy URI, or a country code when the API key has `server_proxy`. |
 | `no_proxy` | boolean | no | Force-disable proxy. |
 | `cdm_type` | string | no | Preferred CDM type. |
+| `credentials`, `cookies`, `cache` | - | no | Your own login material, on a `--remote-only` server only. See [Client cache on a remote-only server](#client-cache-on-a-remote-only-server). |
 
 You can also pass service-specific CLI options as additional keys.
 
+#### Client cache on a remote-only server
+
+A `--remote-only` server keeps nothing that a client sends. When the request
+does not use a server account, the server runs `search`, `list-titles` and
+`list-tracks` on a new service cache directory for that request only, and
+removes the directory when the request ends, on success and on error. A
+`--remote-only` server exposes only `search` of these three endpoints, so the
+rule applies to `list-titles` and `list-tracks` only if a later release exposes
+them in that mode.
+
+`credentials` (`{username, password, extra?}`) and `cookies` (base64 of a
+zlib-compressed Netscape cookie file) are the same as on
+[`POST /api/session/create`](#post-apisessioncreate). `cache` is a map of cache
+files in the same form: each cache key is the file path relative to the service cache
+directory, with `/` separators and no `.json` suffix, and each value is base64
+of the zlib-compressed file. The server writes these files into the request
+cache directory before the service authenticates.
+
+When you send `credentials`, `cookies` or `cache`, a successful response also
+has a `cache` field in the same form, with the updated files (for example a
+refreshed token), but no `titles_` files. Save them to send with the next
+request. A request that uses a server account, or a request to a full-mode
+server, uses the server's own cache and returns no `cache` field.
+
 === "Request"
 
     ```bash
@@ -278,6 +304,8 @@ Show the video, audio, and subtitle tracks for a title. For series, you can scop
 | `episode` | int/string | no | Episode number (combined with `season`). |
 | `part` | int/string | no | Part index of a split episode (combined with `season` and `episode`). |
 | `profile`, `proxy`, `no_proxy`, `cdm_type` | - | no | As on `list-titles`. |
+| `credentials`, `cookies`, `cache` | - | no | As on `list-titles`. |
+| `dl_params` | object | no | The `dl` track selection that the service reads from `ctx.parent.params`: `lang`, `v_lang`, `a_lang`, `acodec` and `forced_subs`, as on [`POST /api/session/create`](#post-apisessioncreate). Some services select the manifests they fetch from these values. |
 
 When you give both `season` and `episode`, the server combines them into a `"{season}x{episode}"` selector. `part` narrows that to `"{season}x{episode}.{part}"`, one part of a [split episode](../creating-a-service.md#split-episodes). Without `part` you get every part of the episode. The server reads `part` only when you give both `season` and `episode`, and `wanted` takes precedence over all three. To scope a part by range, put it in `wanted` instead (`"S01E01.2"`).
 
@@ -857,7 +885,9 @@ DELETE /api/session/{id}        → tear down, harvest updated cache
 
 ### `POST /api/session/create`
 
-Make a remote session for a service and title. Authentication runs asynchronously in the background. This call returns immediately with `status: "authenticating"`, and you then poll the prompt endpoint. The body accepts `service` and `title_id` (both required). It also accepts a broad set of optional keys, because the body allows `additionalProperties`. These are `credentials` (`{username, password, extra?}`), `cookies` (base64 of zlib-compressed Netscape cookie file), `proxy`, `no_proxy`, `profile`, `cache` (map of forwarded cache files, keyed by the file path relative to the service cache directory with `/` separators and no `.json` suffix, so a nested key such as `session_web/<sha1>` lands in a subdirectory), `client_region`, `proxy_region`, `cdm_type`, `range_`, `vcodec`, `quality`, `best_available`, `client`, and any service CLI options. `proxy_region` is the country code the client resolved `proxy` from; the server matches it against its own accounts. `client` is a freeform object the dashboard shows as sent (the CLI puts `version`, `code_hash`, `platform` and a redacted `argv` in it); the server ignores it above 4096 bytes of JSON.
+Make a remote session for a service and title. Authentication runs asynchronously in the background. This call returns immediately with `status: "authenticating"`, and you then poll the prompt endpoint. The body accepts `service` and `title_id` (both required). It also accepts a broad set of optional keys, because the body allows `additionalProperties`. These are `credentials` (`{username, password, extra?}`), `cookies` (base64 of zlib-compressed Netscape cookie file), `proxy`, `no_proxy`, `profile`, `cache` (map of forwarded cache files, keyed by the file path relative to the service cache directory with `/` separators and no `.json` suffix, so a nested key such as `session_web/<sha1>` lands in a subdirectory), `client_region`, `proxy_region`, `cdm_type`, `range_`, `vcodec`, `quality`, `best_available`, `dl_params`, `client`, and any service CLI options. `proxy_region` is the country code the client resolved `proxy` from; the server matches it against its own accounts. `client` is a freeform object the dashboard shows as sent (the CLI puts `version`, `code_hash`, `platform` and a redacted `argv` in it); the server ignores it above 4096 bytes of JSON.
+
+`dl_params` is an object with the `dl` track selection: `lang`, `v_lang` and `a_lang` (arrays of language strings), `acodec` (an array of audio codec names or `dl -a` aliases such as `ddp`) and `forced_subs` (boolean). The service reads these values from `ctx.parent.params`. The server uses the `dl` default for each absent or malformed value, and drops an unknown codec name.
 
 Service CLI options also travel in a nested `service_params` object, which wins over a flat key with the same name. `profile` at the top level always means the credentials profile, never a service's own `--profile` option.
 

docs/dev/rest-api/remote-sessions.md

@@ -124,6 +124,9 @@ hood the client walks a remote session through its lifecycle.
       the client region itself
     - Track-selection hints (`range_`, `vcodec`, `quality`, `best_available`) so
       the server fetches the right manifests
+    - Your language and audio codec selection (`-l`, `-vl`, `-al`, `-a`, `-fs`)
+      in a `dl_params` object. Some services fetch one manifest
+      for each audio language, and pick the languages from these values
     - Your local per-service **cache files** (e.g. refreshed tokens), including
       files in subdirectories of the service cache directory, but only the files
       for the active profile. The client withholds a file whose path, in any
@@ -207,6 +210,9 @@ hood the client walks a remote session through its lifecycle.
     authentication. The login belongs to the client when the client sent
     cookies, credentials, or cache files, or answered a login prompt (a device code
     or an OTP) that led to a successful login. A server-account login never does.
+    The client saves only the returned files that pass the same profile check it
+    applies before it sends them, so a file for another profile cannot overwrite
+    that profile's local file.
 
 !!! tip "Renaming remote titles locally"
     You can rename titles for a remote service you do not have installed locally by
@@ -349,6 +355,28 @@ shows as sent), and the track-selection hints `range_`, `vcodec`,
 `quality`, `best_available`, plus arbitrary service CLI options
 (`additionalProperties: true`).
 
+`dl_params` holds the `dl` track selection: `lang`, `v_lang` and `a_lang`
+(arrays of language strings), `acodec` (an array of audio codec names, such as
+`EC3`), and `forced_subs` (boolean). The server gives these values to the
+service in `ctx.parent.params`, never as service options, so a service option
+with the same name, such as an own `--lang`, does not get them. The server uses
+the `dl` default for each absent or malformed value: `lang` is `["orig"]`,
+`v_lang`, `a_lang` and `acodec` are `[]`, and `forced_subs` is `false`. The
+server drops an unknown codec name. An empty array is a valid value and stays
+empty.
+
+The server does not forward the `dl` mode flags: `video_only`, `audio_only`,
+`subs_only`, `chapters_only`, `list_`, `skip_dl`, `sub_format` and
+`no_attachments`. A service that reads one of them to skip work fetches more on
+the server, but the result is the same.
+
+In a client-login session (not a server account), the server gives the service
+the client's `profile` in `ctx.parent.params`, the same as a local run with that
+profile. The profile name keys the service's token cache, so the cache files
+the client sends and gets back belong to that profile. A service can also use
+the name to select per-profile entries in the server's service config, such as
+a per-profile ESN.
+
 The response returns **before** authentication finishes:
 
 ```json

tests/remote/unit/test_download_manager.py

@@ -182,3 +182,45 @@ def test_worker_write_result_retries_denied_replace(tmp_path: Path, monkeypatch:
     download_worker.write_result(target, {"progress": 42.0})
     assert json.loads(target.read_text(encoding="utf-8")) == {"progress": 42.0}
     assert len(attempts) == 2
+
+
+class _CapturedCtx(Exception):
+    pass
+
+
+@pytest.mark.parametrize(
+    ("params", "expected"),
+    [
+        ({}, (["orig"], [], [], [], False)),
+        (
+            {"lang": ["es-419"], "v_lang": ["en"], "a_lang": ["ja"], "acodec": "AAC,EC3", "forced_subs": True},
+            (["es-419"], ["en"], ["ja"], ["AAC", "EC3"], True),
+        ),
+    ],
+)
+def test_perform_download_puts_the_lang_selection_in_the_service_ctx(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, params: dict, expected: tuple
+) -> None:
+    """Services read the language selection from ctx.parent.params, not from dl.result()."""
+    import click
+
+    import unshackle.commands.dl as dl_module
+    from unshackle.core.api import download_manager, handlers
+    from unshackle.core.services import Services
+
+    class FakeDl:
+        cli = click.Command("dl")
+
+        def __init__(self, ctx: click.Context, **kwargs: object) -> None:
+            raise _CapturedCtx(ctx)
+
+    monkeypatch.setattr(dl_module, "dl", FakeDl)
+    monkeypatch.setattr(Services, "get_path", staticmethod(lambda s: tmp_path))
+    monkeypatch.setattr(handlers, "load_full_cdm", lambda *a: None)
+
+    with pytest.raises(_CapturedCtx) as exc:
+        download_manager.perform_download("job-1", "EXAMPLE", "t1", dict(params))
+
+    ctx = exc.value.args[0]
+    p = ctx.params
+    assert (p["lang"], p["v_lang"], p["a_lang"], [c.name for c in p["acodec"]], p["forced_subs"]) == expected

tests/remote/unit/test_remote_service_helpers.py

@@ -410,6 +410,27 @@ def test_save_returned_cache_writes_nested_keys_and_rejects_escapes(tmp_path, mo
     ]
 
 
+def test_save_returned_cache_drops_other_profiles_files(tmp_path, monkeypatch) -> None:
+    """A returned file for another profile must not overwrite that profile's local file."""
+    import base64
+    import logging
+    import zlib
+    from types import SimpleNamespace
+
+    from unshackle.core.config import config
+    from unshackle.core.remote_service import RemoteService
+
+    tag = "TESTSVC"
+    monkeypatch.setattr(config.directories, "cache", tmp_path)
+    monkeypatch.setattr(config, "credentials", {tag: {"default": "a@example.com:pw", "work": "b@example.com:pw"}})
+    stub = SimpleNamespace(service_tag=tag, log=logging.getLogger("test"))
+    blob = base64.b64encode(zlib.compress(b"x")).decode("ascii")
+
+    RemoteService.save_returned_cache(stub, {"tokens_work": blob, "tokens_default": blob, "tokens": blob}, "work")
+
+    assert sorted(p.name for p in (tmp_path / tag).iterdir()) == ["tokens.json", "tokens_work.json"]
+
+
 def test_save_returned_cache_ignores_oversized_payload(tmp_path, monkeypatch) -> None:
     import logging
     from types import SimpleNamespace
@@ -536,3 +557,54 @@ def test_session_create_writes_nested_cache_and_skips_colliding_keys(tmp_path, m
 
     written = {p.name: p.read_text() for p in tmp_path.rglob("*.json") if p.is_file()}
     assert written == {"x.json": "flat", "keys.v2.json": "dotted"}
+
+
+class _SentCreate(Exception):
+    pass
+
+
+def _sent_create_data(parent_params: dict, service_params: dict) -> dict:
+    import logging
+    from types import SimpleNamespace
+
+    from unshackle.core.remote_service import RemoteService
+
+    def post(path: str, data: dict) -> dict:
+        raise _SentCreate(data)
+
+    svc = object.__new__(RemoteService)
+    svc.ctx = SimpleNamespace(parent=SimpleNamespace(params={"no_proxy": True, **parent_params}), obj=None)
+    svc.service_tag = "EXAMPLE"
+    svc.title_id = "t1"
+    svc._server_accounts = {"global": True}
+    svc._server_cdm = True
+    svc._service_params = service_params
+    svc.log = logging.getLogger("test")
+    svc.client = SimpleNamespace(post=post)
+    with pytest.raises(_SentCreate) as exc:
+        svc.authenticate()
+    return exc.value.args[0]
+
+
+def test_dl_selection_travels_nested_and_a_service_option_cannot_override_it() -> None:
+    parent_params = {
+        "lang": ["orig", "es-419"],
+        "v_lang": [],
+        "a_lang": [],
+        "acodec": [Audio.Codec.EC3],
+        "forced_subs": True,
+    }
+    data = _sent_create_data(parent_params, {"lang": 3, "a_lang": "x"})
+    assert data["dl_params"] == {
+        "lang": ["orig", "es-419"],
+        "v_lang": [],
+        "a_lang": [],
+        "acodec": ["EC3"],
+        "forced_subs": True,
+    }
+    assert data["lang"] == 3
+    assert data["service_params"] == {"lang": 3, "a_lang": "x"}
+
+
+def test_no_dl_params_when_the_ctx_has_no_selection() -> None:
+    assert "dl_params" not in _sent_create_data({}, {})

tests/remote/unit/test_request_cache.py

@@ -0,0 +1,173 @@
+"""List and search requests on a --remote-only server run on a per-request cache directory.
+
+The directory holds only what the client sent, goes back to the client when it logged in
+itself, and is always removed when the request ends.
+"""
+
+import base64
+import json
+import zlib
+
+import click
+import pytest
+
+from unshackle.core.api import handlers
+from unshackle.core.api.stats import stats
+from unshackle.core.cacher import Cacher
+from unshackle.core.config import config
+
+
+class _CacheService:
+    fail_titles = False
+
+    def __init__(self, ctx, title):
+        self.seen = _CacheService.seen
+        self.seen["parent_params"] = ctx.parent.params
+        self.cache = Cacher("EXAMPLE")
+
+    def authenticate(self, cookies, credential):
+        self.seen["seeded"] = self.cache.get("tokens").data
+        self.seen["cache_dir"] = self.cache.get("tokens").path.parent
+        self.cache.get("tokens").set("new")
+        self.cache.get("titles_abc").set("titles")
+
+    def get_titles(self):
+        if self.fail_titles:
+            raise RuntimeError("boom")
+        return []
+
+    def search(self):
+        return iter(())
+
+
+@pytest.fixture
+def seen(tmp_path, monkeypatch):
+    import types
+
+    _CacheService.seen = {}
+    _CacheService.fail_titles = False
+
+    @click.command()
+    @click.argument("title")
+    @click.pass_context
+    def cli(ctx, **kwargs):
+        return _CacheService(ctx, **kwargs)
+
+    module = types.SimpleNamespace(cli=cli)
+    monkeypatch.setattr(config.directories, "cache", tmp_path)
+    monkeypatch.setattr(stats, "mode", "remote_only")
+    monkeypatch.setattr(handlers, "validate_service", lambda tag, request=None: tag)
+    monkeypatch.setattr(handlers, "load_service_yaml", lambda service: {})
+    monkeypatch.setattr(handlers, "resolve_handler_proxy", lambda *args: (None, []))
+    monkeypatch.setattr(handlers, "resolve_server_account", lambda *args: None)
+    monkeypatch.setattr(handlers, "load_full_cdm", lambda *args: None)
+    monkeypatch.setattr(handlers.Services, "load", staticmethod(lambda service: module))
+    monkeypatch.setattr("unshackle.commands.dl.dl.get_cookie_jar", staticmethod(lambda service, profile: None))
+    monkeypatch.setattr("unshackle.commands.dl.dl.get_credentials", staticmethod(lambda service, profile: None))
+    return _CacheService.seen
+
+
+def client_cache(tmp_path, value):
+    Cacher("SRC").get("tokens").set(value)
+    text = (tmp_path / "SRC" / "tokens.json").read_bytes()
+    return {"tokens": base64.b64encode(zlib.compress(text)).decode("ascii")}
+
+
+def request_dirs(tmp_path):
+    return list((tmp_path / "_requests").rglob("*")) if (tmp_path / "_requests").exists() else []
+
+
+CREDENTIALS = {"credentials": {"username": "a@example.com", "password": "pw"}}
+
+
+@pytest.mark.asyncio
+async def test_seeded_cache_is_read_and_returned_then_removed(seen, tmp_path):
+    data = {"service": "EXAMPLE", "title_id": "t", "cache": client_cache(tmp_path, "old")}
+    response = await handlers.list_titles_handler(data)
+    body = json.loads(response.body)
+
+    assert seen["seeded"] == "old"
+    assert seen["cache_dir"].is_relative_to(tmp_path / "_requests")
+    assert set(body["cache"]) == {"tokens"}
+    returned = json.loads(zlib.decompress(base64.b64decode(body["cache"]["tokens"])))
+    assert returned["data"] == "new"
+    assert not seen["cache_dir"].exists()
+    assert not request_dirs(tmp_path)
+
+
+@pytest.mark.asyncio
+async def test_search_returns_cache_to_a_client_with_credentials(seen, tmp_path):
+    response = await handlers.search_handler({"service": "EXAMPLE", "query": "q", **CREDENTIALS})
+    body = json.loads(response.body)
+    assert body["count"] == 0
+    assert set(body["cache"]) == {"tokens"}
+    assert not request_dirs(tmp_path)
+
+
+@pytest.mark.asyncio
+async def test_no_cache_returned_without_client_login(seen, tmp_path):
+    response = await handlers.search_handler({"service": "EXAMPLE", "query": "q"})
+    assert "cache" not in json.loads(response.body)
+    assert seen["cache_dir"].is_relative_to(tmp_path / "_requests")
+    assert not request_dirs(tmp_path)
+
+
+@pytest.mark.asyncio
+async def test_cache_dir_removed_on_error(seen, tmp_path):
+    _CacheService.fail_titles = True
+    data = {"service": "EXAMPLE", "title_id": "t", **CREDENTIALS}
+    response = await handlers.list_tracks_handler(data)
+    assert response.status >= 400
+    assert "cache" not in json.loads(response.body)
+    assert seen["cache_dir"].is_relative_to(tmp_path / "_requests")
+    assert not request_dirs(tmp_path)
+
+
+@pytest.mark.asyncio
+async def test_invalid_client_cache_still_removes_the_dir(seen, tmp_path):
+    data = {"service": "EXAMPLE", "query": "q", "cache": {"tokens": 1}}
+    with pytest.raises(handlers.APIError):
+        await handlers.search_handler(data)
+    assert not request_dirs(tmp_path)
+
+
+@pytest.mark.asyncio
+async def test_server_account_uses_its_own_cache_and_returns_none(seen, tmp_path, monkeypatch):
+    monkeypatch.setattr(handlers, "resolve_server_account", lambda *args: "acct")
+    monkeypatch.setattr(handlers, "server_account_cookies", lambda *args: None)
+    data = {"service": "EXAMPLE", "query": "q", "cache": client_cache(tmp_path, "old"), **CREDENTIALS}
+    response = await handlers.search_handler(data)
+
+    assert "cache" not in json.loads(response.body)
+    assert not seen["seeded"]
+    assert seen["cache_dir"] == tmp_path / "_accounts" / "EXAMPLE" / "acct"
+    assert not (tmp_path / "_requests").exists()
+
+
+@pytest.mark.asyncio
+async def test_full_mode_keeps_the_server_cache(seen, tmp_path, monkeypatch):
+    monkeypatch.setattr(stats, "mode", "full")
+    data = {"service": "EXAMPLE", "query": "q", "cache": client_cache(tmp_path, "old"), **CREDENTIALS}
+    response = await handlers.search_handler(data)
+
+    assert "cache" not in json.loads(response.body)
+    assert not seen["seeded"]
+    assert seen["cache_dir"] == tmp_path / "EXAMPLE"
+    assert not (tmp_path / "_requests").exists()
+
+
+@pytest.mark.asyncio
+async def test_list_tracks_service_sees_dl_params(seen):
+    from unshackle.core.tracks import Audio
+
+    data = {"service": "EXAMPLE", "title_id": "t", "dl_params": {"a_lang": ["es-419"], "acodec": ["ddp"]}}
+    await handlers.list_tracks_handler(data)
+    assert seen["parent_params"]["a_lang"] == ["es-419"]
+    assert seen["parent_params"]["acodec"] == [Audio.Codec.EC3]
+    assert "dl_params" in handlers.LIST_HANDLER_TRANSPORT_KEYS
+
+
+@pytest.mark.asyncio
+async def test_search_service_gets_no_dl_params(seen):
+    await handlers.search_handler({"service": "EXAMPLE", "query": "q", "dl_params": {"a_lang": ["es-419"]}})
+    assert "a_lang" not in seen["parent_params"]

tests/remote/unit/test_server_accounts.py

@@ -133,3 +133,90 @@ def test_server_cdm_ignores_client_cdm_type(monkeypatch):
     handlers.create_service_instance("OTHER", "t", data, None, [], None, server_cdm=True)
     handlers.create_service_instance("OTHER", "t", data, None, [], None)
     assert seen == [None, "playready"]
+
+
+class _LangOptionService:
+    def __init__(self, ctx, title, lang):
+        self.parent_params = ctx.parent.params
+        self.lang = lang
+
+
+def _lang_option_module():
+    import click
+
+    @click.command()
+    @click.argument("title")
+    @click.option("--lang", type=int, default=None)
+    @click.pass_context
+    def cli(ctx, **kwargs):
+        return _LangOptionService(ctx, **kwargs)
+
+    return type("Module", (), {"cli": cli})
+
+
+def _create_with_real_ctx(monkeypatch, data):
+    monkeypatch.setattr(handlers, "load_service_yaml", lambda s: {})
+    monkeypatch.setattr(handlers, "load_full_cdm", lambda *a: None)
+    monkeypatch.setattr(handlers.Services, "load", staticmethod(lambda s: _lang_option_module()))
+    svc, _, _ = handlers.create_service_instance("OTHER", "t", data, None, [], None)
+    return svc
+
+
+def test_nested_dl_params_reach_the_parent_ctx(monkeypatch):
+    from unshackle.core.tracks import Audio
+
+    data = {
+        "dl_params": {
+            "lang": ["es-419", "en"],
+            "v_lang": ["ja"],
+            "a_lang": ["es-419"],
+            "acodec": ["EC3", "NOPE", "AAC"],
+            "forced_subs": True,
+        }
+    }
+    svc = _create_with_real_ctx(monkeypatch, data)
+    assert svc.parent_params["lang"] == ["es-419", "en"]
+    assert svc.parent_params["v_lang"] == ["ja"]
+    assert svc.parent_params["a_lang"] == ["es-419"]
+    assert svc.parent_params["acodec"] == [Audio.Codec.EC3, Audio.Codec.AAC]
+    assert svc.parent_params["forced_subs"] is True
+
+
+def test_absent_dl_params_fall_back_to_the_dl_defaults(monkeypatch):
+    svc = _create_with_real_ctx(monkeypatch, {})
+    selection = {k: svc.parent_params[k] for k in ("lang", "v_lang", "a_lang", "acodec", "forced_subs")}
+    assert selection == {"lang": ["orig"], "v_lang": [], "a_lang": [], "acodec": [], "forced_subs": False}
+
+
+def test_an_empty_lang_list_is_kept(monkeypatch):
+    svc = _create_with_real_ctx(monkeypatch, {"dl_params": {"lang": [], "a_lang": []}})
+    assert svc.parent_params["lang"] == []
+
+
+def test_a_service_lang_option_never_gets_the_dl_lang(monkeypatch):
+    svc = _create_with_real_ctx(monkeypatch, {"dl_params": {"lang": ["en"]}})
+    assert svc.lang is None
+    assert svc.parent_params["lang"] == ["en"]
+    assert "dl_params" in handlers.SESSION_TRANSPORT_KEYS
+
+
+@pytest.mark.parametrize(
+    "dl_params",
+    [
+        "en",
+        {"lang": "en", "v_lang": "ja", "a_lang": {"x": 1}, "acodec": "EC3", "forced_subs": "yes"},
+        {"lang": [1], "v_lang": [None], "a_lang": ["en", None], "acodec": [3], "forced_subs": 1},
+    ],
+)
+def test_malformed_dl_params_are_ignored(monkeypatch, dl_params):
+    svc = _create_with_real_ctx(monkeypatch, {"dl_params": dl_params})
+    selection = {k: svc.parent_params[k] for k in ("lang", "v_lang", "a_lang", "acodec", "forced_subs")}
+    assert selection == {"lang": ["orig"], "v_lang": [], "a_lang": [], "acodec": [], "forced_subs": False}
+
+
+def test_profile_reaches_the_parent_ctx_params(monkeypatch):
+    monkeypatch.setattr(handlers, "load_service_yaml", lambda s: {})
+    monkeypatch.setattr(handlers, "load_full_cdm", lambda *a: None)
+    monkeypatch.setattr(handlers.Services, "load", staticmethod(lambda s: _lang_option_module()))
+    svc, _, _ = handlers.create_service_instance("OTHER", "t", {}, None, [], "work")
+    assert svc.parent_params["profile"] == "work"

unshackle/core/api/download_manager.py

@@ -18,7 +18,7 @@ from typing import Any, Callable, Dict, List, Optional
 
 from unshackle.core.api.events import bus, publish_service_event
 from unshackle.core.api.sanitize import sanitize_log
-from unshackle.core.utils.click_types import VIDEO_CODEC_LIST
+from unshackle.core.utils.click_types import AUDIO_CODEC_LIST, VIDEO_CODEC_LIST
 from unshackle.core.utils.redact import REDACTED, URL_USERINFO_RE, redact_path, redact_text
 
 log = logging.getLogger("download_manager")
@@ -392,6 +392,9 @@ def perform_download(
     else:
         params["vcodec"] = []
 
+    if params.get("acodec"):
+        params["acodec"] = AUDIO_CODEC_LIST.convert(params["acodec"])
+
     range_raw = params.get("range")
     if range_raw:
         if isinstance(range_raw, str):
@@ -474,6 +477,12 @@ def perform_download(
         "vcodec": params.get("vcodec", []),
         "range_": params.get("range", [Video.Range.SDR]),
         "best_available": params.get("best_available", False),
+        # services read these here, not from dl.result(), to pick the manifests they fetch
+        "acodec": params.get("acodec") or [],
+        "lang": params.get("lang", ["orig"]),
+        "v_lang": params.get("v_lang", []),
+        "a_lang": params.get("a_lang", []),
+        "forced_subs": params.get("forced_subs", False),
     }
     # Hand-built context: record parameter sources so service dl overrides
     # apply to defaults but never clobber client-sent values.
@@ -567,7 +576,7 @@ def perform_download(
                 service=service_instance,
                 quality=params.get("quality", []),
                 vcodec=params.get("vcodec", []),
-                acodec=params.get("acodec"),
+                acodec=ctx.params["acodec"],
                 vbitrate=params.get("vbitrate"),
                 abitrate=params.get("abitrate"),
                 vbitrate_range=params.get("vbitrate_range"),
@@ -578,14 +587,14 @@ def perform_download(
                 select_titles=False,
                 wanted=params.get("wanted", []),
                 latest_episode=params.get("latest_episode", False),
-                lang=params.get("lang", ["orig"]),
-                v_lang=params.get("v_lang", []),
-                a_lang=params.get("a_lang", []),
+                lang=ctx.params["lang"],
+                v_lang=ctx.params["v_lang"],
+                a_lang=ctx.params["a_lang"],
                 s_lang=params.get("s_lang", ["all"]),
                 require_audio=params.get("require_audio", []),
                 require_video=params.get("require_video", []),
                 require_subs=params.get("require_subs", []),
-                forced_subs=params.get("forced_subs", False),
+                forced_subs=ctx.params["forced_subs"],
                 forced_s_lang=params.get("forced_s_lang", []),
                 exact_lang=params.get("exact_lang", False),
                 sub_format=params.get("sub_format"),

unshackle/core/api/handlers.py

@@ -13,7 +13,7 @@ from contextlib import suppress
 from datetime import date as date_
 from http.cookiejar import CookieJar, MozillaCookieJar
 from pathlib import Path
-from typing import Any, Callable, Dict, Iterable, List, Optional
+from typing import Any, Callable, Dict, Iterable, List, Optional, cast
 
 import click
 from aiohttp import web
@@ -124,6 +124,8 @@ LIST_HANDLER_TRANSPORT_KEYS = {
     "no_proxy",
     "query",
     "service_params",
+    "cache",
+    "dl_params",
 }
 
 
@@ -286,12 +288,119 @@ def server_login_material(
     return load_client_cookies(data.get("cookies")), credential
 
 
+def api_key_namespace(request: Optional[web.Request]) -> str:
+    """Name of the caller's cache directories, derived from its API key so no caller can guess another's."""
+    import hashlib
+
+    api_key = request.headers.get("X-Secret-Key", "anonymous") if request else "anonymous"
+    return hashlib.pbkdf2_hmac("sha256", api_key.encode(), b"unshackle-session-ns", 100_000).hex()[:12]
+
+
+def write_client_cache(cache_data: Any, cache_tag: str) -> None:
+    """Write a client-sent ``cache`` map into the cache directory ``cache_tag``.
+
+    Each cache key is a file path relative to that directory, and each value is the file as
+    base64 of zlib-compressed bytes. The function skips an unsafe key or a file it cannot write,
+    and raises INVALID_INPUT for a map that is not an object, has too many entries, or holds a
+    value that is not a string.
+    """
+    if not isinstance(cache_data, dict) or len(cache_data) > MAX_SESSION_CACHE_KEYS:
+        raise APIError(APIErrorCode.INVALID_INPUT, f"cache must hold at most {MAX_SESSION_CACHE_KEYS} entries")
+    cache_dir = config.directories.cache / cache_tag
+    cache_dir.mkdir(parents=True, exist_ok=True)
+    for key, content in cache_data.items():
+        safe_name = safe_cache_key(key)
+        if not safe_name:
+            log.warning(f"Rejecting unsafe session cache key: {sanitize_log(key)}")
+            continue
+        if not isinstance(content, str):
+            raise APIError(APIErrorCode.INVALID_INPUT, "cache values must be base64 strings")
+        decompressed = safe_inflate(base64.b64decode(content)).decode("utf-8")
+        target = cache_dir / f"{safe_name}.json"
+        try:
+            target.parent.mkdir(parents=True, exist_ok=True)
+            target.write_text(decompressed, encoding="utf-8")
+        except OSError as e:
+            log.warning(f"Skipping session cache key {sanitize_log(key)}: {e}")
+
+
+def collect_cache_files(cache_tag: str) -> Dict[str, str]:
+    """Read the cache directory ``cache_tag`` back as a ``cache`` map for the client.
+
+    The map has the same form that :func:`write_client_cache` reads. It leaves out the
+    ``titles_`` files, because they hold no login state.
+    """
+    cache_data: Dict[str, str] = {}
+    cache_dir = config.directories.cache / cache_tag
+    if cache_dir.is_dir():
+        for f in sorted(cache_dir.rglob("*.json")):
+            key = f.relative_to(cache_dir).as_posix()[: -len(".json")]
+            if f.stem.startswith("titles_") or not safe_cache_key(key):
+                continue
+            try:
+                cache_data[key] = base64.b64encode(zlib.compress(f.read_bytes())).decode("ascii")
+            except OSError:
+                pass
+    return cache_data
+
+
+class RequestCache:
+    """The service cache of one list or search request on a ``--remote-only`` server.
+
+    Such a server keeps nothing that a client sends, so the request runs on a cache directory
+    of its own that the handler removes when the request ends, on success and on error. Before
+    that, a client that logged in with its own cookies, credentials or cache gets the updated
+    files back, the same as when a remote session ends.
+    """
+
+    def __init__(self) -> None:
+        self.tag: Optional[str] = None
+        self.client_auth = False
+
+    def attach(
+        self,
+        service_instance: Any,
+        data: Dict[str, Any],
+        normalized_service: str,
+        request: Optional[web.Request],
+        client_login: bool,
+    ) -> None:
+        """Give the service a new cache directory, seeded from the client's ``cache``.
+
+        Does nothing on a full-mode server, where the operator's own cache stays in use.
+        """
+        import uuid
+
+        from unshackle.core.api.stats import stats
+
+        if stats.mode != "remote_only":
+            return
+        self.tag = f"_requests/{api_key_namespace(request)}/{uuid.uuid4()}/{normalized_service}"
+        service_instance.cache = Cacher(self.tag)
+        cache_data = data.get("cache")
+        if cache_data:
+            write_client_cache(cache_data, self.tag)
+        self.client_auth = client_login or bool(cache_data)
+
+    def json_response(self, payload: Dict[str, Any]) -> web.Response:
+        """Answer with ``payload``, plus the updated ``cache`` for a client that logged in itself."""
+        if self.tag and self.client_auth:
+            cache_data = collect_cache_files(self.tag)
+            if cache_data:
+                payload["cache"] = cache_data
+        return web.json_response(payload)
+
+    def cleanup(self) -> None:
+        SessionStore.cleanup_cache_dir(self.tag)
+
+
 def setup_list_service(
     data: Dict[str, Any],
     normalized_service: str,
     profile: Optional[str],
     title_id: str,
     request: Optional[web.Request] = None,
+    request_cache: Optional[RequestCache] = None,
 ) -> Any:
     """Assemble and authenticate a service instance for list_titles / list_tracks.
 
@@ -318,13 +427,16 @@ def setup_list_service(
         no_proxy,
         proxy_providers,
         service_config,
-        extra_params={"cookies_supplied": cookies is not None},
+        extra_params={"cookies_supplied": cookies is not None, **forwarded_dl_params(data)},
     )
     service_module = Services.load(normalized_service)
     service_instance = instantiate_service(parent_ctx, service_module, title_id, data, LIST_HANDLER_TRANSPORT_KEYS)
 
     if account:
         service_instance.cache = Cacher(f"_accounts/{normalized_service}/{account}")
+    elif request_cache is not None:
+        client_login = cookies is not None or credential is not None
+        request_cache.attach(service_instance, data, normalized_service, request, client_login)
     service_instance.authenticate(cookies, credential)
     return service_instance
 
@@ -334,6 +446,7 @@ def run_service_search(
     normalized_service: str,
     query: str,
     request: Optional[web.Request] = None,
+    request_cache: Optional[RequestCache] = None,
 ) -> List[Dict[str, Any]]:
     """Assemble and authenticate a service instance, then run its search.
 
@@ -381,6 +494,9 @@ def run_service_search(
 
     if account:
         service_instance.cache = Cacher(f"_accounts/{normalized_service}/{account}")
+    elif request_cache is not None:
+        client_login = cookies is not None or credential is not None
+        request_cache.attach(service_instance, data, normalized_service, request, client_login)
     service_instance.authenticate(cookies, credential)
 
     results: List[Dict[str, Any]] = []
@@ -1297,9 +1413,12 @@ async def search_handler(data: Dict[str, Any], request: Optional[web.Request] =
             details={"service": service_tag},
         )
 
-    results = await asyncio.to_thread(run_service_search, data, normalized_service, query, request)
-
-    return web.json_response({"results": results, "count": len(results)})
+    request_cache = RequestCache()
+    try:
+        results = await asyncio.to_thread(run_service_search, data, normalized_service, query, request, request_cache)
+        return request_cache.json_response({"results": results, "count": len(results)})
+    finally:
+        request_cache.cleanup()
 
 
 async def list_titles_handler(data: Dict[str, Any], request: Optional[web.Request] = None) -> web.Response:
@@ -1317,9 +1436,10 @@ async def list_titles_handler(data: Dict[str, Any], request: Optional[web.Reques
             details={"service": service_tag},
         )
 
+    request_cache = RequestCache()
     try:
         service_instance = await asyncio.to_thread(
-            setup_list_service, data, normalized_service, profile, title_id, request
+            setup_list_service, data, normalized_service, profile, title_id, request, request_cache
         )
         titles = await asyncio.to_thread(service_instance.get_titles)
 
@@ -1328,7 +1448,7 @@ async def list_titles_handler(data: Dict[str, Any], request: Optional[web.Reques
         else:
             title_list = [stamp_service_flags(serialize_title(titles), service_instance)]
 
-        return web.json_response({"titles": title_list})
+        return request_cache.json_response({"titles": title_list})
 
     except APIError:
         raise
@@ -1340,6 +1460,8 @@ async def list_titles_handler(data: Dict[str, Any], request: Optional[web.Reques
             context={"operation": "list_titles", "service": normalized_service, "title_id": title_id},
             debug_mode=debug_mode,
         )
+    finally:
+        request_cache.cleanup()
 
 
 async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Request] = None) -> web.Response:
@@ -1357,9 +1479,10 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques
             details={"service": service_tag},
         )
 
+    request_cache = RequestCache()
     try:
         service_instance = await asyncio.to_thread(
-            setup_list_service, data, normalized_service, profile, title_id, request
+            setup_list_service, data, normalized_service, profile, title_id, request, request_cache
         )
         titles = await asyncio.to_thread(service_instance.get_titles)
 
@@ -1469,7 +1592,7 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques
                         response = {"episodes": episodes_data}
                         if failed_episodes:
                             response["unavailable_episodes"] = failed_episodes
-                        return web.json_response(response)
+                        return request_cache.json_response(response)
                     else:
                         raise APIError(
                             APIErrorCode.NO_CONTENT,
@@ -1500,7 +1623,7 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques
             "subtitles": [serialize_subtitle_track(t) for t in tracks.subtitles],
         }
 
-        return web.json_response(response)
+        return request_cache.json_response(response)
 
     except APIError:
         raise
@@ -1512,6 +1635,8 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques
             context={"operation": "list_tracks", "service": normalized_service, "title_id": title_id},
             debug_mode=debug_mode,
         )
+    finally:
+        request_cache.cleanup()
 
 
 VALID_VCODECS = [choice.upper() for choice in VIDEO_CODEC_LIST.choices]
@@ -2844,9 +2969,47 @@ SESSION_TRANSPORT_KEYS = {
     "vcodec",
     "quality",
     "best_available",
+    "dl_params",
 }
 
 
+def forwarded_dl_params(data: Dict[str, Any]) -> Dict[str, Any]:
+    """Read the dl track selection that a remote-dl client sends under ``dl_params``.
+
+    Services read these values from ``ctx.parent.params`` to pick the manifests they
+    fetch. They travel in their own object so that they never collide with a service option that
+    has the same name. An absent or malformed value gets the dl default, so an old client that
+    sends no ``dl_params`` gets the same result as a ``dl`` run with no selection flags.
+    ``acodec`` holds codec names or aliases, which become ``Audio.Codec`` members the same way
+    ``dl -a`` converts them; an unknown name is dropped.
+    """
+    raw = data.get("dl_params")
+    if not isinstance(raw, dict):
+        raw = {}
+
+    def str_list(key: str) -> Optional[List[str]]:
+        value = raw.get(key)
+        if isinstance(value, list) and all(isinstance(v, str) for v in value):
+            return list(value)
+        if value is not None:
+            log.warning(f"Ignoring dl_params.{key}: it must be an array of strings")
+        return None
+
+    params: Dict[str, Any] = {}
+    for key in ("lang", "v_lang", "a_lang"):
+        langs = str_list(key)
+        params[key] = langs if langs is not None else list(cast(List[str], DEFAULT_DOWNLOAD_PARAMS[key]))
+    # the dl CLI default is [], and some services test membership in it
+    params["acodec"] = []
+    for name in str_list("acodec") or []:
+        try:
+            params["acodec"].extend(AUDIO_CODEC_LIST.convert(name))
+        except click.BadParameter:
+            log.warning(f"Ignoring unknown dl_params.acodec value: {sanitize_log(name)}")
+    params["forced_subs"] = raw.get("forced_subs") is True
+    return params
+
+
 def create_service_instance(
     normalized_service: str,
     title_id: str,
@@ -2891,6 +3054,7 @@ def create_service_instance(
         "vcodec": vcodec_values,
         "quality": data.get("quality"),
         "best_available": data.get("best_available", False),
+        **forwarded_dl_params(data),
     }
 
     if server_account:
@@ -2908,6 +3072,9 @@ def create_service_instance(
         cookies = load_client_cookies(data.get("cookies"))
 
     extra_params["cookies_supplied"] = cookies is not None
+    # Services key their token caches on this. Only sessions get it: they run on a cache
+    # directory of their own, where the list and search handlers share the server's cache.
+    extra_params["profile"] = profile
 
     parent_ctx = build_parent_ctx(
         profile,
@@ -2954,15 +3121,11 @@ async def session_create_handler(data: Dict[str, Any], request: Optional[web.Req
     try:
         proxy_param, proxy_providers = await asyncio.to_thread(resolve_handler_proxy, data, normalized_service, request)
 
-        import hashlib
         import uuid as uuid_mod
 
-        from unshackle.core.config import config as app_config
-
         session_id = str(uuid_mod.uuid4())
         api_key = request.headers.get("X-Secret-Key", "anonymous") if request else "anonymous"
-        api_key_hash = hashlib.pbkdf2_hmac("sha256", api_key.encode(), b"unshackle-session-ns", 100_000).hex()[:12]
-        session_cache_dir = f"_sessions/{api_key_hash}/{session_id}/{normalized_service}"
+        session_cache_dir = f"_sessions/{api_key_namespace(request)}/{session_id}/{normalized_service}"
         session_cache_tag: Optional[str] = session_cache_dir
 
         server_account = server_account_for(request, normalized_service)
@@ -3003,26 +3166,7 @@ async def session_create_handler(data: Dict[str, Any], request: Optional[web.Req
 
         cache_data = data.get("cache", {})
         if cache_data and session_cache_tag:
-            if not isinstance(cache_data, dict) or len(cache_data) > MAX_SESSION_CACHE_KEYS:
-                raise APIError(APIErrorCode.INVALID_INPUT, f"cache must hold at most {MAX_SESSION_CACHE_KEYS} entries")
-            import base64
-
-            cache_dir = app_config.directories.cache / session_cache_tag
-            cache_dir.mkdir(parents=True, exist_ok=True)
-            for key, content in cache_data.items():
-                safe_name = safe_cache_key(key)
-                if not safe_name:
-                    log.warning(f"Rejecting unsafe session cache key: {sanitize_log(key)}")
-                    continue
-                if not isinstance(content, str):
-                    raise APIError(APIErrorCode.INVALID_INPUT, "cache values must be base64 strings")
-                decompressed = safe_inflate(base64.b64decode(content)).decode("utf-8")
-                target = cache_dir / f"{safe_name}.json"
-                try:
-                    target.parent.mkdir(parents=True, exist_ok=True)
-                    target.write_text(decompressed, encoding="utf-8")
-                except OSError as e:
-                    log.warning(f"Skipping session cache key {sanitize_log(key)}: {e}")
+            write_client_cache(cache_data, session_cache_tag)
 
         bridge = InputBridge()
         service_instance._input_bridge = bridge
@@ -4510,11 +4654,7 @@ async def session_info_handler(session_id: str, request: Optional[web.Request] =
 
 async def session_delete_handler(session_id: str, request: Optional[web.Request] = None) -> web.Response:
     """Delete a remote session, return updated cache files, and clean up server-side data."""
-    import base64
-    import zlib
-
     from unshackle.core.api.session_store import get_session_store
-    from unshackle.core.config import config as app_config
 
     session = await get_validated_session(session_id, request)
     store = get_session_store()
@@ -4523,18 +4663,7 @@ async def session_delete_handler(session_id: str, request: Optional[web.Request]
         session.input_bridge.cancel()
 
     cache_tag = session.cache_tag
-    cache_data: Dict[str, str] = {}
-    if cache_tag and session.client_auth:
-        cache_dir = app_config.directories.cache / cache_tag
-        if cache_dir.is_dir():
-            for f in sorted(cache_dir.rglob("*.json")):
-                key = f.relative_to(cache_dir).as_posix()[: -len(".json")]
-                if f.stem.startswith("titles_") or not safe_cache_key(key):
-                    continue
-                try:
-                    cache_data[key] = base64.b64encode(zlib.compress(f.read_bytes())).decode("ascii")
-                except OSError:
-                    pass
+    cache_data = collect_cache_files(cache_tag) if cache_tag and session.client_auth else {}
 
     await store.delete(session_id)
 

unshackle/core/api/routes.py

@@ -408,6 +408,21 @@ async def search(request: web.Request) -> web.Response:
               no_proxy:
                 type: boolean
                 description: Force disable all proxy use (default - false)
+              credentials:
+                type: object
+                additionalProperties: true
+                description: Your own login, on a --remote-only server only ({username, password, extra?})
+              cookies:
+                type: string
+                description: Your own cookies, on a --remote-only server only (base64 of a zlib-compressed Netscape cookie file)
+              cache:
+                type: object
+                additionalProperties:
+                  type: string
+                description: |
+                  Your own cache files, on a --remote-only server only: cache key to base64 of the
+                  zlib-compressed file. The request runs on a cache directory of its own, which the
+                  server removes when the request ends.
     responses:
       '200':
         description: Search results
@@ -439,6 +454,14 @@ async def search(request: web.Request) -> web.Response:
                 count:
                   type: integer
                   description: Number of results returned
+                cache:
+                  type: object
+                  additionalProperties:
+                    type: string
+                  description: |
+                    The updated cache files, in the same form as the request field. Only on a
+                    --remote-only server, without a server account, when you sent credentials,
+                    cookies or cache.
       '400':
         description: Invalid request
     """
@@ -485,9 +508,41 @@ async def list_titles(request: web.Request) -> web.Response:
               title_id:
                 type: string
                 description: Title identifier
+              credentials:
+                type: object
+                additionalProperties: true
+                description: Your own login, on a --remote-only server only ({username, password, extra?})
+              cookies:
+                type: string
+                description: Your own cookies, on a --remote-only server only (base64 of a zlib-compressed Netscape cookie file)
+              cache:
+                type: object
+                additionalProperties:
+                  type: string
+                description: |
+                  Your own cache files, on a --remote-only server only: cache key to base64 of the
+                  zlib-compressed file. The request runs on a cache directory of its own, which the
+                  server removes when the request ends.
     responses:
       '200':
         description: List of titles
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                titles:
+                  type: array
+                  items:
+                    type: object
+                cache:
+                  type: object
+                  additionalProperties:
+                    type: string
+                  description: |
+                    The updated cache files, in the same form as the request field. Only on a
+                    --remote-only server, without a server account, when you sent credentials,
+                    cookies or cache.
       '400':
         description: Invalid request (missing parameters, invalid service)
         content:
@@ -614,9 +669,43 @@ async def list_tracks(request: web.Request) -> web.Response:
               proxy:
                 type: string
                 description: Proxy configuration (optional)
+              dl_params:
+                type: object
+                description: |
+                  The dl track selection (lang, v_lang, a_lang, acodec, forced_subs), in the same
+                  form as on `/api/session/create`. The service reads it from `ctx.parent.params`.
+              credentials:
+                type: object
+                additionalProperties: true
+                description: Your own login, on a --remote-only server only ({username, password, extra?})
+              cookies:
+                type: string
+                description: Your own cookies, on a --remote-only server only (base64 of a zlib-compressed Netscape cookie file)
+              cache:
+                type: object
+                additionalProperties:
+                  type: string
+                description: |
+                  Your own cache files, on a --remote-only server only: cache key to base64 of the
+                  zlib-compressed file. The request runs on a cache directory of its own, which the
+                  server removes when the request ends.
     responses:
       '200':
         description: Track information
+        content:
+          application/json:
+            schema:
+              type: object
+              additionalProperties: true
+              properties:
+                cache:
+                  type: object
+                  additionalProperties:
+                    type: string
+                  description: |
+                    The updated cache files, in the same form as the request field. Only on a
+                    --remote-only server, without a server account, when you sent credentials,
+                    cookies or cache.
       '400':
         description: Invalid request
     """
@@ -1571,6 +1660,37 @@ async def session_create(request: web.Request) -> web.Response:
               proxy_region:
                 type: string
                 description: Two-letter country the client resolved its proxy for; picks a server account
+              dl_params:
+                type: object
+                description: |
+                  The dl track selection. The service reads it from ctx.parent.params, never as
+                  service options. An absent or malformed value gets the dl default, and the
+                  server drops an unknown codec name.
+                properties:
+                  lang:
+                    type: array
+                    items:
+                      type: string
+                    default: ["orig"]
+                  v_lang:
+                    type: array
+                    items:
+                      type: string
+                    default: []
+                  acodec:
+                    type: array
+                    items:
+                      type: string
+                    default: []
+                    description: Audio codec names, such as EC3
+                  a_lang:
+                    type: array
+                    items:
+                      type: string
+                    default: []
+                  forced_subs:
+                    type: boolean
+                    default: false
               client:
                 type: object
                 additionalProperties: true

unshackle/core/remote_service.py

@@ -731,6 +731,29 @@ def cache_stem_is_relevant(
     return not matched_foreign
 
 
+def cache_key_filter(service_tag: str, profile: Optional[str]) -> Callable[[str], bool]:
+    """Return a check for the cache keys that belong to the active credential or profile.
+
+    The check runs :func:`cache_stem_is_relevant` on every path segment of a cache key,
+    because a service may embed the credential digest or profile name in a directory name
+    as well as in the basename. A profile with no credentials entry uses the default one.
+    """
+    from unshackle.commands.dl import dl
+
+    credential = dl.get_credentials(service_tag, profile)
+    allowed = credential_cache_digests(credential) if credential else set()
+    active = profile or "default"
+    profiles = config.credentials.get(service_tag)
+    if isinstance(profiles, dict) and profile and profile not in profiles:
+        active = "default"
+    foreign = set(profiles) - {active} if isinstance(profiles, dict) else set()
+
+    def relevant(key: str) -> bool:
+        return all(cache_stem_is_relevant(part, allowed, active, foreign) for part in key.split("/"))
+
+    return relevant
+
+
 class ServerVault(Vault):
     """The server's vaults as one source seen from the client: it can only be reported against.
 
@@ -944,6 +967,21 @@ class RemoteService:
                 create_data["quality"] = list(quality)
             if self.ctx.parent.params.get("best_available"):
                 create_data["best_available"] = True
+            # Services pick the manifests to fetch from these. They go in their own
+            # object: flat, a service option with the same name would take them (or override them).
+            # An empty list is sent as-is, because only an absent key falls back to the dl default.
+            params = self.ctx.parent.params
+            dl_params: dict[str, Any] = {
+                key: [str(x) for x in params[key]]
+                for key in ("lang", "v_lang", "a_lang")
+                if params.get(key) is not None
+            }
+            if params.get("acodec") is not None:
+                dl_params["acodec"] = [c.name for c in params["acodec"]]
+            if "forced_subs" in params:
+                dl_params["forced_subs"] = bool(params["forced_subs"])
+            if dl_params:
+                create_data["dl_params"] = dl_params
 
         if self._service_params:
             create_data.update(self._service_params)
@@ -1400,18 +1438,21 @@ class RemoteService:
         if session_id:
             try:
                 result = self.client.delete(f"/api/session/{session_id}")
-                self.save_returned_cache(result.get("cache", {}))
+                profile = self.ctx.parent.params.get("profile") if self.ctx.parent else None
+                self.save_returned_cache(result.get("cache", {}), profile)
             except SystemExit:
                 pass
             except Exception as e:
                 self.log.warning(f"Failed to clean up remote session: {e}")
 
-    def save_returned_cache(self, cache_data: Dict[str, str]) -> None:
+    def save_returned_cache(self, cache_data: Dict[str, str], profile: Optional[str] = None) -> None:
         """Save cache files returned by the server to the local cache directory.
 
         The server returns updated cache files (e.g. refreshed tokens) on
         session close. Writing them locally means the next remote session
-        can forward them back, skipping interactive auth.
+        can forward them back, skipping interactive auth. The same filter that
+        :meth:`load_cache_files` applies drops a file that belongs to another
+        credential or profile, so it cannot overwrite that profile's local file.
         """
         if not cache_data:
             return
@@ -1425,59 +1466,56 @@ class RemoteService:
 
         cache_dir = config.directories.cache / self.service_tag
         cache_dir.mkdir(parents=True, exist_ok=True)
+        relevant = cache_key_filter(self.service_tag, profile)
 
+        saved = 0
         for key, content in cache_data.items():
             safe_name = safe_cache_key(key)
             if not safe_name:
                 self.log.warning(f"Rejecting unsafe cache filename from server: {key!r}")
                 continue
+            if not relevant(safe_name):
+                self.log.debug(f"Ignoring cache file from the remote server for another profile: {safe_name}")
+                continue
             try:
                 decompressed = safe_inflate(base64.b64decode(content))
                 # Not with_suffix: that would turn the key "keys.v2" into "keys.json".
                 target = cache_dir / f"{safe_name}.json"
                 target.parent.mkdir(parents=True, exist_ok=True)
                 target.write_bytes(decompressed)
+                saved += 1
             except Exception as e:
                 self.log.warning(f"Failed to save returned cache file '{safe_name}': {e}")
 
-        self.log.info(f"Saved {len(cache_data)} cache file(s) from server")
+        self.log.info(f"Saved {saved} cache file(s) from server")
 
     def load_cache_files(self, profile: Optional[str] = None) -> Dict[str, str]:
         """Collect the cache files to forward, and withhold other profiles' files.
 
         The client cannot rely on the server to filter, so it sends only the files
-        it can tie to the active credential or profile, plus service-global state.
-        At worst, a withheld file makes the server authenticate again.
+        it can tie to the active credential or profile (:func:`cache_key_filter`),
+        plus service-global state. At worst, a withheld file makes the server
+        authenticate again.
 
         Each cache key is the file path relative to the service cache directory,
         with posix separators and no ``.json`` suffix, so it equals the ``Cacher``
-        cache key the service reads it with (``session_web/<sha1>``). The relevance check
-        runs on every path segment, because a service may embed the credential
-        digest or profile name in a directory name as well as in the basename.
+        cache key the service reads it with (``session_web/<sha1>``).
         """
         import zlib
 
-        from unshackle.commands.dl import dl
         from unshackle.core.api.sanitize import safe_cache_key
 
         cache_dir = config.directories.cache / self.service_tag
         if not cache_dir.is_dir():
             return {}
 
-        credential = dl.get_credentials(self.service_tag, profile)
-        allowed = credential_cache_digests(credential) if credential else set()
-        active = profile or "default"
-        profiles = config.credentials.get(self.service_tag)
-        if isinstance(profiles, dict) and profile and profile not in profiles:
-            active = "default"
-        foreign = set(profiles) - {active} if isinstance(profiles, dict) else set()
-
+        relevant = cache_key_filter(self.service_tag, profile)
         files: Dict[str, str] = {}
         for f in sorted(cache_dir.rglob("*.json")):
             key = f.relative_to(cache_dir).as_posix()[: -len(".json")]
             if f.stem.startswith("titles_") or not safe_cache_key(key):
                 continue
-            if not all(cache_stem_is_relevant(part, allowed, active, foreign) for part in key.split("/")):
+            if not relevant(key):
                 self.log.debug(f"Withholding cache file from the remote server: {key}")
                 continue
             files[key] = base64.b64encode(zlib.compress(f.read_bytes())).decode("ascii")