feat(6846): replace_from_backup rolls the store back to a full backup by id

One transaction:
- Refuse unless the backup is a full one and, on a live instance, carries
  this instance's id. A fresh instance (no records, at most the restoring
  account) takes any full backup and adopts its id.
- Clear first: search vectors and the CSS consumer map (derived); every
  backed-up table nothing outside the backup points at, whole; the settings
  the backup carries; and the identity-table rows the backup lacks. A share
  granted by a removed user goes before the user.
- Write the backup through the same _build_* helpers as the copy-in restore,
  now _load_v2(keep_ids=...). Identity tables (users, projects, milestones,
  notes, rulebooks, topics, rules, systems, design systems, code shapes,
  family decisions) go through _place. That keeps the backup's id and
  merges, so API keys, shares and forge connections of a surviving row stay.
- Reset id sequences past the restored max, and sign every session out.

reindex_after_restore runs the four embedding backfills afterwards. A unit
guard checks that no row kept for outside references cascades from a table
replaced whole. Integration tests cover edits, deletions and new rows put
back, a surviving API key, sessions ending, fresh ids afterwards, foreign or
id-less backups refused, a user export refused, a failure part-way changing
nothing, and a fresh instance adopting the id.

Milestone 582, step 2.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/grimoire/services/backup.py

@@ -2328,6 +2328,41 @@ async def _restore_v1(data: dict) -> dict:
     return stats
 
 
+# Columns the restore writes in a LATER pass, once the rows they point at
+# exist: a note's parent and origin, a project's design system and inception.
+# A rollback clears them before overwriting a row in place, so a pointer the
+# backup does not have cannot survive from the current row.
+_SECOND_PASS_COLUMNS: dict[type, tuple[str, ...]] = {
+    Note: ("parent_id", "arose_from_id"),
+    Project: ("design_system_id", "inception"),
+}
+
+
+async def _place(session, obj, source_id: int | None, keep_ids: bool) -> int:
+    """Write one identity-table row and return the id it now has.
+
+    Copy-in: insert, and the database picks the id. Rollback: the row takes
+    `source_id`, the id it had in the backup, and is merged, so a row with
+    that id is overwritten in place (keeping what references it from outside
+    the backup: API keys, shares, forge connections) and a missing one is
+    inserted. A rolled-back row is live: out of the trash, and with no
+    second-pass pointer left from the current row.
+    """
+    if not keep_ids or source_id is None:
+        session.add(obj)
+        await session.flush()
+        return obj.id
+    obj.id = source_id
+    if hasattr(obj, "deleted_at"):
+        obj.deleted_at = None
+        obj.deleted_batch_id = None
+    for column in _SECOND_PASS_COLUMNS.get(type(obj), ()):
+        setattr(obj, column, None)
+    merged = await session.merge(obj)
+    await session.flush()
+    return merged.id
+
+
 async def _restore_v2(data: dict) -> dict:
     """Restore v2/v3 backup with full FK re-mapping.
 
@@ -2342,6 +2377,25 @@ async def _restore_v2(data: dict) -> dict:
     and the id-map bookkeeping, which is the part a split would only make
     harder to read.
     """
+    async with async_session() as session:
+        stats = await _load_v2(session, data, _Maps(), keep_ids=False)
+        await session.commit()
+
+    logger.info("Restored v2/v3 backup: %s", stats)
+    return stats
+
+
+async def _load_v2(session, data: dict, maps: "_Maps", *, keep_ids: bool) -> dict[str, int]:
+    """Write a v2+ payload's rows into `session`, parents first, through the
+    `_build_*` helpers and `maps`. Commits nothing; the caller owns the
+    transaction.
+
+    `keep_ids=False` is the copy-in restore: every row gets a new id and
+    `maps` records old → new. `keep_ids=True` is the rollback
+    (`replace_from_backup`): the identity tables keep the id the row had in
+    the backup, so a row this store already holds is overwritten in place, and
+    `maps` maps each id to itself.
+    """
     stats: dict[str, int] = {
         "users": 0, "projects": 0, "milestones": 0, "notes": 0,
         "task_logs": 0, "note_drafts": 0, "note_versions": 0,
@@ -2361,15 +2415,11 @@ async def _restore_v2(data: dict) -> dict:
         "family_adoptions": 0, "family_decisions": 0,
     }
 
