feat(serve): relay approval messages to jobs and proxy remote downloads

A REST job shows a wait_for_approval message as its notice field. search, list-titles and list-tracks fail at once on a login that needs the user. A remote client downloads through the --proxy it sends.

docs/dev/creating-a-service.md

@@ -272,7 +272,11 @@ token = self.wait_for_approval(
 It raises `TimeoutError` after `timeout` seconds. Under `serve` mode it also stops
 when the client deletes the remote session, so the loop does not continue on the
 server with no client. The message goes out as a log line, so a remote
-session shows it, but a REST download job does not relay it as `input_prompt`.
+session shows it. A REST download job shows it in its `notice` field until the wait ends.
+
+The wait sees a cancel only between two `check` calls, so `check` must return within a few
+seconds. A `check` that reads a socket keeps the socket open across calls and returns `None`
+after each short idle wait.
 
 !!! warning "Make each token cache entry unique"
     The cache above uses `tokens_{device}_{profile}` on purpose. A token cache

docs/dev/rest-api/endpoints.md

@@ -197,6 +197,13 @@ Get the list of titles behind a title ID, for example episodes or a movie, witho
 
 You can also pass service-specific CLI options as additional keys.
 
+#### Logins that need the user
+
+`search`, `list-titles` and `list-tracks` have no channel to the user. A login that asks for
+input or waits for an approval on another device fails at once with `AUTH_FAILED` on these
+endpoints. Complete that login in a download job or a remote session, then send the saved
+login with the request.
+
 #### Client cache on a remote-only server
 
 A `--remote-only` server keeps nothing that a client sends. When the request
@@ -556,7 +563,8 @@ Show download jobs, with optional filtering and sorting.
           "segments_total": 300,
           "speed": "5.2 MB/s",
           "skipped_subtitles": [],
-          "input_prompt": null
+          "input_prompt": null,
+          "notice": null
         }
       ]
     }
@@ -721,6 +729,10 @@ example an OTP code, a PIN, or a device-code confirmation. The job then shows th
 `input_prompt` field and in its `progress` events. Show the prompt to the user and send the
 answer here.
 
+A job that waits for an approval on another device, such as a pair code login, shows the
+message in its `notice` field instead. A notice needs no answer: show it to the user and keep
+polling.
+
 === "Request"
 
     ```json

docs/dev/rest-api/job-lifecycle.md

@@ -163,6 +163,7 @@ full detail:
 | `speed` | string | Transfer speed of the track in progress, e.g. `4.20 MB/s`. |
 | `skipped_subtitles` | list | Subtitles that were skipped as non-fatal failures (when `skip_subtitle_errors` is set). |
 | `input_prompt` | string or `null` | The prompt the service waits on. Answer it with `POST /api/download/jobs/{job_id}/input`. |
+| `notice` | string or `null` | A message the service shows while it waits for an approval on another device, such as a pair code. It needs no answer. |
 
 The full detail adds `started_time`, `completed_time`, `output_files`, the redacted
 `parameters`, and, for failed jobs, `error_message`, `error_details`, and `error_code`.
@@ -368,6 +369,8 @@ Three temporary files bridge the parent and the job worker:
 The job worker has no terminal. When a service asks the user for input, the job worker writes
 the prompt to `progress.json` as `input_prompt` and waits. The answer from
 `POST /api/download/jobs/{job_id}/input` goes to the job worker on its stdin.
+When a service waits for an approval on another device, the job worker writes the message as
+`notice` and sets it back to `null` when the wait ends.
 
 Running each download out-of-process isolates the server from crashes, memory growth,
 and blocking work inside the download engine, and makes hard cancellation (killing the

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

@@ -127,6 +127,9 @@ hood the client walks a remote session through its lifecycle.
       URI and runs its own forwarder for it. A bare region skips Control D, because its
       proxy is a forwarder on the client. See
       [Control D](../../guide/proxies-and-vpn.md#control-d)
+      The client also downloads playlists and segments through the same proxy,
+      as a local run does. `--no-proxy-download` and `--proxy-download` change
+      that for downloads only.
     - 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`)

docs/dev/service-architecture.md

@@ -91,7 +91,8 @@ These are abstract. A service will not load unless it defines all three:
   bridge, the REST download job relay, or the rich console.
 - **`wait_for_approval(message, check, timeout, interval=3.0)`**: shows a message, then
   calls `check` until it returns a value. Use it for a login that the user approves on
