fix(download): stop deleting user data and losing completed downloads (#1511)

Fixes six defects. Each one breaks downloads or deletes user data, and
each has a regression test that fails on 10710ab and on 77b80eb (current
main).

- **Completed-path wait (#1464)**: `_wait_for_completed_path` only
returned a directory after two identical signatures of supported
formats.
- A directory holding only archives (a scene `book.rar` + `.nfo` +
`.sfv`) never got a signature and failed with "Completed download has no
stable supported files yet". Archives (`.zip`/`.rar`, via `is_archive`)
now count toward the signature, so post-processing extracts them again,
as in v1.4.0.
- With `DOWNLOAD_CLIENT_COMPLETED_PATH_TIMEOUT=0`, `max_attempts` is 1,
so every directory failed. One sample with book files or archives is now
accepted when no retry is possible.
- **Remove & Delete Files (#1401)**: `post_process_cleanup` called
`remove(id, delete_files=True)` on torrents matched by `find_existing`
that Shelfmark never added, deleting the user's seeding data.
`_cleanup_refs` now records `added_by_shelfmark`. A pre-existing torrent
(complete, or in progress and joined) is removed but keeps its files.
- **Cancelling an unstarted torrent (#1421)**:
`_remove_unstarted_torrent` deleted with files whenever `progress == 0`.
That included `DownloadStatus.error()`, which always reports 0, and
checking states that are re-verifying data already on disk. The torrent
must now be QUEUED, PAUSED or DOWNLOADING. rTorrent's
`remove(delete_files=True)` always returns False, so a new
`DownloadClient.can_delete_files` flag (False on rTorrent) makes it
remove without files instead.
- **Queue grace overwritten**: when a QUEUED torrent's progress rose on
the same poll, the 900 s queue grace was immediately replaced by a
shorter stall window, because the orchestrator keeps one deadline. The
stall window is now requested only when it is longer than what is left
of the queue grace.
- **`drop_ignored_files` (#1480)**: the suffix rule `w.endswith("/" +
rel)` dropped a kept root `Book.epub` when `Extras/Book.epub` was
skipped. Each skipped entry now drops at most one file: the exact path,
else the file sharing the most trailing components, else a unique
basename, else an AudiobookBay glued `"<folder> <file>"` name (the
fallback `match_plan_to_files` uses, matched on folder plus file so
chaptered `Book 2 01.mp3` picks the right folder). Ties drop nothing.
- **Email and BookLore outputs** ignored `task.ignored_files` and sent
every prepared file. Both now apply `drop_ignored_files` once after
`prepare_output_files`, and report "No book files found" if the user
skipped everything. It is applied in the outputs rather than in
`prepare` because the folder output already filters in
`transfer_book_files`, and filtering twice could match a kept file on
the second pass.

- `make python-checks`: ruff, ruff format, basedpyright (backend +
tests), vulture all pass.
- `uv run pytest tests/ -m "not integration and not e2e"`: 4054 passed
on the rebased branch.
- New tests run against unfixed 77b80eb and 10710ab: all 18 new
regression cases fail, and the 11 existing post-process-cleanup tests
also fail there because they now build the 4-tuple cleanup ref.

shelfmark/download/clients/__init__.py

@@ -195,6 +195,8 @@ class DownloadClient(ABC):
     name: str
     handoff_only = False
     prefers_torrent_file = False
+    # False when remove(delete_files=True) always fails, as on rTorrent.
+    can_delete_files = True
 
     def _log_error(self, method: str, e: Exception, level: str = "error") -> str:
         """Log a client error with consistent formatting.

shelfmark/download/clients/base_handler.py

@@ -65,6 +65,12 @@ POLL_INTERVAL = 2
 QUEUE_GRACE_SECONDS = 900.0
 QUEUE_GRACE_RENEW_SECONDS = 600.0
 QUEUE_MAX_WAIT_SECONDS = 7200.0
+# States in which a torrent at 0% has not started: waiting for a slot, stopped before its
+# first piece, or fetching metadata and peers. Checking and error states are left out
+# because they also report 0% for a torrent whose data is on disk.
+_UNSTARTED_TORRENT_STATES = frozenset(
+    {DownloadState.QUEUED, DownloadState.PAUSED, DownloadState.DOWNLOADING}
+)
 # Torrents are jerky: a swarm with one seed can sit still for several minutes and then carry
 # on, and a magnet can take a while to fetch metadata or find a first peer. The orchestrator
 # ends a download after STALL_TIMEOUT (five minutes) without a change. This setting is the
@@ -203,7 +209,12 @@ def _probe_completed_path(path: Path) -> tuple[bool, OSError | None]:
 def _completed_file_signature(
     path: Path, content_type: str | None
 ) -> tuple[tuple[str, int, int], ...] | None:
-    """Return sizes and mtimes for supported files in a completed path."""
+    """Return sizes and mtimes for supported files and archives in a completed path.
+
+    Archives count because post-processing extracts them: a scene release can hold
+    nothing but `book.rar` and its `.nfo`/`.sfv`.
+    """
+    from shelfmark.download.archive import is_archive
     from shelfmark.download.postprocess.policy import (
         get_supported_audiobook_formats,
         get_supported_formats,
@@ -224,7 +235,7 @@ def _completed_file_signature(
 
         signature = []
         for candidate in candidates:
-            if candidate.suffix.lower() not in supported_suffixes:
+            if candidate.suffix.lower() not in supported_suffixes and not is_archive(candidate):
                 continue
             try:
                 stat = candidate.stat()
@@ -246,8 +257,8 @@ class ExternalClientHandler(DownloadHandler, ABC):
     def __init__(self) -> None:
         """Initialize cleanup tracking for client-managed downloads."""
         # Track downloads that may need client-side cleanup after Shelfmark completes import.
-        # task_id -> (client, download_id, protocol)
-        self._cleanup_refs: dict[str, tuple[DownloadClient, str, str]] = {}
+        # task_id -> (client, download_id, protocol, added_by_shelfmark)
+        self._cleanup_refs: dict[str, tuple[DownloadClient, str, str, bool]] = {}
 
     @abstractmethod
     def _resolve_download(
@@ -369,7 +380,7 @@ class ExternalClientHandler(DownloadHandler, ABC):
         if client_ref is None:
             return
 
-        client, download_id, protocol = client_ref
+        client, download_id, protocol, added_by_shelfmark = client_ref
 
         if protocol == "usenet":
             # "Move" means copy into ingest then let the usenet client delete its own files.
@@ -389,12 +400,18 @@ class ExternalClientHandler(DownloadHandler, ABC):
         elif protocol == "torrent":
             torrent_action = config.get("PROWLARR_TORRENT_ACTION", "keep")
             if torrent_action in ("remove", "remove_and_delete"):
-                # Import succeeded; the client handles its own download data.
-                try:
-                    removed = client.remove(
+                # Import succeeded; the client handles its own download data. Only data this
+                # task downloaded is Shelfmark's to delete: a torrent the client already had
+                # may be the user's own seed, so it is removed but keeps its files.
+                delete_files = torrent_action == "remove_and_delete" and added_by_shelfmark
+                if torrent_action == "remove_and_delete" and not delete_files:
+                    logger.info(
+                        "Keeping files of torrent %s in %s: Shelfmark did not add it",
                         download_id,
-                        delete_files=torrent_action == "remove_and_delete",
+                        getattr(client, "name", "client"),
                     )
+                try:
+                    removed = client.remove(download_id, delete_files=delete_files)
                     if not removed:
                         logger.warning(
                             "Failed to remove torrent %s from %s",
@@ -615,12 +632,21 @@ class ExternalClientHandler(DownloadHandler, ABC):
         Torrents are normally left in the client after a cancel so an already-downloaded one
         can keep seeding. One at 0% has nothing to seed and nothing to resume, and leaving it
         queued only holds a download slot and piles up dead entries.
+
+        Progress alone does not prove that: an error status (a failed lookup, say) and a
+        client re-checking existing data both report 0%. Only a torrent that is queued,
+        paused or fetching its first pieces counts as unstarted.
         """
         try:
             status = client.get_status(download_id)
             if status.progress > 0:  # it has data, so it may be seeding or resumable
                 return False
-            return bool(client.remove(download_id, delete_files=True))
+            if status.state not in _UNSTARTED_TORRENT_STATES:
+                return False
+            # rTorrent refuses to remove a torrent with its files, and one that never
+            # started has none worth deleting.
+            delete_files = client.can_delete_files is not False
+            return bool(client.remove(download_id, delete_files=delete_files))
         except _CLIENT_CLEANUP_ERRORS as e:
             logger.warning(
                 "Could not remove unstarted torrent %s from %s: %s", download_id, client.name, e
@@ -866,7 +892,8 @@ class ExternalClientHandler(DownloadHandler, ABC):
                     result.path,
                     content_type,
                 )
-                if signature is not None and signature == previous_signature:
+                # With no wait configured there is no second sample to compare against.
+                if signature is not None and (signature == previous_signature or max_attempts == 1):
                     return result.path, None
                 previous_signature = signature
                 last_error = "Completed download has no stable supported files yet."
@@ -1017,7 +1044,12 @@ class ExternalClientHandler(DownloadHandler, ABC):
 
                     if result:
                         self._on_download_complete(task)
-                        self._cleanup_refs[task.task_id] = (client, download_id, request.protocol)
+                        self._cleanup_refs[task.task_id] = (
+                            client,
+                            download_id,
+                            request.protocol,
+                            False,  # added_by_shelfmark: the client already had it
+                        )
                     return result
 
                 # Existing but still downloading - join the progress polling
@@ -1124,7 +1156,7 @@ class ExternalClientHandler(DownloadHandler, ABC):
         """Poll the download client for progress and handle completion.
 
         ``added_by_shelfmark`` is False when the task joined a torrent the client already had;
-        such a torrent is never removed on cancel.
+        such a torrent is never removed on cancel, and its files are never deleted.
         """
         poll_interval = self._poll_interval()
         queued_since: float | None = None
@@ -1228,6 +1260,9 @@ class ExternalClientHandler(DownloadHandler, ABC):
                 # Reset not-found counter on successful status check
                 not_found_count = 0
 
+                # The orchestrator keeps one grace deadline and a new request replaces it, so
+                # a movement window must not cut short a longer queue grace still running.
+                queue_grace_left = 0.0
                 if status.state == DownloadState.QUEUED:
                     now = time.monotonic()
                     if queued_since is None:
@@ -1238,6 +1273,8 @@ class ExternalClientHandler(DownloadHandler, ABC):
                     ):
                         request_activity_grace(status_callback, QUEUE_GRACE_SECONDS)
                         grace_requested_at = now
+                    if grace_requested_at is not None:
+                        queue_grace_left = grace_requested_at + QUEUE_GRACE_SECONDS - now
                 elif queued_since is not None:
                     release_activity_grace(status_callback)
                     queued_since = None
@@ -1304,7 +1341,9 @@ class ExternalClientHandler(DownloadHandler, ABC):
                         return None
 
                 if status.progress > last_progress:
-                    if stall_window > STALL_TIMEOUT_MIN_MINUTES * SECONDS_PER_MINUTE:
+                    if stall_window > max(
+                        STALL_TIMEOUT_MIN_MINUTES * SECONDS_PER_MINUTE, queue_grace_left
+                    ):
                         request_activity_grace(status_callback, stall_window)
                     start_window_given = True
                 elif not start_window_given and status.state != DownloadState.QUEUED:
@@ -1380,7 +1419,7 @@ class ExternalClientHandler(DownloadHandler, ABC):
         # Clean up on success
         if result:
             self._on_download_complete(task)
-            self._cleanup_refs[task.task_id] = (client, download_id, protocol)
+            self._cleanup_refs[task.task_id] = (client, download_id, protocol, added_by_shelfmark)
 
         return result
 

shelfmark/download/clients/rtorrent.py

@@ -102,6 +102,7 @@ class RTorrentClient(DownloadClient):
 
     protocol = "torrent"
     name = "rtorrent"
+    can_delete_files = False
 
     def __init__(self) -> None:
         """Initialize rTorrent client with settings from config."""

shelfmark/download/outputs/booklore.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import os
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
@@ -302,6 +302,7 @@ def _post_process_booklore(
     *,
     preserve_source_on_failure: bool = False,
 ) -> str | None:
+    from shelfmark.download.postprocess.packs import drop_ignored_files
     from shelfmark.download.postprocess.pipeline import (
         CustomScriptContext,
         OutputPlan,
@@ -354,6 +355,8 @@ def _post_process_booklore(
     )
     if not prepared:
         return None
+    # Files the user skipped in the pack review are not uploaded.
+    prepared = replace(prepared, files=drop_ignored_files(prepared.files, task.ignored_files))
 
     logger.debug(
         "Task %s: prepared %d file(s) for Booklore upload",
@@ -363,6 +366,10 @@ def _post_process_booklore(
 
     success = False
     try:
+        if not prepared.files:
+            status_callback("error", "No book files found")
+            return None
+
         unsupported_files = [
             file_path
             for file_path in prepared.files

shelfmark/download/outputs/email.py

@@ -6,7 +6,7 @@ import mimetypes
 import smtplib
 import ssl
 from contextlib import suppress
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from email.message import EmailMessage
 from email.utils import formatdate, make_msgid, parseaddr
 from typing import TYPE_CHECKING, Any
@@ -311,6 +311,7 @@ def _post_process_email(
     *,
     preserve_source_on_failure: bool = False,
 ) -> str | None:
+    from shelfmark.download.postprocess.packs import drop_ignored_files
     from shelfmark.download.postprocess.pipeline import (
         CustomScriptContext,
         OutputPlan,
@@ -373,9 +374,15 @@ def _post_process_email(
     )
     if not prepared:
         return None
+    # Files the user skipped in the pack review are not sent.
+    prepared = replace(prepared, files=drop_ignored_files(prepared.files, task.ignored_files))
 
     success = False
     try:
+        if not prepared.files:
+            status_callback("error", "No book files found")
+            return None
+
         limit_mb_raw = core_config.config.get("EMAIL_ATTACHMENT_SIZE_LIMIT_MB", 25)
         attachment_limit_mb = _parse_attachment_limit_mb(limit_mb_raw)
 

shelfmark/download/postprocess/packs.py

@@ -357,32 +357,72 @@ def group_files_into_books(
     ]
 
 
+def _single_best(scores: dict[Path, int]) -> Path | None:
+    """Return the highest-scoring path, or None when nothing scored or the best ties."""
+    if not scores:
+        return None
+    best = max(scores.values())
+    top = [path for path, score in scores.items() if score == best]
+    return top[0] if len(top) == 1 else None
+
+
+def _match_ignored_entry(entry: str, parts_by_path: dict[Path, tuple[str, ...]]) -> Path | None:
+    """Find the one on-disk file a skipped release path refers to (see `drop_ignored_files`)."""
+    wanted = PurePosixPath(entry).parts
+    if not wanted:
+        return None
+    for path, parts in parts_by_path.items():
+        if parts == wanted:
+            return path
+
+    # The root can shift after extraction, so one path may end with the other. The file
+    # sharing the most trailing components wins: a skipped `Extras/Book.epub` is the
+    # subfolder copy, not a kept `Book.epub` in the root.
+    shared: dict[Path, int] = {}
+    for path, parts in parts_by_path.items():
+        depth = min(len(parts), len(wanted))
+        if parts[-depth:] == wanted[-depth:]:
+            shared[path] = depth
+    if (match := _single_best(shared)) is not None:
+        return match
+
+    same_name = [path for path, parts in parts_by_path.items() if parts[-1] == wanted[-1]]
+    if len(same_name) == 1:
+        return same_name[0]
+
+    # AudiobookBay glues "<folder> <file>" into one name; the file whose trailing
+    # components, joined the same way, end that name on the most components wins.
+    glued = wanted[-1].lower()
+    tails: dict[Path, int] = {}
+    for path, parts in parts_by_path.items():
+        for depth in range(len(parts), 0, -1):
+            tail = " ".join(parts[-depth:]).lower()
+            if glued == tail or glued.endswith(" " + tail):
+                tails[path] = depth
+                break
+    return _single_best(tails)
+
+
 def drop_ignored_files(book_files: list[Path], ignored: list[str] | None) -> list[Path]:
     """Remove files the user chose to skip in the pack review.
 
-    Ignored entries are release-relative paths. A file matches by relative path first,
-    then by one path being a suffix of the other (the root can shift after extraction),
-    then by basename when that basename is unique among the files.
+    Ignored entries are release-relative paths, and each one drops at most one file:
+    the file at that relative path, else the one sharing the most trailing components
+    with it (the root can shift after extraction), else the only file with that
+    basename, else the one its AudiobookBay-style glued "<folder> <file>" name ends
+    with. A tie drops nothing, since keeping a skipped file beats losing a kept one.
     """
     if not book_files or not ignored:
         return book_files
     _root, rel_by_path = _relative_paths(book_files)
+    parts_by_path = {path: PurePosixPath(rel).parts for path, rel in rel_by_path.items()}
     wanted = {i.replace("\\", "/").lstrip("./") for i in ignored if i and i.strip()}
-    name_counts: dict[str, int] = {}
-    for path in book_files:
-        name_counts[path.name] = name_counts.get(path.name, 0) + 1
-
-    def is_ignored(path: Path) -> bool:
-        rel = rel_by_path[path]
-        if rel in wanted:
-            return True
-        if any(w.endswith("/" + rel) or rel.endswith("/" + w) for w in wanted):
-            return True
-        return name_counts[path.name] == 1 and any(
-            PurePosixPath(w).name == path.name for w in wanted
-        )
-
-    return [p for p in book_files if not is_ignored(p)]
+    dropped = {
+        match
+        for entry in wanted
+        if (match := _match_ignored_entry(entry, parts_by_path)) is not None
+    }
+    return [p for p in book_files if p not in dropped]
 
 
 def match_plan_to_files(

tests/download/test_packs.py

@@ -404,3 +404,40 @@ class TestDropIgnoredFiles:
 
         files = [tmp_path / "a.m4b"]
         assert drop_ignored_files(files, None) is files
+
+    @pytest.mark.parametrize("skipped", ["Extras/Book.epub", "Pack/Extras/Book.epub"])
+    def test_a_skipped_subfolder_file_keeps_the_root_file_of_the_same_name(
+        self, tmp_path: Path, skipped: str
+    ):
+        from shelfmark.download.postprocess.packs import drop_ignored_files
+
+        kept = tmp_path / "Pack" / "Book.epub"
+        files = [kept, tmp_path / "Pack" / "Extras" / "Book.epub"]
+        assert drop_ignored_files(files, [skipped]) == [kept]
+
+    def test_an_ambiguous_skip_drops_nothing(self, tmp_path: Path):
+        from shelfmark.download.postprocess.packs import drop_ignored_files
+
+        files = [tmp_path / "X" / "Book.epub", tmp_path / "Y" / "Book.epub"]
+        assert drop_ignored_files(files, ["Book.epub"]) == files
+
+    def test_drops_audiobookbay_glued_folder_and_file_names(self, tmp_path: Path):
+        from shelfmark.download.postprocess.packs import drop_ignored_files
+
+        root = tmp_path / "Cradle - Will Wight Books 1-10"
+        skysworn = root / "Will Wight - Skysworn Cradle, Book 4" / "Skysworn Cradle, Book 4.m4b"
+        uncrowned = root / "Uncrowned Cradle, Book 7.m4b"
+        files = [skysworn, uncrowned]
+        glued = "Will Wight - Skysworn Cradle, Book 4 Skysworn Cradle, Book 4.m4b"
+        assert drop_ignored_files(files, [glued]) == [uncrowned]
+
+    def test_glued_names_pick_the_right_folder_for_chaptered_tracks(self, tmp_path: Path):
+        from shelfmark.download.postprocess.packs import drop_ignored_files
+
+        files = [
+            tmp_path / "Pack" / "Book 1" / "01.mp3",
+            tmp_path / "Pack" / "Book 1" / "02.mp3",
+            tmp_path / "Pack" / "Book 2" / "01.mp3",
+            tmp_path / "Pack" / "Book 2" / "02.mp3",
+        ]
+        assert drop_ignored_files(files, ["Book 2 01.mp3", "Book 2 02.mp3"]) == files[:2]

tests/download/test_torrent_queue_stall.py

@@ -155,6 +155,19 @@ def test_a_torrent_leaving_the_queue_at_zero_gets_a_fresh_start_window() -> None
     assert rec.graces == [bh.QUEUE_GRACE_SECONDS, 0.0, WINDOW]
 
 
+def test_a_queued_torrent_that_moves_keeps_its_longer_queue_grace() -> None:
+    # The orchestrator keeps one deadline; a 10-minute window would cut the queue grace short.
+    rec = _poll([_status(Q, 5.0), _status(Q, 6.0)], clock=[0.0, 1.0, 2.0, 3.0], minutes=10)
+
+    assert rec.graces == [bh.QUEUE_GRACE_SECONDS]
+
+
+def test_a_queued_torrent_that_moves_gets_a_window_longer_than_its_queue_grace() -> None:
+    rec = _poll([_status(Q, 5.0)], clock=[0.0, 1.0, 2.0], minutes=30)
+
+    assert rec.graces == [bh.QUEUE_GRACE_SECONDS, 1800.0]
+
+
 def test_a_queued_torrent_is_not_given_the_start_window_yet() -> None:
     assert _poll([_status(Q)] * 4).graces == [bh.QUEUE_GRACE_SECONDS]  # the queue grace only
 

tests/download/test_torrent_removal.py

@@ -1,7 +1,8 @@
-"""A cancelled torrent that never started is removed from the client."""
+"""Which torrents Shelfmark removes from the client, and when it deletes their files."""
 
 from __future__ import annotations
 
+from pathlib import Path
 from threading import Event
 from unittest.mock import MagicMock, patch
 
@@ -98,3 +99,121 @@ def test_a_failed_removal_does_not_break_the_cancel() -> None:
     ProwlarrHandler()._handle_cancelled_download(
         client, "dl", "torrent", lambda *_a: None, remove_if_unstarted=True
     )  # does not raise
+
+
+@pytest.mark.parametrize(
+    "status",
+    [
+        # A failed lookup reports 0% whatever the torrent holds.
+        DownloadStatus.error("qbittorrent get_status failed (ConnectionError): offline"),
+        # Re-checking data already on disk starts from 0% too.
+        _status(DownloadState.CHECKING, 0.0),
+        _status(DownloadState.UNKNOWN, 0.0),
+    ],
+)
+def test_a_torrent_at_zero_is_not_removed_unless_it_really_has_not_started(
+    status: DownloadStatus,
+) -> None:
+    client = MagicMock()
+    client.name = "qbittorrent"
+    client.get_status.return_value = status  # what the client reports once the user cancels
+
+    ProwlarrHandler()._handle_cancelled_download(
+        client, "dl", "torrent", lambda *_a: None, remove_if_unstarted=True
+    )
+
+    client.remove.assert_not_called()
+
+
+def test_a_paused_torrent_that_downloaded_nothing_is_removed() -> None:
+    client = _cancel(_status(DownloadState.PAUSED, 0.0), existing=False)
+
+    client.remove.assert_called_once_with("dl", delete_files=True)
+
+
+def test_an_unstarted_torrent_is_removed_without_files_where_the_client_cannot_delete_them() -> (
+    None
+):
+    from shelfmark.download.clients.rtorrent import RTorrentClient
+
+    client = MagicMock()
+    client.name = "rtorrent"
+    client.can_delete_files = RTorrentClient.can_delete_files
+    client.get_status.return_value = _status(Q, 0.0)
+    client.remove.return_value = True
+
+    removed = ProwlarrHandler()._remove_unstarted_torrent(client, "dl")
+
+    assert removed is True
+    client.remove.assert_called_once_with("dl", delete_files=False)
+
+
+# --- Remove & Delete Files only deletes what this task downloaded ------------------------
+
+
+def _import_then_clean_up(existing: DownloadStatus | None, torrent_action: str) -> MagicMock:
+    """Download through the handler, import successfully, then run the completion action."""
+    client = MagicMock()
+    client.name = "qbittorrent"
+    client.find_existing.return_value = ("dl", existing) if existing else None
+    client.add_download.return_value = "dl"
+    client.get_status.return_value = _status(DownloadState.COMPLETE, 100.0, complete=True)
+    client.remove.return_value = True
+    handler = ProwlarrHandler()
+    task = DownloadTask(task_id="r", source="prowlarr", title="Book")
+    release = {"protocol": "torrent", "magnetUrl": "magnet:?xt=urn:btih:abc123"}
+    with (
+        patch("shelfmark.release_sources.prowlarr.handler.get_release", return_value=release),
+        patch("shelfmark.release_sources.prowlarr.handler.get_client", return_value=client),
+        patch("shelfmark.release_sources.prowlarr.handler.remove_release"),
+        patch("shelfmark.release_sources.prowlarr.handler.POLL_INTERVAL", 0.01),
+        patch.object(
+            ProwlarrHandler,
+            "_wait_for_completed_path",
+            return_value=(Path("/downloads/Book"), None),
+        ),
+        patch(
+            "shelfmark.download.clients.base_handler.config.get",
+            side_effect=lambda key, default=None, **_kw: (
+                torrent_action if key == "PROWLARR_TORRENT_ACTION" else default
+            ),
+        ),
+    ):
+        result = handler.download(
+            task=task,
+            cancel_flag=Event(),
+            progress_callback=lambda _p: None,
+            status_callback=_Recorder().status,
+        )
+        assert result == str(Path("/downloads/Book"))
+        handler.post_process_cleanup(task, success=True)
+    return client
+
+
+def test_remove_and_delete_deletes_the_files_of_a_torrent_this_task_added() -> None:
+    client = _import_then_clean_up(None, "remove_and_delete")
+
+    client.add_download.assert_called_once()
+    client.remove.assert_called_once_with("dl", delete_files=True)
+
+
+@pytest.mark.parametrize(
+    "existing",
+    [
+        _status(DownloadState.SEEDING, 100.0, complete=True),  # the user's own finished seed
+        _status(D, 40.0),  # an in-progress torrent the task joined
+    ],
+)
+def test_remove_and_delete_keeps_the_files_of_a_torrent_the_client_already_had(
+    existing: DownloadStatus,
+) -> None:
+    client = _import_then_clean_up(existing, "remove_and_delete")
+
+    client.add_download.assert_not_called()
+    client.remove.assert_called_once_with("dl", delete_files=False)
+
+
+def test_remove_still_removes_a_torrent_the_client_already_had() -> None:
+    client = _import_then_clean_up(_status(DownloadState.SEEDING, 100.0, complete=True), "remove")
+
+    client.remove.assert_called_once_with("dl", delete_files=False)

tests/prowlarr/test_handler.py

@@ -1535,7 +1535,7 @@ class TestProwlarrHandlerPostProcessCleanup:
         mock_client = MagicMock()
         mock_client.name = "qbittorrent"
         mock_client.set_category.return_value = True
-        handler._cleanup_refs[task.task_id] = (mock_client, "abc123", "torrent")
+        handler._cleanup_refs[task.task_id] = (mock_client, "abc123", "torrent", True)
 
         config_values = {
             "PROWLARR_TORRENT_ACTION": "change_category",
@@ -1555,7 +1555,7 @@ class TestProwlarrHandlerPostProcessCleanup:
         task = DownloadTask(task_id="torrent-no-category", source="prowlarr", title="Test")
 
         mock_client = MagicMock()
-        handler._cleanup_refs[task.task_id] = (mock_client, "abc123", "torrent")
+        handler._cleanup_refs[task.task_id] = (mock_client, "abc123", "torrent", True)
 
         with patch("shelfmark.download.clients.base_handler.config.get", return_value="keep"):
             handler.post_process_cleanup(task, success=True)
@@ -1568,7 +1568,7 @@ class TestProwlarrHandlerPostProcessCleanup:
         task = DownloadTask(task_id="torrent-empty-category", source="prowlarr", title="Test")
 
         mock_client = MagicMock()
-        handler._cleanup_refs[task.task_id] = (mock_client, "abc123", "torrent")
+        handler._cleanup_refs[task.task_id] = (mock_client, "abc123", "torrent", True)
 
         config_values = {
             "PROWLARR_TORRENT_ACTION": "change_category",
@@ -1589,7 +1589,7 @@ class TestProwlarrHandlerPostProcessCleanup:
 
         mock_client = MagicMock()
         mock_client.name = "nzbget"
-        handler._cleanup_refs[task.task_id] = (mock_client, "123", "usenet")
+        handler._cleanup_refs[task.task_id] = (mock_client, "123", "usenet", True)
 
         with patch("shelfmark.download.clients.base_handler.config.get", return_value="move"):
             handler.post_process_cleanup(task, success=True)
@@ -1602,7 +1602,7 @@ class TestProwlarrHandlerPostProcessCleanup:
 
         mock_client = MagicMock()
         mock_client.name = "nzbget"
-        handler._cleanup_refs[task.task_id] = (mock_client, "123", "usenet")
+        handler._cleanup_refs[task.task_id] = (mock_client, "123", "usenet", True)
 
         with patch("shelfmark.download.clients.base_handler.config.get", return_value="copy"):
             handler.post_process_cleanup(task, success=True)
@@ -1615,7 +1615,7 @@ class TestProwlarrHandlerPostProcessCleanup:
 
         mock_client = MagicMock()
         mock_client.name = "nzbget"
-        handler._cleanup_refs[task.task_id] = (mock_client, "123", "usenet")
+        handler._cleanup_refs[task.task_id] = (mock_client, "123", "usenet", True)
         handler._delete_local_download_data = MagicMock(side_effect=ConnectionError("offline"))
 
         with (
@@ -1641,7 +1641,7 @@ class TestProwlarrHandlerPostProcessCleanup:
 
         mock_client = MagicMock()
         mock_client.name = "qbittorrent"
-        handler._cleanup_refs[task.task_id] = (mock_client, "abc123", "torrent")
+        handler._cleanup_refs[task.task_id] = (mock_client, "abc123", "torrent", True)
 
         with patch(
             "shelfmark.download.clients.base_handler.config.get", return_value=torrent_action
@@ -1656,7 +1656,7 @@ class TestProwlarrHandlerPostProcessCleanup:
         task = DownloadTask(task_id="torrent-import-failed", source="prowlarr", title="Test")
 
         mock_client = MagicMock()
-        handler._cleanup_refs[task.task_id] = (mock_client, "abc123", "torrent")
+        handler._cleanup_refs[task.task_id] = (mock_client, "abc123", "torrent", True)
 
         with patch(
             "shelfmark.download.clients.base_handler.config.get",
@@ -1674,7 +1674,7 @@ class TestProwlarrHandlerPostProcessCleanup:
         mock_client = MagicMock()
         mock_client.name = "qbittorrent"
         mock_client.remove.side_effect = ConnectionError("offline")
-        handler._cleanup_refs[task.task_id] = (mock_client, "123", "torrent")
+        handler._cleanup_refs[task.task_id] = (mock_client, "123", "torrent", True)
 
         with (
             patch("shelfmark.download.clients.base_handler.config.get", return_value="remove"),
@@ -1696,7 +1696,7 @@ class TestProwlarrHandlerPostProcessCleanup:
         mock_client = MagicMock()
         mock_client.name = "rtorrent"
         mock_client.remove.return_value = False
-        handler._cleanup_refs[task.task_id] = (mock_client, "abc123", "torrent")
+        handler._cleanup_refs[task.task_id] = (mock_client, "abc123", "torrent", True)
 
         with (
             patch(

tests/prowlarr/test_remote_path_mappings.py

@@ -365,6 +365,70 @@ def test_waits_for_supported_file_to_stop_growing_in_remapped_directory(monkeypa
     assert recorder.status_updates.count(("locating", "Waiting for completed files...")) == 3
 
 
+def _wait_for_local_directory(monkeypatch, directory: Path, *, timeout: object):
+    import shelfmark.download.clients.base_handler as base_handler
+
+    mock_client = MagicMock()
+    mock_client.name = "qbittorrent"
+    mock_client.get_download_path.return_value = str(directory)
+
+    def config_get(key: str, default=""):
+        if key == "PROWLARR_REMOTE_PATH_MAPPINGS":
+            return []
+        if key == "DOWNLOAD_CLIENT_COMPLETED_PATH_TIMEOUT":
+            return timeout
+        return default
+
+    handler = ProwlarrHandler()
+    monkeypatch.setattr(base_handler.config, "get", config_get)
+    monkeypatch.setattr(handler, "_completed_path_retry_interval", lambda: 0.01)
+    monkeypatch.setattr(base_handler.time, "sleep", lambda _seconds: None)
+    return handler._wait_for_completed_path(
+        mock_client,
+        "download_id",
+        cancel_flag=None,
+        status_callback=ProgressRecorder().status_callback,
+    )
+
+
+def test_a_directory_holding_only_an_archive_is_returned_for_extraction(monkeypatch, tmp_path):
+    # A scene release: the book is inside the .rar, which post-processing extracts.
+    release = tmp_path / "Book.Title-GROUP"
+    release.mkdir()
+    (release / "book.rar").write_bytes(b"Rar!archive")
+    (release / "book.nfo").write_text("nfo")
+    (release / "book.sfv").write_text("sfv")
+
+    resolved_path, error = _wait_for_local_directory(monkeypatch, release, timeout=1)
+
+    assert resolved_path == release
+    assert error is None
+
+
+def test_a_completed_directory_is_returned_when_no_wait_is_configured(monkeypatch, tmp_path):
+    release = tmp_path / "Book"
+    release.mkdir()
+    (release / "book.epub").write_bytes(b"epub")
+
+    resolved_path, error = _wait_for_local_directory(monkeypatch, release, timeout=0)
+
+    assert resolved_path == release
+    assert error is None
+
+
+def test_a_directory_without_book_files_still_fails_when_no_wait_is_configured(
+    monkeypatch, tmp_path
+):
+    release = tmp_path / "Book"
+    release.mkdir()
+    (release / "cover.jpg").write_bytes(b"jpg")
+
+    resolved_path, error = _wait_for_local_directory(monkeypatch, release, timeout=0)
+
+    assert resolved_path is None
+    assert error == "Completed download has no stable supported files yet."
+
+
 def test_wait_for_completed_path_fails_fast_for_unsafe_mapping(monkeypatch, tmp_path):
     import shelfmark.download.clients.base_handler as base_handler