-    async with async_session() as session:
-        maps = _Maps()
 
     # 1. Users
     for u_data in data.get("users", []):
         user = _build_user(u_data, maps)
-            session.add(user)
+        maps.users[u_data["id"]] = await _place(session, user, u_data["id"], keep_ids)
-            await session.flush()
-            maps.users[u_data["id"]] = user.id
         stats["users"] += 1
 
     # 2. Projects
@@ -2377,9 +2427,7 @@ async def _restore_v2(data: dict) -> dict:
         proj = _build_project(p_data, maps)
         if proj is None:
             continue
-            session.add(proj)
+        maps.projects[p_data["id"]] = await _place(session, proj, p_data["id"], keep_ids)
-            await session.flush()
-            maps.projects[p_data["id"]] = proj.id
         stats["projects"] += 1
 
     # 3. Milestones
@@ -2387,9 +2435,7 @@ async def _restore_v2(data: dict) -> dict:
         ms = _build_milestone(m_data, maps)
         if ms is None:
             continue
-            session.add(ms)
+        maps.milestones[m_data["id"]] = await _place(session, ms, m_data["id"], keep_ids)
-            await session.flush()
-            maps.milestones[m_data["id"]] = ms.id
         stats["milestones"] += 1
 
     # 4a. Notes — first pass (no parent_id yet)
@@ -2399,13 +2445,12 @@ async def _restore_v2(data: dict) -> dict:
         note = _build_note(n_data, maps)
         if note is None:
             continue
-            session.add(note)
+        note_id = await _place(session, note, n_data["id"], keep_ids)
-            await session.flush()
+        maps.notes[n_data["id"]] = note_id
-            maps.notes[n_data["id"]] = note.id
         if n_data.get("parent_id"):
-                notes_with_parents.append((note.id, n_data["parent_id"]))
+            notes_with_parents.append((note_id, n_data["parent_id"]))
         if n_data.get("arose_from_id"):
-                notes_with_origins.append((note.id, n_data["arose_from_id"]))
+            notes_with_origins.append((note_id, n_data["arose_from_id"]))
         stats["notes"] += 1
 
     # 4b. Patch the note->note edges. Deferred for the same reason
@@ -2482,9 +2527,7 @@ async def _restore_v2(data: dict) -> dict:
         rb = _build_rulebook(rb_data, maps)
         if rb is None:
             continue
-            session.add(rb)
+        maps.rulebooks[rb_data["id"]] = await _place(session, rb, rb_data["id"], keep_ids)
-            await session.flush()
-            maps.rulebooks[rb_data["id"]] = rb.id
         stats["rulebooks"] += 1
 
     # 10. Topics (v3)
@@ -2492,9 +2535,7 @@ async def _restore_v2(data: dict) -> dict:
         topic = _build_topic(t_data, maps)
         if topic is None:
             continue
-            session.add(topic)
+        maps.topics[t_data["id"]] = await _place(session, topic, t_data["id"], keep_ids)
-            await session.flush()
-            maps.topics[t_data["id"]] = topic.id
         stats["rulebook_topics"] += 1
 
     # 11. Rules (v3) — topic-rule (topic_id) XOR project-rule (project_id)
@@ -2502,9 +2543,7 @@ async def _restore_v2(data: dict) -> dict:
         rule = _build_rule(r_data, maps)
         if rule is None:
             continue  # orphaned — its parent didn't restore
-            session.add(rule)
+        maps.rules[r_data["id"]] = await _place(session, rule, r_data["id"], keep_ids)
-            await session.flush()
-            maps.rules[r_data["id"]] = rule.id
         stats["rules"] += 1
 
     # 12-14. Rulebook subscriptions, rule and topic suppressions (v3-v14)
