fix(backup): six tables were silently absent, and nothing would catch a seventh

services/backup.py enumerated its tables as hand-maintained literals with
nothing tying them to the schema. Tables added since that list was last
extended were absent from every backup — no error, no warning, and a restore
that reports success.

Missing: systems, record_systems (0065), note_usage_events (0071),
design_systems, design_tokens (0072), and repo_bindings — which the issue
itself had not spotted, found only by diffing the model tablenames against the
two lists instead of trusting either.

_NOT_INCLUDED was worse than incomplete: it named "embeddings", "invitations"
and "password_resets", none of which are tables. It read as coverage while
naming nothing the schema could confirm. Now real names, plus retrieval_logs —
observational telemetry that grows per query and that nothing reads for
correctness.

THE DELIVERABLE IS THE GUARD, not the six sections. Extending a list fixes
today and changes nothing about the next table; a new one now fails a test
until someone either backs it up or states that it shouldn't be. It checks
both directions — an unaccounted table, and a listed name that no longer
exists, which is what the three phantom entries above would have tripped.

Design systems need ordering care: parent_id is a self-FK. The export orders
parent-first (parent_id NULLS FIRST, then id — a parent always has the smaller
id), so restore resolves each parent from the map as it goes, with no second
pass. A child whose parent is missing lands as a root rather than failing the
whole restore.

Usage events are kept because pull-through is the evidence base for whether
recall works, and it only ever accumulates — a restore that dropped it would
reset that measurement to zero while everything still looked fine.

BACKUP_VERSION 4 -> 5. Every new restore section is data.get()-guarded, so
v2/v3/v4 payloads restore unchanged.

Closes #2293.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UaYUaouG9jjhATyuxCKrQs

src/scribe/services/backup.py

@@ -8,7 +8,10 @@ from scribe.models.milestone import Milestone
 from scribe.models.note import Note
 from scribe.models.note_draft import NoteDraft
 from scribe.models.note_version import NoteVersion
+from scribe.models.design_system import DesignSystem, DesignToken
+from scribe.models.note_usage import NoteUsageEvent
 from scribe.models.project import Project
