feat(dl): add --latest-episodes N to download the N newest episodes

Split episodes count once. Exposed as latest_episodes on the REST API. Closes #159.

docs/dev/rest-api/endpoints.md

@@ -429,6 +429,7 @@ Make a download job. It requires `service` and `title_id`. Every other field is
 | `no_atmos` | boolean | `false` | Exclude Atmos tracks. |
 | `wanted` | string / string[] | `[]` | Episode/season selectors, as a list or as one comma-separated string. Accepts the part form, `"S01E01.2"`, and the air-date form, `"2026-08-11"` or `"2026-08-01:2026-08-31"`. For a music release, a selector is a track number, `"1-5"` or `"1,3,7"`, or `"{disc}x{track}"` such as `"2x3"`. |
 | `latest_episode` | boolean | `false` | Only the newest episode. |
+| `latest_episodes` | integer | `null` | Only the N newest episodes. A positive integer. When set, it overrides `latest_episode`. |
 | `lang` / `v_lang` / `a_lang` / `s_lang` | string[] | `["orig"]` / `[]` / `[]` / `["all"]` | Language filters. |
 | `require_audio` | string[] | `[]` | Audio languages that must exist. The job fails if one is missing, even with `best_available`. |
 | `require_video` | string[] | `[]` | Video languages that must exist. The job fails if one is missing, even with `best_available`. |

docs/guide/cli-reference.md