@@ -2652,10 +2691,10 @@ async def _restore_v2(data: dict) -> dict:
         decision = _build_family_decision(fd, maps)
         if decision is None:
             continue
-            session.add(decision)
+        source_id = int(fd["id"]) if fd.get("id") else None
-            await session.flush()
+        decision_id = await _place(session, decision, source_id, keep_ids)
-            if fd.get("id"):
+        if source_id:
-                maps.decisions[int(fd["id"])] = decision.id
+            maps.decisions[source_id] = decision_id
         stats["family_decisions"] += 1
 
     # A rule's edit history. Must come after the rules themselves — the
@@ -2673,9 +2712,7 @@ async def _restore_v2(data: dict) -> dict:
         system = _build_system(sy_data, maps)
         if system is None:
             continue
-            session.add(system)
+        maps.systems[sy_data["id"]] = await _place(session, system, sy_data["id"], keep_ids)
-            await session.flush()
-            maps.systems[sy_data["id"]] = system.id
         stats["systems"] += 1
 
     # 16. Record<->system links
@@ -2700,9 +2737,7 @@ async def _restore_v2(data: dict) -> dict:
         design = _build_design_system(ds_data, maps)
         if design is None:
             continue
-            session.add(design)
+        maps.design_systems[ds_data["id"]] = await _place(session, design, ds_data["id"], keep_ids)
-            await session.flush()
-            maps.design_systems[ds_data["id"]] = design.id
         stats["design_systems"] += 1
 
     # 18. Design tokens
@@ -2751,10 +2786,10 @@ async def _restore_v2(data: dict) -> dict:
         shape = _build_code_shape(cs_data, maps)
         if shape is None:
             continue
-            session.add(shape)
+        source_id = int(cs_data["id"]) if cs_data.get("id") else None
-            await session.flush()
+        shape_id = await _place(session, shape, source_id, keep_ids)
-            if cs_data.get("id"):
+        if source_id:
-                maps.shapes[int(cs_data["id"])] = shape.id
+            maps.shapes[source_id] = shape_id
         stats["code_shapes"] += 1
 
     # v8: the ledger's history rides its shapes.
@@ -2806,7 +2841,197 @@ async def _restore_v2(data: dict) -> dict:
             choices["design_system_id"] = maps.design_systems.get(ds) if ds else None
             proj.inception = {**inception, "choices": choices}
 