+from scribe.models.repo_binding import RepoBinding
 from scribe.models.rulebook import (
     Rule,
     Rulebook,
@@ -18,6 +21,7 @@ from scribe.models.rulebook import (
     project_topic_suppressions,
 )
 from scribe.models.setting import Setting
+from scribe.models.system import RecordSystem, System
 from scribe.models.task_log import TaskLog
 from scribe.models.user import User
 
@@ -26,17 +30,44 @@ logger = logging.getLogger(__name__)
 # Backup format version. v3 (2026-06) added rulebooks/topics/rules + their
 # project subscription/suppression join tables. v4 (2026-07) dropped events
 # 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.
 # Bump when the serialized schema changes.
-BACKUP_VERSION = 4
+BACKUP_VERSION = 5
+
+# 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
+# what tests/test_services_backup.py asserts against Base.metadata.
+#
+# The point is the ABSENCE case. A new table gets a model and a migration, both
+# of which fail loudly if wrong, and then silently never gets a backup section:
+# no error, no warning, and a restore that reports success. Naming the coverage
+# explicitly turns "someone forgot" into a failing test (#2293).
+_BACKED_UP = [
+    "users", "projects", "milestones", "notes", "task_logs", "note_drafts",
+    "note_versions", "settings", "rulebooks", "rulebook_topics", "rules",
+    "project_rulebook_subscriptions", "project_rule_suppressions",
+    "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",
+]
 
 # Tables intentionally NOT in the backup, surfaced in the payload so the gap is
 # explicit rather than silent. ACL (groups/shares) is a coherent follow-up;
-# embeddings are derived (regenerated from note bodies); api_keys are sensitive
+# note_embeddings are derived (regenerated from note bodies); api_keys are
-# credentials; the rest are transient/operational.
+# sensitive credentials; retrieval_logs is observational telemetry that nothing
+# reads for correctness and that grows per query; the rest are
+# transient/operational.
+#
+# REAL table names, deliberately. This list used to read "embeddings",
+# "invitations", "password_resets" — none of which are tables — so it looked
+# like coverage while naming nothing the schema could confirm.
 _NOT_INCLUDED = [
     "groups", "group_memberships", "project_shares", "note_shares",
-    "api_keys", "embeddings", "app_logs", "notifications", "invitations",
+    "api_keys", "note_embeddings", "app_logs", "notifications",
-    "password_resets", "user_profiles",
+    "invitation_tokens", "password_reset_tokens", "user_profiles",
+    "retrieval_logs",
 ]
 
 
@@ -60,12 +91,73 @@ def _topic_suppression_rows(rows) -> list[dict]:
     return [{"project_id": r.project_id, "topic_id": r.topic_id} for r in rows]
 
 
+# The v5 sections. Pure row-builders like the join-table helpers above, for the
+# same reason: CI has no database, so a serialiser that is a plain function is
+# one that can actually be tested.
+
+def _system_rows(rows) -> list[dict]:
+    return [
+        {
+            "id": r.id, "user_id": r.user_id, "project_id": r.project_id,
+            "name": r.name, "description": r.description, "color": r.color,
+            "status": r.status, "order_index": r.order_index,
+        }
+        for r in rows
+    ]
+
+
+def _record_system_rows(rows) -> list[dict]:
+    return [{"note_id": r.note_id, "system_id": r.system_id} for r in rows]
+
+
+def _design_system_rows(rows) -> list[dict]:
+    return [
+        {
+            "id": r.id, "owner_user_id": r.owner_user_id, "title": r.title,
+            "description": r.description, "guidance": r.guidance,
+            "parent_id": r.parent_id,
+        }
+        for r in rows
+    ]
+
+
+def _design_token_rows(rows) -> list[dict]:
+    return [
+        {
+            "id": r.id, "design_system_id": r.design_system_id, "name": r.name,
+            "value_by_mode": r.value_by_mode or {},
+            "group_name": r.group_name, "purpose": r.purpose,
+            "rationale": r.rationale, "supersedes": r.supersedes or [],
+            "order_index": r.order_index,
+        }
+        for r in rows
+    ]
+
+
+def _usage_event_rows(rows) -> list[dict]:
+    return [
+        {
+            "user_id": r.user_id, "note_id": r.note_id, "event": r.event,
+            "source": r.source,
+            "created_at": r.created_at.isoformat() if r.created_at else None,
+        }
+        for r in rows
+    ]
+
+
+def _repo_binding_rows(rows) -> list[dict]:
+    return [
+        {"user_id": r.user_id, "project_id": r.project_id, "repo_key": r.repo_key}
+        for r in rows
+    ]
+
+
 # ---------------------------------------------------------------------------
 # Export
 # ---------------------------------------------------------------------------
 
 async def export_full_backup() -> dict:
-    """Export all data as a version-3 JSON backup."""
+    """Export all data as a version-5 JSON backup."""
     async with async_session() as session:
         users = (await session.execute(select(User))).scalars().all()
         projects = (await session.execute(select(Project))).scalars().all()
@@ -77,6 +169,18 @@ async def export_full_backup() -> dict:
             select(NoteVersion).order_by(NoteVersion.note_id, NoteVersion.id)
         )).scalars().all()
         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()
+        # 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.
+        design_systems = (await session.execute(
+            select(DesignSystem).order_by(DesignSystem.parent_id.nullsfirst(),
+                                          DesignSystem.id)
+        )).scalars().all()
+        design_tokens = (await session.execute(select(DesignToken))).scalars().all()
+        usage_events = (await session.execute(select(NoteUsageEvent))).scalars().all()
+        repo_bindings = (await session.execute(select(RepoBinding))).scalars().all()
         rulebooks = (await session.execute(select(Rulebook))).scalars().all()
         topics = (await session.execute(select(RulebookTopic))).scalars().all()
         rules = (await session.execute(select(Rule))).scalars().all()
@@ -244,11 +348,17 @@ async def export_full_backup() -> dict:
         "rulebook_subscriptions": _subscription_rows(subscriptions),
         "rule_suppressions": _rule_suppression_rows(rule_suppressions),
         "topic_suppressions": _topic_suppression_rows(topic_suppressions),
+        "systems": _system_rows(systems),
+        "record_systems": _record_system_rows(record_systems),
+        "design_systems": _design_system_rows(design_systems),
+        "design_tokens": _design_token_rows(design_tokens),
+        "note_usage_events": _usage_event_rows(usage_events),
+        "repo_bindings": _repo_binding_rows(repo_bindings),
     }
 
 
 async def export_user_backup(user_id: int) -> dict:
