src/scribe/services/backup.py

@@ -7,6 +7,7 @@ from scribe.models import async_session
 from scribe.models.milestone import Milestone
 from scribe.models.note import Note
 from scribe.models.note_draft import NoteDraft
+from scribe.models.note_supersession import NoteSupersession
 from scribe.models.note_version import NoteVersion
 from scribe.models.design_system import DesignSystem, DesignToken
 from scribe.models.note_usage import NoteUsageEvent
@@ -32,8 +33,11 @@ logger = logging.getLogger(__name__)
 # when the calendar surface was retired — old v3 events are skipped on restore.
 # v5 (2026-08) added the six tables that had accumulated outside the backup
 # entirely (#2293), and the coverage guard that stops the seventh.
+# v6 (2026-08) added note_supersessions — and the guard did stop the seventh:
+# the table shipped without a backup section and the coverage test failed the
+# build, which is the whole reason that list was written.
 # Bump when the serialized schema changes.
-BACKUP_VERSION = 5
+BACKUP_VERSION = 6
 
 # Every table this backup carries, by its REAL name. Paired with _NOT_INCLUDED
 # below, these two lists must together account for the entire schema — which is
@@ -50,7 +54,7 @@ _BACKED_UP = [
     "project_topic_suppressions",
     # v5 (2026-08): the five-year gap this list was written to stop.
     "systems", "record_systems", "design_systems", "design_tokens",
-    "note_usage_events", "repo_bindings",
+    "note_usage_events", "repo_bindings", "note_supersessions",
 ]
 
 # Tables intentionally NOT in the backup, surfaced in the payload so the gap is
@@ -110,6 +114,17 @@ def _record_system_rows(rows) -> list[dict]:
     return [{"note_id": r.note_id, "system_id": r.system_id} for r in rows]
 
 