+
+    return stats
+
+
+# ── Rollback: put this store back to a full backup ─────────────────────────
+
+
+class RestoreRefused(Exception):
+    """Why a rollback will not run, in words an operator can act on."""
+
+
+# Tables whose rows are pointed at BY ID, from inside the backup and from
+# tables it does not carry (API keys, shares, forge connections, embeddings).
+# A rollback keeps each row the backup holds under the same id, overwriting it
+# in place, and removes the rest. Each is written through `_place`.
+_IDENTITY_TABLES = (
+    "users", "projects", "milestones", "notes", "rulebooks", "rulebook_topics",
+    "rules", "systems", "design_systems", "code_shapes", "family_decisions",
+)
+# Global vocabularies matched on slug, seeded by migrations: a rollback adds
+# what the backup names and removes nothing.
+_CATALOG_TABLES = ("canonical_systems", "platforms")
+# Rebuilt from the records. A rolled-back record's text may be older than its
+# vectors, which would then look current while encoding text the record no
+# longer holds, so a rollback clears them and the backfills rebuild them.
+_DERIVED_TABLES = (
+    "note_embeddings", "rule_embeddings", "milestone_embeddings",
+    "system_embeddings", "code_shape_consumers",
+)
+
+
+async def _count(session, model) -> int:
+    from sqlalchemy import func
+    return (await session.execute(select(func.count()).select_from(model))).scalar_one()
+
+
+async def _is_fresh(session) -> bool:
+    """No records yet, and at most the one account that is doing the
+    restore. Trashed records count: they are still this instance's."""
+    if await _count(session, User) > 1:
+        return False
+    for model in (Project, Note, Rule, Rulebook):
+        if await _count(session, model):
+            return False
+    return True
+
+
+def _ids_in(data: dict, table: str) -> list[int]:
+    return sorted({int(row["id"]) for row in data.get(table) or [] if row.get("id")})
+
+
+async def _clear_for_rollback(session, data: dict) -> None:
+    """Remove what the backup does not hold, before its rows are written.
+
+    Removing first, not after, is what lets a name the backup gives one id
+    (a username, a slug) be free when it is written: the row holding it now
+    under another id is gone by then. Everything foreign-keyed to a removed
+    row with ON DELETE CASCADE goes with it, which is only ever a row the
+    backup does not hold either, or one this rollback rewrites.
+    """
+    from sqlalchemy import BigInteger, any_, bindparam, not_
+    from sqlalchemy.dialects.postgresql import ARRAY
+
+    from grimoire.models import Base
+
+    tables = Base.metadata.tables
+    for name in _DERIVED_TABLES:
+        await session.execute(tables[name].delete())
+
+    # Every other backed-up table is replaced whole: nothing outside the
+    # backup points at its rows, and their ids carry no meaning of their own.
+    keep_whole = set(_IDENTITY_TABLES) | set(_CATALOG_TABLES) | {"settings"}
+    for table in reversed(Base.metadata.sorted_tables):
+        if table.name in _BACKED_UP and table.name not in keep_whole:
+            await session.execute(table.delete())
+    # Settings the backup never carries (the backup secret, the run history)
+    # are this instance's own and stay.
+    settings_t = tables["settings"]
+    await session.execute(
+        settings_t.delete().where(settings_t.c.key.notin_(sorted(_SETTINGS_NOT_EXPORTED)))
+    )
+
+    for name in reversed(_IDENTITY_TABLES):
+        table = tables[name]
+        keep = _ids_in(data, name)
+        doomed = (
+            not_(table.c.id == any_(bindparam(f"keep_{name}", keep, type_=ARRAY(BigInteger))))
+            if keep else table.c.id.isnot(None)
+        )
+        # A reference from outside the backup with no ON DELETE (a share's
+        # granted_by) would refuse the delete: those rows go first.
+        for other in Base.metadata.sorted_tables:
+            if other.name in _BACKED_UP:
+                continue
+            for fk in other.foreign_keys:
+                if fk.column.table is table and not fk.ondelete:
+                    await session.execute(
+                        other.delete().where(fk.parent.in_(select(table.c.id).where(doomed)))
+                    )
+        await session.execute(table.delete().where(doomed))
+
+
+async def _reset_sequences(session) -> None:
+    """Point every id sequence past the highest id now in its table, so the
+    next insert does not collide with a restored row."""
+    from sqlalchemy import text
+
+    from grimoire.models import Base
+
+    for name in _BACKED_UP:
+        table = Base.metadata.tables[name]
+        if "id" not in table.c:
+            continue
+        await session.execute(text(
+            f"SELECT setval(pg_get_serial_sequence('{name}', 'id'), "
+            f"COALESCE((SELECT MAX(id) FROM {name}), 0) + 1, false) "
+            f"WHERE pg_get_serial_sequence('{name}', 'id') IS NOT NULL"
+        ))
+
+
+async def replace_from_backup(data: dict) -> dict:
+    """Put this store back to a full backup, in one transaction.
+
+    On a live instance the backup must come from this instance (the same
+    instance id): every row then keeps the id it had in the backup, a row the
+    backup holds is overwritten in place, and one it does not hold is
+    removed. Records edited since go back, deleted ones return, ones made
+    since are gone. A fresh instance (no records, at most the restoring
+    account) takes any full backup the same way and adopts its instance id.
+
+    What the backup does not carry stays where its owner survives: API keys,
+    forge connections, shares, logs, the backup destination's secret. Every
+    session is signed out, because each account is now the backup's. Search
+    vectors are cleared, so `reindex_after_restore` has to run afterwards.
+
+    Raises RestoreRefused, with nothing changed, when the backup cannot be
+    applied here. Any failure part-way rolls the transaction back.
+    """
+    from sqlalchemy import update
+
+    from grimoire.services.instance import adopt_instance_id, get_instance_id
+
+    if (data.get("version") or 1) < 2 or data.get("scope", "full") != "full":
+        raise RestoreRefused(
+            "This file isn't a full backup of the instance (it is a single user's "
+            "export, or a very old format), so it can't put the whole instance back."
+        )
+    backup_instance = data.get("instance_id")
+
+    async with async_session() as session:
+        fresh = await _is_fresh(session)
+        own_instance = await get_instance_id(session)
+        if not fresh:
+            if not backup_instance:
+                raise RestoreRefused(
+                    "This backup was made before backups recorded which instance they "
+                    "came from, so it can only be restored into a new, empty instance."
+                )
+            if backup_instance != own_instance:
+                raise RestoreRefused(
+                    "This backup comes from a different Grimoire instance. Rolling back "
+                    "matches records by id, and its ids belong to that instance, so it "
+                    "can only be restored into a new, empty instance."
+                )
+
+        await _clear_for_rollback(session, data)
+        stats = await _load_v2(session, data, _Maps(), keep_ids=True)
+        await _reset_sequences(session)
+        await session.execute(update(User).values(session_version=User.session_version + 1))
+        if fresh and backup_instance:
+            await adopt_instance_id(session, backup_instance)
         await session.commit()
 
