feat(trash): daily retention purge scheduler (03:30 UTC) wired into app lifecycle

src/fabledassistant/app.py

@@ -175,6 +175,10 @@ def create_app() -> Quart:
         )
         start_version_pinning_scheduler(asyncio.get_running_loop())
 
+        # Trash retention scheduler (daily expired-trash purge at 03:30 UTC)
+        from fabledassistant.services.trash_scheduler import start_trash_scheduler
+        start_trash_scheduler(asyncio.get_running_loop())
+
         # Diagnostic instrumentation — heartbeat, signal handlers, asyncio
         # exception hook. Cheap (~1 log line/min), high diagnostic value when
         # the app crashes mysteriously. See services/diagnostics.py.
@@ -189,6 +193,8 @@ def create_app() -> Quart:
             stop_version_pinning_scheduler,
         )
         stop_version_pinning_scheduler()
+        from fabledassistant.services.trash_scheduler import stop_trash_scheduler
+        stop_trash_scheduler()
         from fabledassistant.services.diagnostics import stop_diagnostics
         stop_diagnostics()
 

src/fabledassistant/services/trash_scheduler.py

@@ -0,0 +1,70 @@
+"""Daily APScheduler cron that purges expired trash.
+
+Mirrors version_pinning_scheduler.py: a single global BackgroundScheduler job
+at 03:30 UTC bridges into the asyncio loop to run the async purge. Reads the
+operator's `trash_retention_days` setting (single-tenant: user 1); 0 disables
+auto-purge.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from apscheduler.schedulers.background import BackgroundScheduler
+from apscheduler.triggers.cron import CronTrigger
+
+from fabledassistant.services import trash as trash_svc
+from fabledassistant.services.settings import get_setting
+
+logger = logging.getLogger(__name__)
+
+_scheduler: BackgroundScheduler | None = None
+_loop: asyncio.AbstractEventLoop | None = None
+
+
+def _run_purge_threadsafe() -> None:
+    """APScheduler invokes this from a worker thread; bridge into the loop."""
+    if _loop is None:
+        logger.warning("trash scheduler: no loop registered")
+        return
+
+    async def _runner():
+        try:
+            raw = await get_setting(1, "trash_retention_days", "90")
+            try:
+                days = int(raw)
+            except (TypeError, ValueError):
+                days = 90
+            purged = await trash_svc.purge_expired(days)
+            if purged:
+                logger.info("trash purge: removed %d expired row(s)", purged)
+            else:
+                logger.debug("trash purge: nothing expired")
+        except Exception:
+            logger.exception("trash purge run failed")
+
+    asyncio.run_coroutine_threadsafe(_runner(), _loop)
+
+
+def start_trash_scheduler(loop: asyncio.AbstractEventLoop) -> None:
+    global _scheduler, _loop
+    if _scheduler is not None:
+        return
+    _loop = loop
+    _scheduler = BackgroundScheduler()
+    _scheduler.add_job(
+        _run_purge_threadsafe,
+        trigger=CronTrigger(hour=3, minute=30, timezone="UTC"),
+        id="trash_retention_purge",
+        replace_existing=True,
+    )
+    _scheduler.start()
+    logger.info("Trash retention scheduler started (daily 03:30 UTC)")
+
+
+def stop_trash_scheduler() -> None:
+    global _scheduler
+    if _scheduler is not None:
+        _scheduler.shutdown(wait=False)
+        _scheduler = None
+        logger.info("Trash retention scheduler stopped")