diff --git a/src/grimoire/services/backup.py b/src/grimoire/services/backup.py
index d96317c7..60451fc9 100644
--- a/src/grimoire/services/backup.py
+++ b/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,452 +2415,623 @@ 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)
-            await session.flush()
-            maps.users[u_data["id"]] = user.id
-            stats["users"] += 1
+    # 1. Users
+    for u_data in data.get("users", []):
+        user = _build_user(u_data, maps)
+        maps.users[u_data["id"]] = await _place(session, user, u_data["id"], keep_ids)
+        stats["users"] += 1
 
-        # 2. Projects
-        for p_data in data.get("projects", []):
-            proj = _build_project(p_data, maps)
-            if proj is None:
-                continue
-            session.add(proj)
-            await session.flush()
-            maps.projects[p_data["id"]] = proj.id
-            stats["projects"] += 1
+    # 2. Projects
+    for p_data in data.get("projects", []):
+        proj = _build_project(p_data, maps)
+        if proj is None:
+            continue
+        maps.projects[p_data["id"]] = await _place(session, proj, p_data["id"], keep_ids)
+        stats["projects"] += 1
 
-        # 3. Milestones
-        for m_data in data.get("milestones", []):
-            ms = _build_milestone(m_data, maps)
-            if ms is None:
-                continue
-            session.add(ms)
-            await session.flush()
-            maps.milestones[m_data["id"]] = ms.id
-            stats["milestones"] += 1
+    # 3. Milestones
+    for m_data in data.get("milestones", []):
+        ms = _build_milestone(m_data, maps)
+        if ms is None:
+            continue
+        maps.milestones[m_data["id"]] = await _place(session, ms, m_data["id"], keep_ids)
+        stats["milestones"] += 1
 
-        # 4a. Notes — first pass (no parent_id yet)
-        notes_with_parents: list[tuple[int, int]] = []  # (new_note_id, old_parent_id)
-        notes_with_origins: list[tuple[int, int]] = []  # (new_note_id, old_arose_from_id)
-        for n_data in data.get("notes", []):
-            note = _build_note(n_data, maps)
-            if note is None:
-                continue
-            session.add(note)
-            await session.flush()
-            maps.notes[n_data["id"]] = note.id
-            if n_data.get("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"]))
-            stats["notes"] += 1
+    # 4a. Notes — first pass (no parent_id yet)
+    notes_with_parents: list[tuple[int, int]] = []  # (new_note_id, old_parent_id)
+    notes_with_origins: list[tuple[int, int]] = []  # (new_note_id, old_arose_from_id)
+    for n_data in data.get("notes", []):
+        note = _build_note(n_data, maps)
+        if note is None:
+            continue
+        note_id = await _place(session, note, n_data["id"], keep_ids)
+        maps.notes[n_data["id"]] = note_id
+        if n_data.get("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"]))
+        stats["notes"] += 1
 
-        # 4b. Patch the note->note edges. Deferred for the same reason
-        # parent_id is: these are ids in the SOURCE database. An edge whose
-        # target did not survive stays NULL.
-        for new_note_id, old_parent_id in notes_with_parents:
-            new_parent_id = maps.notes.get(old_parent_id)
-            if new_parent_id:
-                note_row = await session.get(Note, new_note_id)
-                if note_row:
-                    note_row.parent_id = new_parent_id
-        for new_note_id, old_origin_id in notes_with_origins:
-            new_origin_id = maps.notes.get(old_origin_id)
-            if new_origin_id:
-                note_row = await session.get(Note, new_note_id)
-                if note_row:
-                    note_row.arose_from_id = new_origin_id
+    # 4b. Patch the note->note edges. Deferred for the same reason
+    # parent_id is: these are ids in the SOURCE database. An edge whose
+    # target did not survive stays NULL.
+    for new_note_id, old_parent_id in notes_with_parents:
+        new_parent_id = maps.notes.get(old_parent_id)
+        if new_parent_id:
+            note_row = await session.get(Note, new_note_id)
+            if note_row:
+                note_row.parent_id = new_parent_id
+    for new_note_id, old_origin_id in notes_with_origins:
+        new_origin_id = maps.notes.get(old_origin_id)
+        if new_origin_id:
+            note_row = await session.get(Note, new_note_id)
+            if note_row:
+                note_row.arose_from_id = new_origin_id
 
