fix(downloads): keep terminal tasks visible during active batch + concurrent pool

Three fixes from on-device testing of best-quality mode:

1. clear_completed_local no longer prunes terminal tasks that belong to a
   STILL-ACTIVE batch (one with non-terminal work remaining). The 5-min
   "Clean Completed Downloads" automation was yanking completed/failed/
   unverified rows out of download_tasks mid-run — and failed/cancelled
   aren't in library_history — so they only reappeared after the batch
   ended. Now the whole active batch stays intact until it finishes.

2. search_all_sources runs every source CONCURRENTLY (asyncio.gather)
   instead of sequentially, so the pool waits only for the slowest source
   (e.g. usenet/Prowlarr) in parallel rather than summing all latencies.

3. The pool log now reports per-source contribution counts
   (e.g. "usenet=0, hifi=11, soulseek=1") instead of just echoing the
   chain, so a release-level source that returns nothing for a track-title
   query is visible rather than appearing to have been searched.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

core/download_engine/engine.py

@@ -25,6 +25,7 @@ big-bang switchover.
 
 from __future__ import annotations
 
+import asyncio
 import threading
 from typing import Any, Dict, Iterator, List, Optional, Tuple
 
@@ -477,29 +478,55 @@ class DownloadEngine:
         excluded = {s.lower() for s in (exclude_sources or []) if s}
         pooled_tracks = []
         pooled_albums = []
+        # Per-source contribution for an honest pool log — e.g. a release-level
+        # source like usenet/torrent that returns nothing for a track-title
+        # query should read "usenet=0", not silently hide behind the chain name.
+        contributions = []
 
+        # Decide which sources to actually query, recording why the rest were
+        # skipped. Searches then run CONCURRENTLY so the pool waits only for the
+        # slowest source (e.g. usenet/Prowlarr, which can be slow) rather than
+        # the sum of every source's latency.
+        to_search = []  # (source_name, plugin)
         for source_name in source_chain:
             if source_name.lower() in excluded:
+                contributions.append(f"{source_name}=excluded")
                 continue
             plugin = self._plugins.get(source_name)
             if plugin is None:
                 logger.info(f"Skipping {source_name} (not available)")
+                contributions.append(f"{source_name}=unavailable")
                 continue
             if hasattr(plugin, 'is_configured') and not plugin.is_configured():
                 logger.info(f"Skipping {source_name} (not configured)")
+                contributions.append(f"{source_name}=unconfigured")
                 continue
-            try:
-                tracks, albums = await plugin.search(query, timeout, progress_callback)
-                if tracks:
-                    pooled_tracks.extend(tracks)
-                if albums:
-                    pooled_albums.extend(albums)
-            except Exception as e:
-                logger.warning(f"{source_name} search failed: {e}")
+            to_search.append((source_name, plugin))
+
+        async def _one(plugin):
+            return await plugin.search(query, timeout, progress_callback)
+
+        results = await asyncio.gather(
+            *[_one(plugin) for _, plugin in to_search],
+            return_exceptions=True,
+        )
+
+        for (source_name, _), result in zip(to_search, results):
+            if isinstance(result, Exception):
+                logger.warning(f"{source_name} search failed: {result}")
+                contributions.append(f"{source_name}=error")
+                continue
+            tracks, albums = result
+            n = len(tracks) if tracks else 0
+            if tracks:
+                pooled_tracks.extend(tracks)
+            if albums:
+                pooled_albums.extend(albums)
+            contributions.append(f"{source_name}={n}")
 
         logger.info(
-            "Best-quality pool: %d candidates across %s for: %s",
-            len(pooled_tracks), ', '.join(source_chain), query,
+            "Best-quality pool: %d candidates [%s] for: %s",
+            len(pooled_tracks), ', '.join(contributions), query,
         )
         return (pooled_tracks, pooled_albums)
 

core/downloads/cancel.py