-    """Export a single user's data as a version-3 JSON backup."""
+    """Export a single user's data as a version-5 JSON backup."""
     async with async_session() as session:
         user = await session.get(User, user_id)
         projects = (await session.execute(
@@ -274,6 +384,32 @@ async def export_user_backup(user_id: int) -> dict:
         settings = (await session.execute(
             select(Setting).where(Setting.user_id == user_id)
         )).scalars().all()
+        systems = (await session.execute(
+            select(System).where(System.user_id == user_id)
+        )).scalars().all()
+        system_ids = [sy.id for sy in systems]
+        note_ids = [n.id for n in notes]
+        # Scoped by the user's SYSTEMS, not their notes: a shared note carrying
+        # this user's system tag belongs in their backup, and a note of theirs
+        # tagged with someone else's system does not — that row is the other
+        # user's to keep.
+        record_systems = (await session.execute(
+            select(RecordSystem).where(RecordSystem.system_id.in_(system_ids))
+        )).scalars().all() if system_ids else []
+        design_systems = (await session.execute(
+            select(DesignSystem).where(DesignSystem.owner_user_id == user_id)
+            .order_by(DesignSystem.parent_id.nullsfirst(), DesignSystem.id)
+        )).scalars().all()
+        ds_ids = [d.id for d in design_systems]
+        design_tokens = (await session.execute(
+            select(DesignToken).where(DesignToken.design_system_id.in_(ds_ids))
+        )).scalars().all() if ds_ids else []
+        usage_events = (await session.execute(
+            select(NoteUsageEvent).where(NoteUsageEvent.note_id.in_(note_ids))
+        )).scalars().all() if note_ids else []
+        repo_bindings = (await session.execute(
+            select(RepoBinding).where(RepoBinding.user_id == user_id)
+        )).scalars().all()
         rulebooks = (await session.execute(
             select(Rulebook).where(Rulebook.owner_user_id == user_id)
         )).scalars().all()
@@ -455,6 +591,12 @@ async def export_user_backup(user_id: int) -> dict:
         "rulebook_subscriptions": _subscription_rows(subscriptions),
         "rule_suppressions": _rule_suppression_rows(rule_suppressions),
         "topic_suppressions": _topic_suppression_rows(topic_suppressions),
+        "systems": _system_rows(systems),
+        "record_systems": _record_system_rows(record_systems),
+        "design_systems": _design_system_rows(design_systems),
+        "design_tokens": _design_token_rows(design_tokens),
+        "note_usage_events": _usage_event_rows(usage_events),
+        "repo_bindings": _repo_binding_rows(repo_bindings),
     }
 
 
@@ -556,6 +698,8 @@ async def _restore_v2(data: dict) -> dict:
         "settings": 0, "rulebooks": 0, "rulebook_topics": 0, "rules": 0,
         "rulebook_subscriptions": 0, "rule_suppressions": 0,
         "topic_suppressions": 0,
+        "systems": 0, "record_systems": 0, "design_systems": 0,
+        "design_tokens": 0, "note_usage_events": 0, "repo_bindings": 0,
     }
 
     async with async_session() as session:
@@ -814,6 +958,106 @@ async def _restore_v2(data: dict) -> dict:
             ))
             stats["topic_suppressions"] += 1
 
+        # --- v5 sections. Every one is data.get()-guarded, so a v2/v3/v4
+        # payload restores without them rather than failing on an absent key.
+
+        # 15. Systems
+        system_id_map: dict[int, int] = {}
+        for sy_data in data.get("systems", []):
+            mapped_uid = user_id_map.get(sy_data.get("user_id", 0))
+            mapped_pid = project_id_map.get(sy_data.get("project_id", 0))
+            if mapped_uid is None or mapped_pid is None:
+                continue
+            system = System(
+                user_id=mapped_uid, project_id=mapped_pid,
+                name=sy_data.get("name", ""),
+                description=sy_data.get("description"),
+                color=sy_data.get("color"),
+                status=sy_data.get("status", "active"),
+                order_index=sy_data.get("order_index", 0),
+            )
+            session.add(system)
+            await session.flush()
+            system_id_map[sy_data["id"]] = system.id
+            stats["systems"] += 1
+
+        # 16. Record↔system links
+        for rs in data.get("record_systems", []):
+            mapped_nid = note_id_map.get(rs.get("note_id", 0))
+            mapped_sid = system_id_map.get(rs.get("system_id", 0))
+            if mapped_nid is None or mapped_sid is None:
+                continue
+            session.add(RecordSystem(note_id=mapped_nid, system_id=mapped_sid))
+            stats["record_systems"] += 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
+        # rather than failing the whole restore.
+        design_system_id_map: dict[int, int] = {}
+        for ds_data in data.get("design_systems", []):
+            mapped_uid = user_id_map.get(ds_data.get("owner_user_id", 0))
+            if mapped_uid is None:
+                continue
+            design = DesignSystem(
+                owner_user_id=mapped_uid,
+                title=ds_data.get("title", ""),
+                description=ds_data.get("description"),
+                guidance=ds_data.get("guidance"),
+                parent_id=design_system_id_map.get(ds_data.get("parent_id") or 0),
+            )
+            session.add(design)
+            await session.flush()
+            design_system_id_map[ds_data["id"]] = design.id
+            stats["design_systems"] += 1
+
+        # 18. Design tokens
+        for t_data in data.get("design_tokens", []):
+            mapped_dsid = design_system_id_map.get(t_data.get("design_system_id", 0))
+            if mapped_dsid is None:
+                continue
+            session.add(DesignToken(
+                design_system_id=mapped_dsid,
+                name=t_data.get("name", ""),
+                value_by_mode=t_data.get("value_by_mode") or {},
+                group_name=t_data.get("group_name"),
+                purpose=t_data.get("purpose"),
+                rationale=t_data.get("rationale"),
+                supersedes=t_data.get("supersedes") or [],
+                order_index=t_data.get("order_index", 0),
+            ))
+            stats["design_tokens"] += 1
+
+        # 19. Usage events. Kept because pull-through is the evidence base for
+        # whether recall works at all, and it is only ever accumulated — a
+        # restore that dropped it would silently reset that measurement to zero
+        # while everything still looked fine.
+        for ev in data.get("note_usage_events", []):
+            mapped_nid = note_id_map.get(ev.get("note_id", 0))
+            if mapped_nid is None:
+                continue
+            session.add(NoteUsageEvent(
+                user_id=user_id_map.get(ev.get("user_id") or 0),
+                note_id=mapped_nid,
+                event=ev.get("event", ""),
+                source=ev.get("source", ""),
+                created_at=_dt(ev.get("created_at")),
+            ))
+            stats["note_usage_events"] += 1
+
+        # 20. Repo bindings — small, but losing them means every bound repo
+        # quietly stops loading its project at session start.
+        for rb_data in data.get("repo_bindings", []):
+            mapped_uid = user_id_map.get(rb_data.get("user_id", 0))
+            mapped_pid = project_id_map.get(rb_data.get("project_id", 0))
+            if mapped_uid is None or mapped_pid is None:
+                continue
+            session.add(RepoBinding(
+                user_id=mapped_uid, project_id=mapped_pid,
+                repo_key=rb_data.get("repo_key", ""),
+            ))
+            stats["repo_bindings"] += 1
+
         await session.commit()
 
     logger.info("Restored v2/v3 backup: %s", stats)