@@ -117,6 +117,7 @@ unshackle honours the special language tokens `orig`, `all`, and `best` everywhe
 | `-w`, `--wanted` | Wanted episodes, e.g. `S01-S05,S07`, `S01E01-S02E03`. Supports exclusions with a leading `-` (e.g. `-S03`). For a [split episode](downloading.md#split-episodes), `.N` picks one part (`S01E01.2`), a range must stay inside the episode (`S01E01.1-S01E01.3`), and `S01E01` on its own takes every part. For a [dated episode](downloading.md#daily-and-date-based-content), a token can also be an ISO air date (`2026-08-11`) or a date range with a colon (`2026-08-01:2026-08-31`). For a [music release](downloading.md#music-tracks), a token is a track number (`1-5`, `1,3,7`), or `{disc}x{track}` (`2x3`) when the release has more than one disc. |
 | `--select-titles` | Interactively select what to download: episodes of a series, or films when a title has more than one. **Cannot combine with `-w`.** |
 | `--latest-episode` | Download only the single most recent episode. |
+| `--latest-episodes N` | Download only the N most recent episodes. |
 | `--list-titles` | List titles only; do not download. |
 
 ### Track-type inclusion / exclusion

docs/guide/downloading.md

@@ -55,7 +55,7 @@ with `--skip-dl`.
 2. **Authenticate**: sign in to the service with your profile's cookies/credentials.
 3. **Fetch titles**: retrieve the movie, episode list, or album (cached unless
    `--no-cache`/`--reset-cache`).
-4. **Filter titles**: apply `--wanted`, `--latest-episode`, or `--select-titles`.
+4. **Filter titles**: apply `--wanted`, `--latest-episode`, `--latest-episodes`, or `--select-titles`.
 5. **Get tracks**: parse the manifest into video, audio, subtitle, and chapter tracks.
 6. **Select tracks**: narrow each type down using your quality, codec, range, language,
    and bitrate flags.
@@ -65,7 +65,7 @@ with `--skip-dl`.
    `mkvmerge`, and move the finished file to your downloads directory.
 
 `--list` prints the tracks a title exposes and stops before selection. `--list-titles`
-prints every title the service returned and stops before `--wanted`/`--latest-episode`
+prints every title the service returned and stops before `--wanted`/`--latest-episode`/`--latest-episodes`
 filtering. `--skip-dl` runs the license step but skips the actual segment download.
 
 ## Choosing quality
@@ -600,12 +600,19 @@ when you want a part on its own.
 | --- | --- |
 | `--select-titles` | Interactively pick episodes of a series, or films when a title has more than one. **Cannot combine with `-w`.** |
 | `--latest-episode` | Download only the single most recent episode. |
-| `--list-titles` | List every title the service returned, then stop. `-w`/`--latest-episode` are not applied to this listing. |
+| `--latest-episodes N` | Download only the N most recent episodes. The newest episodes can come from more than one season. A [split episode](#split-episodes) counts as one episode, and all its parts download. |
+| `--list-titles` | List every title the service returned, then stop. `-w`, `--latest-episode` and `--latest-episodes` are not applied to this listing. |
+
+When `--latest-episode` or `--latest-episodes` is set, `-w` has no effect.
 
 ```shell title="Grab just the newest episode"
 unshackle dl --latest-episode EXAMPLE 81234567
 ```
 
+```shell title="Grab the three newest episodes"
+unshackle dl --latest-episodes 3 EXAMPLE 81234567
+```
+
 ## Including and excluding track types
 
 You can restrict a download to certain track categories, either positively (only these)
@@ -653,7 +660,7 @@ Before committing to a long download, inspect what unshackle *would* do:
 | Flag | Effect |
 | --- | --- |
 | `--list` | List the tracks the service exposes for each title, then stop. No selection, no download. |
-| `--list-titles` | List every title the service returned, then stop. `-w`/`--latest-episode` are not applied to this listing. |
+| `--list-titles` | List every title the service returned, then stop. `-w`, `--latest-episode` and `--latest-episodes` are not applied to this listing. |
 | `--skip-dl` | Skip downloading but still acquire the decryption keys. |
 
 ```shell title="See the track selection without downloading"
@@ -1002,6 +1009,7 @@ authoritative list.
 | `--daily` | | Fill missing air dates from TVDB during `--enrich`. |
 | `--select-titles` | | Interactively pick episodes or films. |
 | `--latest-episode` | | Only the newest episode. |
+| `--latest-episodes N` | | Only the N newest episodes. |
 | `--video-only` / `--audio-only` / `--subs-only` | `-V` / `-A` / `-S` | Restrict track types. |
 | `--no-video` / `--no-audio` / `--no-subs` / `--no-chapters` / `--no-attachments` | `-nv` / `-na` / `-ns` / `-nc` / `-nt` | Skip track types. |
 | `--worst` | | Lowest bitrate within `-q`. |

tests/core/test_latest_episodes.py

@@ -0,0 +1,61 @@
+"""Selection for --latest-episode / --latest-episodes and its REST API validation."""
+
+from __future__ import annotations
+
+from typing import Optional
+
+import pytest
+
+from unshackle.commands.dl import latest_episode_keys
+from unshackle.core.api.handlers import validate_download_parameters
+from unshackle.core.titles import Movie, Movies, Series
+from unshackle.core.titles.episode import Episode
+
+
+class Svc:
+    pass
+
+
+def episode(season: int, number: int, part: Optional[int] = None) -> Episode:
+    return Episode(id_=f"ep-{season}x{number}.{part}", service=Svc, title="T", season=season, number=number, part=part)
+
+
+def series() -> Series:
+    # out of order on purpose: the Series sorts itself
+    return Series(episode(s, n) for s, n in [(2, 1), (1, 1), (2, 3), (1, 2), (2, 2)])
+
+
+@pytest.mark.parametrize(
+    ("latest_episode", "latest_episodes", "expected"),
+    [
+        (False, None, []),
+        (True, None, [(2, 3)]),
+        (False, 3, [(2, 1), (2, 2), (2, 3)]),
+        (True, 2, [(2, 2), (2, 3)]),
+        (False, 99, [(1, 1), (1, 2), (2, 1), (2, 2), (2, 3)]),
+    ],
+)
+def test_the_newest_episodes_are_picked_across_seasons(
+    latest_episode: bool, latest_episodes: Optional[int], expected: list[tuple[int, int]]
+) -> None:
+    assert latest_episode_keys(series(), latest_episode, latest_episodes) == expected
+
+
+def test_a_split_episode_takes_one_slot() -> None:
+    titles = Series([episode(1, 4), episode(1, 5, part=1), episode(1, 5, part=2)])
+    assert latest_episode_keys(titles, False, 2) == [(1, 4), (1, 5)]
+    assert latest_episode_keys(titles, True, None) == [(1, 5)]
+
+
+def test_titles_that_are_not_a_series_are_left_alone() -> None:
+    movies = Movies([Movie(id_="movie-1", service=Svc, name="M")])
+    assert latest_episode_keys(movies, True, 3) == []
+
+
+@pytest.mark.parametrize("value", [0, -1, True, "3", 1.5])
+def test_a_non_positive_or_non_integer_count_is_refused(value: object) -> None:
+    assert validate_download_parameters({"latest_episodes": value}) == "latest_episodes must be a positive integer"
+
+
+def test_a_positive_count_is_accepted() -> None:
+    assert validate_download_parameters({"latest_episodes": 3}) is None

tests/remote/unit/test_dashboard.py

@@ -4,13 +4,16 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import sys
+import threading
 import time
+from collections.abc import Callable
 from datetime import datetime, timedelta, timezone
 
 import pytest
 from aiohttp import web
 
-from unshackle.core.api.events import bus
+from unshackle.core.api.events import EventBus, bus
 from unshackle.core.api.handlers import api_key_authentication, dashboard_authentication
 from unshackle.core.api.routes import cors_middleware, setup_routes
 from unshackle.core.api.session_store import SessionStore
@@ -72,6 +75,56 @@ def test_ring_since_and_level() -> None:
     assert [r["msg"] for r in ring.since(0, "warning")] == ["m3"]
 
 
+def _hammer(write: Callable[[], object], read: Callable[[], object]) -> None:
+    """Call ``read`` on this thread while four threads call ``write`` in a loop."""
+    stop = threading.Event()
+
+    def spin() -> None:
+        while not stop.is_set():
+            write()
+
+    old_interval = sys.getswitchinterval()
+    sys.setswitchinterval(1e-6)
+    threads = [threading.Thread(target=spin, daemon=True) for _ in range(4)]
+    try:
+        for t in threads:
+            t.start()
+        for _ in range(300):
+            read()
+    finally:
+        stop.set()
+        sys.setswitchinterval(old_interval)
+        for t in threads:
+            t.join()
+
+
+def test_bus_since_survives_concurrent_publish() -> None:
+    local = EventBus(history=5000)
+    _hammer(lambda: local.publish("log", {}), lambda: local.since(0))
+    seqs = [item["seq"] for item in local.history]
+    assert seqs == list(range(seqs[0], seqs[0] + len(seqs)))
+
+
+def test_bus_publish_reenters_its_lock() -> None:
+    local = EventBus()
+
+    def nested() -> None:
+        with local._lock:
+            local.publish("log", {})
+
+    worker = threading.Thread(target=nested, daemon=True)
+    worker.start()
+    worker.join(5)
+    assert not worker.is_alive(), "publish() deadlocked re-entering its own lock"
+    assert [item["seq"] for item in local.since(0)] == [1]
+
+
+def test_ring_since_survives_concurrent_emit() -> None:
+    ring = RingLogHandler(maxlen=5000)
+    record = logging.LogRecord("test.ring", logging.INFO, __file__, 0, "m", None, None)
+    _hammer(lambda: ring.handle(record), lambda: ring.since(0))
+
+
 def test_ring_masks_host_paths_and_secrets_in_records() -> None:
     """The ring feeds both /api/logs and the SSE stream, so it must mask on the way in."""
     from pathlib import Path

tests/remote/unit/test_download_manager.py

@@ -227,6 +227,37 @@ def test_perform_download_puts_the_lang_selection_in_the_service_ctx(
     assert (p["lang"], p["v_lang"], p["a_lang"], [c.name for c in p["acodec"]], p["forced_subs"]) == expected
 
 
+def test_perform_download_passes_latest_episodes_to_dl_result(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    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:
+            pass
+
+        def result(self, **kwargs: object) -> None:
+            raise _CapturedCtx(kwargs)
+
+    class FakeService:
+        def __init__(self, ctx: click.Context) -> None:
+            pass
+
+    monkeypatch.setattr(dl_module, "dl", FakeDl)
+    monkeypatch.setattr(Services, "get_path", staticmethod(lambda s: tmp_path))
+    monkeypatch.setattr(Services, "load", staticmethod(lambda s: FakeService))
+    monkeypatch.setattr(handlers, "load_full_cdm", lambda *a: None)
+
+    with pytest.raises(_CapturedCtx) as exc:
+        download_manager.perform_download("job-1", "EXAMPLE", "t1", {"latest_episodes": 3})
+
+    assert exc.value.args[0]["latest_episodes"] == 3
+
+
 def test_worker_relays_a_prompt_and_reads_the_answer(monkeypatch: pytest.MonkeyPatch) -> None:
     import json
     import os

unshackle/commands/dl.py

@@ -214,6 +214,18 @@ def title_wanted(candidate: Any, wanted: Collection[str]) -> bool:
     return bool(candidate.matches_wanted(wanted))
 
 
+def latest_episode_keys(titles: Any, latest_episode: bool, latest_episodes: Optional[int]) -> list[tuple[int, int]]:
+    """The (season, number) of each episode ``--latest-episode(s)`` keeps, oldest first.
+
+    Counts distinct episodes, not Series entries, so a split episode takes one slot and all its parts download.
+    """
+    count = latest_episodes or (1 if latest_episode else 0)
+    if not count or not isinstance(titles, Series):
+        return []
+    # a Series keeps itself sorted, so the newest episodes are last
+    return list(dict.fromkeys((ep.season, ep.number) for ep in titles))[-count:]
+
+
 def server_url(server_name: Optional[str]) -> str:
     """The configured URL of a remote server, or an empty string when the config cannot be read."""
     from unshackle.core.remote_service import resolve_server
@@ -644,6 +656,12 @@ class dl:
         default=False,
         help="Download only the single most recent episode available.",
     )
+    @click.option(
+        "--latest-episodes",
+        type=click.IntRange(min=1),
+        default=None,
+        help="Download only the N most recent episodes available.",
+    )
     @click.option(
         "-vl",
         "--v-lang",
@@ -1431,6 +1449,7 @@ class dl:
         select_titles: bool,
         wanted: list[str],
         latest_episode: bool,
+        latest_episodes: Optional[int],
         lang: list[str],
         v_lang: list[str],
         a_lang: list[str],
@@ -1898,13 +1917,12 @@ class dl:
                 count = len(titles)
                 console.print(Padding(f"[text]Total selected: {count}[/]", (0, 5)))
 
-        latest_episode_id = None
-        if latest_episode and isinstance(titles, Series) and len(titles) > 0:
-            # Series is already sorted by (season, number, year)
-            # The last episode in the sorted list is the latest
-            latest_ep = titles[-1]
-            latest_episode_id = f"{latest_ep.season}x{latest_ep.number}"
-            self.log.info(f"Latest episode mode: Selecting S{latest_ep.season:02}E{latest_ep.number:02}")
+        latest_keys = latest_episode_keys(titles, latest_episode, latest_episodes)
+        if latest_keys:
+            picked = ", ".join(f"S{season:02}E{number:02}" for season, number in latest_keys)
+            self.log.info(f"Latest episode mode: Selecting {picked}")
+            # replaces -w; a base SxE key selects every part of a split episode
+            wanted = [f"{season}x{number}" for season, number in latest_keys]
 
         def select_best_audio(
             audio: list[Audio],
@@ -1955,15 +1973,9 @@ class dl:
 
         base_selection = (v_lang, a_lang, s_lang, range_)
 
-        def post_script_queued(candidate: Any) -> bool:
-            """Mirror of the filters the loop below applies, so the season counter matches it."""
-            if isinstance(candidate, Episode) and latest_episode and latest_episode_id:
-                return f"{candidate.season}x{candidate.number}" == latest_episode_id
-            return title_wanted(candidate, wanted)
-
         post_script_pending: dict[Any, int] = {}
         for candidate in titles:
-            if post_script_queued(candidate):
+            if title_wanted(candidate, wanted):
                 key = post_script_group(candidate)
                 post_script_pending[key] = post_script_pending.get(key, 0) + 1
         post_script_last: dict[Any, dict[Path, dict[str, str]]] = {}
@@ -1972,11 +1984,7 @@ class dl:
 
         for i, title in enumerate(titles):
             v_lang, a_lang, s_lang, range_ = base_selection
-            if isinstance(title, Episode) and latest_episode and latest_episode_id:
-                # If --latest-episode is set, only process the latest episode
-                if f"{title.season}x{title.number}" != latest_episode_id:
-                    continue
-            elif not title_wanted(title, wanted):
+            if not title_wanted(title, wanted):
                 continue
 
             if progress_sink:

unshackle/core/api/download_manager.py

@@ -591,6 +591,7 @@ def perform_download(
                 select_titles=False,
                 wanted=params.get("wanted", []),
                 latest_episode=params.get("latest_episode", False),
+                latest_episodes=params.get("latest_episodes"),
                 lang=ctx.params["lang"],
                 v_lang=ctx.params["v_lang"],
                 a_lang=ctx.params["a_lang"],

unshackle/core/api/events.py

@@ -18,6 +18,7 @@ class EventBus:
         self.history: Deque[Dict[str, Any]] = deque(maxlen=history)
         self._loop: Optional[asyncio.AbstractEventLoop] = None
         self._loop_thread: Optional[threading.Thread] = None
+        self._lock = threading.RLock()
 
     def subscribe(self) -> asyncio.Queue:
         self._loop = asyncio.get_running_loop()
@@ -31,12 +32,25 @@ class EventBus:
             self._subs.remove(queue)
 
     def since(self, seq: int) -> List[Dict[str, Any]]:
-        return [item for item in self.history if item["seq"] > seq]
+        """Return the history after ``seq``.
+
+        ``deque.copy()`` is one C call, so it is atomic against an append from another thread.
+        Do not take the lock here: a finalizer that logs on the loop can then wait for the log
+        handler's lock while a worker that holds it waits for this lock, and the server stops.
+        """
+        return [item for item in self.history.copy() if item["seq"] > seq]
 
     def publish(self, event: str, data: Dict[str, Any]) -> None:
-        self.seq += 1
-        item = {"seq": self.seq, "event": event, "data": data}
-        self.history.append(item)
+        """Record and fan out one event; safe to call from any thread.
+
+        The lock is re-entrant because a log call inside it publishes again. ``seq``
+        moves only after the append, so a reader that sees it also finds the event in history.
+        """
+        with self._lock:
+            seq = self.seq + 1
+            item = {"seq": seq, "event": event, "data": data}
+            self.history.append(item)
+            self.seq = seq
         if not self._subs:
             return
         on_loop = self._loop is None or threading.current_thread() is self._loop_thread

unshackle/core/api/handlers.py

@@ -55,6 +55,7 @@ DEFAULT_DOWNLOAD_PARAMS = {
     "no_atmos": False,
     "wanted": [],
     "latest_episode": False,
+    "latest_episodes": None,
     "lang": ["orig"],
     "v_lang": [],
     "a_lang": [],
@@ -1871,7 +1872,7 @@ def validate_download_parameters(data: Dict[str, Any]) -> Optional[str]:
         if not isinstance(data["downloads"], int) or data["downloads"] <= 0:
             return "downloads must be a positive integer"
 
-    for name in ("tmdb_id", "tvdb_id"):
+    for name in ("tmdb_id", "tvdb_id", "latest_episodes"):
         if data.get(name) is not None:
             value = data[name]
             if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
@@ -2302,6 +2303,14 @@ async def dashboard_jobs_handler(request: web.Request) -> web.Response:
     return web.json_response([job.to_dict(include_full_details=True) for job in get_download_manager().list_jobs()])
 
 
+def _poll_cursor(before: int, items: List[Dict[str, Any]]) -> int:
+    """Cursor for the next poll, from the buffer's seq read *before* its ``since()`` call.
+
+    This poll or the next one returns a record added between the two reads; no poll skips it.
+    """
+    return max(before, items[-1]["seq"]) if items else before
+
+
 async def dashboard_logs_handler(request: web.Request) -> web.Response:
     """Recent log records; `since` (seq) and `level` filter the ring buffer."""
     from unshackle.core.api.stats import ring
@@ -2310,8 +2319,9 @@ async def dashboard_logs_handler(request: web.Request) -> web.Response:
         since = int(request.query.get("since", 0))
     except ValueError:
         since = 0
+    before = ring.seq
     records = ring.since(since, request.query.get("level"), request.query.get("logger"))
-    return web.json_response({"seq": ring.seq, "records": records})
+    return web.json_response({"seq": _poll_cursor(before, records), "records": records})
 
 
 async def dashboard_session_logs_handler(request: web.Request) -> web.Response:
@@ -2332,13 +2342,9 @@ async def dashboard_session_logs_handler(request: web.Request) -> web.Response:
     except ValueError:
         since = 0
     buffer = session.log_buffer
-    return web.json_response(
-        {
-            "session_id": session_id,
-            "records": buffer.since(since) if buffer else [],
-            "last_seq": buffer.last_seq if buffer else 0,
-        }
-    )
+    before = buffer.last_seq if buffer else 0
+    records = buffer.since(since) if buffer else []
+    return web.json_response({"session_id": session_id, "records": records, "last_seq": _poll_cursor(before, records)})
 
 
 async def dashboard_keys_handler(request: web.Request) -> web.Response:
@@ -2630,7 +2636,11 @@ async def dashboard_events_handler(request: web.Request) -> web.StreamResponse:
         except ValueError:
             since = 0
         status = await dashboard_status_handler(request)
-        return web.json_response({"seq": bus.seq, "stats": json.loads(status.text or "{}"), "events": bus.since(since)})
+        before = bus.seq
+        events = bus.since(since)
+        return web.json_response(
+            {"seq": _poll_cursor(before, events), "stats": json.loads(status.text or "{}"), "events": events}
+        )
 
     response = web.StreamResponse(
         headers={

unshackle/core/api/routes.py

@@ -853,6 +853,10 @@ async def download(request: web.Request) -> web.Response:
               latest_episode:
                 type: boolean
                 description: Download only the single most recent episode (default - false)
+              latest_episodes:
+                type: integer
+                minimum: 1
+                description: Download only the N most recent episodes; overrides latest_episode (default - null)
               lang:
                 type: array
                 items:

unshackle/core/api/stats.py

@@ -224,24 +224,27 @@ class RingLogHandler(logging.Handler):
         self.seq = 0
 
     def emit(self, record: logging.LogRecord) -> None:
-        self.seq += 1
+        """Buffer and publish one record; ``seq`` moves only after the append, as in ``EventBus``."""
+        seq = self.seq + 1
         item = {
-            "seq": self.seq,
+            "seq": seq,
             "ts": record.created,
             "level": record.levelname,
             "logger": record.name,
             "msg": redact_secrets(self.format(record)),
         }
         self.records.append(item)
+        self.seq = seq
         bus.publish("log", item)
 
     def since(self, seq: int = 0, level: Optional[str] = None, logger: Optional[str] = None) -> List[Dict[str, Any]]:
+        """Filter a copy of the records: ``emit()`` appends from worker threads."""
         min_level = logging.getLevelName(level.upper()) if level else 0
         if not isinstance(min_level, int):
             min_level = 0
         return [
             r
-            for r in self.records
+            for r in self.records.copy()
             if r["seq"] > seq
             and logging.getLevelName(r["level"]) >= min_level
             and (not logger or r["logger"] == logger or r["logger"].startswith(logger + "."))