-  another device, such as a pair code.
+  another device, such as a pair code. A remote-dl session shows the message as a log
+  line, and a REST download job shows it as the job's `notice`.
 - **Event callbacks**: `on_segment_downloaded`, `on_track_downloaded`,
   `on_track_decrypted`, `on_track_repacked`, `on_track_multiplex` let a service react to
   pipeline events (for example, MonaLisa decrypts each segment in `on_segment_downloaded`).

tests/core/test_service_wait_for_approval.py

@@ -1,9 +1,11 @@
 import logging
+from contextlib import nullcontext
 from types import SimpleNamespace
 
 import pytest
 
-from unshackle.core.api.input_bridge import InputBridge
+from unshackle.core import console
+from unshackle.core.api.input_bridge import NO_USER_MESSAGE, InputBridge, NoUserBridge
 from unshackle.core.service import Service
 
 
@@ -32,3 +34,45 @@ def test_stops_when_the_remote_session_is_cancelled():
     with pytest.raises(RuntimeError):
         Service.wait_for_approval(_service(bridge), "enter code", check, timeout=5, interval=0)
     assert len(calls) == 1
+
+
+def _raise_value_error():
+    raise ValueError("denied")
+
+
+def test_logs_the_message_for_a_remote_session(caplog):
+    """A remote session reads the message only from the service log."""
+    with caplog.at_level(logging.INFO, logger="test"):
+        Service.wait_for_approval(_service(), "enter code", lambda: "token", timeout=5, interval=0)
+    assert "enter code" in caplog.messages
+
+
+def test_a_request_with_no_user_fails_at_once():
+    calls = []
+    with pytest.raises(RuntimeError, match="needs the user"):
+        Service.wait_for_approval(_service(NoUserBridge()), "enter code", lambda: calls.append(1), timeout=5)
+    assert not calls
+    with pytest.raises(RuntimeError, match="needs the user"):
+        Service.request_input(_service(NoUserBridge()), "Enter OTP")
+
+
+def test_no_user_message_is_an_authentication_error():
+    from unshackle.core.api.errors import APIErrorCode, categorize_exception
+
+    assert categorize_exception(RuntimeError(NO_USER_MESSAGE)).error_code == APIErrorCode.AUTH_FAILED
+
+
+@pytest.mark.parametrize(
+    "check, expectation",
+    [
+        (lambda: "token", nullcontext()),
+        (lambda: None, pytest.raises(TimeoutError)),
+        (_raise_value_error, pytest.raises(ValueError)),
+    ],
+)
+def test_shows_the_message_as_a_notice_and_clears_it(monkeypatch, check, expectation):
+    notices = []
+    monkeypatch.setattr(console, "_notice_handler", notices.append)
+    with expectation:
+        Service.wait_for_approval(_service(), "enter code", check, timeout=0.05, interval=0.01)
+    assert notices == ["enter code", None]

tests/remote/unit/test_download_manager.py