-    logger.info("Restored v2/v3 backup: %s", stats)
+    mode = "fresh" if fresh else "rollback"
-    return stats
+    logger.info("Replaced the store from a backup (%s): %s", mode, stats)
+    return {"mode": mode, "restored": stats, "exported_at": data.get("exported_at")}
+
+
+async def reindex_after_restore() -> None:
+    """Rebuild the search vectors a rollback cleared. Long-running; start it
+    in the background (the startup backfill does the same after a restart)."""
+    from grimoire.services.embeddings import (
+        backfill_milestone_embeddings, backfill_note_embeddings,
+        backfill_rule_embeddings, backfill_system_embeddings,
+    )
+
+    for backfill in (
+        backfill_note_embeddings, backfill_rule_embeddings,
+        backfill_milestone_embeddings, backfill_system_embeddings,
+    ):
+        try:
+            await backfill()
+        except Exception:
+            logger.exception("reindex after restore: %s failed", backfill.__name__)

tests/test_integration_backup_rollback.py

@@ -0,0 +1,178 @@
+"""Real-Postgres proof that a full backup rolls this instance back by id.
+
+The lane shares one database, so every test here takes its backup of the
+whole store, changes its own rows, and rolls back to that backup: what it
+puts back is exactly what was there when it started.
+"""
+from unittest.mock import AsyncMock, patch
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import func, select
+
+from grimoire.models import async_session
+from grimoire.models.note import Note
+from grimoire.models.project import Project
+from grimoire.models.user import User
+from grimoire.services import backup
+from grimoire.services.api_keys import create_api_key, lookup_key
+from grimoire.services.instance import adopt_instance_id, get_instance_id
+from tests.helpers import ensure_user
+
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
+
+OWNER = "rollback_owner"
+
+
+@pytest_asyncio.fixture
+async def seeded():
+    """One user with a project and two tasks, committed, and a full backup
+    taken right after."""
+    async with async_session() as s:
+        user = await ensure_user(s, OWNER)
+        # A clean slate for this user: an earlier test's rows would otherwise
+        # be in this test's backup too.
+        for model in (Note, Project):
+            for row in (await s.execute(select(model).where(model.user_id == user.id))).scalars():
+                await s.delete(row)
+        await s.flush()
+        project = Project(user_id=user.id, title="Rollback drill")
+        s.add(project)
+        await s.flush()
+        kept = Note(user_id=user.id, project_id=project.id, title="before",
+                    body="as backed up", note_type="note", task_kind="work", status="todo")
+        doomed = Note(user_id=user.id, project_id=project.id, title="deleted later",
+                      body="", note_type="note", task_kind="work", status="todo")
+        s.add_all([kept, doomed])
+        await s.commit()
+        ids = {"user": user.id, "project": project.id, "kept": kept.id, "doomed": doomed.id}
+    key, _ = await create_api_key(ids["user"], "rollback drill", "read")
+    payload = await backup.export_full_backup()
+    return ids, payload, key
+
+
+def _own(payload: dict, table: str, user_id: int) -> list[dict]:
+    return sorted(
+        (row for row in payload[table] if row.get("user_id") == user_id),
+        key=lambda row: row["id"],
+    )
+
+
+async def _change_everything(ids: dict) -> int:
+    """Edit one task, delete the other, add a task and a project. Returns the
+    new task's id."""
+    async with async_session() as s:
+        kept = await s.get(Note, ids["kept"])
+        kept.title = "after"
+        kept.body = "edited since the backup"
+        await s.delete(await s.get(Note, ids["doomed"]))
+        extra = Project(user_id=ids["user"], title="made after the backup")
+        s.add(extra)
+        await s.flush()
+        new = Note(user_id=ids["user"], project_id=extra.id, title="new since",
+                   body="", note_type="note", task_kind="work", status="todo")
+        s.add(new)
+        await s.commit()
+        return new.id
+
+
+async def test_a_rollback_puts_back_edits_and_deletions_and_removes_new_records(seeded):
+    ids, payload, _ = seeded
+    new_id = await _change_everything(ids)
+
+    result = await backup.replace_from_backup(payload)
+
+    assert result["mode"] == "rollback"
+    async with async_session() as s:
+        kept = await s.get(Note, ids["kept"])
+        assert (kept.title, kept.body) == ("before", "as backed up")
+        assert await s.get(Note, ids["doomed"]) is not None
+        assert await s.get(Note, new_id) is None
+        titles = (await s.execute(
+            select(Project.title).where(Project.user_id == ids["user"])
+        )).scalars().all()
+        assert titles == ["Rollback drill"]
+
+    after = await backup.export_full_backup()
+    for table in ("projects", "milestones", "notes"):
+        assert _own(after, table, ids["user"]) == _own(payload, table, ids["user"]), table
+
+
+async def test_a_surviving_users_api_key_still_works_and_their_sessions_end(seeded):
+    ids, payload, key = seeded
+    async with async_session() as s:
+        version_before = (await s.get(User, ids["user"])).session_version
+    await _change_everything(ids)
+
+    await backup.replace_from_backup(payload)
+
+    found = await lookup_key(key)
+    assert found is not None and found.user_id == ids["user"]
+    async with async_session() as s:
+        assert (await s.get(User, ids["user"])).session_version == version_before + 1
+
+
+async def test_new_records_after_a_rollback_take_fresh_ids(seeded):
+    ids, payload, _ = seeded
+    await _change_everything(ids)
+    await backup.replace_from_backup(payload)
+
+    async with async_session() as s:
+        highest = (await s.execute(select(func.max(Note.id)))).scalar_one()
+        note = Note(user_id=ids["user"], project_id=ids["project"], title="after rollback",
+                    body="", note_type="note", task_kind="work", status="todo")
+        s.add(note)
+        await s.commit()
+        assert note.id > highest
+        await s.delete(note)
+        await s.commit()
+
+
+@pytest.mark.parametrize("instance_id", ["another-instance", None])
+async def test_a_backup_from_elsewhere_is_refused_on_a_live_instance(seeded, instance_id):
+    ids, payload, _ = seeded
+    foreign = {**payload, "instance_id": instance_id}
+    new_id = await _change_everything(ids)
+
+    with pytest.raises(backup.RestoreRefused):
+        await backup.replace_from_backup(foreign)
+
+    async with async_session() as s:
+        assert (await s.get(Note, ids["kept"])).title == "after"
+        assert await s.get(Note, new_id) is not None
+    await backup.replace_from_backup(payload)
+
+
+async def test_a_users_own_export_is_refused(seeded):
+    _, payload, _ = seeded
+    with pytest.raises(backup.RestoreRefused):
+        await backup.replace_from_backup({**payload, "scope": "user"})
+
+
+async def test_a_failure_part_way_changes_nothing(seeded):
+    ids, payload, _ = seeded
+    new_id = await _change_everything(ids)
+
+    with patch.object(backup, "_reset_sequences", AsyncMock(side_effect=RuntimeError("boom"))):
+        with pytest.raises(RuntimeError):
+            await backup.replace_from_backup(payload)
+
+    async with async_session() as s:
+        assert (await s.get(Note, ids["kept"])).title == "after"
+        assert await s.get(Note, ids["doomed"]) is None
+        assert await s.get(Note, new_id) is not None
+    await backup.replace_from_backup(payload)
+
+
+async def test_a_fresh_instance_adopts_the_backups_instance_id(seeded):
+    _, payload, _ = seeded
+    own = await get_instance_id()
+    try:
+        with patch.object(backup, "_is_fresh", AsyncMock(return_value=True)):
+            result = await backup.replace_from_backup({**payload, "instance_id": "a-lineage"})
+        assert result["mode"] == "fresh"
+        assert await get_instance_id() == "a-lineage"
+    finally:
+        async with async_session() as s:
+            await adopt_instance_id(s, own)
+            await s.commit()

