fix: wait for synced files before processing completed downloads (#1464)

## Summary

```text
wait for completed download path
  regular file -> keep existing immediate handling
  directory -> inspect files enabled for this task's content type
              wait until the supported-file signature is stable across checks
```

Fixes the remote-sync race in #1463. A mapped torrent folder can appear
before its large media file finishes copying, so path existence alone
does not mean the download is ready.

## Evidence

- **Before:** In #1463, qBittorrent reported a multi-file audiobook
complete while the destination contained only `.cue` and `.jpg`;
Shelfmark immediately post-processed the folder and failed with “No book
files found in download.” Retrying after the `.m4b` sync finished
imported it.
- **After:** A regression test starts with the cue sheet and cover
present, then simulates the `.m4b` file appearing and growing. Path
waiting returns only after the supported file's size and modification
time remain stable across checks.
- Credential-free isolated `python -m pytest
tests/prowlarr/test_remote_path_mappings.py -q -n 0 -o addopts=''`: 11
passed.
- `ruff check shelfmark/download/clients/base_handler.py
tests/prowlarr/test_remote_path_mappings.py`: passed.

## Merge Danger

**Door:** Two-way

**Blast Radius:** Download handling

The wait uses the existing completed-path timeout and leaves regular
single-file downloads on their current path.

Co-authored-by: Sean Li <22557931+sean0407@users.noreply.github.com>

shelfmark/download/clients/base_handler.py

@@ -194,6 +194,46 @@ def _probe_completed_path(path: Path) -> tuple[bool, OSError | None]:
     return True, 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."""
+    from shelfmark.download.postprocess.policy import (
+        get_supported_audiobook_formats,
+        get_supported_formats,
+    )
+
+    formats = (
+        get_supported_audiobook_formats()
+        if is_audiobook(content_type)
+        else get_supported_formats()
+    )
+    supported_suffixes = {f".{fmt.lstrip('.').lower()}" for fmt in formats}
+
+    try:
+        if path.is_file():
+            candidates = (path,)
+        elif path.is_dir():
+            candidates = path.rglob("*")
+        else:
+            return None
+
+        signature = []
+        for candidate in candidates:
+            if candidate.suffix.lower() not in supported_suffixes:
+                continue
+            try:
+                stat = candidate.stat()
+            except OSError:
+                continue
+            if not candidate.is_file() or stat.st_size <= 0:
+                continue
+            name = candidate.relative_to(path).as_posix() if path.is_dir() else candidate.name
+            signature.append((name, stat.st_size, stat.st_mtime_ns))
+    except OSError:
+        return None
+
+    return tuple(sorted(signature)) or None
+
+
 class ExternalClientHandler(DownloadHandler, ABC):
     """Shared lifecycle handler for sources that hand off to torrent/usenet clients."""
 
@@ -752,9 +792,11 @@ class ExternalClientHandler(DownloadHandler, ABC):
         *,
         cancel_flag: Event | None,
         status_callback: Callable[[str, str | None], None],
+        content_type: str | None = None,
     ) -> tuple[Path | None, str | None]:
-        """Wait briefly for completed files to appear on disk."""
+        """Wait for supported files to appear and stop changing on disk."""
         last_error: str | None = None
+        previous_signature: tuple[tuple[str, int, int], ...] | None = None
         retry_interval = self._completed_path_retry_interval()
         timeout_seconds = self._completed_path_timeout_seconds()
         if retry_interval <= 0 or timeout_seconds <= 0:
@@ -773,7 +815,32 @@ class ExternalClientHandler(DownloadHandler, ABC):
                 log_details=log_details,
             )
             if result.path:
-                return result.path, None
+                if run_blocking_io(result.path.is_file):
+                    return result.path, None
+                signature = run_blocking_io(
+                    _completed_file_signature,
+                    result.path,
+                    content_type,
+                )
+                if signature is not None and signature == previous_signature:
+                    return result.path, None
+                previous_signature = signature
+                last_error = "Completed download has no stable supported files yet."
+                if attempt < max_attempts:
+                    status_callback("locating", "Waiting for completed files...")
+                    logger.debug(
+                        "Supported files are missing or still changing for %s (%s) (attempt %d/%d)",
+                        client.name,
+                        download_id,
+                        attempt,
+                        max_attempts,
+                    )
+                    if cancel_flag:
+                        if cancel_flag.wait(timeout=retry_interval):
+                            return None, last_error
+                    else:
+                        time.sleep(retry_interval)
+                continue
 
             last_error = result.error
 
@@ -874,6 +941,7 @@ class ExternalClientHandler(DownloadHandler, ABC):
                         download_id=download_id,
                         cancel_flag=cancel_flag,
                         status_callback=status_callback,
+                        content_type=task.content_type,
                     )
                     if not source_path_obj:
                         if cancel_flag.is_set():
@@ -1161,6 +1229,7 @@ class ExternalClientHandler(DownloadHandler, ABC):
                 download_id=download_id,
                 cancel_flag=cancel_flag,
                 status_callback=status_callback,
+                content_type=task.content_type,
             )
             if not source_path_obj:
                 if cancel_flag.is_set():

tests/prowlarr/test_remote_path_mappings.py

@@ -309,6 +309,62 @@ def test_wait_for_completed_path_allows_delayed_remapped_file(monkeypatch, tmp_p
     assert ("locating", "Waiting for completed files...") in recorder.status_updates
 
 
+def test_waits_for_supported_file_to_stop_growing_in_remapped_directory(monkeypatch, tmp_path):
+    import shelfmark.download.clients.base_handler as base_handler
+
+    local_dir = tmp_path / "local" / "downloads" / "book"
+    local_dir.mkdir(parents=True)
+    (local_dir / "book.cue").write_text("cue")
+    (local_dir / "cover.jpg").write_text("cover")
+
+    mock_client = MagicMock()
+    mock_client.name = "qbittorrent"
+    mock_client.get_download_path.return_value = "/remote/downloads/book"
+
+    def config_get(key: str, default=""):
+        if key == "PROWLARR_REMOTE_PATH_MAPPINGS":
+            return [
+                {
+                    "host": "qbittorrent",
+                    "remotePath": "/remote/downloads",
+                    "localPath": str(local_dir.parent),
+                }
+            ]
+        if key == "DOWNLOAD_CLIENT_COMPLETED_PATH_TIMEOUT":
+            return 1
+        return default
+
+    media_file = local_dir / "book.m4b"
+    sleep_calls = 0
+
+    def advance_sync(_seconds: float) -> None:
+        nonlocal sleep_calls
+        sleep_calls += 1
+        if sleep_calls == 1:
+            media_file.write_bytes(b"a")
+        elif sleep_calls == 2:
+            media_file.write_bytes(b"audio")
+
+    handler = ProwlarrHandler()
+    recorder = ProgressRecorder()
+    monkeypatch.setattr(base_handler.config, "get", config_get)
+    monkeypatch.setattr(handler, "_completed_path_retry_interval", lambda: 0.01)
+    monkeypatch.setattr(base_handler.time, "sleep", advance_sync)
+
+    resolved_path, error = handler._wait_for_completed_path(
+        mock_client,
+        "download_id",
+        cancel_flag=None,
+        status_callback=recorder.status_callback,
+        content_type="audiobook",
+    )
+
+    assert resolved_path == local_dir
+    assert error is None
+    assert sleep_calls == 3
+    assert recorder.status_updates.count(("locating", "Waiting for completed files...")) == 3
+
+
 def test_wait_for_completed_path_fails_fast_for_unsafe_mapping(monkeypatch, tmp_path):
     import shelfmark.download.clients.base_handler as base_handler