-        # 5. TaskLogs
-        for tl_data in data.get("task_logs", []):
-            tl = _build_task_log(tl_data, maps)
-            if tl is None:
-                continue
-            session.add(tl)
-            stats["task_logs"] += 1
+    # 5. TaskLogs
+    for tl_data in data.get("task_logs", []):
+        tl = _build_task_log(tl_data, maps)
+        if tl is None:
+            continue
+        session.add(tl)
+        stats["task_logs"] += 1
 
-        # 6. NoteDrafts
-        for nd_data in data.get("note_drafts", []):
-            nd = _build_note_draft(nd_data, maps)
-            if nd is None:
-                continue
-            session.add(nd)
-            stats["note_drafts"] += 1
+    # 6. NoteDrafts
+    for nd_data in data.get("note_drafts", []):
+        nd = _build_note_draft(nd_data, maps)
+        if nd is None:
+            continue
+        session.add(nd)
+        stats["note_drafts"] += 1
 
-        # 7. NoteVersions
-        for nv_data in data.get("note_versions", []):
-            nv = _build_note_version(nv_data, maps)
-            if nv is None:
-                continue
-            session.add(nv)
-            stats["note_versions"] += 1
+    # 7. NoteVersions
+    for nv_data in data.get("note_versions", []):
+        nv = _build_note_version(nv_data, maps)
+        if nv is None:
+            continue
+        session.add(nv)
+        stats["note_versions"] += 1
 
-        # 8. Settings
-        for s_data in data.get("settings", []):
-            setting = _build_setting(s_data, maps)
-            if setting is None:
-                continue
-            session.add(setting)
-            stats["settings"] += 1
+    # 8. Settings
+    for s_data in data.get("settings", []):
+        setting = _build_setting(s_data, maps)
+        if setting is None:
+            continue
+        session.add(setting)
+        stats["settings"] += 1
 
-        # 8b. Retrieval tuning history (v16) — restored beside the settings it
-        # explains, and for the same reason: the numbers above are the state,
-        # these rows are the argument for it. Those dials move on the
-        # operator's behalf, so an install restored with the values
-        # and without the reasons is one tuned by nobody it can name.
-        for t_data in data.get("retrieval_tuning_events", []):
-            event = _build_retrieval_tuning_event(t_data, maps)
-            if event is None:
-                continue
-            session.add(event)
-            stats["retrieval_tuning_events"] += 1
+    # 8b. Retrieval tuning history (v16) — restored beside the settings it
+    # explains, and for the same reason: the numbers above are the state,
+    # these rows are the argument for it. Those dials move on the
+    # operator's behalf, so an install restored with the values
+    # and without the reasons is one tuned by nobody it can name.
+    for t_data in data.get("retrieval_tuning_events", []):
+        event = _build_retrieval_tuning_event(t_data, maps)
+        if event is None:
+            continue
+        session.add(event)
+        stats["retrieval_tuning_events"] += 1
 
-        # 8c. Moment mappings (v21) — the install's corrections to which
-        # actions reach which moment. Names only, so only the user remaps.
-        for mm_data in data.get("moment_mappings", []):
-            mapping = _build_moment_mapping(mm_data, maps)
-            if mapping is None:
-                continue
-            session.add(mapping)
-            stats["moment_mappings"] += 1
+    # 8c. Moment mappings (v21) — the install's corrections to which
+    # actions reach which moment. Names only, so only the user remaps.
+    for mm_data in data.get("moment_mappings", []):
+        mapping = _build_moment_mapping(mm_data, maps)
+        if mapping is None:
+            continue
+        session.add(mapping)
+        stats["moment_mappings"] += 1
 
-        # 9. Rulebooks (v3)
-        for rb_data in data.get("rulebooks", []):
-            rb = _build_rulebook(rb_data, maps)
-            if rb is None:
-                continue
-            session.add(rb)
-            await session.flush()
-            maps.rulebooks[rb_data["id"]] = rb.id
-            stats["rulebooks"] += 1
+    # 9. Rulebooks (v3)
+    for rb_data in data.get("rulebooks", []):
+        rb = _build_rulebook(rb_data, maps)
+        if rb is None:
+            continue
+        maps.rulebooks[rb_data["id"]] = await _place(session, rb, rb_data["id"], keep_ids)
+        stats["rulebooks"] += 1
 