@@ -286,6 +286,30 @@ def test_worker_relays_a_prompt_and_reads_the_answer(monkeypatch: pytest.MonkeyP
     assert updates == [{"input_prompt": "Enter PIN"}, {"input_prompt": None}]
 
 
+def test_worker_relays_a_notice_as_progress(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A show_notice() message reaches the parent as ``notice`` progress, and None clears it."""
+    import os
+    import sys
+    from types import SimpleNamespace
+
+    from unshackle.core.api import download_worker
+    from unshackle.core.console import set_notice_handler, set_prompt_handler, show_notice
+
+    read_fd, write_fd = os.pipe()
+    monkeypatch.setattr(sys, "stdin", SimpleNamespace(fileno=lambda: read_fd))
+    updates: list[dict] = []
+
+    download_worker.relay_prompts(updates.append)
+    try:
+        show_notice("Enter pair code ABCD")
+        show_notice(None)
+    finally:
+        set_prompt_handler(None)
+        set_notice_handler(None)
+        os.close(write_fd)
+    assert updates == [{"notice": "Enter pair code ABCD"}, {"notice": None}]
+
+
 def test_a_prompt_discards_an_answer_sent_before_it(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     """A late answer to an earlier prompt must not answer the next one; the timeout is not retryable."""
     import sys
@@ -420,6 +444,46 @@ async def test_a_repeated_prompt_reaches_an_event_listener(
     assert job.input_prompt is None
 
 
+async def test_a_notice_reaches_the_job_and_an_event_listener(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, manager: DownloadQueueManager
+) -> None:
+    """The job carries the worker's notice while the worker waits, and drops it at the end."""
+    import asyncio
+
+    from unshackle.core.api import download_manager
+
+    worker = tmp_path / "worker.py"
+    # The prompt holds the worker open until the test has seen the notice.
+    source = FAKE_PROMPT_WORKER.replace("import prompt_user", "import prompt_user, show_notice").replace(
+        'answers = [prompt_user("Enter OTP"), prompt_user("Enter OTP")]',
+        'show_notice("Enter pair code ABCD")\nanswers = [prompt_user("Continue")]',
+    )
+    assert source.count("show_notice") == 2, "FAKE_PROMPT_WORKER changed; update the replacements"
+    worker.write_text(source, encoding="utf-8")
+    spawn = asyncio.create_subprocess_exec
+
+    async def spawn_fake_worker(executable: str, *args: str, **kwargs: Any) -> Any:
+        return await spawn(executable, str(worker), *args[2:], **kwargs)
+
+    monkeypatch.setattr(download_manager.asyncio, "create_subprocess_exec", spawn_fake_worker)
+    job = manager.create_job("EXAMPLE", "t")
+    events = manager.subscribe(job.job_id)
+    run = asyncio.create_task(manager.run_download_async(job))
+
+    async def answer_after_notice() -> None:
+        while True:
+            event = await events.get()
+            if event and event["data"]["input_prompt"]:
+                assert event["data"]["notice"] == "Enter pair code ABCD"
+                assert job.to_dict()["notice"] == "Enter pair code ABCD"
+                assert manager.submit_input(job, "ok") is True
+                return
+
+    await asyncio.wait_for(answer_after_notice(), timeout=20)
+    assert await asyncio.wait_for(run, timeout=20) == ["ok"]
+    assert job.notice is None
+
+
 async def test_input_handler_rejects_bad_requests(
     monkeypatch: pytest.MonkeyPatch, manager: DownloadQueueManager
 ) -> None:

tests/remote/unit/test_remote_service_helpers.py

@@ -7,6 +7,7 @@ from enum import Enum
 import pytest
 
 from unshackle.core.remote_service import (
+    RemoteService,
     build_title,
     build_tracks,
     deserialize_audio,
@@ -563,11 +564,12 @@ class _SentCreate(Exception):
     pass
 
 
-def _sent_create_data(parent_params: dict, service_params: dict) -> dict:
+def _remote_service(parent_params: dict, service_params: dict) -> RemoteService:
+    """A RemoteService whose session create raises _SentCreate with the request body."""
     import logging
     from types import SimpleNamespace
 
-    from unshackle.core.remote_service import RemoteService
+    import requests
 
     def post(path: str, data: dict) -> dict:
         raise _SentCreate(data)
@@ -579,11 +581,32 @@ def _sent_create_data(parent_params: dict, service_params: dict) -> dict:
     svc._server_accounts = {"global": True}
     svc._server_cdm = True
     svc._service_params = service_params
+    svc._session = requests.Session()
     svc.log = logging.getLogger("test")
-    svc.client = SimpleNamespace(post=post)
+    svc.client = SimpleNamespace(post=post, server_url="https://server.example")
+    return svc
+
+
+def _sent_create_data(parent_params: dict, service_params: dict) -> dict:
+    with pytest.raises(_SentCreate) as exc:
+        _remote_service(parent_params, service_params).authenticate()
+    return exc.value.args[0]
+
+
+@pytest.mark.parametrize("no_proxy, expected", [(False, {"all": "http://proxy.example:8080"}), (True, {})])
+def test_the_client_session_downloads_through_the_proxy_it_sends(
+    monkeypatch: pytest.MonkeyPatch, no_proxy: bool, expected: dict
+) -> None:
+    """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("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:
         svc.authenticate()
-    return exc.value.args[0]
+    assert dict(svc._session.proxies) == expected
+    assert exc.value.args[0].get("proxy") == (None if no_proxy else "http://proxy.example:8080")
 
 
 def test_dl_selection_travels_nested_and_a_service_option_cannot_override_it() -> None:

unshackle/core/api/download_manager.py

@@ -126,6 +126,8 @@ class DownloadJob:
 
     # Prompt a service raised through prompt_user() and waits on; answered by submit_input().
     input_prompt: Optional[str] = None
+    # Message a service shows through show_notice() while it waits for an approval; needs no answer.
+    notice: Optional[str] = None
 
     # Subtitles skipped under skip_subtitle_errors (non-fatal). Each entry is a dl.SkippedSubtitle
     # dict (id / language / title) so a client can report which weren't available.
@@ -160,6 +162,7 @@ class DownloadJob:
             "speed": self.speed,
             "skipped_subtitles": self.skipped_subtitles,
             "input_prompt": self.input_prompt,
+            "notice": self.notice,
         }
 
         if include_full_details:
@@ -1109,6 +1112,8 @@ class DownloadQueueManager:
                                 job.speed = str(progress_data["speed"])
                             if "input_prompt" in progress_data:
                                 job.input_prompt = progress_data["input_prompt"]
+                            if "notice" in progress_data:
+                                job.notice = progress_data["notice"]
                             if progress_data.get("skipped_subtitles"):
                                 job.skipped_subtitles = progress_data["skipped_subtitles"]
                             if "progress" in progress_data:
@@ -1195,6 +1200,7 @@ class DownloadQueueManager:
 
             self._download_processes.pop(job.job_id, None)
             job.input_prompt = None
+            job.notice = None
 
             temp_paths = self._job_temp_files.pop(job.job_id, {})
             for path in temp_paths.values():

unshackle/core/api/download_worker.py

@@ -16,7 +16,7 @@ from typing import Any, Callable, Dict
 from uuid import uuid4
 
 from unshackle.core.api.input_bridge import AUTH_INPUT_TIMEOUT
-from unshackle.core.console import set_prompt_handler
+from unshackle.core.console import set_notice_handler, set_prompt_handler
 
 from .download_manager import perform_download
 
@@ -64,7 +64,7 @@ def relay_prompts(progress_callback: Callable[[Dict[str, Any]], None]) -> None:
     The parent writes each answer as one JSON string per line, so an answer can hold a newline. The reader
     uses unbuffered stdin: a daemon thread blocked in ``sys.stdin.buffer`` aborts the interpreter at shutdown.
     Prompts go out one at a time, and a prompt discards answers that arrived before it, such as a late
-    answer to a prompt that timed out.
+    answer to a prompt that timed out. A show_notice() message goes out as ``notice`` progress and needs no answer.
     """
     answers: queue.Queue[str] = queue.Queue()
     one_prompt = threading.Lock()
@@ -92,6 +92,7 @@ def relay_prompts(progress_callback: Callable[[Dict[str, Any]], None]) -> None:
                 progress_callback({"input_prompt": None})
 
     set_prompt_handler(ask)
+    set_notice_handler(lambda message: progress_callback({"notice": message}))
 
 
 def main(argv: list[str]) -> int:

unshackle/core/api/handlers.py

@@ -20,7 +20,7 @@ from aiohttp import web
 
 from unshackle.core.api.compression import safe_inflate
 from unshackle.core.api.errors import APIError, APIErrorCode, categorize_exception, handle_api_exception
-from unshackle.core.api.input_bridge import AuthStatus, InputBridge
+from unshackle.core.api.input_bridge import AuthStatus, InputBridge, NoUserBridge
 from unshackle.core.api.sanitize import MAX_SESSION_CACHE_KEYS, safe_cache_key, sanitize_log
 from unshackle.core.api.session_log import SessionLogBuffer, SessionLogMirror, capture_service_logs
 from unshackle.core.api.session_store import SessionStore
@@ -443,6 +443,7 @@ def setup_list_service(
     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._input_bridge = NoUserBridge()
     service_instance.authenticate(cookies, credential)
     return service_instance
 
@@ -503,6 +504,7 @@ def run_service_search(
     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._input_bridge = NoUserBridge()
     service_instance.authenticate(cookies, credential)
 
     results: List[Dict[str, Any]] = []

unshackle/core/api/input_bridge.py

@@ -185,11 +185,11 @@ class InputBridge:
         self._picked_up.set()
         self._response_ready.set()
 
-    @property
+    def ensure_active(self) -> None:
-    def cancelled(self) -> bool:
+        """Raise RuntimeError once the server cancelled the bridge, so a login loop stops."""
-        """``True`` once the server cancelled the bridge, so a login loop must stop."""
         with self._lock:
-            return self._cancelled
+            if self._cancelled:
+                raise RuntimeError("Session was cancelled")
 
     @property
     def status(self) -> AuthStatus:
@@ -200,3 +200,19 @@ class InputBridge:
     def status(self, value: AuthStatus) -> None:
         with self._lock:
             self._status = value
+
+
+NO_USER_MESSAGE = (
+    "This login needs the user, and this request cannot reach the user. "
+    "Start a download job or a remote session to complete the authentication"
+)
+
+
+class NoUserBridge(InputBridge):
+    """Bridge for a request with no channel to the user: a login step that needs the user fails at once."""
+
+    def request_input(self, prompt: str, timeout: float = AUTH_INPUT_TIMEOUT) -> str:
+        raise RuntimeError(NO_USER_MESSAGE)
+
+    def ensure_active(self) -> None:
+        raise RuntimeError(NO_USER_MESSAGE)

unshackle/core/api/routes.py

@@ -1375,7 +1375,8 @@ async def download_job_input(request: web.Request) -> web.Response:
     description: >
       Submit the answer to the prompt a running job waits on (an OTP code, a PIN, a device-code
       confirmation). The job shows the prompt in its `input_prompt` field and in its SSE `progress`
-      events. The job fails if nobody answers within the prompt timeout.
+      events. The job fails if nobody answers within the prompt timeout. A job that waits for an
+      approval on another device shows the message in its `notice` field, which takes no answer.
     parameters:
       - name: job_id
         in: path

unshackle/core/console.py

@@ -490,6 +490,21 @@ def prompt_user(prompt: str) -> str:
     return console.input(body)
 
 
+_notice_handler: Optional[Callable[[Optional[str]], None]] = None
+
+
+def set_notice_handler(handler: Optional[Callable[[Optional[str]], None]]) -> None:
+    """Route every :func:`show_notice` call to *handler*, for a process whose log output no user reads."""
+    global _notice_handler
+    _notice_handler = handler
+
+
+def show_notice(message: Optional[str]) -> None:
+    """Give *message* to the notice handler, or None to clear it. Without a handler this does nothing."""
+    if _notice_handler is not None:
+        _notice_handler(message)
+
+
 __all__ = (
     "ComfyLogRenderer",
     "ComfyRichHandler",
@@ -501,5 +516,7 @@ __all__ = (
     "listing_table",
     "print_wide",
     "prompt_user",
+    "set_notice_handler",
     "set_prompt_handler",
+    "show_notice",
 )

unshackle/core/remote_service.py

@@ -1020,6 +1020,9 @@ class RemoteService:
         if create_data.get("proxy") and is_loopback(create_data["proxy"]) and not is_loopback(self.client.server_url):
             # the local CLI resolves the proxy, so a local-only URI is useless to a server elsewhere
             raise click.ClickException("That proxy is on your machine, so --remote cannot use it.")
+        local_proxy = None if no_proxy else (proxy or create_data.get("proxy"))
+        if local_proxy:
+            self._session.proxies["all"] = local_proxy
 
         if not no_proxy and not proxy and client_region:
             create_data["client_region"] = client_region

unshackle/core/service.py

@@ -19,7 +19,7 @@ from rich.rule import Rule
 
 from unshackle.core.cacher import Cacher
 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 DRM_T
@@ -489,20 +489,25 @@ class Service(metaclass=ABCMeta):
         """Show ``message``, then call ``check`` until it returns a value.
 
         Use it for a login that the user approves on another device, such as a pair code.
-        ``check`` returns None while the approval is pending. Raises TimeoutError after
+        ``check`` returns None while the approval is pending, and must return within a few
-        ``timeout`` seconds, and RuntimeError when the remote session no longer exists.
+        seconds. Raises TimeoutError after ``timeout`` seconds, and RuntimeError when the
+        remote session no longer exists. A REST download job shows ``message`` as its ``notice``.
         """
         self.log.info(message)
+        show_notice(message)
         deadline = time.monotonic() + timeout
-        while True:
+        try:
-            if self._input_bridge is not None and self._input_bridge.cancelled:
+            while True:
-                raise RuntimeError("Session was cancelled")
+                if self._input_bridge is not None:
-            result = check()
+                    self._input_bridge.ensure_active()
-            if result is not None:
+                result = check()
-                return result
+                if result is not None:
-            if time.monotonic() + interval >= deadline:
+                    return result
-                raise TimeoutError(f"No approval within {timeout:.0f}s")
+                if time.monotonic() + interval >= deadline:
-            time.sleep(interval)
+                    raise TimeoutError(f"No approval within {timeout:.0f}s")
+                time.sleep(interval)
+        finally:
+            show_notice(None)
 
     def search(self) -> Generator[SearchResult, None, None]:
         """