+def _note_supersession_rows(rows) -> list[dict]:
+    """Which record has overtaken which. Carried because it is a JUDGEMENT —
+    someone decided this note replaced that one, and nothing in either note's
+    text records the decision. Lose it and the corpus silently reverts to
+    ranking stale material alongside current material."""
+    return [
+        {"superseder_id": r.superseder_id, "superseded_id": r.superseded_id}
+        for r in rows
+    ]
+
+
 def _design_system_rows(rows) -> list[dict]:
     return [
         {
@@ -171,6 +186,9 @@ async def export_full_backup() -> dict:
         settings = (await session.execute(select(Setting))).scalars().all()
         systems = (await session.execute(select(System))).scalars().all()
         record_systems = (await session.execute(select(RecordSystem))).scalars().all()
+        supersessions = (
+            await session.execute(select(NoteSupersession))
+        ).scalars().all()
         # Parent-first, so a restore can resolve parent_id as it goes rather
         # than needing a second pass — the self-FK is the only ordering
         # constraint in this payload.
@@ -354,6 +372,7 @@ async def export_full_backup() -> dict:
         "design_tokens": _design_token_rows(design_tokens),
         "note_usage_events": _usage_event_rows(usage_events),
         "repo_bindings": _repo_binding_rows(repo_bindings),
+        "note_supersessions": _note_supersession_rows(supersessions),
     }
 
 
@@ -396,6 +415,17 @@ async def export_user_backup(user_id: int) -> dict:
         record_systems = (await session.execute(
             select(RecordSystem).where(RecordSystem.system_id.in_(system_ids))
         )).scalars().all() if system_ids else []
+        # BOTH ends must be this user's notes. A claim spanning out to someone
+        # else's record cannot be restored into a single-user import — the far
+        # id would not be in the map — so carrying it would export a row that
+        # silently vanishes on the way back in. Whole-instance backups have no
+        # such problem and take every row.
+        supersessions = (await session.execute(
+            select(NoteSupersession).where(
+                NoteSupersession.superseder_id.in_(note_ids),
+                NoteSupersession.superseded_id.in_(note_ids),
+            )
+        )).scalars().all() if note_ids else []
         design_systems = (await session.execute(
             select(DesignSystem).where(DesignSystem.owner_user_id == user_id)
             .order_by(DesignSystem.parent_id.nullsfirst(), DesignSystem.id)
@@ -597,6 +627,7 @@ async def export_user_backup(user_id: int) -> dict:
         "design_tokens": _design_token_rows(design_tokens),
         "note_usage_events": _usage_event_rows(usage_events),
         "repo_bindings": _repo_binding_rows(repo_bindings),
+        "note_supersessions": _note_supersession_rows(supersessions),
     }
 
 
@@ -700,6 +731,7 @@ async def _restore_v2(data: dict) -> dict:
         "topic_suppressions": 0,
         "systems": 0, "record_systems": 0, "design_systems": 0,
         "design_tokens": 0, "note_usage_events": 0, "repo_bindings": 0,
+        "note_supersessions": 0,
     }
 
     async with async_session() as session:
@@ -990,6 +1022,24 @@ async def _restore_v2(data: dict) -> dict:
             session.add(RecordSystem(note_id=mapped_nid, system_id=mapped_sid))
             stats["record_systems"] += 1
 
+        # 16b. Supersession claims. Guarded by `data.get` like every other
+        # post-v2 section, so a v5 or older payload restores cleanly without it.
+        #
+        # Both ends must map. A claim is about a PAIR — half of one is not a
+        # weaker claim, it is a dangling row pointing at whatever note happens
+        # to hold that id next.
+        for sup in data.get("note_supersessions", []):
+            mapped_new = note_id_map.get(sup.get("superseder_id", 0))
+            mapped_old = note_id_map.get(sup.get("superseded_id", 0))
+            if mapped_new is None or mapped_old is None or mapped_new == mapped_old:
+                continue
+            session.add(
+                NoteSupersession(
+                    superseder_id=mapped_new, superseded_id=mapped_old
+                )
+            )
+            stats["note_supersessions"] += 1
+
         # 17. Design systems. The export orders these parent-first, so a
         # parent's new id is always in the map by the time a child needs it —
         # no second pass, and a child whose parent is missing lands as a root

tests/test_services_backup.py

@@ -13,8 +13,11 @@ import pytest
 from scribe.services import backup
 
 
-def test_backup_version_is_v5():
-    assert backup.BACKUP_VERSION == 5
+def test_backup_version_is_v6():
+    """v6 added note_supersessions (#278). The bump is the point of the test —
+    a payload section added without moving the version produces backups that
+    are structurally different and indistinguishable by inspection."""
+    assert backup.BACKUP_VERSION == 6
 
 
 def test_not_included_lists_the_known_gaps():
@@ -102,11 +105,26 @@ async def test_export_full_backup_contains_every_declared_section():
     assert out["version"] == backup.BACKUP_VERSION
     assert out["scope"] == "full"
     assert "api_keys" in out["_not_included"]
-    # The sections v2 silently dropped, plus the six v5 added (empty here).
+    # The sections v2 silently dropped, the six v5 added, and v6's
+    # note_supersessions (all empty here).
     for key in ("rulebooks", "rulebook_topics", "rules",
                 "rulebook_subscriptions", "rule_suppressions",
                 "topic_suppressions",
                 "systems", "record_systems", "design_systems",
-                "design_tokens", "note_usage_events", "repo_bindings"):
+                "design_tokens", "note_usage_events", "repo_bindings",
+                "note_supersessions"):
         assert key in out, f"missing export section: {key}"
         assert out[key] == []
+
+
+def test_supersession_rows_serialise_the_pair():
+    """The row builder is a plain function precisely so it can be tested with
+    no database — same reason as the other v5/v6 builders."""
+    class _Row:
+        def __init__(self, a, b):
+            self.superseder_id, self.superseded_id = a, b
+
+    assert backup._note_supersession_rows([_Row(9, 4), _Row(9, 5)]) == [
+        {"superseder_id": 9, "superseded_id": 4},
+        {"superseder_id": 9, "superseded_id": 5},
+    ]