fix(download): stop stale aria2 GIDs from spamming errors after queue clears

- Log expected "GID not found" tellStatus probes at DEBUG, and treat a
  forgotten GID as permanent so the poll loop recovers immediately
  instead of burning 4 retries x 3s of ERROR lines per cycle
- cancel_download tolerates a forgotten GID and always pops the
  in-memory transfer so concurrent polls cannot re-register a
  cancelled download
- Restore sweep deletes aria2 state records with no resolvable target
  path instead of skipping them forever
- Clearing the download queue now also cancels in-memory tasks, removes
  live aria2 transfers and drops persisted state for the cleared ids
  (partial files on disk are preserved)

py/services/aria2_downloader.py

@@ -217,8 +217,9 @@ class Aria2Downloader:
         """Call get_status with retry for transient RPC failures.
 
         Only retries on :exc:`Aria2Error` (RPC-level failure).  Returns
-        ``None`` immediately when the download_id is not tracked (a missing
-        transfer is not a transient condition, so retrying is pointless).
+        ``None`` immediately when the transfer is not tracked or its GID is
+        gone from the daemon (a missing transfer is not a transient
+        condition, so retrying is pointless).
 
         A single failed RPC call should not immediately fail the download,
         because aria2 may be temporarily busy (e.g. finalizing multiple
@@ -332,7 +333,13 @@ class Aria2Downloader:
         return transfer
 
     async def get_status(self, download_id: str) -> Optional[Dict[str, Any]]:
-        """Return the raw aria2 status payload for a known download."""
+        """Return the raw aria2 status payload for a known download.
+
+        Returns ``None`` when the download_id is not tracked or the daemon no
+        longer knows the transfer's GID (daemon restart / forceRemove).  A
+        forgotten GID is permanent, not transient, so the caller's recovery
+        path handles it instead of burning retry attempts on a dead GID.
+        """
 
         transfer = self._transfers.get(download_id)
         if transfer is None:
@@ -348,8 +355,17 @@ class Aria2Downloader:
             "files",
         ]
         try:
-            status = await self._rpc_call("aria2.tellStatus", [transfer.gid, keys])
+            status = await self._rpc_call(
+                "aria2.tellStatus", [transfer.gid, keys], log_errors=False
+            )
         except Exception as exc:
+            if "not found" in str(exc).lower():
+                logger.debug(
+                    "aria2 GID %s for download %s is gone; treating as lost transfer",
+                    transfer.gid,
+                    download_id,
+                )
+                return None
             raise Aria2Error(f"Failed to query aria2 download status: {exc}") from exc
 
         if isinstance(status, dict):
@@ -367,7 +383,9 @@ class Aria2Downloader:
             "files",
         ]
         try:
-            status = await self._rpc_call("aria2.tellStatus", [gid, keys])
+            status = await self._rpc_call(
+                "aria2.tellStatus", [gid, keys], log_errors=False
+            )
         except Exception as exc:
             message = str(exc)
             if "cannot be found" in message.lower() or "not found" in message.lower():
@@ -434,8 +452,19 @@ class Aria2Downloader:
         try:
             await self._rpc_call("aria2.forceRemove", [transfer.gid])
         except Exception as exc:
-            return {"success": False, "error": str(exc)}
+            if "not found" not in str(exc).lower():
+                return {"success": False, "error": str(exc)}
+            # The daemon already forgot this GID (restart / prior removal),
+            # so the transfer is effectively cancelled.
+            logger.debug(
+                "aria2 GID %s for download %s already gone during cancel",
+                transfer.gid,
+                download_id,
+            )
 
+        # Drop the in-memory entry as well so a concurrent poll loop does
+        # not mistake the removal for a lost transfer and re-register it.
+        self._transfers.pop(download_id, None)
         await self._state_store.remove(download_id)
         return {"success": True, "message": "Download cancelled successfully"}
 
@@ -725,7 +754,9 @@ class Aria2Downloader:
 
         return isinstance(result, dict)
 
-    async def _rpc_call(self, method: str, params: list[Any]) -> Any:
+    async def _rpc_call(
+        self, method: str, params: list[Any], *, log_errors: bool = True
+    ) -> Any:
         if not self._rpc_url:
             raise Aria2Error("aria2 RPC endpoint is not initialized")
 
@@ -756,7 +787,10 @@ class Aria2Downloader:
             error = body["error"] or {}
             code = error.get("code") if isinstance(error, dict) else None
             message = error.get("message") if isinstance(error, dict) else str(error)
-            logger.error(
+            # Probing calls (e.g. tellStatus for a GID the daemon may have
+            # forgotten) pass log_errors=False: an expected "not found" must
+            # not spam the log at ERROR level.
+            (logger.error if log_errors else logger.debug)(
                 "aria2 RPC %s failed with HTTP %s, code=%s, message=%s",
                 method,
                 response.status,
@@ -771,7 +805,7 @@ class Aria2Downloader:
             raise Aria2Error(status_message or "Unknown aria2 RPC error")
 
         if response.status != 200:
-            logger.error(
+            (logger.error if log_errors else logger.debug)(
                 "aria2 RPC %s returned unexpected HTTP status %s without error payload: %s",
                 method,
                 response.status,

py/services/download_coordinator.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, Awaitable, Callable, Dict, Optional
+from typing import Any, Awaitable, Callable, Dict, Iterable, Optional
 
 from .downloader import DownloadProgress
 
@@ -186,6 +186,14 @@ class DownloadCoordinator:
         download_manager = await self._download_manager_factory()
         return await download_manager.get_active_downloads()
 
+    async def discard_cleared_downloads(self, download_ids: Iterable[str]) -> int:
+        """Tear down in-memory/aria2 tracking for queue-cleared downloads."""
+
+        if not download_ids:
+            return 0
+        download_manager = await self._download_manager_factory()
+        return await download_manager.discard_cleared_downloads(download_ids)
+
     def _parse_optional_int(self, value: Any, field: str) -> Optional[int]:
         """Parse an optional integer from user input."""
 

py/services/download_manager.py

@@ -13,7 +13,7 @@ import zipfile
 from concurrent.futures import ThreadPoolExecutor
 from collections import OrderedDict
 import uuid
-from typing import Any, Dict, List, Optional, Set, Tuple, cast
+from typing import Any, Dict, Iterable, List, Optional, Set, Tuple, cast
 from urllib.parse import urlparse
 from ..utils.models import LoraMetadata, CheckpointMetadata, EmbeddingMetadata
 from ..utils.constants import (
@@ -1100,6 +1100,11 @@ class DownloadManager:
 
                 save_path = self._resolve_save_path_from_persisted_record(record)
                 if save_path is None:
+                    # No resolvable target path (e.g. a queued download whose
+                    # paths were never resolved before shutdown): the record
+                    # can never be restored, so drop it instead of letting it
+                    # accumulate in the state store forever.
+                    await self._aria2_state_store.remove(download_id)
                     continue
 
                 if (
@@ -2897,6 +2902,64 @@ class DownloadManager:
             # Preserve aria2 state store entry so the partial download
             # info survives restarts and can be resumed later
 
+    async def discard_cleared_downloads(self, download_ids: Iterable[str]) -> int:
+        """Stop in-memory tracking for downloads cleared from the queue.
+
+        Cancels asyncio tasks, removes live aria2 transfers and drops the
+        persisted aria2 state so cleared downloads cannot keep polling the
+        daemon or be resurrected as ghost entries on the next restart.
+        Partial files on disk are preserved; unlike ``cancel_download`` no
+        files are deleted.
+
+        Returns the number of downloads that had any in-memory or persisted
+        tracking removed.
+        """
+        discarded = 0
+        aria2_downloader = None
+
+        for download_id in download_ids:
+            task = self._download_tasks.get(download_id)
+            info = self._active_downloads.get(download_id)
+            persisted = await self._aria2_state_store.get(download_id)
+            if task is None and info is None and persisted is None:
+                continue
+
+            discarded += 1
+
+            if task is not None:
+                task.cancel()
+
+            pause_control = self._pause_events.pop(download_id, None)
+            if pause_control is not None:
+                pause_control.resume()
+
+            if task is not None:
+                try:
+                    await asyncio.wait_for(asyncio.shield(task), timeout=2.0)
+                except (asyncio.CancelledError, asyncio.TimeoutError):
+                    pass
+
+            self._download_tasks.pop(download_id, None)
+            self._active_downloads.pop(download_id, None)
+
+            backend = (info or persisted or {}).get("transfer_backend") or "python"
+            if backend == "aria2":
+                if aria2_downloader is None:
+                    aria2_downloader = await get_aria2_downloader()
+                if await aria2_downloader.has_transfer(download_id):
+                    try:
+                        await aria2_downloader.cancel_download(download_id)
+                    except Exception as exc:
+                        logger.warning(
+                            "Failed to remove aria2 transfer for cleared download %s: %s",
+                            download_id,
+                            exc,
+                        )
+
+            await self._aria2_state_store.remove(download_id)
+
+        return discarded
+
     async def pause_download(self, download_id: str) -> Dict[str, Any]:
         """Pause an active download without losing progress."""
 

py/services/download_queue_service.py

@@ -6,7 +6,7 @@ import logging
 import os
 import sqlite3
 import time
-from typing import Any, Optional
+from typing import Any, List, Optional
 
 from ..utils.cache_paths import get_cache_base_dir
 
@@ -390,23 +390,31 @@ class DownloadQueueService:
             conn.commit()
         return True
 
-    async def clear_queue(self, status_filter: Optional[str] = None) -> int:
+    async def clear_queue(self, status_filter: Optional[str] = None) -> List[str]:
         """Remove items from the queue.
 
         When *status_filter* is provided only items with that status are
-        deleted.  Returns the number of deleted rows.
+        deleted.  Returns the ``download_id`` values of the deleted rows so
+        callers can also tear down any in-memory tracking for them.
         """
         async with self._lock:
             conn = self._get_conn()
             if status_filter is not None:
-                cursor = conn.execute(
+                rows = conn.execute(
+                    "SELECT download_id FROM download_queue WHERE status = ?",
+                    (status_filter,),
+                ).fetchall()
+                conn.execute(
                     "DELETE FROM download_queue WHERE status = ?",
                     (status_filter,),
                 )
             else:
-                cursor = conn.execute("DELETE FROM download_queue")
+                rows = conn.execute(
+                    "SELECT download_id FROM download_queue"
+                ).fetchall()
+                conn.execute("DELETE FROM download_queue")
             conn.commit()
-        return cursor.rowcount
+        return [row["download_id"] for row in rows]
 
     async def complete_download(
         self,