tests/test_services_backup.py

@@ -13,16 +13,54 @@ import pytest
 from scribe.services import backup
 
 
-def test_backup_version_is_v4():
+def test_backup_version_is_v5():
-    assert backup.BACKUP_VERSION == 4
+    assert backup.BACKUP_VERSION == 5
 
 
 def test_not_included_lists_the_known_gaps():
     # The deferred tables must be surfaced explicitly, not silently dropped.
-    for table in ("groups", "project_shares", "note_shares", "api_keys", "embeddings"):
+    for table in ("groups", "project_shares", "note_shares", "api_keys",
+                  "note_embeddings", "retrieval_logs"):
         assert table in backup._NOT_INCLUDED
 
 
+def test_every_table_is_either_backed_up_or_explicitly_excluded():
+    """THE GUARD (#2293), and the only shape of test that catches an ABSENCE.
+
+    A new table gets a model and a migration — both fail loudly if wrong — and
+    then silently never gets a backup section. No error, no warning, and a
+    restore that reports success. That is how `systems`, `record_systems`,
+    `note_usage_events`, `design_systems`, `design_tokens` and `repo_bindings`
+    all went missing, over five migrations, with nothing to notice.
+
+    Extending the export fixes today. THIS fixes the next one: adding a table
+    now fails here until someone either backs it up or states in
+    `_NOT_INCLUDED` that it shouldn't be. Either is fine; silence is not.
+    """
+    from scribe.models import Base
+
+    schema = set(Base.metadata.tables)
+    accounted = set(backup._BACKED_UP) | set(backup._NOT_INCLUDED)
+
+    unaccounted = schema - accounted
+    assert not unaccounted, (
+        f"{len(unaccounted)} table(s) are neither backed up nor explicitly "
+        f"excluded: {sorted(unaccounted)}. Add each to backup._BACKED_UP (and "
+        f"give it an export + restore section) or to backup._NOT_INCLUDED with "
+        f"a reason in the comment above it."
+    )
+
+    # And the reverse: a name in either list that no longer exists is a lie the
+    # guard would otherwise keep telling. This half is what caught "embeddings",
+    # "invitations" and "password_resets" — three entries that named nothing.
+    phantom = accounted - schema
+    assert not phantom, (
+        f"backup lists table(s) that are not in the schema: {sorted(phantom)}. "
+        f"Renamed or dropped — fix the list rather than leaving it to read as "
+        f"coverage."
+    )
+
+
 def test_join_table_row_helpers_are_pure():
     subs = [SimpleNamespace(project_id=1, rulebook_id=2)]
     rsup = [SimpleNamespace(project_id=1, rule_id=9)]