-        # 10. Topics (v3)
-        for t_data in data.get("rulebook_topics", []):
-            topic = _build_topic(t_data, maps)
-            if topic is None:
-                continue
-            session.add(topic)
-            await session.flush()
-            maps.topics[t_data["id"]] = topic.id
-            stats["rulebook_topics"] += 1
+    # 10. Topics (v3)
+    for t_data in data.get("rulebook_topics", []):
+        topic = _build_topic(t_data, maps)
+        if topic is None:
+            continue
+        maps.topics[t_data["id"]] = await _place(session, topic, t_data["id"], keep_ids)
+        stats["rulebook_topics"] += 1
 
-        # 11. Rules (v3) — topic-rule (topic_id) XOR project-rule (project_id)
-        for r_data in data.get("rules", []):
-            rule = _build_rule(r_data, maps)
-            if rule is None:
-                continue  # orphaned — its parent didn't restore
-            session.add(rule)
-            await session.flush()
-            maps.rules[r_data["id"]] = rule.id
-            stats["rules"] += 1
+    # 11. Rules (v3) — topic-rule (topic_id) XOR project-rule (project_id)
+    for r_data in data.get("rules", []):
+        rule = _build_rule(r_data, maps)
+        if rule is None:
+            continue  # orphaned — its parent didn't restore
+        maps.rules[r_data["id"]] = await _place(session, rule, r_data["id"], keep_ids)
+        stats["rules"] += 1
 
-        # 12-14. Rulebook subscriptions, rule and topic suppressions (v3-v14)
-        # `rulebook_subscriptions`, `rule_suppressions` and `topic_suppressions`
-        # are READ BY NOBODY: their tables do not exist. An archive carrying
-        # them still imports, for the reason 14b gives.
+    # 12-14. Rulebook subscriptions, rule and topic suppressions (v3-v14)
+    # `rulebook_subscriptions`, `rule_suppressions` and `topic_suppressions`
+    # are READ BY NOBODY: their tables do not exist. An archive carrying
+    # them still imports, for the reason 14b gives.
 
-        # 14b. Always-on rulebook exclusions (v10)
-        # `rulebook_exclusions` is a v10 section READ BY NOBODY: its table
-        # does not exist. An archive carrying it still imports — the key is
-        # simply not looked at — because refusing a backup for remembering
-        # something deleted would make every such archive unrestorable.
+    # 14b. Always-on rulebook exclusions (v10)
+    # `rulebook_exclusions` is a v10 section READ BY NOBODY: its table
+    # does not exist. An archive carrying it still imports — the key is
+    # simply not looked at — because refusing a backup for remembering
+    # something deleted would make every such archive unrestorable.
 
-        # --- v5 sections. Every one is data.get()-guarded, so a v2/v3/v4
-        # payload restores without them rather than failing on an absent key.
+    # --- v5 sections. Every one is data.get()-guarded, so a v2/v3/v4
+    # payload restores without them rather than failing on an absent key.
 
