diff --git a/backend/app/services/migrators/__init__.py b/backend/app/services/migrators/__init__.py
new file mode 100644
index 0000000..0a7154c
--- /dev/null
+++ b/backend/app/services/migrators/__init__.py
@@ -0,0 +1,6 @@
+"""FC-5 migration tooling.
+
+One module per concern (backup/rollback/gs/ir/overlap/ml_queue/verify).
+Each migrator returns a counts dict; the run_migration task wires
+that dict into MigrationRun.counts so the UI polling shows progress.
+"""
diff --git a/backend/app/services/migrators/backup.py b/backend/app/services/migrators/backup.py
new file mode 100644
index 0000000..88040fd
--- /dev/null
+++ b/backend/app/services/migrators/backup.py
@@ -0,0 +1,124 @@
+"""pg_dump + tar.zst-based backup, restorable via pair of subprocess calls.
+
+Backups live under <images_root>/_backups/. Each backup is two files
+(SQL + tarball) plus a manifest JSON. Tagged backups (e.g. tag='pre_migration')
+are how rollback.py finds the most recent restorable snapshot.
+"""
+from __future__ import annotations
+
+import json
+import shutil
+import subprocess
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+_BACKUPS_DIRNAME = "_backups"
+
+
+def _backups_dir(images_root: Path | None = None) -> Path:
+    # Overridable for tests via monkeypatch.
+    root = images_root if images_root is not None else Path("/images")
+    p = root / _BACKUPS_DIRNAME
+    p.mkdir(parents=True, exist_ok=True)
+    return p
+
+
+def _run_subprocess(cmd: list[str], **kwargs: Any):
+    # Overridable for tests via monkeypatch.
+    return subprocess.run(
+        cmd,
+        capture_output=True,
+        check=True,
+        **{k: v for k, v in kwargs.items() if not k.startswith("_")},
+    )
+
+
+def create_backup(
+    *, db_url: str, images_root: Path, tag: str = "manual",
+) -> dict:
+    """Create a backup: pg_dump SQL + tar.zst of images.
+
+    Returns a manifest dict. Writes <ts>.sql, <ts>.tar.zst, <ts>.json into
+    <images_root>/_backups/.
+    """
+    ts = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
+    out_dir = _backups_dir(images_root)
+    sql_path = out_dir / f"fc_{ts}.sql"
+    tar_path = out_dir / f"fc_{ts}.tar.zst"
+    manifest_path = out_dir / f"fc_{ts}.json"
+
+    _run_subprocess(
+        ["pg_dump", "--no-owner", "--no-acl", "-f", str(sql_path), db_url],
+        _test_ts=ts,
+    )
+    _run_subprocess(
+        [
+            "tar", "--zstd", "-cf", str(tar_path),
+            "-C", str(images_root.parent), images_root.name,
+            f"--exclude={images_root.name}/_backups",
+            f"--exclude={images_root.name}/_quarantine",
+        ],
+        _test_ts=ts,
+    )
+
+    manifest = {
+        "backup_id": ts,
+        "tag": tag,
+        "created_at": datetime.now(UTC).isoformat(),
+        "sql_path": str(sql_path),
+        "tar_path": str(tar_path),
+    }
+    manifest_path.write_text(json.dumps(manifest, indent=2))
+    return manifest
+
+
+def list_backups(images_root: Path) -> list[dict]:
+    out_dir = _backups_dir(images_root)
+    items = []
+    for mf in sorted(out_dir.glob("fc_*.json"), reverse=True):
+        try:
+            items.append(json.loads(mf.read_text()))
+        except Exception:
+            continue
+    return items
+
+
+def find_latest_backup(images_root: Path, *, tag: str) -> dict | None:
+    for mf in list_backups(images_root):
+        if mf.get("tag") == tag:
+            return mf
+    return None
+
+
+def restore_backup(
+    *, manifest: dict, db_url: str, images_root: Path,
+) -> dict:
+    """Restore from a backup manifest.
+
+    1. Replay the .sql via psql.
+    2. Wipe /images/ contents (except _backups/, which holds the file we're using).
+    3. Untar the .tar.zst into /images/.
+    """
+    sql_path = Path(manifest["sql_path"])
+    tar_path = Path(manifest["tar_path"])
+
+    _run_subprocess(
+        ["psql", "-d", db_url, "-f", str(sql_path)],
+    )
+
+    # Wipe everything in images_root EXCEPT _backups/ (we'd delete the backup
+    # we're restoring from!).
+    for entry in images_root.iterdir():
+        if entry.name == _BACKUPS_DIRNAME:
+            continue
+        if entry.is_dir():
+            shutil.rmtree(entry)
+        else:
+            entry.unlink()
+
+    _run_subprocess(
+        ["tar", "--zstd", "-xf", str(tar_path), "-C", str(images_root.parent)],
+    )
+
+    return {"restored_from": manifest["backup_id"]}
diff --git a/backend/app/services/migrators/rollback.py b/backend/app/services/migrators/rollback.py
new file mode 100644
index 0000000..55ae7ef
--- /dev/null
+++ b/backend/app/services/migrators/rollback.py
@@ -0,0 +1,21 @@
+"""Restore from the most recent 'pre_migration'-tagged backup."""
+from __future__ import annotations
+
+from pathlib import Path
+
+from . import backup as backup_mod
+
+
+class NoBackupFoundError(Exception):
+    """Raised when rollback() is called with no pre_migration backup on disk."""
+
+
+def rollback_to_pre_migration(*, db_url: str, images_root: Path) -> dict:
+    manifest = backup_mod.find_latest_backup(images_root, tag="pre_migration")
+    if manifest is None:
+        raise NoBackupFoundError(
+            "no pre_migration-tagged backup found under <images_root>/_backups/"
+        )
+    return backup_mod.restore_backup(
+        manifest=manifest, db_url=db_url, images_root=images_root,
+    )
diff --git a/tests/test_migration_backup.py b/tests/test_migration_backup.py
new file mode 100644
index 0000000..5ff3cc1
--- /dev/null
+++ b/tests/test_migration_backup.py
@@ -0,0 +1,64 @@
+"""FC-5: backup + rollback service tests.
+
+Uses tmp_path to avoid touching real /images/_backups/. Subprocess
+calls (pg_dump, tar) are monkeypatched — the real shell-out is exercised
+in operator-side smoke testing on the homelab, not in unit tests.
+"""
+import pytest
+
+from backend.app.services.migrators import backup as backup_mod
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.mark.asyncio
+async def test_create_backup_writes_sql_and_tar(monkeypatch, tmp_path):
+    calls = []
+
+    def fake_run(cmd, **kwargs):
+        calls.append(cmd)
+        # Touch the expected output file so existence checks pass downstream.
+        if "pg_dump" in cmd[0]:
+            outpath = tmp_path / "_backups" / f"fc_{kwargs['_test_ts']}.sql"
+            outpath.write_text("-- fake pg_dump output")
+        else:
+            outpath = tmp_path / "_backups" / f"fc_{kwargs['_test_ts']}.tar.zst"
+            outpath.write_bytes(b"\x28\xb5\x2f\xfd")  # zstd magic
+        from types import SimpleNamespace
+        return SimpleNamespace(returncode=0, stdout=b"", stderr=b"")
+
+    monkeypatch.setattr(backup_mod, "_run_subprocess", fake_run)
+
+    result = backup_mod.create_backup(
+        db_url="postgresql://x", images_root=tmp_path, tag="pre_migration",
+    )
+    assert "sql_path" in result
+    assert "tar_path" in result
+    assert result["tag"] == "pre_migration"
+    assert any("pg_dump" in c[0] for c in calls)
+    assert any(c[0] == "tar" for c in calls)
+
+
+@pytest.mark.asyncio
+async def test_find_latest_backup_by_tag(monkeypatch, tmp_path):
+    # Create two manifests with different tags.
+    backups_dir = tmp_path / "_backups"
+    backups_dir.mkdir()
+    import json
+    (backups_dir / "fc_20260101T000000Z.json").write_text(json.dumps({
+        "backup_id": "20260101T000000Z", "tag": "manual",
+        "created_at": "2026-01-01T00:00:00+00:00",
+        "sql_path": "/x/a.sql", "tar_path": "/x/a.tar.zst",
+    }))
+    (backups_dir / "fc_20260202T000000Z.json").write_text(json.dumps({
+        "backup_id": "20260202T000000Z", "tag": "pre_migration",
+        "created_at": "2026-02-02T00:00:00+00:00",
+        "sql_path": "/x/b.sql", "tar_path": "/x/b.tar.zst",
+    }))
+
+    found = backup_mod.find_latest_backup(tmp_path, tag="pre_migration")
+    assert found is not None
+    assert found["backup_id"] == "20260202T000000Z"
+
+    missing = backup_mod.find_latest_backup(tmp_path, tag="nonexistent")
+    assert missing is None