diff --git a/backend/app/tasks/backup.py b/backend/app/tasks/backup.py
new file mode 100644
index 0000000..3c40afe
--- /dev/null
+++ b/backend/app/tasks/backup.py
@@ -0,0 +1,129 @@
+"""FC-3h: backup/restore Celery tasks.
+
+All tasks live on the maintenance queue (per celery_app.task_routes).
+task_run lifecycle tracking is automatic via FC-3i signals — these
+tasks just record the operator-facing artifact metadata into
+BackupRun.
+"""
+from __future__ import annotations
+
+import logging
+from datetime import UTC, datetime
+from pathlib import Path
+
+from celery.exceptions import SoftTimeLimitExceeded
+from sqlalchemy.exc import DBAPIError, OperationalError
+
+from ..celery_app import celery
+from ..config import get_config
+from ..models import BackupRun
+from ..services import backup_service
+from ._sync_engine import sync_session_factory as _sync_session_factory
+
+log = logging.getLogger(__name__)
+IMAGES_ROOT = Path("/images")
+
+
+def _mark_failed(session, row: BackupRun, exc: BaseException) -> None:
+    """Flip a BackupRun row from running/restoring to error with a
+    truncated error message and finished_at. Caller already holds the
+    session open."""
+    row.status = "error"
+    row.error = f"{type(exc).__name__}: {exc}"[:2000]
+    row.finished_at = datetime.now(UTC)
+    session.add(row)
+    session.commit()
+
+
+@celery.task(
+    name="backend.app.tasks.backup.backup_db_task",
+    bind=True,
+    autoretry_for=(OperationalError, DBAPIError),
+    retry_backoff=10, retry_backoff_max=120, max_retries=2,
+    soft_time_limit=600, time_limit=720,
+)
+def backup_db_task(self, *, tag: str | None = None,
+                   triggered_by: str = "manual") -> dict:
+    """Create one DB backup. Returns {'backup_run_id': N}."""
+    SessionLocal = _sync_session_factory()
+    cfg = get_config()
+    now = datetime.now(UTC)
+    with SessionLocal() as session:
+        row = BackupRun(
+            kind="db", status="running", tag=tag,
+            triggered_by=triggered_by, started_at=now, manifest={},
+        )
+        session.add(row); session.commit(); session.refresh(row)
+        run_id = row.id
+
+    try:
+        result = backup_service.backup_db(
+            db_url=cfg.database_url_sync, images_root=IMAGES_ROOT,
+            tag=tag, triggered_by=triggered_by,
+        )
+    except (SoftTimeLimitExceeded, Exception) as exc:
+        with SessionLocal() as session:
+            row = session.get(BackupRun, run_id)
+            if row is not None:
+                _mark_failed(session, row, exc)
+        raise
+
+    with SessionLocal() as session:
+        row = session.get(BackupRun, run_id)
+        row.status = "ok"
+        row.finished_at = datetime.now(UTC)
+        row.sql_path = result["sql_path"]
+        row.size_bytes = result["size_bytes"]
+        row.manifest = {
+            "manifest_path": result["manifest_path"],
+            "ts": result["ts"],
+        }
+        session.commit()
+    return {"backup_run_id": run_id}
+
+
+@celery.task(
+    name="backend.app.tasks.backup.backup_images_task",
+    bind=True,
+    autoretry_for=(OperationalError, DBAPIError),
+    retry_backoff=30, retry_backoff_max=300, max_retries=1,
+    soft_time_limit=21600, time_limit=23400,
+)
+def backup_images_task(self, *, tag: str | None = None,
+                       triggered_by: str = "manual") -> dict:
+    """Create one images backup. Same shape as backup_db_task; uses
+    tar_path instead of sql_path."""
+    SessionLocal = _sync_session_factory()
+    now = datetime.now(UTC)
+    with SessionLocal() as session:
+        row = BackupRun(
+            kind="images", status="running", tag=tag,
+            triggered_by=triggered_by, started_at=now, manifest={},
+        )
+        session.add(row); session.commit(); session.refresh(row)
+        run_id = row.id
+
+    try:
+        result = backup_service.backup_images(
+            images_root=IMAGES_ROOT,
+            tag=tag, triggered_by=triggered_by,
+        )
+    except (SoftTimeLimitExceeded, Exception) as exc:
+        with SessionLocal() as session:
+            row = session.get(BackupRun, run_id)
+            if row is not None:
+                _mark_failed(session, row, exc)
+        raise
+
+    with SessionLocal() as session:
+        row = session.get(BackupRun, run_id)
+        row.status = "ok"
+        row.finished_at = datetime.now(UTC)
+        row.tar_path = result["tar_path"]
+        row.size_bytes = result["size_bytes"]
+        row.manifest = {
+            "manifest_path": result["manifest_path"],
+            "ts": result["ts"],
+        }
+        session.commit()
+    return {"backup_run_id": run_id}