feat(dashboard): smarter Retry All Failed with per-source dedup and recent-success skip

The Dashboard's "Retry All Failed" button used to fetch every non-superseded
failure, then fire one retry HTTP call per row. Two missing guards:

- No per-source dedup — a source with N historical failures enqueued N jobs
  for the same URL.
- No "recent success" check — a source that succeeded two hours ago could
  still have stale un-superseded failures from last week that got re-queued.

Adds POST /api/downloads/retry-failed-bulk:

  • Selects non-superseded failures, ordered (source_id ASC, created_at DESC)
  • Per-source dedup keeps only the most recent failure per source
  • Skips sources whose last_success is within the configurable recent window
    (default 24h, via recent_window_hours body param)
  • Resets + enqueues in a single DB transaction, single Celery dispatch loop
  • Returns a per-reason skip breakdown: duplicate_source, recent_success,
    no_source (orphaned rows)

The Dashboard button now calls the bulk endpoint and surfaces the full
skip breakdown in the toast, so the user knows exactly what was (and
wasn't) queued. Logic is extracted into a pure _plan_bulk_retry helper
and unit-tested against SimpleNamespace fixtures — 7 new tests covering
the dedup, window, and orphan paths.

backend/app/api/downloads.py

@@ -114,6 +114,109 @@ async def get_download(download_id: int):
             return jsonify(response)
 
 
+def _plan_bulk_retry(sorted_failures, sources_by_id, recent_cutoff):
+    """Decide which failed downloads to retry given per-source dedup + recent-success filter.
+
+    Args:
+        sorted_failures: Download rows sorted by (source_id ASC, created_at DESC) —
+            the first occurrence of each source_id is the most recent failure.
+        sources_by_id: dict mapping source_id -> Source (for `last_success` lookup).
+        recent_cutoff: datetime — any source whose last_success is >= this is skipped.
+
+    Returns (retry_ids, summary) where summary has skip counts per reason.
+    """
+    seen_sources = set()
+    retry_ids = []
+    skipped_dup = 0
+    skipped_recent = 0
+    skipped_no_source = 0
+
+    for failure in sorted_failures:
+        if failure.source_id is None:
+            skipped_no_source += 1
+            continue
+        if failure.source_id in seen_sources:
+            skipped_dup += 1
+            continue
+        seen_sources.add(failure.source_id)
+
+        source = sources_by_id.get(failure.source_id)
+        if source and source.last_success and source.last_success >= recent_cutoff:
+            skipped_recent += 1
+            continue
+
+        retry_ids.append(failure.id)
+
+    return retry_ids, {
+        "retried": len(retry_ids),
+        "skipped_duplicate_source": skipped_dup,
+        "skipped_recent_success": skipped_recent,
+        "skipped_no_source": skipped_no_source,
+    }
+
+
+@bp.route("/retry-failed-bulk", methods=["POST"])
+async def retry_failed_bulk():
+    """Bulk-retry non-superseded failed downloads, with smart dedup.
+
+    Applies two filters that the old per-row Dashboard loop missed:
+    1. Per-source dedup — a source with N historical failures enqueues once.
+    2. Recent-success skip — a source whose `last_success` is within the
+       recent window (default 24h, overridable via `recent_window_hours`)
+       is considered healthy; we leave its stale failures alone rather
+       than pile on redundant retry work.
+    """
+    data = await request.get_json() or {}
+    recent_window_hours = max(0, int(data.get("recent_window_hours", 24)))
+    recent_cutoff = utcnow() - timedelta(hours=recent_window_hours)
+
+    async with current_app.db_engine.connect() as conn:
+        async with AsyncSession(bind=conn) as session:
+            failures_result = await session.execute(
+                select(Download)
+                .where(
+                    Download.status == DownloadStatus.FAILED,
+                    Download.superseded == False,
+                )
+                .order_by(Download.source_id.asc(), Download.created_at.desc())
+            )
+            failures = list(failures_result.scalars().all())
+
+            source_ids = {f.source_id for f in failures if f.source_id is not None}
+            sources_by_id = {}
+            if source_ids:
+                sources_result = await session.execute(
+                    select(Source).where(Source.id.in_(source_ids))
+                )
+                sources_by_id = {s.id: s for s in sources_result.scalars().all()}
+
+            retry_ids, summary = _plan_bulk_retry(failures, sources_by_id, recent_cutoff)
+
+            for failure in failures:
+                if failure.id in retry_ids:
+                    failure.status = DownloadStatus.QUEUED
+                    failure.error_type = None
+                    failure.error_message = None
+                    failure.started_at = None
+                    failure.completed_at = None
+
+            await session.commit()
+
+    for download_id in retry_ids:
+        process_download.delay(download_id)
+
+    current_app.logger.info(
+        f"retry-failed-bulk: {summary} (recent_window_hours={recent_window_hours})"
+    )
+
+    return jsonify({
+        "message": f"Queued {summary['retried']} retries",
+        **summary,
+        "total_failures_considered": len(failures),
+        "recent_window_hours": recent_window_hours,
+    })
+
+
 @bp.route("/<int:download_id>/retry", methods=["POST"])
 async def retry_download(download_id: int):
     """Retry a failed download."""