-        # 14c. The global area catalog, matched on SLUG. Runs BEFORE systems,
-        # which resolve their mapping through this map. The map is seeded from
-        # what this install already has, so the common case creates nothing.
-        existing_canonical = (await session.execute(
-            select(CanonicalSystem).where(CanonicalSystem.deleted_at.is_(None))
-        )).scalars().all()
-        for entry in existing_canonical:
-            maps.canonical_by_slug[entry.slug] = entry.id
-        for cs_data in data.get("canonical_systems", []):
-            entry = _build_canonical_system(cs_data, maps)
-            if entry is None:
-                continue
-            session.add(entry)
-            await session.flush()
-            maps.canonical_by_slug[entry.slug] = entry.id
-            stats["canonical_systems"] += 1
-
-        # 14d. A rule's area tags and its typed edges. Runs HERE, not beside the
-        # rules in section 11, because it needs both maps: the rule ids from
-        # there and the canonical slugs from 14c just above.
-        #
-        # This one is a raw insert on a join table with no model class, so it
-        # has no builder and no column guard — both ends ARE the row.
-        for rs in data.get("rule_systems", []):
-            mapped_rule = maps.rules.get(rs.get("rule_id", 0))
-            canonical_id = maps.canonical_by_slug.get(rs.get("canonical_slug") or "")
-            if mapped_rule is None or canonical_id is None:
-                continue
-            await session.execute(rule_systems_t.insert().values(
-                rule_id=mapped_rule, canonical_id=canonical_id,
-            ))
-            stats["rule_systems"] += 1
-
-        # The same shape for a rule's mounts (v22): a join table with no
-        # model, so no builder — the rule id is the only thing to remap.
-        for rm in data.get("rule_moments", []):
-            mapped_rule = maps.rules.get(rm.get("rule_id", 0))
-            if mapped_rule is None or not rm.get("moment"):
-                continue
-            await session.execute(rule_moments_t.insert().values(
-                rule_id=mapped_rule, moment=rm["moment"],
-            ))
-            stats["rule_moments"] += 1
-
-        # Proposals and judgments about those mounts (v23); archives before
-        # v23 carry none, and every rule restores unjudged.
-        for rj in data.get("rule_moment_judgments", []):
-            judgment = _build_rule_moment_judgment(rj, maps)
-            if judgment is None:
-                continue
-            session.add(judgment)
-            stats["rule_moment_judgments"] += 1
-
-        for rr in data.get("rule_relations", []):
-            relation = _build_rule_relation(rr, maps)
-            if relation is None:
-                continue
-            session.add(relation)
-            stats["rule_relations"] += 1
-
-        # Lesson → rule links: after both notes and rules are
-        # mapped, which is why they sit beside the rule edges. Archives before
-        # v18 carry no section and restore with none.
-        for lr in data.get("lesson_rule_links", []):
-            link = _build_lesson_rule_link(lr, maps)
-            if link is None:
-                continue
-            session.add(link)
-            stats["lesson_rule_links"] += 1
-        # "No rule fits" answers; archives before v19 carry none.
-        for nr in data.get("lesson_no_rule", []):
-            answer = _build_lesson_no_rule(nr, maps)
-            if answer is None:
-                continue
-            session.add(answer)
-            stats["lesson_no_rule"] += 1
-
-        # Family canon (v25). Here because it needs projects, notes and topics
-        # all mapped. The platform catalog first, matched on slug and seeded
-        # from what this install already has — the 14c shape.
-        existing_platforms = (await session.execute(
-            select(Platform).where(Platform.deleted_at.is_(None))
-        )).scalars().all()
-        for platform in existing_platforms:
-            maps.platform_by_slug[platform.slug] = platform.id
-        for p_data in data.get("platforms", []):
-            platform = _build_platform(p_data, maps)
-            if platform is None:
-                continue
-            session.add(platform)
-            await session.flush()
-            maps.platform_by_slug[platform.slug] = platform.id
-            stats["platforms"] += 1
-        for pp in data.get("project_platforms", []):
-            membership = _build_project_platform(pp, maps)
-            if membership is None:
-                continue
-            session.add(membership)
-            stats["project_platforms"] += 1
-        for fi in data.get("family_ideas", []):
-            idea = _build_family_idea(fi, maps)
-            if idea is None:
-                continue
-            session.add(idea)
-            stats["family_ideas"] += 1
-        # The ideas must exist before anything foreign-keys them.
+    # 14c. The global area catalog, matched on SLUG. Runs BEFORE systems,
+    # which resolve their mapping through this map. The map is seeded from
+    # what this install already has, so the common case creates nothing.
+    existing_canonical = (await session.execute(
+        select(CanonicalSystem).where(CanonicalSystem.deleted_at.is_(None))
+    )).scalars().all()
+    for entry in existing_canonical:
+        maps.canonical_by_slug[entry.slug] = entry.id
+    for cs_data in data.get("canonical_systems", []):
+        entry = _build_canonical_system(cs_data, maps)
+        if entry is None:
+            continue
+        session.add(entry)
         await session.flush()
