diff --git a/tests/test_tasks_backup.py b/tests/test_tasks_backup.py
new file mode 100644
index 0000000..fc9de00
--- /dev/null
+++ b/tests/test_tasks_backup.py
@@ -0,0 +1,278 @@
+"""FC-3h: backup/restore Celery task integration tests.
+
+Uses task_always_eager for synchronous in-test execution.
+Subprocess + IMAGES_ROOT are monkeypatched so tests don't touch
+real /images or shell out to pg_dump.
+"""
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+import pytest
+from sqlalchemy import select
+
+from backend.app.celery_app import celery
+from backend.app.models import BackupRun, ImportSettings
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture(autouse=True)
+def _eager_celery(monkeypatch):
+    monkeypatch.setattr(celery.conf, "task_always_eager", True)
+    monkeypatch.setattr(celery.conf, "task_eager_propagates", False)
+
+
+@pytest.fixture(autouse=True)
+def fake_subprocess_and_images_root(monkeypatch, tmp_path):
+    monkeypatch.setattr("backend.app.tasks.backup.IMAGES_ROOT", tmp_path)
+    monkeypatch.setattr(
+        "backend.app.services.backup_service._DB_SUBPROCESS_TIMEOUT_S", 5,
+    )
+
+    class _FakeProc:
+        returncode = 0; stdout = b""; stderr = b""
+
+    def _fake_run(cmd, **kwargs):
+        if cmd[0] == "pg_dump":
+            i = cmd.index("-f")
+            Path(cmd[i + 1]).write_bytes(b"-- fake pg_dump\n")
+        elif cmd[0] == "tar" and "-cf" in cmd:
+            i = cmd.index("-cf")
+            Path(cmd[i + 1]).write_bytes(b"fake tar payload")
+        return _FakeProc()
+
+    monkeypatch.setattr("subprocess.run", _fake_run)
+
+
+def _seed_backup(db_sync, *, kind, status, started_at, tag=None,
+                 finished_at=None):
+    row = BackupRun(
+        kind=kind, status=status, tag=tag,
+        triggered_by="manual", started_at=started_at,
+        finished_at=finished_at or (started_at + timedelta(seconds=10)),
+        sql_path="/tmp/fake.sql" if kind == "db" else None,
+        tar_path="/tmp/fake.tar.zst" if kind == "images" else None,
+        manifest={},
+    )
+    db_sync.add(row); db_sync.flush()
+    return row.id
+
+
+# --- backup_db_task --------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_backup_db_task_creates_backup_run_row_status_ok(db_sync):
+    from backend.app.tasks.backup import backup_db_task
+    result = backup_db_task.delay(tag=None, triggered_by="manual").get()
+    run_id = result["backup_run_id"]
+    row = db_sync.execute(
+        select(
+            BackupRun.kind, BackupRun.status, BackupRun.sql_path,
+            BackupRun.size_bytes, BackupRun.finished_at, BackupRun.error,
+        ).where(BackupRun.id == run_id)
+    ).one()
+    assert row.kind == "db"
+    assert row.status == "ok"
+    assert row.sql_path and row.sql_path.endswith(".sql")
+    assert row.size_bytes is not None and row.size_bytes > 0
+    assert row.finished_at is not None
+    assert row.error is None
+
+
+@pytest.mark.asyncio
+async def test_backup_db_task_records_failure_on_subprocess_error(db_sync, monkeypatch):
+    from backend.app.tasks.backup import backup_db_task
+
+    def _boom(*a, **kw):
+        raise RuntimeError("synthetic pg_dump fail")
+    monkeypatch.setattr("subprocess.run", _boom)
+
+    with pytest.raises(RuntimeError):
+        backup_db_task.delay().get()
+
+    row = db_sync.execute(
+        select(BackupRun.status, BackupRun.error)
+        .where(BackupRun.kind == "db")
+        .order_by(BackupRun.id.desc()).limit(1)
+    ).one()
+    assert row.status == "error"
+    assert "synthetic" in (row.error or "")
+
+
+@pytest.mark.asyncio
+async def test_backup_db_task_persists_tag(db_sync):
+    from backend.app.tasks.backup import backup_db_task
+    result = backup_db_task.delay(tag="pre-cutover").get()
+    tag = db_sync.execute(
+        select(BackupRun.tag).where(BackupRun.id == result["backup_run_id"])
+    ).scalar_one()
+    assert tag == "pre-cutover"
+
+
+# --- backup_images_task ---------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_backup_images_task_creates_backup_run(db_sync):
+    from backend.app.tasks.backup import backup_images_task
+    result = backup_images_task.delay().get()
+    row = db_sync.execute(
+        select(BackupRun.kind, BackupRun.status, BackupRun.tar_path)
+        .where(BackupRun.id == result["backup_run_id"])
+    ).one()
+    assert row.kind == "images"
+    assert row.status == "ok"
+    assert row.tar_path and row.tar_path.endswith(".tar.zst")
+
+
+# --- restore_db_task ------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_restore_db_task_creates_restoring_marker_then_restored(db_sync):
+    from backend.app.tasks.backup import backup_db_task, restore_db_task
+    src = backup_db_task.delay().get()
+    src_id = src["backup_run_id"]
+
+    restore_db_task.delay(source_backup_run_id=src_id).get()
+
+    rows = db_sync.execute(
+        select(BackupRun.status, BackupRun.triggered_by)
+        .where(BackupRun.restored_from_id == src_id)
+    ).all()
+    assert len(rows) == 1
+    assert rows[0].status == "restored"
+    assert rows[0].triggered_by == "restore"
+
+
+@pytest.mark.asyncio
+async def test_restore_db_task_rejects_non_db_source(db_sync):
+    from backend.app.tasks.backup import restore_db_task
+
+    now = datetime.now(UTC)
+    img_id = _seed_backup(db_sync, kind="images", status="ok", started_at=now)
+    db_sync.commit()
+
+    with pytest.raises(ValueError):
+        restore_db_task.delay(source_backup_run_id=img_id).get()
+
+
+# --- prune_backups --------------------------------------------------
+
+
+def test_prune_backups_keeps_last_n_untagged_per_kind(db_sync):
+    from backend.app.tasks.backup import prune_backups
+
+    s = db_sync.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    s.backup_db_keep_last_n = 2
+    s.backup_images_keep_last_n = 1
+    db_sync.commit()
+
+    now = datetime.now(UTC)
+    for i in range(4):
+        _seed_backup(db_sync, kind="db", status="ok",
+                     started_at=now - timedelta(hours=i))
+    for i in range(3):
+        _seed_backup(db_sync, kind="images", status="ok",
+                     started_at=now - timedelta(hours=i))
+    db_sync.commit()
+
+    result = prune_backups.apply().get()
+    assert result["db_deleted"] == 2
+    assert result["images_deleted"] == 2
+
+
+def test_prune_backups_protects_tagged_rows(db_sync):
+    from backend.app.tasks.backup import prune_backups
+
+    s = db_sync.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    s.backup_db_keep_last_n = 1
+    db_sync.commit()
+
+    now = datetime.now(UTC)
+    _seed_backup(db_sync, kind="db", status="ok", started_at=now)
+    tagged_id = _seed_backup(
+        db_sync, kind="db", status="ok",
+        started_at=now - timedelta(days=30), tag="forever",
+    )
+    _seed_backup(db_sync, kind="db", status="ok",
+                 started_at=now - timedelta(days=1))
+    db_sync.commit()
+
+    prune_backups.apply().get()
+
+    surviving_tag = db_sync.execute(
+        select(BackupRun.tag).where(BackupRun.id == tagged_id)
+    ).scalar_one_or_none()
+    assert surviving_tag == "forever"
+
+
+def test_prune_backups_never_deletes_running_or_restoring(db_sync):
+    from backend.app.tasks.backup import prune_backups
+
+    s = db_sync.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    s.backup_db_keep_last_n = 0
+    db_sync.commit()
+
+    ancient = datetime.now(UTC) - timedelta(days=30)
+    running_id = _seed_backup(db_sync, kind="db", status="running", started_at=ancient)
+    restoring_id = _seed_backup(db_sync, kind="db", status="restoring", started_at=ancient)
+    db_sync.commit()
+
+    prune_backups.apply().get()
+
+    statuses = db_sync.execute(
+        select(BackupRun.status).where(BackupRun.id.in_([running_id, restoring_id]))
+    ).scalars().all()
+    assert set(statuses) == {"running", "restoring"}
+
+
+# --- backup_db_nightly ----------------------------------------------
+
+
+def test_nightly_skips_when_disabled(db_sync):
+    from backend.app.tasks.backup import backup_db_nightly
+
+    s = db_sync.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    s.backup_db_nightly_enabled = False
+    db_sync.commit()
+
+    result = backup_db_nightly.apply().get()
+    assert "skipped" in result and "disabled" in result["skipped"]
+
+
+def test_nightly_skips_when_hour_mismatch(db_sync):
+    from backend.app.tasks.backup import backup_db_nightly
+
+    s = db_sync.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    s.backup_db_nightly_enabled = True
+    s.backup_db_nightly_hour_utc = (datetime.now(UTC).hour + 12) % 24
+    db_sync.commit()
+
+    result = backup_db_nightly.apply().get()
+    assert "skipped" in result and "hour=" in result["skipped"]
+
+
+def test_nightly_dispatches_when_enabled_at_configured_hour(db_sync):
+    from backend.app.tasks.backup import backup_db_nightly
+
+    s = db_sync.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    s.backup_db_nightly_enabled = True
+    s.backup_db_nightly_hour_utc = datetime.now(UTC).hour
+    db_sync.commit()
+
+    result = backup_db_nightly.apply().get()
+    assert "dispatched" in result