fix(asyncio): track strong refs on orphan create_task sites

  asyncio holds only a weak reference to tasks returned by
  ``create_task``. Fire-and-forget callers that discard the return value
  let the event loop GC the task before it finishes, logging
  ``Task was destroyed but it is pending!`` with no traceback. The #1648
  support-bundle review surfaced 94 such warnings in 8 days of v0.2.4.5
  -- the silently-vanished exceptions reach support bundles as opaque
  GC notices instead of actionable errors.

  New backend/app/core/tasks.py::spawn_background_task(coro, *, name=None)
  is the one place in the codebase that calls asyncio.create_task. It
  stores the task in a module-level set, attaches a done-callback that
  auto-removes on completion AND surfaces any uncaught exception via the
  logger with the originating traceback, and accepts name= so a leak
  source is traceable through /tracebacks and the log line. Cancelled
  tasks don't log (a shutting-down service is not an error).

  Migrated the 16 truly-orphan create_task call sites to the helper:

    main.py (8):
      reconcile-stale, cooldown-poweroff, energy calc, smart-plug,
      maintenance-check, photo-then-notify, layer-timelapse,
      scan-timelapse, print-scheduler, notify-no-archive (the last one
      was hand-rolling the same pattern with task + no-op done_callback)
    printers.py:3123        apply-pa-after-refresh
    print_queue.py:1034     queue cooldown-poweroff
    firmware_update.py:261  firmware upload
    archive.py:1514         timelapse mp4 convert
    print_scheduler.py:2199 watchdog print-start
    library.py:1614         STL backfill
    smart_plugs.py:259      tasmota scan
    discovery.py:159        subnet scan
    smart_plug_manager.py   x3 plug auto-off-pending
    background_dispatch.py  x2 (lambda-wrapped inside
                            loop.call_soon_threadsafe) upload progress

  Sites that already kept strong refs are unchanged:
    self._tasks.append(asyncio.create_task(...)) -- VP manager,
      tcp_proxy, mqtt_server
    self._x_task = asyncio.create_task(...) on service instances --
      mqtt_bridge, obico_detection, github_backup, archive_purge,
      local_backup, library_trash, discovery service
    Locally assigned + awaited/gathered -- tcp_proxy bidirectional
      pumps, camera_fanout, slice_dispatch, slicer_api progress_task,
      manager._finish_release_task, main.py module-level cleanup loops

backend/app/api/routes/discovery.py

@@ -12,6 +12,7 @@ from pydantic import BaseModel
 
 from backend.app.core.auth import RequirePermissionIfAuthEnabled
 from backend.app.core.permissions import Permission