-        for fp in data.get("family_idea_platforms", []):
-            scope = _build_family_idea_platform(fp, maps)
-            if scope is None:
-                continue
-            session.add(scope)
-            stats["family_idea_platforms"] += 1
-        for fr in data.get("family_idea_references", []):
-            ref = _build_family_idea_reference(fr, maps)
-            if ref is None:
-                continue
-            session.add(ref)
-            stats["family_idea_references"] += 1
-        for fa in data.get("family_adoptions", []):
-            adoption = _build_family_adoption(fa, maps)
-            if adoption is None:
-                continue
-            session.add(adoption)
-            stats["family_adoptions"] += 1
-        # Oldest first, flushed one at a time: each decision's new id goes into
-        # the map before a later decision can name it as a precedent.
-        for fd in sorted(data.get("family_decisions", []), key=lambda r: r.get("id") or 0):
-            decision = _build_family_decision(fd, maps)
-            if decision is None:
-                continue
-            session.add(decision)
-            await session.flush()
-            if fd.get("id"):
-                maps.decisions[int(fd["id"])] = decision.id
-            stats["family_decisions"] += 1
+        maps.canonical_by_slug[entry.slug] = entry.id
+        stats["canonical_systems"] += 1
 
-        # A rule's edit history. Must come after the rules themselves — the
-        # rule map is only populated above — and both ids are ids in the
-        # SOURCE database, the same trap as a note's arose_from_id.
-        for rv in data.get("rule_versions", []):
-            version = _build_rule_version(rv, maps)
-            if version is None:
-                continue
-            session.add(version)
-            stats["rule_versions"] += 1
+    # 14d. A rule's area tags and its typed edges. Runs HERE, not beside the
+    # rules in section 11, because it needs both maps: the rule ids from
+    # there and the canonical slugs from 14c just above.
+    #
+    # This one is a raw insert on a join table with no model class, so it
+    # has no builder and no column guard — both ends ARE the row.
+    for rs in data.get("rule_systems", []):
+        mapped_rule = maps.rules.get(rs.get("rule_id", 0))
+        canonical_id = maps.canonical_by_slug.get(rs.get("canonical_slug") or "")
+        if mapped_rule is None or canonical_id is None:
+            continue
+        await session.execute(rule_systems_t.insert().values(
+            rule_id=mapped_rule, canonical_id=canonical_id,
+        ))
+        stats["rule_systems"] += 1
 
-        # 15. Systems
-        for sy_data in data.get("systems", []):
-            system = _build_system(sy_data, maps)
-            if system is None:
-                continue
-            session.add(system)
-            await session.flush()
-            maps.systems[sy_data["id"]] = system.id
-            stats["systems"] += 1
+    # The same shape for a rule's mounts (v22): a join table with no
+    # model, so no builder — the rule id is the only thing to remap.
+    for rm in data.get("rule_moments", []):
+        mapped_rule = maps.rules.get(rm.get("rule_id", 0))
+        if mapped_rule is None or not rm.get("moment"):
+            continue
+        await session.execute(rule_moments_t.insert().values(
+            rule_id=mapped_rule, moment=rm["moment"],
+        ))
+        stats["rule_moments"] += 1
 
-        # 16. Record<->system links
-        for rs in data.get("record_systems", []):
-            link = _build_record_system(rs, maps)
-            if link is None:
-                continue
-            session.add(link)
-            stats["record_systems"] += 1
+    # Proposals and judgments about those mounts (v23); archives before
+    # v23 carry none, and every rule restores unjudged.
+    for rj in data.get("rule_moment_judgments", []):
+        judgment = _build_rule_moment_judgment(rj, maps)
+        if judgment is None:
+            continue
+        session.add(judgment)
+        stats["rule_moment_judgments"] += 1
 
-        # 16b. Supersession claims. Guarded by `data.get` like every other
-        # post-v2 section, so a v5 or older payload restores cleanly without it.
-        for sup in data.get("note_supersessions", []):
-            claim = _build_note_supersession(sup, maps)
-            if claim is None:
-                continue
-            session.add(claim)
-            stats["note_supersessions"] += 1
+    for rr in data.get("rule_relations", []):
+        relation = _build_rule_relation(rr, maps)
+        if relation is None:
+            continue
+        session.add(relation)
+        stats["rule_relations"] += 1
 