tests/test_services_backup.py

@@ -685,3 +685,41 @@ def test_rule_rows_carry_the_kind_so_a_preference_does_not_restore_as_a_rule():
         arose_from_id=None, created_at=stamp, updated_at=stamp,
     )
     assert backup._rule_rows([row])[0]["kind"] == "preference"
+
+
+def test_a_rollback_never_cascades_into_a_row_it_keeps_for_outside_references():
+    """A rollback replaces most backed-up tables whole, and keeps an identity
+    table's rows in place so what points at them from OUTSIDE the backup (API
+    keys, shares, forge connections) survives. If such a kept row had an
+    ON DELETE CASCADE foreign key into a table replaced whole, clearing that
+    table would delete the kept row, and its keys and shares with it."""
+    from grimoire.models import Base
+
+    tables = Base.metadata.tables
+    backed_up = set(backup._BACKED_UP)
+    referenced_from_outside = {
+        fk.column.table.name
+        for table in Base.metadata.sorted_tables if table.name not in backed_up
+        for fk in table.foreign_keys if fk.column.table.name in backed_up
+    }
+    assert {"users", "projects", "notes"} <= referenced_from_outside
+    assert referenced_from_outside <= set(backup._IDENTITY_TABLES)
+    safe_targets = set(backup._IDENTITY_TABLES) | set(backup._CATALOG_TABLES) | (
+        set(tables) - backed_up
+    )
+    unsafe = sorted(
+        f"{name}.{fk.parent.name} -> {fk.column.table.name}"
+        for name in referenced_from_outside
+        for fk in tables[name].foreign_keys
+        if (fk.ondelete or "").upper() == "CASCADE" and fk.column.table.name not in safe_targets
+    )
+    assert unsafe == []
+
+
+def test_every_rollback_table_list_names_real_backed_up_tables():
+    from grimoire.models import Base
+
+    assert set(backup._IDENTITY_TABLES) <= set(backup._BACKED_UP)
+    assert set(backup._CATALOG_TABLES) <= set(backup._BACKED_UP)
+    assert set(backup._DERIVED_TABLES) <= set(backup._NOT_INCLUDED)
+    assert set(backup._DERIVED_TABLES) <= set(Base.metadata.tables)