+from backend.app.core.tasks import spawn_background_task
 from backend.app.models.user import User
 from backend.app.services.discovery import (
     discovery_service,
@@ -154,9 +155,10 @@ async def start_subnet_scan(
         request: Subnet to scan in CIDR notation (e.g., "192.168.1.0/24")
     """
     # Start scan in background
-    import asyncio
-
-    asyncio.create_task(subnet_scanner.scan_subnet(request.subnet, request.timeout))
+    spawn_background_task(
+        subnet_scanner.scan_subnet(request.subnet, request.timeout),
+        name=f"subnet-scan-{request.subnet}",
+    )
 
     # Return immediate status
     scanned, total = subnet_scanner.progress

backend/app/api/routes/library.py

@@ -1,6 +1,5 @@
 """API routes for File Manager (Library) functionality."""
 
-import asyncio
 import base64
 import binascii
 import contextlib
@@ -30,6 +29,7 @@ from backend.app.core.auth import (
 from backend.app.core.config import settings as app_settings
 from backend.app.core.database import async_session, get_db
 from backend.app.core.permissions import Permission
+from backend.app.core.tasks import spawn_background_task
 from backend.app.models.archive import PrintArchive
 from backend.app.models.library import LibraryFile, LibraryFolder
 from backend.app.models.print_queue import PrintQueueItem
@@ -1611,7 +1611,7 @@ async def scan_external_folder(
     # folder_cache.values() covers the root + every pre-existing subfolder
     # + every subfolder created during this scan. all_folder_ids on its own
     # would miss the newly-created ones (it's snapshotted before the walk).
-    asyncio.create_task(
+    spawn_background_task(
         _backfill_external_stl_thumbnails(list(set(folder_cache.values()))),
         name=f"stl-backfill-folder-{folder_id}",
     )

backend/app/api/routes/print_queue.py

@@ -16,6 +16,7 @@ from backend.app.core.auth import RequirePermissionIfAuthEnabled, require_owners
 from backend.app.core.config import settings
 from backend.app.core.database import get_db
 from backend.app.core.permissions import Permission
+from backend.app.core.tasks import spawn_background_task
 from backend.app.models.archive import PrintArchive
 from backend.app.models.library import LibraryFile
 from backend.app.models.print_batch import PrintBatch
@@ -961,7 +962,6 @@ async def stop_queue_item(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_UPDATE_ALL),
 ):
     """Stop an actively printing queue item."""
-    import asyncio
 
     from backend.app.models.smart_plug import SmartPlug
     from backend.app.services.printer_manager import printer_manager
@@ -1031,7 +1031,7 @@ async def stop_queue_item(
                     logger.info("Auto-off: Powering off printer %s", printer_id)
                     await tasmota_service.turn_off(plug)
 
-        asyncio.create_task(cooldown_and_poweroff())
+        spawn_background_task(cooldown_and_poweroff(), name=f"queue-cooldown-poweroff-{printer_id}")
 
     return {"message": "Print stopped" if stop_sent else "Queue item cancelled (printer was offline)"}
 

backend/app/api/routes/printers.py

@@ -12,6 +12,7 @@ from backend.app.core.auth import RequireCameraStreamTokenIfAuthEnabled, Require
 from backend.app.core.config import settings
 from backend.app.core.database import get_db
 from backend.app.core.permissions import Permission
+from backend.app.core.tasks import spawn_background_task
 from backend.app.models.ams_label import AmsLabel
 from backend.app.models.printer import Printer
 from backend.app.models.slot_preset import SlotPresetMapping
@@ -3120,7 +3121,10 @@ async def refresh_ams_slot(
         raise HTTPException(400, message)
 
     # Apply PA profile after delay (RFID re-read takes a few seconds)
-    asyncio.create_task(_apply_pa_after_refresh(printer_id, ams_id, slot_id))
+    spawn_background_task(
+        _apply_pa_after_refresh(printer_id, ams_id, slot_id),
+        name=f"apply-pa-after-refresh-{printer_id}-{ams_id}-{slot_id}",
+    )
 
     return {"success": True, "message": message}
 

backend/app/api/routes/smart_plugs.py

@@ -12,6 +12,7 @@ from backend.app.api.routes.settings import get_setting
 from backend.app.core.auth import RequirePermissionIfAuthEnabled
 from backend.app.core.database import get_db
 from backend.app.core.permissions import Permission
+from backend.app.core.tasks import spawn_background_task
 from backend.app.models.printer import Printer
 from backend.app.models.smart_plug import SmartPlug
 from backend.app.models.user import User
@@ -249,14 +250,16 @@ async def start_tasmota_scan(
 
     Auto-detects local network if no IP range provided.
     """
-    import asyncio
 
     # Auto-detect network
     from_ip, to_ip = get_local_network_range()
     timeout = request.timeout if request else 1.0
 
     # Start scan in background
-    asyncio.create_task(tasmota_scanner.scan_range(from_ip, to_ip, timeout))
+    spawn_background_task(
+        tasmota_scanner.scan_range(from_ip, to_ip, timeout),
+        name="tasmota-scan",
+    )
 
     # Return immediate status
     scanned, total = tasmota_scanner.progress

backend/app/core/tasks.py

@@ -0,0 +1,85 @@
+"""Background-task helper that keeps a strong reference to fire-and-forget tasks.
+
+asyncio holds only a weak reference to tasks returned by ``create_task`` --
+when the caller discards the return value (the "fire and forget" pattern),
+the task can be garbage-collected mid-execution and the event loop logs
+``Task was destroyed but it is pending!`` with no traceback. A support
+bundle review under #1648 surfaced 94 such warnings in 8 days of v0.2.4.5.
+
+``spawn_background_task`` is the one place in the codebase that calls
+``asyncio.create_task``. It stores the task in a module-level set, removes
+it when the task completes, and surfaces any uncaught exception through
+the logger so a silently-swallowed error becomes a visible WARNING with
+the originating traceback instead of an opaque GC warning.
+
+Use this for any work that should run in the background without being
+awaited inline. For tasks that the service owns and needs to cancel on
+shutdown, store the returned ``asyncio.Task`` on the service instance
+instead (the helper still adds the strong reference, so storing it twice
+is redundant but harmless).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from collections.abc import Coroutine
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# Strong-reference holder. Tasks live here from creation through completion.
+# Module-level so the set survives across spawn calls; the done-callback
+# removes each task as it finishes so the set doesn't grow without bound
+# (the event loop's GC can't reap an entry the callback still holds, but
+# the discard breaks the cycle immediately).
+_background_tasks: set[asyncio.Task[Any]] = set()
+
+
+def spawn_background_task(
+    coro: Coroutine[Any, Any, Any],
+    *,
+    name: str | None = None,
+) -> asyncio.Task[Any]:
+    """Schedule ``coro`` on the running loop without losing the task reference.
+
+    Args:
+        coro: The coroutine to run. Must not already be a Task.
+        name: Optional task name surfaced in /tracebacks and the
+            done-callback log line so a leaked task is traceable to its
+            spawn site.
+
+    Returns:
+        The created ``asyncio.Task``. Most callers ignore it -- the helper
+        keeps its own strong reference. Callers that need to ``await`` or
+        cancel later can store it on a service instance.
+    """
+    task = asyncio.create_task(coro, name=name)
+    _background_tasks.add(task)
+    task.add_done_callback(_on_task_done)
+    return task
+
+
+def _on_task_done(task: asyncio.Task[Any]) -> None:
+    """Discard the strong reference and surface any uncaught exception.
+
+    Without this, an exception raised inside a fire-and-forget task is
+    silently retrieved by ``Task.__del__`` and never reaches the logger.
+    Surface it here as a WARNING with the task name so support bundles
+    capture the originating error instead of an opaque GC notice.
+    """
+    _background_tasks.discard(task)
+    if task.cancelled():
+        return
+    exc = task.exception()
+    if exc is not None:
+        logger.warning(
+            "Background task %r raised an uncaught exception",
+            task.get_name(),
+            exc_info=exc,
+        )
+
+
+def active_task_count() -> int:
+    """Number of background tasks currently in flight. Used by tests."""
+    return len(_background_tasks)

backend/app/main.py

@@ -72,6 +72,7 @@ from backend.app.api.routes.maintenance import _get_printer_maintenance_internal
 from backend.app.api.routes.support import init_debug_logging
 from backend.app.core.config import APP_VERSION, settings as app_settings
 from backend.app.core.database import async_session, engine, init_db
+from backend.app.core.tasks import spawn_background_task
 from backend.app.core.websocket import ws_manager
 from backend.app.models.smart_plug import SmartPlug
 from backend.app.services.archive import ArchiveService, peek_plate_index_in_3mf, swap_plate_suffix
@@ -823,7 +824,10 @@ async def on_printer_status_change(printer_id: int, state: PrinterState):
     # the same connection don't re-trigger reconciliation.
     if state.connected and not _printer_reconciled_since_connect.get(printer_id, False):
         _printer_reconciled_since_connect[printer_id] = True
-        asyncio.create_task(reconcile_stale_active_prints(printer_id))
+        spawn_background_task(
+            reconcile_stale_active_prints(printer_id),
+            name=f"reconcile-stale-prints-{printer_id}",
+        )
     elif not state.connected and _printer_reconciled_since_connect.get(printer_id, False):
         # Re-arm so the next reconnect triggers reconciliation again.
         _printer_reconciled_since_connect[printer_id] = False
@@ -3839,7 +3843,10 @@ async def on_print_complete(printer_id: int, data: dict):
                                 except Exception as e:
                                     logger.warning("Failed to power off plug %s for printer %s: %s", plug_id, pid, e)
 
-                    asyncio.create_task(cooldown_and_poweroff(printer_id, [p.id for p in enabled_plugs]))
+                    spawn_background_task(
+                        cooldown_and_poweroff(printer_id, [p.id for p in enabled_plugs]),
+                        name=f"cooldown-poweroff-{printer_id}",
+                    )
     except Exception as e:
         logging.getLogger(__name__).warning(f"Queue item update failed: {e}")
 
@@ -4024,8 +4031,7 @@ async def on_print_complete(printer_id: int, data: dict):
             except Exception as e:
                 logger.warning("[NOTIFY-BG] Failed to send notification without archive: %s", e, exc_info=True)
 
-        task = asyncio.create_task(_notify_no_archive())
-        task.add_done_callback(lambda _t: None)
+        spawn_background_task(_notify_no_archive(), name="notify-no-archive")
         return
 
     log_timing("Archive lookup")
@@ -4363,9 +4369,9 @@ async def on_print_complete(printer_id: int, data: dict):
             logger.warning("[PHOTO-BG] Failed: %s", e)
             return None
 
-    asyncio.create_task(_background_energy_calculation())
+    spawn_background_task(_background_energy_calculation(), name="background-energy-calc")
     # Photo capture task - result will be used by notifications
-    photo_task = asyncio.create_task(_background_finish_photo())
+    photo_task = spawn_background_task(_background_finish_photo(), name="background-finish-photo")
     log_timing("Background tasks scheduled (energy, photo)")
 
     # Also run smart plug, notifications, and maintenance as background tasks
@@ -4544,8 +4550,8 @@ async def on_print_complete(printer_id: int, data: dict):
         except Exception as e:
             logger.warning("[MAINT-BG] Failed: %s", e)
 
-    asyncio.create_task(_background_smart_plug())
-    asyncio.create_task(_background_maintenance_check())
+    spawn_background_task(_background_smart_plug(), name="background-smart-plug")
+    spawn_background_task(_background_maintenance_check(), name="background-maintenance-check")
 
     # Notification task waits for photo capture to complete first (with timeout).
     # When a timelapse was recording, photo sourcing polls the per-print
@@ -4572,7 +4578,7 @@ async def on_print_complete(printer_id: int, data: dict):
         except Exception as e:
             logger.error("[PHOTO-NOTIFY] Notification sending failed: %s", e, exc_info=True)
 
-    asyncio.create_task(_photo_then_notify())
+    spawn_background_task(_photo_then_notify(), name="photo-then-notify")
 
     # Stitch external camera layer timelapse if session was active
     print_status = data.get("status", "completed")
@@ -4611,7 +4617,7 @@ async def on_print_complete(printer_id: int, data: dict):
             except Exception:
                 pass  # Best-effort timelapse session cancellation on error
 
-    asyncio.create_task(_background_layer_timelapse())
+    spawn_background_task(_background_layer_timelapse(), name="background-layer-timelapse")
 
     log_timing("All background tasks scheduled")
 
@@ -4621,7 +4627,10 @@ async def on_print_complete(printer_id: int, data: dict):
         # Schedule timelapse scan as background task with retries
         # The printer needs time to encode the video after print completion
         baseline = _timelapse_baselines.pop(printer_id, None)
-        asyncio.create_task(_scan_for_timelapse_with_retries(archive_id, baseline))
+        spawn_background_task(
+            _scan_for_timelapse_with_retries(archive_id, baseline),
+            name=f"scan-timelapse-{archive_id}",
+        )
         log_timing("Timelapse scan scheduled")
 
     logger.info("[CALLBACK] on_print_complete finished for printer %s, archive %s", printer_id, archive_id)
@@ -5442,7 +5451,7 @@ async def lifespan(app: FastAPI):
                 logging.warning("Failed to auto-connect to Spoolman: %s", e)
 
     # Start the print scheduler
-    asyncio.create_task(print_scheduler.run())
+    spawn_background_task(print_scheduler.run(), name="print-scheduler")
 
     # Start background dispatch worker for send/start operations
     await background_dispatch.start()

backend/app/services/archive.py

@@ -13,6 +13,7 @@ from sqlalchemy import and_, or_, select, text
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.app.core.config import settings
+from backend.app.core.tasks import spawn_background_task
 from backend.app.models.archive import PrintArchive
 from backend.app.models.filament import Filament
 from backend.app.models.printer import Printer
@@ -1511,7 +1512,7 @@ class ArchiveService:
 
         # For non-MP4 videos (e.g. AVI from P1S), kick off background conversion
         if not filename.lower().endswith(".mp4"):
-            asyncio.create_task(
+            spawn_background_task(
                 _convert_timelapse_to_mp4(archive_id, timelapse_file),
                 name=f"timelapse-convert-{archive_id}",
             )

backend/app/services/background_dispatch.py

@@ -20,6 +20,7 @@ from sqlalchemy import select
 
 from backend.app.core.config import settings
 from backend.app.core.database import async_session
+from backend.app.core.tasks import spawn_background_task
 from backend.app.core.websocket import ws_manager
 from backend.app.models.library import LibraryFile
 from backend.app.models.printer import Printer
@@ -629,7 +630,10 @@ class BackgroundDispatchService:
                         progress_state["last_emit"] = now
                         progress_state["last_bytes"] = uploaded
                         loop.call_soon_threadsafe(
-                            lambda u=uploaded, t=total: asyncio.create_task(self._set_active_upload_progress(job, u, t))
+                            lambda u=uploaded, t=total: spawn_background_task(
+                                self._set_active_upload_progress(job, u, t),
+                                name=f"upload-progress-{job.id}",
+                            )
                         )
 
                 if ftp_retry_enabled:
@@ -828,7 +832,10 @@ class BackgroundDispatchService:
                         progress_state["last_emit"] = now
                         progress_state["last_bytes"] = uploaded
                         loop.call_soon_threadsafe(
-                            lambda u=uploaded, t=total: asyncio.create_task(self._set_active_upload_progress(job, u, t))
+                            lambda u=uploaded, t=total: spawn_background_task(
+                                self._set_active_upload_progress(job, u, t),
+                                name=f"upload-progress-{job.id}",
+                            )
                         )
 
                 if ftp_retry_enabled:

backend/app/services/firmware_update.py

@@ -16,6 +16,7 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.app.core.compat import StrEnum
+from backend.app.core.tasks import spawn_background_task
 from backend.app.core.websocket import ws_manager
 from backend.app.models.printer import Printer
 from backend.app.services.bambu_ftp import (
@@ -258,14 +259,15 @@ class FirmwareUpdateService:
         await self._broadcast_progress(printer_id, state)
 
         # Run the upload in background
-        asyncio.create_task(
+        spawn_background_task(
             self._do_upload(
                 printer_id=printer_id,
                 ip_address=printer.ip_address,
                 access_code=printer.access_code,
                 model=model,
                 target_version=target_version,
-            )
+            ),
+            name=f"firmware-upload-{printer_id}",
         )
 
         return True

backend/app/services/print_scheduler.py

@@ -13,6 +13,7 @@ from sqlalchemy.orm import selectinload
 
 from backend.app.core.config import settings
 from backend.app.core.database import async_session, run_with_retry
+from backend.app.core.tasks import spawn_background_task
 from backend.app.models.archive import PrintArchive
 from backend.app.models.library import LibraryFile
 from backend.app.models.print_queue import PrintQueueItem
@@ -2196,14 +2197,15 @@ class PrintScheduler:
             # that would otherwise cause the item to re-dispatch as a reprint
             # of the just-finished job (#1078).
             if pre_state:
-                asyncio.create_task(
+                spawn_background_task(
                     self._watchdog_print_start(
                         item.id,
                         item.printer_id,
                         pre_state,
                         pre_subtask_id,
                         pre_gcode_file,
-                    )
+                    ),
+                    name=f"watchdog-print-start-{item.id}",
                 )
 
             # Get estimated time for notification

backend/app/services/smart_plug_manager.py

@@ -8,6 +8,7 @@ from typing import TYPE_CHECKING
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from backend.app.core.tasks import spawn_background_task
 from backend.app.services.homeassistant import homeassistant_service
 from backend.app.services.printer_manager import printer_manager
 from backend.app.services.rest_smart_plug import rest_smart_plug_service
@@ -340,7 +341,7 @@ class SmartPlugManager:
         logger.info("Scheduling turn-off for plug '%s' in %s seconds", plug.name, delay_seconds)
 
         # Mark as pending in database (survives restarts)
-        asyncio.create_task(self._mark_auto_off_pending(plug.id, True))
+        spawn_background_task(self._mark_auto_off_pending(plug.id, True), name=f"plug-auto-off-pending-{plug.id}")
 
         task = asyncio.create_task(
             self._delayed_off(
@@ -419,7 +420,7 @@ class SmartPlugManager:
         logger.info("Scheduling temperature-based turn-off for plug '%s' (threshold: %s°C)", plug.name, temp_threshold)
 
         # Mark as pending in database (survives restarts)
-        asyncio.create_task(self._mark_auto_off_pending(plug.id, True))
+        spawn_background_task(self._mark_auto_off_pending(plug.id, True), name=f"plug-auto-off-pending-{plug.id}")
 
         task = asyncio.create_task(
             self._temp_based_off(
@@ -579,7 +580,7 @@ class SmartPlugManager:
             self._pending_off[plug_id].cancel()
             del self._pending_off[plug_id]
             # Clear pending state in database
-            asyncio.create_task(self._mark_auto_off_pending(plug_id, False))
+            spawn_background_task(self._mark_auto_off_pending(plug_id, False), name=f"plug-auto-off-pending-{plug_id}")
 
     def cancel_all_pending(self):
         """Cancel all pending turn-off tasks."""

backend/tests/unit/core/test_tasks.py

@@ -0,0 +1,104 @@
+"""Unit tests for the spawn_background_task helper (#1648 follow-up).
+
+asyncio holds only weak references to tasks, so a fire-and-forget
+create_task whose return value is discarded can be GC'd mid-flight and
+log ``Task was destroyed but it is pending!`` with no traceback.
+``spawn_background_task`` is the central helper that fixes this: it
+stores a strong reference until completion, surfaces uncaught exceptions
+through the logger, and auto-removes finished tasks.
+"""
+
+import asyncio
+import logging
+
+import pytest
+
+from backend.app.core.tasks import active_task_count, spawn_background_task
+
+
+@pytest.mark.asyncio
+async def test_holds_strong_ref_until_completion():
+    """Discarding the returned task must not let asyncio reap it mid-flight.
+    Pre-fix, ``asyncio.create_task(coro)`` with no caller-side reference
+    would let GC swallow short tasks before they finished."""
+    finished = asyncio.Event()
+
+    async def work() -> None:
+        await asyncio.sleep(0)
+        finished.set()
+
+    # Note: NOT storing the returned task -- this is exactly the
+    # pattern the helper exists to support.
+    spawn_background_task(work())
+    await asyncio.wait_for(finished.wait(), timeout=1.0)
+
+
+@pytest.mark.asyncio
+async def test_removes_from_strong_ref_set_after_completion():
+    """The strong-ref set must shrink as tasks complete; otherwise a
+    long-running process accumulates one entry per spawned task and the
+    helper itself becomes a leak."""
+    before = active_task_count()
+
+    async def work() -> None:
+        await asyncio.sleep(0)
+
+    spawn_background_task(work())
+    spawn_background_task(work())
+    # Yield enough times for both tasks + their done-callbacks to run.
+    for _ in range(5):
+        await asyncio.sleep(0)
+    assert active_task_count() == before
+
+
+@pytest.mark.asyncio
+async def test_uncaught_exception_logged_as_warning(caplog):
+    """A fire-and-forget task that raises must surface the exception via
+    the logger with the traceback attached -- otherwise the error vanishes
+    silently and only an opaque ``Task was destroyed`` notice reaches the
+    support bundle."""
+
+    async def boom() -> None:
+        raise RuntimeError("synthetic failure for test")
+
+    with caplog.at_level(logging.WARNING, logger="backend.app.core.tasks"):
+        spawn_background_task(boom(), name="boom-task")
+        for _ in range(5):
+            await asyncio.sleep(0)
+
+    # One WARNING with the task name and the exception info.
+    boom_records = [r for r in caplog.records if "boom-task" in r.message]
+    assert len(boom_records) == 1
+    assert boom_records[0].levelno == logging.WARNING
+    assert boom_records[0].exc_info is not None
+    assert isinstance(boom_records[0].exc_info[1], RuntimeError)
+
+
+@pytest.mark.asyncio
+async def test_cancelled_task_does_not_log_exception(caplog):
+    """Explicit cancellation isn't an error -- a service shutting down
+    its background loops should not be reported as 'uncaught exception'."""
+
+    async def long_running() -> None:
+        await asyncio.sleep(10.0)
+
+    with caplog.at_level(logging.WARNING, logger="backend.app.core.tasks"):
+        task = spawn_background_task(long_running(), name="cancel-me")
+        await asyncio.sleep(0)  # Let it start.
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+    assert not any("cancel-me" in r.message for r in caplog.records)
+
+
+@pytest.mark.asyncio
+async def test_task_name_propagates():
+    """Named tasks make the leak source visible in tracebacks and the
+    done-callback log line. Pin that ``name`` reaches the underlying
+    Task so support bundles surface the spawn site."""
+    task = spawn_background_task(asyncio.sleep(0), name="named-spawn-test")
+    assert task.get_name() == "named-spawn-test"
+    await task