-        # 17. Design systems, parent-first (the export orders them that way).
-        for ds_data in data.get("design_systems", []):
-            design = _build_design_system(ds_data, maps)
-            if design is None:
-                continue
-            session.add(design)
-            await session.flush()
-            maps.design_systems[ds_data["id"]] = design.id
-            stats["design_systems"] += 1
+    # Lesson → rule links: after both notes and rules are
+    # mapped, which is why they sit beside the rule edges. Archives before
+    # v18 carry no section and restore with none.
+    for lr in data.get("lesson_rule_links", []):
+        link = _build_lesson_rule_link(lr, maps)
+        if link is None:
+            continue
+        session.add(link)
+        stats["lesson_rule_links"] += 1
+    # "No rule fits" answers; archives before v19 carry none.
+    for nr in data.get("lesson_no_rule", []):
+        answer = _build_lesson_no_rule(nr, maps)
+        if answer is None:
+            continue
+        session.add(answer)
+        stats["lesson_no_rule"] += 1
 
-        # 18. Design tokens
-        for t_data in data.get("design_tokens", []):
-            token = _build_design_token(t_data, maps)
-            if token is None:
-                continue
-            session.add(token)
-            stats["design_tokens"] += 1
+    # Family canon (v25). Here because it needs projects, notes and topics
+    # all mapped. The platform catalog first, matched on slug and seeded
+    # from what this install already has — the 14c shape.
+    existing_platforms = (await session.execute(
+        select(Platform).where(Platform.deleted_at.is_(None))
+    )).scalars().all()
+    for platform in existing_platforms:
+        maps.platform_by_slug[platform.slug] = platform.id
+    for p_data in data.get("platforms", []):
+        platform = _build_platform(p_data, maps)
+        if platform is None:
+            continue
+        session.add(platform)
+        await session.flush()
+        maps.platform_by_slug[platform.slug] = platform.id
+        stats["platforms"] += 1
+    for pp in data.get("project_platforms", []):
+        membership = _build_project_platform(pp, maps)
+        if membership is None:
+            continue
+        session.add(membership)
+        stats["project_platforms"] += 1
+    for fi in data.get("family_ideas", []):
+        idea = _build_family_idea(fi, maps)
+        if idea is None:
+            continue
+        session.add(idea)
+        stats["family_ideas"] += 1
+    # The ideas must exist before anything foreign-keys them.
+    await session.flush()
+    for fp in data.get("family_idea_platforms", []):
+        scope = _build_family_idea_platform(fp, maps)
+        if scope is None:
+            continue
+        session.add(scope)
+        stats["family_idea_platforms"] += 1
+    for fr in data.get("family_idea_references", []):
+        ref = _build_family_idea_reference(fr, maps)
+        if ref is None:
+            continue
+        session.add(ref)
+        stats["family_idea_references"] += 1
+    for fa in data.get("family_adoptions", []):
+        adoption = _build_family_adoption(fa, maps)
+        if adoption is None:
+            continue
+        session.add(adoption)
+        stats["family_adoptions"] += 1
+    # Oldest first, flushed one at a time: each decision's new id goes into
+    # the map before a later decision can name it as a precedent.
+    for fd in sorted(data.get("family_decisions", []), key=lambda r: r.get("id") or 0):
+        decision = _build_family_decision(fd, maps)
+        if decision is None:
+            continue
+        source_id = int(fd["id"]) if fd.get("id") else None
+        decision_id = await _place(session, decision, source_id, keep_ids)
+        if source_id:
+            maps.decisions[source_id] = decision_id
+        stats["family_decisions"] += 1
 
-        # 19. Usage events
-        for ev in data.get("note_usage_events", []):
-            event = _build_usage_event(ev, maps)
-            if event is None:
-                continue
-            session.add(event)
-            stats["note_usage_events"] += 1
+    # A rule's edit history. Must come after the rules themselves — the
+    # rule map is only populated above — and both ids are ids in the
+    # SOURCE database, the same trap as a note's arose_from_id.
+    for rv in data.get("rule_versions", []):
+        version = _build_rule_version(rv, maps)
+        if version is None:
+            continue
+        session.add(version)
+        stats["rule_versions"] += 1
 
