diff --git a/backend/app/tasks/backup.py b/backend/app/tasks/backup.py
index 4f374b2..c5181ce 100644
--- a/backend/app/tasks/backup.py
+++ b/backend/app/tasks/backup.py
@@ -12,11 +12,12 @@ from datetime import UTC, datetime
 from pathlib import Path
 
 from celery.exceptions import SoftTimeLimitExceeded
+from sqlalchemy import select
 from sqlalchemy.exc import DBAPIError, OperationalError
 
 from ..celery_app import celery
 from ..config import get_config
-from ..models import BackupRun
+from ..models import BackupRun, ImportSettings
 from ..services import backup_service
 from ._sync_engine import sync_session_factory as _sync_session_factory
 
@@ -220,3 +221,72 @@ def restore_images_task(self, *, source_backup_run_id: int) -> dict:
         row.finished_at = datetime.now(UTC)
         session.commit()
     return {"backup_run_id": marker_id}
+
+
+@celery.task(
+    name="backend.app.tasks.backup.prune_backups",
+    soft_time_limit=300, time_limit=600,
+)
+def prune_backups() -> dict:
+    """Daily Beat. Per-kind retention from ImportSettings.
+
+    Returns {"db_deleted": N, "images_deleted": M, "files_unlinked": K}.
+    Tagged rows (tag IS NOT NULL) are never pruned.
+    Status='running' / 'restoring' rows are never pruned (recovery
+    sweep from FC-3i handles those via task_run).
+    """
+    SessionLocal = _sync_session_factory()
+    counts = {"db_deleted": 0, "images_deleted": 0, "files_unlinked": 0}
+    with SessionLocal() as session:
+        s = session.execute(
+            select(ImportSettings).where(ImportSettings.id == 1)
+        ).scalar_one()
+        for kind, keep in (
+            ("db", s.backup_db_keep_last_n),
+            ("images", s.backup_images_keep_last_n),
+        ):
+            candidates = session.execute(
+                select(BackupRun)
+                .where(BackupRun.kind == kind)
+                .where(BackupRun.tag.is_(None))
+                .where(BackupRun.status.in_(["ok", "error"]))
+                .order_by(BackupRun.started_at.desc())
+                .offset(keep)
+            ).scalars().all()
+            for row in candidates:
+                result = backup_service.unlink_artifact_files(
+                    sql_path=row.sql_path,
+                    tar_path=row.tar_path,
+                    manifest_path=(row.manifest or {}).get("manifest_path"),
+                )
+                counts["files_unlinked"] += sum(
+                    1 for v in result.values() if v
+                )
+                session.delete(row)
+                counts[f"{kind}_deleted"] += 1
+        session.commit()
+    return counts
+
+
+@celery.task(
+    name="backend.app.tasks.backup.backup_db_nightly",
+    soft_time_limit=60, time_limit=120,
+)
+def backup_db_nightly() -> dict:
+    """Hourly tick. Dispatches a real backup ONLY if the configured
+    UTC hour matches and the nightly setting is enabled. Returns
+    either {'skipped': '<reason>'} or {'dispatched': '<task_id>'}."""
+    SessionLocal = _sync_session_factory()
+    with SessionLocal() as session:
+        s = session.execute(
+            select(ImportSettings).where(ImportSettings.id == 1)
+        ).scalar_one()
+        nightly_enabled = s.backup_db_nightly_enabled
+        configured_hour = s.backup_db_nightly_hour_utc
+    if not nightly_enabled:
+        return {"skipped": "nightly disabled"}
+    now_hour = datetime.now(UTC).hour
+    if now_hour != configured_hour:
+        return {"skipped": f"hour={now_hour} != configured={configured_hour}"}
+    res = backup_db_task.delay(triggered_by="nightly")
+    return {"dispatched": res.id}