src/scribe/services/backup.py

@@ -51,11 +51,17 @@ logger = logging.getLogger(__name__)
 # v10 (2026-08) added projects.inception + project_rulebook_exclusions
 # (milestone 297): the WHY a project inherits what it does, and its opt-outs.
 # v11 (2026-08) added the note verification trio — notes.verify_with /
-# expires_when / verified_at (milestone 317). NOT an audit of the notes
-# section: it carries 16 of the 27 `notes` columns, and #3182 tracks the nine that
-# have been missing since long before this.
+# expires_when / verified_at (milestone 317).
+# v12 (2026-08) closed #3182: the nine Note columns that had been missing for
+# years (note_type, task_kind, arose_from_id, data, description, the
+# recurrence pair, started_at, completed_at), plus milestones.body — which IS
+# the plan — and repo_bindings.ref. Until v12 a restore reported success and
+# handed back a corpus with every snippet and process flattened into a plain
+# note, every issue and spike into `work`, and every plan reduced to a title.
+# _COLUMN_EXCLUSIONS and its guard landed with it, so the next such column
+# fails the build instead.
 # Bump when the serialized schema changes.
-BACKUP_VERSION = 11
+BACKUP_VERSION = 12
 
 # 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
@@ -112,6 +118,82 @@ _NOT_INCLUDED = [
 ]
 
 
+# Columns a backed-up table deliberately does NOT export, per table. Paired
+# with the column-coverage guard in tests/test_services_backup.py, this is
+# _NOT_INCLUDED's shape one level down — and it exists because the table guard
+# could not see the failure it was written to stop.
+#
+# #2293 was six whole tables missing. #3182 was NINE COLUMNS missing from a
+# table that had been "covered" for years: note_type and task_kind, so every
+# snippet and process restored as a plain note and every issue and spike as
+# `work`; arose_from_id, so every provenance edge vanished; the recurrence
+# pair, so recurring tasks stopped recurring; milestones.body, which IS the
+# plan; repo_bindings.ref, the branch a ledger follows. Every one arrived the
+# same way — a column added to the model and the migration, both of which fail
+# loudly, and then never added to the serialiser, which fails silently.
+#
+# So: a new column on a backed-up table now fails the build unless it is either
+# exported or named here with a reason. "I forgot" is no longer expressible.
+_COLUMN_EXCLUSIONS: dict[str, set[str]] = {
+    # Trash is not exported at all, so neither is the batch id that groups a
+    # deletion for restore(). Uniform across every soft-deletable table.
+    "users": set(),
+    "projects": {
+        "deleted_at", "deleted_batch_id",
+        # Credentials-adjacent, same reasoning as api_keys and
+        # forge_connections: a restored project falls back to keyring-by-host
+        # resolution, which is the documented unpinned behaviour (#2778).
+        "forge_connection_id",
+    },
+    "milestones": {"deleted_at", "deleted_batch_id"},
+    "notes": {"deleted_at", "deleted_batch_id"},
+    "task_logs": set(),
+    "note_drafts": set(),
+    "note_versions": set(),
+    "settings": set(),
+    "rulebooks": {"deleted_at", "deleted_batch_id"},
+    "rulebook_topics": {"deleted_at", "deleted_batch_id"},
+    "rules": {"deleted_at", "deleted_batch_id"},
+    "systems": {
+        "deleted_at", "deleted_batch_id",
+        # Travels as `canonical_slug`: the catalog is global and its ids are
+        # per-install, so an id would restore pointing at whatever area
+        # happened to land on that number.
+        "canonical_id",
+        "created_at", "updated_at",
+    },
+    "canonical_systems": {
+        "deleted_at", "deleted_batch_id",
+        # Matched on SLUG at restore, so a target install that already seeded
+        # the standard vocabulary reuses its own rows rather than colliding.
+        "id", "created_at", "updated_at",
+    },
+    # Edge tables: the id is regenerated on insert, and the pair IS the row.
+    "record_systems": {"id", "created_at"},
+    "note_supersessions": {"id", "created_at"},
+    "rule_relations": {"id", "created_at"},
+    "note_usage_events": {"id"},
+    "design_systems": {"deleted_at", "deleted_batch_id", "created_at", "updated_at"},
+    "design_tokens": {"deleted_at", "deleted_batch_id", "created_at", "updated_at"},
+    "repo_bindings": {"id", "created_at", "updated_at"},
+    # Serialised via the model's own to_dict(), so a column reaches the backup
+    # the moment it reaches that method — and the guard still catches one that
+    # reaches neither.
+    "code_shapes": {
+        # The PROPOSER's standing suggestion for an unclassified row, not a
+        # judgment: "looks like an instance of #N", or "repeats with no canon".
+        # Every refresh recomputes it and a judgment clears it, so carrying it
+        # would restore stale machine guesses over a tree the proposer has not
+        # seen. Same reasoning as code_shape_consumers in _NOT_INCLUDED —
+        # derived data is regenerated, never restored.
+        "proposed_snippet_id", "proposal_basis", "proposal_score",
+        "proposal_group", "proposed_at", "proposed_sha",
+    },
+    "code_shape_events": set(),
+    "code_shape_uses": set(),
+}
+
+
 def _dt(val: str | None) -> datetime:
     return datetime.fromisoformat(val) if val else datetime.now(timezone.utc)
 