-        # The rule twin. Must come after the rules themselves; the rule map is
-        # populated there.
-        for ev in data.get("rule_usage_events", []):
-            event = _build_rule_usage_event(ev, maps)
-            if event is None:
-                continue
-            session.add(event)
-            stats["rule_usage_events"] += 1
+    # 15. Systems
+    for sy_data in data.get("systems", []):
+        system = _build_system(sy_data, maps)
+        if system is None:
+            continue
+        maps.systems[sy_data["id"]] = await _place(session, system, sy_data["id"], keep_ids)
+        stats["systems"] += 1
 
-        # And the System twin, after the Systems (step 15 fills their map).
-        for ev in data.get("system_usage_events", []):
-            event = _build_system_usage_event(ev, maps)
-            if event is None:
-                continue
-            session.add(event)
-            stats["system_usage_events"] += 1
+    # 16. Record<->system links
+    for rs in data.get("record_systems", []):
+        link = _build_record_system(rs, maps)
+        if link is None:
+            continue
+        session.add(link)
+        stats["record_systems"] += 1
 
-        # 20. Repo bindings
-        for rb_data in data.get("repo_bindings", []):
-            binding = _build_repo_binding(rb_data, maps)
-            if binding is None:
-                continue
-            session.add(binding)
-            stats["repo_bindings"] += 1
+    # 16b. Supersession claims. Guarded by `data.get` like every other
+    # post-v2 section, so a v5 or older payload restores cleanly without it.
+    for sup in data.get("note_supersessions", []):
+        claim = _build_note_supersession(sup, maps)
+        if claim is None:
+            continue
+        session.add(claim)
+        stats["note_supersessions"] += 1
 
-        # 21. Code shapes (v7)
-        for cs_data in data.get("code_shapes", []):
-            shape = _build_code_shape(cs_data, maps)
-            if shape is None:
-                continue
-            session.add(shape)
-            await session.flush()
-            if cs_data.get("id"):
-                maps.shapes[int(cs_data["id"])] = shape.id
-            stats["code_shapes"] += 1
+    # 17. Design systems, parent-first (the export orders them that way).
+    for ds_data in data.get("design_systems", []):
+        design = _build_design_system(ds_data, maps)
+        if design is None:
+            continue
+        maps.design_systems[ds_data["id"]] = await _place(session, design, ds_data["id"], keep_ids)
+        stats["design_systems"] += 1
 
-        # v8: the ledger's history rides its shapes.
-        for ev in data.get("code_shape_events", []):
-            event = _build_code_shape_event(ev, maps)
-            if event is None:
-                continue
-            session.add(event)
-            stats["code_shape_events"] += 1
+    # 18. Design tokens
+    for t_data in data.get("design_tokens", []):
+        token = _build_design_token(t_data, maps)
+        if token is None:
+            continue
+        session.add(token)
+        stats["design_tokens"] += 1
 
-        # v9: consumption edges.
-        for use in data.get("code_shape_uses", []):
-            edge = _build_code_shape_use(use, maps)
-            if edge is None:
-                continue
-            session.add(edge)
-            stats["code_shape_uses"] += 1
+    # 19. Usage events
+    for ev in data.get("note_usage_events", []):
+        event = _build_usage_event(ev, maps)
+        if event is None:
+            continue
+        session.add(event)
+        stats["note_usage_events"] += 1
 