@@ -83,9 +83,26 @@ def clear_completed_local() -> int:
     """
     cleared = 0
     with tasks_lock:
+        # Protect tasks belonging to a still-active batch. A batch is "active"
+        # while any of its queued tasks is non-terminal (still searching /
+        # downloading / queued / post-processing). Pruning a batch's completed
+        # or failed tasks mid-run would yank them out of the Downloads page —
+        # and failed/cancelled rows aren't recoverable from library_history —
+        # so the user would never see them until the batch ended. Keep the whole
+        # active batch intact; it gets cleaned by a later run once it finishes.
+        protected_task_ids: set = set()
+        for batch in download_batches.values():
+            queue = batch.get('queue', []) if isinstance(batch, dict) else []
+            batch_active = any(
+                download_tasks.get(tid, {}).get('status') not in _TERMINAL_STATUSES
+                for tid in queue if tid in download_tasks
+            )
+            if batch_active:
+                protected_task_ids.update(queue)
+
         task_ids_to_remove = [
             tid for tid, task in download_tasks.items()
-            if task.get('status') in _TERMINAL_STATUSES
+            if task.get('status') in _TERMINAL_STATUSES and tid not in protected_task_ids
         ]
         for tid in task_ids_to_remove:
             del download_tasks[tid]

tests/downloads/test_downloads_cancel.py

@@ -176,15 +176,30 @@ def test_clear_completed_drops_empty_batches():
     assert download_batches['b3']['queue'] == ['t3']
 
 
-def test_clear_completed_prunes_terminal_task_ids_from_batch_queues():
-    """Batch with mix of terminal + active tasks gets queue trimmed, not deleted."""
+def test_clear_completed_keeps_terminal_tasks_in_active_batch():
+    """A completed/failed task in a batch that still has active work is KEPT, so
+    the Downloads page shows it for the whole run (the 5-min Clean Completed
+    automation must not yank terminal tasks out from under a live batch)."""
     download_tasks['t1'] = {'status': 'completed'}
-    download_tasks['t2'] = {'status': 'downloading'}
+    download_tasks['t2'] = {'status': 'downloading'}  # batch still active
     download_batches['b1'] = {'queue': ['t1', 't2']}
 
-    cancel.clear_completed_local()
+    cleared = cancel.clear_completed_local()
+    assert cleared == 0  # nothing removed — batch is live
     assert 'b1' in download_batches
-    assert download_batches['b1']['queue'] == ['t2']
+    assert download_batches['b1']['queue'] == ['t1', 't2']  # queue intact
+
+
+def test_clear_completed_clears_terminal_tasks_once_batch_is_finished():
+    """When every task in a batch is terminal, the batch is done — its tasks are
+    cleared and the empty batch dropped."""
+    download_tasks['t1'] = {'status': 'completed'}
+    download_tasks['t2'] = {'status': 'failed'}
+    download_batches['b1'] = {'queue': ['t1', 't2']}
+
+    cleared = cancel.clear_completed_local()
+    assert cleared == 2
+    assert 'b1' not in download_batches
 
 
 def test_clear_completed_drops_batch_locks_for_deleted_batches():

tests/quality/test_search_all_sources.py

@@ -72,6 +72,37 @@ def test_skips_unconfigured_and_swallows_raising_source():
     assert {t.name for t in tracks} == {'c1'}  # a skipped, b raised, c survives
 
 
+def test_searches_run_concurrently_and_wait_for_slow_source():
+    import time
+
+    class _SlowPlugin:
+        def __init__(self, name, delay):
+            self._name = name
+            self._delay = delay
+            self.searched = False
+
+        def is_configured(self):
+            return True
+
+        async def search(self, query, timeout=None, progress_callback=None):
+            self.searched = True
+            await asyncio.sleep(self._delay)
+            return ([_Cand(self._name)], [])
+
+    slow = _SlowPlugin('usenet', 0.20)   # a slow release-level source
+    fast = _SlowPlugin('hifi', 0.05)
+    eng = _engine_with({'usenet': slow, 'hifi': fast})
+
+    start = time.monotonic()
+    tracks, _ = asyncio.run(eng.search_all_sources('q', ['usenet', 'hifi']))
+    elapsed = time.monotonic() - start
+
+    # Both included — the pool waits for the slow source.
+    assert {t.name for t in tracks} == {'usenet', 'hifi'}
+    # Concurrent: total ≈ max(0.20, 0.05), well under the sequential sum 0.25.
+    assert elapsed < 0.20 + 0.04
+
+
 def test_empty_when_all_sources_empty():
     a = _FakePlugin([])
     eng = _engine_with({'a': a})