@@ -244,7 +326,13 @@ def _code_shape_use_rows(rows) -> list[dict]:
 
 def _repo_binding_rows(rows) -> list[dict]:
     return [
-        {"user_id": r.user_id, "project_id": r.project_id, "repo_key": r.repo_key}
+        # `ref` is the branch this binding's ledger follows (#2873). Without
+        # it a restored binding silently falls back to the default branch, and
+        # the shape ledger starts accounting for a different tree (#3182).
+        {
+            "user_id": r.user_id, "project_id": r.project_id,
+            "repo_key": r.repo_key, "ref": r.ref,
+        }
         for r in rows
     ]
 
@@ -286,6 +374,11 @@ def _milestone_rows(rows) -> list[dict]:
         {
             "id": m.id, "user_id": m.user_id, "project_id": m.project_id,
             "title": m.title, "description": m.description, "status": m.status,
+            # THE PLAN. A milestone IS the plan (0066) and `body` is its
+            # design and intent; `description` is only the one-line summary.
+            # Dropping this restored every plan as a title with no reasoning
+            # behind it (#3182).
+            "body": m.body,
             "order_index": m.order_index,
             "created_at": m.created_at.isoformat(),
             "updated_at": m.updated_at.isoformat(),
@@ -295,31 +388,45 @@ def _milestone_rows(rows) -> list[dict]:
 
 
 def _note_rows(rows) -> list[dict]:
-    # INCOMPLETE, and knowingly so — see #3182. This carries 16 of the
-    # `notes` table's 27 columns. `note_type`, `task_kind`, `arose_from_id`, `data`,
-    # `description`, `recurrence_rule`, `recurrence_next_spawn_at`,
-    # `started_at` and `completed_at` are all absent, so a restore flattens
-    # every snippet and process into a plain note and every issue and spike
-    # into `work`. That predates the verification trio below and is tracked
-    # separately rather than fixed in passing: `arose_from_id` points at
-    # another note and needs the same second pass `parent_id` gets, which is
-    # a change with its own trap and deserves its own tests.
-    #
-    # The table-coverage guard cannot see this. It asserts that every TABLE in
-    # Base.metadata is either backed up or declared not-included; nothing
-    # checks columns, which is exactly how nine of them went missing quietly.
     return [
         {
             "id": n.id, "user_id": n.user_id, "title": n.title, "body": n.body,
+            "description": n.description,
             "tags": n.tags or [], "parent_id": n.parent_id,
             "project_id": n.project_id, "milestone_id": n.milestone_id,
             "status": n.status, "priority": n.priority,
             "due_date": n.due_date.isoformat() if n.due_date else None,
             "created_at": n.created_at.isoformat(),
             "updated_at": n.updated_at.isoformat(),
-            # The verification trio (milestone 317, migration 0092). These
-            # travel because they are operator judgment — "somebody checked
-            # this fact, and this is when" — which nothing can recompute.
+            # WHAT KIND OF RECORD THIS IS — both typing axes (#3182). Missing
+            # until now, which meant a restore reported success and handed back
+            # a corpus with every snippet and process flattened into a plain
+            # note and every issue and spike into `work`. Nothing recomputes
+            # these; the vocabulary is simply gone.
+            "note_type": n.note_type,
+            "task_kind": n.task_kind,
+            # Provenance — which record caused this one. Re-mapped in the
+            # second pass beside parent_id, never here: the value is an id in
+            # the SOURCE database.
+            "arose_from_id": n.arose_from_id,
+            # The queryable mirror. The only one of these that would self-heal
+            # (backfill_snippet_data rebuilds it from the body at startup), but
+            # a restore should not hand back a corpus that needs a restart to
+            # become searchable by location.
+            "data": n.data,
+            # Lifecycle: when the work actually started and finished, and the
+            # recurrence rule that makes a task come back. Without these a
+            # restored recurring task simply stops recurring.
+            "started_at": n.started_at.isoformat() if n.started_at else None,
+            "completed_at": n.completed_at.isoformat() if n.completed_at else None,
+            "recurrence_rule": n.recurrence_rule,
+            "recurrence_next_spawn_at": (
+                n.recurrence_next_spawn_at.isoformat()
+                if n.recurrence_next_spawn_at else None
+            ),
+            # The verification trio (milestone 317, migration 0092). Operator
+            # judgment — "somebody checked this fact, and this is when" —
+            # which nothing can recompute.
             "verify_with": n.verify_with,
             "expires_when": n.expires_when,
             "verified_at": n.verified_at.isoformat() if n.verified_at else None,
@@ -770,6 +877,17 @@ async def _restore_v1(data: dict) -> dict:
                 body=n_data.get("body", ""),
                 tags=n_data.get("tags", []),
                 parent_id=None,  # patched below
+                arose_from_id=None,  # patched below, same reason
+                description=n_data.get("description"),
+                note_type=n_data.get("note_type") or "note",
+                task_kind=n_data.get("task_kind") or "work",
+                data=n_data.get("data"),
+                started_at=_dt_or_none(n_data.get("started_at")),
+                completed_at=_dt_or_none(n_data.get("completed_at")),
+                recurrence_rule=n_data.get("recurrence_rule"),
+                recurrence_next_spawn_at=_dt_or_none(
+                    n_data.get("recurrence_next_spawn_at")
+                ),
                 status=n_data.get("status"),
                 priority=n_data.get("priority"),
                 due_date=_d(n_data.get("due_date")),
@@ -789,14 +907,25 @@ async def _restore_v1(data: dict) -> dict:
                 note_id_map[old_id] = note.id
             stats["notes"] += 1
 
-        # Patch parent_id now that all notes have new IDs
+        # Patch the two note->note edges now that every note has a new id.
+        # Both are ids in the SOURCE database, so writing either straight into
+        # the constructor would point at whatever record happens to hold that
+        # number here — a restore that succeeds and silently re-parents (#3182).
+        # An edge whose target did not survive the import is left NULL rather
+        # than guessed at.
         for n_data in data.get("notes", []):
             old_id = n_data.get("id")
+            if not old_id or old_id not in note_id_map:
+                continue
+            note_row = await session.get(Note, note_id_map[old_id])
+            if note_row is None:
+                continue
             old_parent = n_data.get("parent_id")
-            if old_id and old_parent and old_id in note_id_map and old_parent in note_id_map:
-                note_row = await session.get(Note, note_id_map[old_id])
-                if note_row:
-                    note_row.parent_id = note_id_map[old_parent]
+            if old_parent and old_parent in note_id_map:
+                note_row.parent_id = note_id_map[old_parent]
+            old_origin = n_data.get("arose_from_id")
+            if old_origin and old_origin in note_id_map:
+                note_row.arose_from_id = note_id_map[old_origin]
 
         for s_data in data.get("settings", []):
             mapped_user_id = user_id_map.get(s_data.get("user_id", 0))
@@ -889,6 +1018,7 @@ async def _restore_v2(data: dict) -> dict:
                 project_id=mapped_pid,
                 title=m_data.get("title", ""),
                 description=m_data.get("description"),
+                body=m_data.get("body"),
                 status=m_data.get("status", "active"),
                 order_index=m_data.get("order_index", 0),
                 created_at=_dt(m_data.get("created_at")),
@@ -901,6 +1031,7 @@ async def _restore_v2(data: dict) -> dict:
 
         # 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", []):
             mapped_uid = user_id_map.get(n_data.get("user_id", 0))
             if mapped_uid is None:
@@ -911,6 +1042,17 @@ async def _restore_v2(data: dict) -> dict:
                 body=n_data.get("body", ""),
                 tags=n_data.get("tags", []),
                 parent_id=None,
+                arose_from_id=None,  # patched below, same reason as parent_id
+                description=n_data.get("description"),
+                note_type=n_data.get("note_type") or "note",
+                task_kind=n_data.get("task_kind") or "work",
+                data=n_data.get("data"),
+                started_at=_dt_or_none(n_data.get("started_at")),
+                completed_at=_dt_or_none(n_data.get("completed_at")),
+                recurrence_rule=n_data.get("recurrence_rule"),
+                recurrence_next_spawn_at=_dt_or_none(
+                    n_data.get("recurrence_next_spawn_at")
+                ),
                 project_id=project_id_map.get(n_data["project_id"]) if n_data.get("project_id") else None,
                 milestone_id=milestone_id_map.get(n_data["milestone_id"]) if n_data.get("milestone_id") else None,
                 status=n_data.get("status"),
@@ -928,15 +1070,25 @@ async def _restore_v2(data: dict) -> dict:
             note_id_map[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 parent_id
+        # 4b. Patch the note->note edges. Deferred for the same reason
+        # parent_id always has been: these are ids in the SOURCE database
+        # (#3182). An edge whose target did not survive stays NULL.
         for new_note_id, old_parent_id in notes_with_parents:
             new_parent_id = note_id_map.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 = note_id_map.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", []):
@@ -1289,6 +1441,7 @@ async def _restore_v2(data: dict) -> dict:
             session.add(RepoBinding(
                 user_id=mapped_uid, project_id=mapped_pid,
                 repo_key=rb_data.get("repo_key", ""),
+                ref=rb_data.get("ref"),
             ))
             stats["repo_bindings"] += 1
 

tests/test_integration_backup_note_roundtrip.py

@@ -0,0 +1,207 @@
+"""Real-Postgres round trip for the note fields #3182 restored.
+
+The unit lane can prove a serialiser EMITS a key. It cannot prove a restore
+puts the value back on the right row, and the note->note edges are exactly
+where that distinction bites: `parent_id` and `arose_from_id` hold ids from
+the SOURCE database, so a restore that writes them straight through succeeds,
+reports success, and silently points every edge at whatever record happens to
+hold that number here.
+
+That is why #3182 was not fixed in passing.
+
+These drive the REAL `restore_full_backup`, not a reimplementation of its
+loop. A test that re-derives the remap it is checking would agree with
+whatever the product does, including nothing.
+"""
+import pytest
+import pytest_asyncio
+from sqlalchemy import select
+
+from scribe.models import async_session
+from scribe.models.note import Note
+from scribe.models.user import User
+from scribe.services import backup
+from tests.helpers import ensure_user
+
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
+
+RESTORED_USERNAME = "backup_roundtrip_restored"
+
+
+@pytest_asyncio.fixture
+async def source():
+    """A snippet, a process, and an issue that arose from a work task — the
+    shapes whose identity the backup used to drop — exported as backup rows.
+
+    Deleted children-first: `arose_from_id` is a real FK, so removing the
+    origin while the issue still points at it is asking the database a
+    question the test has no reason to ask.
+    """
+    async with async_session() as s:
+        owner = await ensure_user(s, "backup_roundtrip_owner")
+        uid = owner.id
+        await s.commit()
+
+    async with async_session() as s:
+        origin = Note(
+            user_id=uid, title="the work that broke", body="",
+            status="done", task_kind="work",
+        )
+        snippet = Note(
+            user_id=uid, title="debounce — rate-limit", body="```js\n1\n```",
+            note_type="snippet", data={"name": "debounce", "language": "js"},
+        )
+        process = Note(
+            user_id=uid, title="DRY pass", body="steps", note_type="process",
+        )
+        s.add_all([origin, snippet, process])
+        await s.commit()
+        origin_id = origin.id
+
+        issue = Note(
+            user_id=uid, title="the fix", body="symptom -> cause -> fix",
+            status="done", task_kind="issue", arose_from_id=origin_id,
+            description="one-liner",
+        )
+        s.add(issue)
+        await s.commit()
+        order = [issue.id, origin_id, snippet.id, process.id]
+
+    async with async_session() as s:
+        rows = (await s.execute(select(Note).where(Note.id.in_(order)))).scalars().all()
+        note_rows = backup._note_rows(list(rows))
+        user_rows = backup._user_rows(
+            [(await s.execute(select(User).where(User.id == uid))).scalars().one()]
+        )
+    # Restore mints a NEW user from the payload, so the restored corpus is
+    # entirely separate from the source — which is what makes the id
+    # assertions meaningful. Renamed here rather than in a sibling fixture:
+    # `restored` depends on this one, and a rename elsewhere might not have
+    # run by the time the restore does.
+    user_rows[0]["username"] = RESTORED_USERNAME
+
+    yield {
+        "payload": {
+            "version": backup.BACKUP_VERSION,
+            "users": user_rows,
+            "notes": note_rows,
+        },
+        "origin_id": origin_id,
+        "owner_id": uid,
+    }
+
+    async with async_session() as s:
+        for nid in order:
+            row = await s.get(Note, nid)
+            if row is not None:
+                await s.delete(row)
+        await s.commit()
+
+
+@pytest_asyncio.fixture
+async def restored(source):
+    """Run the real restore, then hand back the new rows by title.
+
+    The restore mints a NEW user from the payload, so the restored corpus is
+    entirely separate from the source one — which is what makes the id
+    assertions below meaningful.
+    """
+    await backup.restore_full_backup(source["payload"])
+
+    async with async_session() as s:
+        user = (await s.execute(
+            select(User).where(User.username == RESTORED_USERNAME)
+        )).scalars().first()
+        assert user is not None, "the payload's user was not restored"
+        rows = (await s.execute(
+            select(Note).where(Note.user_id == user.id)
+        )).scalars().all()
+        by_title = {n.title: n for n in rows}
+        new_user_id = user.id
+
+    yield by_title, source
+
+    async with async_session() as s:
+        fresh = [await s.get(Note, r.id) for r in by_title.values()]
+        for row in fresh:
+            if row is not None:
+                row.arose_from_id = None
+                row.parent_id = None
+        await s.flush()
+        for row in fresh:
+            if row is not None:
+                await s.delete(row)
+        await s.commit()
+    async with async_session() as s:
+        user = await s.get(User, new_user_id)
+        if user is not None:
+            await s.delete(user)
+        await s.commit()
+
+
+@pytest_asyncio.fixture(autouse=True)
+async def _no_leftover_restored_user():
+    """The payload's username is fixed, so a previous failed run would leave a
+    row that makes `restored` pick the wrong user. Clear it first."""
+    async with async_session() as s:
+        stale = (await s.execute(
+            select(User).where(User.username == RESTORED_USERNAME)
+        )).scalars().all()
+        for user in stale:
+            notes = (await s.execute(
+                select(Note).where(Note.user_id == user.id)
+            )).scalars().all()
+            for n in notes:
+                n.arose_from_id = None
+                n.parent_id = None
+            await s.flush()
+            for n in notes:
+                await s.delete(n)
+            await s.delete(user)
+        await s.commit()
+
+
+async def test_a_restored_record_keeps_what_it_IS(restored):
+    """#3182's headline. Without note_type and task_kind a restore reported
+    success and handed back a corpus where every snippet and process was a
+    plain note and every issue and spike was `work` — the whole vocabulary
+    milestone 312 and #3128 were about, gone, with nothing to notice it by."""
+    by_title, _ = restored
+
+    assert by_title["debounce — rate-limit"].note_type == "snippet"
+    assert by_title["DRY pass"].note_type == "process"
+    assert by_title["the fix"].task_kind == "issue"
+    assert by_title["the work that broke"].task_kind == "work"
+
+
+async def test_a_restored_snippet_keeps_its_queryable_mirror(restored):
+    """The one field that would self-heal — backfill_snippet_data rebuilds it
+    from the body at startup — but a restore should not hand back a corpus
+    that needs a restart before it is findable by location."""
+    by_title, _ = restored
+    assert by_title["debounce — rate-limit"].data == {
+        "name": "debounce", "language": "js",
+    }
+
+
+async def test_the_provenance_edge_is_remapped_not_copied(restored):
+    """THE regression, and the reason this needed a real database.
+
+    The payload's `arose_from_id` is an id in the SOURCE database. Copying it
+    through would leave the restored issue pointing at whatever record happens
+    to hold that number — a restore that succeeds and silently rewires
+    history. The edge must land on the RESTORED origin instead.
+    """
+    by_title, src = restored
+    issue = by_title["the fix"]
+    origin = by_title["the work that broke"]
+
+    assert issue.arose_from_id == origin.id
+    # ...and that is a different row from the one the payload named.
+    assert issue.arose_from_id != src["origin_id"]
+
+
+async def test_description_and_status_survive(restored):
+    by_title, _ = restored
+    assert by_title["the fix"].description == "one-liner"
+    assert by_title["the fix"].status == "done"

tests/test_services_backup.py

@@ -23,7 +23,7 @@ def test_backup_version_is_current():
 
     (Named for the number it asserted until v10, which is exactly the drift a
     name-carrying-a-value invites; it now says what it checks.)"""
-    assert backup.BACKUP_VERSION == 11
+    assert backup.BACKUP_VERSION == 12
 
 
 def _exportable_note(**over):
@@ -31,10 +31,14 @@ def _exportable_note(**over):
     MagicMock: `_note_rows` calls .isoformat() on the timestamps, and a mock
     would happily return another mock instead of failing."""
     base = dict(
-        id=1, user_id=7, title="t", body="b", tags=["x"], parent_id=None,
+        id=1, user_id=7, title="t", body="b", description=None, tags=["x"],
+        parent_id=None, arose_from_id=None,
         project_id=None, milestone_id=None, status=None, priority=None,
         due_date=None, created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
         updated_at=datetime(2026, 1, 2, tzinfo=timezone.utc),
+        note_type="note", task_kind="work", data=None,
+        started_at=None, completed_at=None,
+        recurrence_rule=None, recurrence_next_spawn_at=None,
         verify_with=None, expires_when=None, verified_at=None,
     )
     base.update(over)
@@ -66,40 +70,188 @@ def test_a_never_checked_note_exports_a_null_stamp_and_restores_as_one():
     assert backup._dt(row["verified_at"]) is not None
 
 
-def test_the_note_section_gap_is_pinned_rather_than_silent():
-    """#3182. `_note_rows` carries 16 of the `notes` table's 27 columns, and the
-    absences are not harmless: without `note_type` every snippet and process
-    restores as a plain note, and without `task_kind` every issue and spike
-    restores as `work`.
+def test_the_record_type_and_kind_survive_the_export():
+    """#3182's headline. Without these two columns a restore reported success
+    and handed back a corpus where all 90 snippets and 3 processes were plain
+    notes and all 435 issues and the spike were `work` — the entire vocabulary
+    milestone 312 and #3128 were about, gone, with nothing to notice it by."""
+    [snippet] = backup._note_rows([_exportable_note(note_type="snippet")])
+    [issue] = backup._note_rows([_exportable_note(task_kind="issue", status="done")])
+    assert snippet["note_type"] == "snippet"
+    assert issue["task_kind"] == "issue"
 
-    The table-coverage guard cannot see this — it asserts that every TABLE is
-    backed up or declared excluded, and nothing checks COLUMNS, which is how
-    these went missing quietly.
 
-    This test exists to make the gap loud and to make fixing it visible: when
-    #3182 lands, this list shrinks, and a reviewer sees exactly which fields
-    started travelling. It is not an endorsement of the omissions.
-    """
+def test_provenance_and_lifecycle_travel():
+    started = datetime(2026, 3, 1, tzinfo=timezone.utc)
+    [row] = backup._note_rows([_exportable_note(
+        arose_from_id=42,
+        description="one-liner",
+        started_at=started,
+        recurrence_rule={"freq": "weekly"},
+    )])
+    assert row["arose_from_id"] == 42
+    assert row["description"] == "one-liner"
+    assert row["started_at"] == started.isoformat()
+    assert row["recurrence_rule"] == {"freq": "weekly"}
+    # Absent lifecycle stamps stay absent — a note that never started must not
+    # restore as one that started at restore time.
+    assert row["completed_at"] is None
+
+
+def test_a_milestone_carries_its_plan():
+    """A milestone IS the plan (0066); `body` is its design and intent and
+    `description` is only the one-line summary. Dropping it restored every
+    plan as a title with no reasoning behind it (#3182)."""
+    m = SimpleNamespace(
+        id=1, user_id=7, project_id=2, title="t", description="d",
+        body="## Goal\n\nthe actual plan", status="active", order_index=0,
+        created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
+        updated_at=datetime(2026, 1, 2, tzinfo=timezone.utc),
+    )
+    [row] = backup._milestone_rows([m])
+    assert row["body"] == "## Goal\n\nthe actual plan"
+
+
+def test_a_repo_binding_carries_the_branch_its_ledger_follows():
+    """#2873. Without `ref` a restored binding silently falls back to the
+    default branch and the shape ledger starts accounting for a different
+    tree — a wrong answer that looks like a working one."""
+    b = SimpleNamespace(user_id=7, project_id=2, repo_key="Scribe", ref="dev")
+    [row] = backup._repo_binding_rows([b])
+    assert row["ref"] == "dev"
+
+
+# The table -> (model, row helper) registry the column guard walks. Kept here
+# rather than in the service because it exists only to be introspected: the
+# product code already knows these pairings by calling them.
+def _column_guard_targets():
+    from scribe.models.canonical_system import CanonicalSystem
+    from scribe.models.code_shape import CodeShape, CodeShapeEvent, CodeShapeUse
+    from scribe.models.design_system import DesignSystem, DesignToken
+    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_usage import NoteUsageEvent
+    from scribe.models.note_version import NoteVersion
+    from scribe.models.project import Project
+    from scribe.models.repo_binding import RepoBinding
+    from scribe.models.rulebook import Rule, Rulebook, RulebookTopic, RuleRelation
+    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
 
-    carried = set(backup._note_rows([_exportable_note()])[0])
-    missing = {c.name for c in Note.__table__.columns} - carried
-
-    assert missing == {
-        # Deliberate: trashed rows are not exported, so neither is the batch
-        # id that groups them for restore().
-        "deleted_at", "deleted_batch_id",
-        # NOT deliberate — the #3182 gap, in the order they hurt.
-        "note_type",       # snippets and processes flatten into notes
-        "task_kind",       # issues and spikes flatten into work
-        "arose_from_id",   # every issue -> origin edge is dropped
-        "recurrence_rule", "recurrence_next_spawn_at",  # recurring tasks stop
-        "started_at", "completed_at",                   # lifecycle history
-        "description",
-        "data",            # self-heals: backfill_snippet_data rebuilds it
+    return {
+        "users": (User, backup._user_rows),
+        "projects": (Project, backup._project_rows),
+        "milestones": (Milestone, backup._milestone_rows),
+        "notes": (Note, backup._note_rows),
+        "task_logs": (TaskLog, backup._task_log_rows),
+        "note_drafts": (NoteDraft, backup._note_draft_rows),
+        "note_versions": (NoteVersion, backup._note_version_rows),
+        "settings": (Setting, backup._setting_rows),
+        "rulebooks": (Rulebook, backup._rulebook_rows),
+        "rulebook_topics": (RulebookTopic, backup._topic_rows),
+        "rules": (Rule, backup._rule_rows),
+        "systems": (System, lambda rows: backup._system_rows(rows, {})),
+        "canonical_systems": (CanonicalSystem, backup._canonical_system_rows),
+        "record_systems": (RecordSystem, backup._record_system_rows),
+        "note_supersessions": (NoteSupersession, backup._note_supersession_rows),
+        "rule_relations": (RuleRelation, backup._rule_relation_rows),
+        "note_usage_events": (NoteUsageEvent, backup._usage_event_rows),
+        "design_systems": (DesignSystem, backup._design_system_rows),
+        "design_tokens": (DesignToken, backup._design_token_rows),
+        "repo_bindings": (RepoBinding, backup._repo_binding_rows),
+        "code_shapes": (CodeShape, backup._code_shape_rows),
+        "code_shape_events": (CodeShapeEvent, backup._code_shape_event_rows),
+        "code_shape_uses": (CodeShapeUse, backup._code_shape_use_rows),
     }
 
 
+def _stand_in(model):
+    """A real instance of `model` with every column set to a value of roughly
+    the right type, so the serialiser runs and we can read which KEYS it
+    produced. Values are meaningless; only the shape of the output dict is
+    under test.
+
+    A real instance rather than a MagicMock because several helpers delegate to
+    the model's own `to_dict()`, and a mock would return another mock instead
+    of a dict. Typed rather than a bare instance because the helpers call
+    `.isoformat()` on the timestamps, which `None` does not have.
+    """
+    import sqlalchemy as sa
+
+    row = model()
+    for column in model.__table__.columns:
+        t = column.type
+        if isinstance(t, sa.DateTime):
+            value = datetime(2026, 1, 1, tzinfo=timezone.utc)
+        elif isinstance(t, sa.Date):
+            value = datetime(2026, 1, 1).date()
+        elif isinstance(t, sa.Boolean):
+            value = False
+        elif isinstance(t, sa.Integer):
+            value = 1
+        elif isinstance(t, sa.ARRAY) or isinstance(getattr(t, "impl", None), sa.ARRAY):
+            value = []
+        elif isinstance(t, (sa.Text, sa.String)):
+            value = "x"
+        else:
+            # JSON/JSONB and anything exotic. None is what these actually hold
+            # most of the time, and no serialiser calls a method on one.
+            value = None
+        setattr(row, column.name, value)
+    return row
+
+
+@pytest.mark.parametrize("table", sorted(_column_guard_targets()))
+def test_every_column_is_exported_or_declared_excluded(table):
+    """THE COLUMN GUARD (#3182) — _NOT_INCLUDED's shape, one level down.
+
+    The table guard below catches a whole table going missing. It cannot catch
+    a COLUMN going missing from a table it already considers covered, which is
+    how nine of them vanished from `notes` alone: note_type and task_kind, so
+    every snippet and process restored as a plain note and every issue and
+    spike as `work`; arose_from_id, so every provenance edge went; the
+    recurrence pair, so recurring tasks stopped recurring. Plus milestones.body
+    — which IS the plan — and repo_bindings.ref.
+
+    Each arrived the same way: added to the model and the migration, both of
+    which fail loudly, and never to the serialiser, which fails silently.
+
+    A new column must now be exported or named in _COLUMN_EXCLUSIONS with a
+    reason. Forgetting is no longer expressible.
+    """
+    model, helper = _column_guard_targets()[table]
+    [row] = helper([_stand_in(model)])
+
+    columns = {c.name for c in model.__table__.columns}
+    missing = columns - set(row)
+    declared = backup._COLUMN_EXCLUSIONS[table]
+
+    assert missing == declared, (
+        f"{table}: exported columns and _COLUMN_EXCLUSIONS disagree.\n"
+        f"  dropped but not declared: {sorted(missing - declared)}\n"
+        f"  declared but exported anyway: {sorted(declared - missing)}"
+    )
+
+
+def test_the_column_guard_covers_every_table_with_a_row_helper():
+    """The guard is only as good as its registry — a table added to _BACKED_UP
+    with a new helper, and not to the registry, would be unguarded and look
+    guarded. Join tables have no model class and carry both their columns by
+    construction, so they are the only permitted absences."""
+    join_tables = {
+        "rulebook_subscriptions", "rule_suppressions",
+        "topic_suppressions", "rulebook_exclusions", "rule_systems",
+    }
+    covered = set(_column_guard_targets()) | join_tables
+    assert set(backup._BACKED_UP) - covered == set()
+    # And no stale entries: every declaration must name a real target.
+    assert set(backup._COLUMN_EXCLUSIONS) == set(_column_guard_targets())
+
+
 def test_not_included_lists_the_known_gaps():
     # The deferred tables must be surfaced explicitly, not silently dropped.
     # forge_connections is excluded as CREDENTIALS (api_keys reasoning): a