-        # v10: a project's design-system pointer and its inception record ride
-        # the project but point at design systems and rulebooks restored AFTER
-        # it — so they are written last, with ids re-mapped. An id that did
-        # not survive drops out of the record rather than dangling.
-        #
-        # An UPDATE on a row this restore already wrote, not a construction,
-        # which is why it has no builder: there is no set of kwargs for a
-        # column guard to read.
-        for p_data in data.get("projects", []):
-            new_pid = maps.projects.get(p_data.get("id") or 0)
-            if new_pid is None:
-                continue
-            proj = await session.get(Project, new_pid)
-            if proj is None:
-                continue
-            old_ds = p_data.get("design_system_id")
-            if old_ds:
-                proj.design_system_id = maps.design_systems.get(old_ds)
-            inception = p_data.get("inception")
-            if isinstance(inception, dict):
-                choices = dict(inception.get("choices") or {})
-                # DROPPED, not remapped. An older archive may carry the
-                # retired exclusion choice; restoring it would put
-                # a key back that `validate_inception` now rejects as unknown,
-                # so the next edit to that project would fail on data this
-                # importer wrote.
-                choices.pop("exclude_always_on_rulebooks", None)
-                # Same for the retired subscribe_rulebooks choice.
-                choices.pop("subscribe_rulebooks", None)
-                ds = choices.get("design_system_id")
-                choices["design_system_id"] = maps.design_systems.get(ds) if ds else None
-                proj.inception = {**inception, "choices": choices}
+    # The rule twin. Must come after the rules themselves; the rule map is
+    # populated there.
+    for ev in data.get("rule_usage_events", []):
+        event = _build_rule_usage_event(ev, maps)
+        if event is None:
+            continue
+        session.add(event)
+        stats["rule_usage_events"] += 1
 
+    # And the System twin, after the Systems (step 15 fills their map).
+    for ev in data.get("system_usage_events", []):
+        event = _build_system_usage_event(ev, maps)
+        if event is None:
+            continue
+        session.add(event)
+        stats["system_usage_events"] += 1
+
+    # 20. Repo bindings
+    for rb_data in data.get("repo_bindings", []):
+        binding = _build_repo_binding(rb_data, maps)
+        if binding is None:
+            continue
+        session.add(binding)
+        stats["repo_bindings"] += 1
+
+    # 21. Code shapes (v7)
+    for cs_data in data.get("code_shapes", []):
+        shape = _build_code_shape(cs_data, maps)
+        if shape is None:
+            continue
+        source_id = int(cs_data["id"]) if cs_data.get("id") else None
+        shape_id = await _place(session, shape, source_id, keep_ids)
+        if source_id:
+            maps.shapes[source_id] = shape_id
+        stats["code_shapes"] += 1
+
+    # v8: the ledger's history rides its shapes.
+    for ev in data.get("code_shape_events", []):
+        event = _build_code_shape_event(ev, maps)
+        if event is None:
+            continue
+        session.add(event)
+        stats["code_shape_events"] += 1
+
+    # v9: consumption edges.
+    for use in data.get("code_shape_uses", []):
+        edge = _build_code_shape_use(use, maps)
+        if edge is None:
+            continue
+        session.add(edge)
+        stats["code_shape_uses"] += 1
+
+    # v10: a project's design-system pointer and its inception record ride
+    # the project but point at design systems and rulebooks restored AFTER
+    # it — so they are written last, with ids re-mapped. An id that did
+    # not survive drops out of the record rather than dangling.
+    #
+    # An UPDATE on a row this restore already wrote, not a construction,
+    # which is why it has no builder: there is no set of kwargs for a
+    # column guard to read.
+    for p_data in data.get("projects", []):
+        new_pid = maps.projects.get(p_data.get("id") or 0)
+        if new_pid is None:
+            continue
+        proj = await session.get(Project, new_pid)
+        if proj is None:
+            continue
+        old_ds = p_data.get("design_system_id")
+        if old_ds:
+            proj.design_system_id = maps.design_systems.get(old_ds)
+        inception = p_data.get("inception")
+        if isinstance(inception, dict):
+            choices = dict(inception.get("choices") or {})
+            # DROPPED, not remapped. An older archive may carry the
+            # retired exclusion choice; restoring it would put
+            # a key back that `validate_inception` now rejects as unknown,
+            # so the next edit to that project would fail on data this
+            # importer wrote.
+            choices.pop("exclude_always_on_rulebooks", None)
+            # Same for the retired subscribe_rulebooks choice.
+            choices.pop("subscribe_rulebooks", None)
+            ds = choices.get("design_system_id")
+            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)
-    return stats
+    mode = "fresh" if fresh else "rollback"
+    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__)
diff --git a/tests/test_integration_backup_rollback.py b/tests/test_integration_backup_rollback.py
new file mode 100644
index 00000000..c6c80924
--- /dev/null
+++ b/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()
diff --git a/tests/test_services_backup.py b/tests/test_services_backup.py
index 8a11721b..031682c7 100644
--- a/tests/test_services_backup.py
+++ b/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)