FabledScribe/src/scribe/services/backup.py

import logging
from datetime import date, datetime, timezone

from sqlalchemy import or_, select

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.rule_version import RuleVersion
from scribe.models.design_system import DesignSystem, DesignToken
from scribe.models.note_usage import NoteUsageEvent
from scribe.models.rule_usage import RuleUsageEvent
from scribe.models.retrieval_tuning import RetrievalTuningEvent
from scribe.models.canonical_system import CanonicalSystem
from scribe.models.rulebook import RuleRelation, rule_systems as rule_systems_t
from scribe.models.code_shape import CodeShape, CodeShapeEvent, CodeShapeUse
from scribe.models.project import Project
from scribe.models.repo_binding import RepoBinding
from scribe.models.rulebook import (
    Rule,
    Rulebook,
    RulebookTopic,
)
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

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.
# 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.
# v7 (2026-08) added code_shapes — the shape ledger (#2787). Classifications
# are judgment data worth carrying; a restore keeps a judgment only when its
# snippet target survives the id re-mapping, else the row rejoins the todo.
# v8 (2026-08) added code_shape_events — the ledger's history (#2793): what
# was used where, when, and why is not recomputable, so it travels.
# v9 (2026-08) added code_shape_uses — the ledger's consumption edges (#2870):
# judgment-grade edges (agent/audit/import) are operator records; mechanical
# ones (reference/hook) travel too, cheaply, and the next refresh refreshes them.
# 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).
# 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.
# v13 (2026-08) added rule_versions — a rule's edit history (milestone 323).
# v14 (2026-09) added rule_usage_events — the rule twin of note_usage_events
# (milestone 333). Carrying it is the WHOLE REASON the table is separate: the
# note importer maps note_id through note_id_map, so a rule id parked there
# would restore attached to whatever note took that number.
# v15 (2026-09) dropped rulebook_subscriptions / rule_suppressions /
# topic_suppressions with their tables (milestone 414): a rule's scope is its
# home now. Older archives carrying those sections still restore — the keys are
# simply not read — as do the subscribe_rulebooks inception choices they hold.
# v16 (2026-09) added retrieval_tuning_events (milestone 416): the REASON each
# retrieval floor and budget is where it is. `settings` already carried the
# numbers, so leaving this behind would restore six moved dials with the
# argument for them silently dropped — and from this step on those dials are
# moved by the model, which is exactly the case where the operator needs the
# argument to review.
# v17 (2026-09) added retrieval_tuning_events.embedding_model / shape_version
# (milestone 416 step 6): a floor is a distance in ONE embedding model's
# geometry over documents cut one particular way, so the number alone cannot
# say whether it still measures anything. Both travel NULLABLE and unfilled —
# a row written before the stamp existed restores unstamped, because inventing
# the model it was measured under would turn "unknown" into a stated fact.
# Bump when the serialized schema changes.
BACKUP_VERSION = 17

# 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",
    # 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_supersessions",
    # v7 (2026-08): the shape ledger (#2787); v8: its history (#2793).
    "code_shapes", "code_shape_events", "code_shape_uses",
    # v9 (2026-08): the global area catalog (milestone 307). Global, not
    # user-scoped, so it rides in EVERY export — including a single-user
    # one, whose Systems would otherwise restore unmapped.
    "canonical_systems",
    # v10 (2026-08): a rule's area tag and its typed edges (milestone 307).
    "rule_systems", "rule_relations",
    # v13 (2026-08): a rule's edit history (milestone 323). note_versions has
    # always travelled; its sibling has no excuse not to.
    "rule_versions",
    # v14 (2026-09): rule usage telemetry (milestone 333). Same argument
    # note_usage_events makes for itself — pull-through is only ever
    # accumulated, so a restore that dropped it would silently reset the
    # measurement to zero while everything still looked fine.
    "rule_usage_events",
    # v16 (2026-09): why each retrieval floor and budget is where it is
    # (milestone 416). The values live in `settings` and already travelled; the
    # argument for them had nowhere to go. Now that the model moves these dials
    # on the operator's behalf, a restore that kept the numbers and dropped the
    # reasons would leave an install tuned by nobody it can name.
    "retrieval_tuning_events",
]

# 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;
# the four *_embeddings tables are derived (regenerated at startup
# from the records themselves, which is also how a chunker bump is handled); api_keys are
# 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", "note_embeddings", "rule_embeddings", "milestone_embeddings",
    "system_embeddings",
    "app_logs", "notifications",
    "invitation_tokens", "password_reset_tokens", "user_profiles",
    "retrieval_logs",
    # Sensitive credentials, same reasoning as api_keys: a backup that carries
    # forge tokens is a token-exfiltration file. Users re-add connections
    # after a restore; the per-project pin (projects.forge_connection_id) is
    # deliberately not exported either, so restored projects fall back to
    # keyring-by-host resolution — the documented unpinned behavior (#2778).
    "forge_connections",
    # Derived, like note_embeddings: the CSS consumer map (milestone 302) is
    # rebuilt from the repo archive by every coverage sync, and carries no
    # judgment — the first refresh after a restore recreates it exactly.
    "code_shape_consumers",
]


# 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"},
    # Same as the note twin: the surrogate key is re-issued on insert.
    "rule_usage_events": {"id"},
    # Same again — and everything else travels, because each remaining column
    # is part of the argument: what moved, from what, to what, by whom, why.
    "retrieval_tuning_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"},
    # Everything travels. A version row IS the audit trail, so a column left
    # behind is a fact about a binding instruction that no longer exists
    # anywhere.
    "rule_versions": set(),
    # 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(),
}


# The same accounting for the way BACK IN (#4197). `_COLUMN_EXCLUSIONS` above
# makes a dropped column unexpressible on the way out; this one does it for the
# import, where the same mistake is worse — an export gap leaves an obviously
# thin backup, an import gap means holding a complete, correct file and
# restoring an incomplete database from it, with a success message.
#
# A name here means the `_build_*` helper for that table deliberately does not
# pass it. THREE REASONS ARE LEGITIMATE and nothing else is:
#
#   1. The column is not in the backup at all, because `_COLUMN_EXCLUSIONS`
#      above leaves it out. Nothing can restore what was never written.
#   2. The database issues the value: a surrogate `id`, which must be re-issued
#      because the destination's ids are its own.
#   3. It is written LATER in the same restore, by something that is not a
#      construction — the project's design-system pointer and inception record
#      point at rows created after the project itself.
#
# "We forgot" is not on the list, which is the whole point. `code_shapes`
# carried three that had no reason at all — reason_code, recheck_at and
# diverges_from, exported and dropped — and this table is how they were found.
_IMPORT_COLUMN_EXCLUSIONS: dict[str, set[str]] = {
    "users": {"id"},
    "projects": {
        "id",
        # Not exported (see above), so not restorable.
        "deleted_at", "deleted_batch_id", "forge_connection_id",
        # Reason 3: both point at rows this restore creates AFTER the project,
        # so they are patched in the final pass with their ids re-mapped.
        "design_system_id", "inception",
    },
    "milestones": {"id", "deleted_at", "deleted_batch_id"},
    "notes": {"id", "deleted_at", "deleted_batch_id"},
    "task_logs": {"id"},
    "note_drafts": {"id"},
    "note_versions": {"id"},
    # A composite primary key (user_id, key) — no surrogate to re-issue, and
    # every column travels.
    "settings": set(),
    "rulebooks": {"id", "deleted_at", "deleted_batch_id"},
    "rulebook_topics": {"id", "deleted_at", "deleted_batch_id"},
    "rules": {"id", "deleted_at", "deleted_batch_id"},
    "rule_versions": {"id"},
    # `canonical_id` IS set, from the exported slug. The timestamps are not
    # exported, so the row is re-stamped as of the restore.
    "systems": {
        "id", "deleted_at", "deleted_batch_id", "created_at", "updated_at",
    },
    "canonical_systems": {
        "id", "deleted_at", "deleted_batch_id", "created_at", "updated_at",
    },
    "record_systems": {"id", "created_at"},
    "note_supersessions": {"id", "created_at"},
    "rule_relations": {"id", "created_at"},
    "note_usage_events": {"id"},
    "rule_usage_events": {"id"},
    "retrieval_tuning_events": {"id"},
    "design_systems": {
        "id", "deleted_at", "deleted_batch_id", "created_at", "updated_at",
    },
    "design_tokens": {
        "id", "deleted_at", "deleted_batch_id", "created_at", "updated_at",
    },
    "repo_bindings": {"id", "created_at", "updated_at"},
    "code_shapes": {
        "id",
        # Not exported: the proposer's standing suggestion is derived, and the
        # next refresh recomputes it against the restored snippet ids.
        "proposed_snippet_id", "proposal_basis", "proposal_score",
        "proposal_group", "proposed_at", "proposed_sha",
    },
    "code_shape_events": {"id"},
    "code_shape_uses": {"id"},
}


def _dt(val: str | None) -> datetime:
    return datetime.fromisoformat(val) if val else datetime.now(timezone.utc)


def _dt_or_none(val: str | None) -> datetime | None:
    """Like _dt, but keeps an absent timestamp absent.

    _dt substitutes now() because created_at/updated_at must not be null.
    For a nullable column that MEANS something by being empty, that default
    is a lie: a rule nobody ever verified would restore looking verified at
    the moment of the restore, and drop straight to the bottom of the sweep
    it should have topped.
    """
    return datetime.fromisoformat(val) if val else None


def _d(val: str | None) -> date | None:
    return date.fromisoformat(val) if val else None




# 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 _canonical_system_rows(rows) -> list[dict]:
    """The global area catalog. Carried WITHOUT ids: a restore matches on slug,
    so a target install that already seeded the standard vocabulary reuses its
    own rows and only gains the entries an admin added here."""
    return [
        {
            "name": r.name, "slug": r.slug, "description": r.description,
            "order_index": r.order_index,
        }
        for r in rows
    ]


def _system_rows(rows, canonical_slugs: dict[int, str]) -> list[dict]:
    """A project's Systems. The canonical mapping travels as a SLUG, not an id
    — 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."""
    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,
            "canonical_slug": canonical_slugs.get(r.canonical_id or 0),
        }
        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 _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 [
        {
            "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, "project_id": r.project_id,
            "created_at": r.created_at.isoformat() if r.created_at else None,
        }
        for r in rows
    ]


def _rule_usage_event_rows(rows) -> list[dict]:
    return [
        {
            "user_id": r.user_id, "rule_id": r.rule_id, "event": r.event,
            "source": r.source,
            # The reason a rule was departed from (#4212). Exported because
            # it is the only field on this table that cannot be recomputed:
            # counts can be re-derived from a fresh install's own use, a
            # stated reason cannot, and a `departed` row that comes back
            # without one is indistinguishable from a rule that was missed.
            "detail": r.detail,
            "created_at": r.created_at.isoformat() if r.created_at else None,
        }
        for r in rows
    ]


def _retrieval_tuning_event_rows(rows) -> list[dict]:
    """The record of why a retrieval dial is where it is (#4102).

    Not `to_dict()`: that method serves the MCP reader, which already knows
    whose install it is asking about, so it omits `user_id`. A restore has to
    remap it, and a row that arrived without it would have to be dropped or
    guessed at.
    """
    return [
        {
            "user_id": r.user_id, "surface": r.surface, "dial": r.dial,
            "old_value": r.old_value, "new_value": r.new_value,
            "actor": r.actor, "reason": r.reason,
            # Carried, and NOT defaulted to the current model on the way out
            # (#4104): a row that was unstamped when it was written is still
            # unstamped after a round trip, and a backup that quietly filled
            # the gap would turn "we don't know" into a stated fact.
            "embedding_model": r.embedding_model,
            "shape_version": r.shape_version,
            "created_at": r.created_at.isoformat() if r.created_at else None,
        }
        for r in rows
    ]


def _code_shape_rows(rows) -> list[dict]:
    return [r.to_dict() for r in rows]


def _code_shape_event_rows(rows) -> list[dict]:
    return [r.to_dict() for r in rows]


def _code_shape_use_rows(rows) -> list[dict]:
    return [r.to_dict() for r in rows]


def _repo_binding_rows(rows) -> list[dict]:
    return [
        # `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
    ]


# Row builders for the sections both exporters carry. Pure, like the join-table
# helpers above; the full and per-user exports used to restate every one of
# these comprehensions side by side, and a column added to one and not the
# other is a backup that silently drops it (#2293's shape, one layer down).

def _user_rows(rows) -> list[dict]:
    return [
        {
            "id": u.id, "username": u.username, "email": u.email,
            "password_hash": u.password_hash, "oauth_sub": u.oauth_sub,
            "role": u.role, "session_version": u.session_version,
            "created_at": u.created_at.isoformat(),
        }
        for u in rows
    ]


def _project_rows(rows) -> list[dict]:
    return [
        {
            "id": p.id, "user_id": p.user_id, "title": p.title,
            "description": p.description, "goal": p.goal, "status": p.status,
            "color": p.color,
            "design_system_id": p.design_system_id,
            "inception": p.inception,
            "created_at": p.created_at.isoformat(),
            "updated_at": p.updated_at.isoformat(),
        }
        for p in rows
    ]


def _milestone_rows(rows) -> list[dict]:
    return [
        {
            "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(),
        }
        for m in rows
    ]


def _note_rows(rows) -> list[dict]:
    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(),
            # 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,
        }
        for n in rows
    ]


def _task_log_rows(rows) -> list[dict]:
    return [
        {
            "id": tl.id, "user_id": tl.user_id, "task_id": tl.task_id,
            "content": tl.content, "duration_minutes": tl.duration_minutes,
            "created_at": tl.created_at.isoformat(),
            "updated_at": tl.updated_at.isoformat(),
        }
        for tl in rows
    ]


def _note_draft_rows(rows) -> list[dict]:
    return [
        {
            "id": nd.id, "user_id": nd.user_id, "note_id": nd.note_id,
            "proposed_body": nd.proposed_body, "original_body": nd.original_body,
            "instruction": nd.instruction, "scope": nd.scope,
            "created_at": nd.created_at.isoformat(),
            "updated_at": nd.updated_at.isoformat(),
        }
        for nd in rows
    ]


def _note_version_rows(rows) -> list[dict]:
    return [
        {
            "id": nv.id, "user_id": nv.user_id, "note_id": nv.note_id,
            "title": nv.title, "body": nv.body, "tags": nv.tags or [],
            "pin_kind": nv.pin_kind, "pin_label": nv.pin_label,
            "created_at": nv.created_at.isoformat(),
        }
        for nv in rows
    ]


def _rule_version_rows(rows) -> list[dict]:
    """A rule's edit history. Sibling of _note_version_rows, and it travels for
    the same reason: a version is the only record of what a binding
    instruction used to say, and nothing can recompute it."""
    return [
        {
            "id": rv.id, "rule_id": rv.rule_id, "user_id": rv.user_id,
            "title": rv.title, "statement": rv.statement, "why": rv.why,
            "how_to_apply": rv.how_to_apply, "when_to_apply": rv.when_to_apply,
            "kind": rv.kind, "verify_with": rv.verify_with,
            "expires_when": rv.expires_when,
            "created_at": rv.created_at.isoformat(),
        }
        for rv in rows
    ]


def _setting_rows(rows) -> list[dict]:
    return [{"user_id": s.user_id, "key": s.key, "value": s.value} for s in rows]


def _rulebook_rows(rows) -> list[dict]:
    return [
        {
            "id": rb.id, "owner_user_id": rb.owner_user_id, "title": rb.title,
            "description": rb.description,
            "created_at": rb.created_at.isoformat(),
            "updated_at": rb.updated_at.isoformat(),
        }
        for rb in rows
    ]


def _topic_rows(rows) -> list[dict]:
    return [
        {
            "id": t.id, "rulebook_id": t.rulebook_id, "title": t.title,
            "description": t.description, "order_index": t.order_index,
            "created_at": t.created_at.isoformat(),
            "updated_at": t.updated_at.isoformat(),
        }
        for t in rows
    ]


def _rule_system_rows(rows) -> list[dict]:
    """A rule's area tags, carried by canonical SLUG for the same reason the
    Systems are: the catalog is global and its ids are per-install."""
    return [{"rule_id": rule_id, "canonical_slug": slug} for rule_id, slug in rows]


def _rule_relation_rows(rows) -> list[dict]:
    """The typed edges between rules. Carried because they are a JUDGEMENT —
    someone decided these two fail together, or that one supersedes the other,
    and nothing in either rule's text records the decision. Lose them and a
    split rule silently starts arriving half at a time again."""
    return [
        {
            "from_rule_id": r.from_rule_id, "to_rule_id": r.to_rule_id,
            "kind": r.kind, "note": r.note,
        }
        for r in rows
    ]


def _rule_rows(rows) -> list[dict]:
    return [
        {
            "id": r.id, "topic_id": r.topic_id, "project_id": r.project_id,
            "title": r.title, "statement": r.statement, "why": r.why,
            "how_to_apply": r.how_to_apply, "order_index": r.order_index,
            "when_to_apply": r.when_to_apply, "kind": r.kind,
            "verify_with": r.verify_with, "expires_when": r.expires_when,
            "verified_at": r.verified_at.isoformat() if r.verified_at else None,
            "arose_from_id": r.arose_from_id,
            "created_at": r.created_at.isoformat(),
            "updated_at": r.updated_at.isoformat(),
        }
        for r in rows
    ]


# ---------------------------------------------------------------------------
# Export
# ---------------------------------------------------------------------------

async def export_full_backup() -> dict:
    """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()
        milestones = (await session.execute(select(Milestone))).scalars().all()
        notes = (await session.execute(select(Note))).scalars().all()
        task_logs = (await session.execute(select(TaskLog))).scalars().all()
        note_drafts = (await session.execute(select(NoteDraft))).scalars().all()
        note_versions = (await session.execute(
            select(NoteVersion).order_by(NoteVersion.note_id, NoteVersion.id)
        )).scalars().all()
        rule_versions = (await session.execute(
            select(RuleVersion).order_by(RuleVersion.rule_id, RuleVersion.id)
        )).scalars().all()
        settings = (await session.execute(select(Setting))).scalars().all()
        systems = (await session.execute(select(System))).scalars().all()
        canonical_systems = (await session.execute(
            select(CanonicalSystem).where(CanonicalSystem.deleted_at.is_(None))
            .order_by(CanonicalSystem.order_index)
        )).scalars().all()
        rule_system_rows = (await session.execute(
            select(rule_systems_t.c.rule_id, CanonicalSystem.slug)
            .join(CanonicalSystem, CanonicalSystem.id == rule_systems_t.c.canonical_id)
        )).all()
        rule_relations = (await session.execute(select(RuleRelation))).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.
        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()
        rule_usage_events = (
            await session.execute(select(RuleUsageEvent))
        ).scalars().all()
        # Oldest first, so a restored history reads in the order the dials
        # actually moved — the sequence IS the argument when a surface has been
        # walked up and down.
        retrieval_tuning_events = (await session.execute(
            select(RetrievalTuningEvent).order_by(RetrievalTuningEvent.id)
        )).scalars().all()
        repo_bindings = (await session.execute(select(RepoBinding))).scalars().all()
        code_shapes = (await session.execute(select(CodeShape))).scalars().all()
        code_shape_events = (await session.execute(
            select(CodeShapeEvent).order_by(CodeShapeEvent.at, CodeShapeEvent.id)
        )).scalars().all()
        code_shape_uses = (await session.execute(
            select(CodeShapeUse).order_by(CodeShapeUse.shape_id, CodeShapeUse.snippet_id)
        )).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()

    return {
        "version": BACKUP_VERSION,
        "scope": "full",
        "exported_at": datetime.now(timezone.utc).isoformat(),
        "_security_notice": (
            "This backup contains hashed passwords. "
            "Store it securely and restrict access."
        ),
        "_not_included": _NOT_INCLUDED,
        "users": _user_rows(users),
        "projects": _project_rows(projects),
        "milestones": _milestone_rows(milestones),
        "notes": _note_rows(notes),
        "task_logs": _task_log_rows(task_logs),
        "note_drafts": _note_draft_rows(note_drafts),
        "note_versions": _note_version_rows(note_versions),
        "rule_versions": _rule_version_rows(rule_versions),
        "settings": _setting_rows(settings),
        "rulebooks": _rulebook_rows(rulebooks),
        "rulebook_topics": _topic_rows(topics),
        "rules": _rule_rows(rules),
        "canonical_systems": _canonical_system_rows(canonical_systems),
        "rule_systems": _rule_system_rows(rule_system_rows),
        "rule_relations": _rule_relation_rows(rule_relations),
        "systems": _system_rows(
            systems, {c.id: c.slug for c in canonical_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),
        "rule_usage_events": _rule_usage_event_rows(rule_usage_events),
        "retrieval_tuning_events": _retrieval_tuning_event_rows(
            retrieval_tuning_events
        ),
        "repo_bindings": _repo_binding_rows(repo_bindings),
        "note_supersessions": _note_supersession_rows(supersessions),
        "code_shapes": _code_shape_rows(code_shapes),
        "code_shape_events": _code_shape_event_rows(code_shape_events),
        "code_shape_uses": _code_shape_use_rows(code_shape_uses),
    }


async def export_user_backup(user_id: int) -> dict:
    """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(
            select(Project).where(Project.user_id == user_id)
        )).scalars().all()
        project_ids = [p.id for p in projects]
        milestones = (await session.execute(
            select(Milestone).where(Milestone.user_id == user_id)
        )).scalars().all()
        notes = (await session.execute(
            select(Note).where(Note.user_id == user_id)
        )).scalars().all()
        task_logs = (await session.execute(
            select(TaskLog).where(TaskLog.user_id == user_id)
        )).scalars().all()
        note_drafts = (await session.execute(
            select(NoteDraft).where(NoteDraft.user_id == user_id)
        )).scalars().all()
        note_versions = (await session.execute(
            select(NoteVersion).where(NoteVersion.user_id == user_id)
            .order_by(NoteVersion.note_id, NoteVersion.id)
        )).scalars().all()
        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()
        # Global: taken whole even in a per-user export, because the Systems
        # above reference it and a partial catalog restores partial mappings.
        canonical_systems = (await session.execute(
            select(CanonicalSystem).where(CanonicalSystem.deleted_at.is_(None))
            .order_by(CanonicalSystem.order_index)
        )).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 []
        # 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)
        )).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()
        # The ledger has no user_id of its own — rows belong to the project
        # they account for, so a user's export carries their projects' rows.
        code_shapes = (await session.execute(
            select(CodeShape).where(CodeShape.project_id.in_(project_ids))
        )).scalars().all() if project_ids else []
        code_shape_events = (await session.execute(
            select(CodeShapeEvent).where(CodeShapeEvent.project_id.in_(project_ids))
            .order_by(CodeShapeEvent.at, CodeShapeEvent.id)
        )).scalars().all() if project_ids else []
        code_shape_uses = (await session.execute(
            select(CodeShapeUse).join(CodeShape, CodeShape.id == CodeShapeUse.shape_id)
            .where(CodeShape.project_id.in_(project_ids))
            .order_by(CodeShapeUse.shape_id, CodeShapeUse.snippet_id)
        )).scalars().all() if project_ids else []
        rulebooks = (await session.execute(
            select(Rulebook).where(Rulebook.owner_user_id == user_id)
        )).scalars().all()
        rulebook_ids = [rb.id for rb in rulebooks]
        topics = (await session.execute(
            select(RulebookTopic).where(RulebookTopic.rulebook_id.in_(rulebook_ids))
        )).scalars().all() if rulebook_ids else []
        topic_ids = [t.id for t in topics]
        # Rules owned by the user = rules in the user's topics OR project-rules
        # on the user's projects.
        rule_filters = []
        if topic_ids:
            rule_filters.append(Rule.topic_id.in_(topic_ids))
        if project_ids:
            rule_filters.append(Rule.project_id.in_(project_ids))
        rules = (await session.execute(
            select(Rule).where(or_(*rule_filters))
        )).scalars().all() if rule_filters else []
        # Scoped to the rules this export already carries: an edge whose far
        # end is absent would restore pointing at nothing.
        _rule_ids = [r.id for r in rules]
        rule_system_rows = (await session.execute(
            select(rule_systems_t.c.rule_id, CanonicalSystem.slug)
            .join(CanonicalSystem, CanonicalSystem.id == rule_systems_t.c.canonical_id)
            .where(rule_systems_t.c.rule_id.in_(_rule_ids))
        )).all() if _rule_ids else []
        # Scoped through the RULE, not the version's user_id. That column is
        # the ACTOR (milestone 323), so filtering on it would carry the
        # versions this user wrote on someone ELSE's rule and drop the ones
        # someone else wrote on theirs — the opposite of a per-user export.
        rule_versions = (await session.execute(
            select(RuleVersion).where(RuleVersion.rule_id.in_(_rule_ids))
            .order_by(RuleVersion.rule_id, RuleVersion.id)
        )).scalars().all() if _rule_ids else []
        # Scoped through the RULE for the same reason the versions above are,
        # and it is worth restating because the column that looks right is
        # wrong: `user_id` here is whoever the arm fired FOR, not who owns the
        # rule. Filtering on it would carry this user's surfacings of someone
        # ELSE's rule and drop the ones fired for someone else on theirs.
        rule_usage_events = (await session.execute(
            select(RuleUsageEvent).where(RuleUsageEvent.rule_id.in_(_rule_ids))
        )).scalars().all() if _rule_ids else []
        # Scoped on user_id, and here that IS the right column — unlike the
        # rule usage events directly above. These record changes to this user's
        # OWN retrieval settings, which is what `user_id` means on this table;
        # there is no second owner to route around.
        retrieval_tuning_events = (await session.execute(
            select(RetrievalTuningEvent)
            .where(RetrievalTuningEvent.user_id == user_id)
            .order_by(RetrievalTuningEvent.id)
        )).scalars().all()
        rule_relations = (await session.execute(
            select(RuleRelation).where(
                RuleRelation.from_rule_id.in_(_rule_ids),
                RuleRelation.to_rule_id.in_(_rule_ids),
            )
        )).scalars().all() if _rule_ids else []

    return {
        "version": BACKUP_VERSION,
        "scope": "user",
        "exported_at": datetime.now(timezone.utc).isoformat(),
        "_not_included": _NOT_INCLUDED,
        "user": {
            "id": user.id,
            "username": user.username,
            "email": user.email,
            "role": user.role,
            "created_at": user.created_at.isoformat(),
        } if user else None,
        "projects": _project_rows(projects),
        "milestones": _milestone_rows(milestones),
        "notes": _note_rows(notes),
        "task_logs": _task_log_rows(task_logs),
        "note_drafts": _note_draft_rows(note_drafts),
        "note_versions": _note_version_rows(note_versions),
        "rule_versions": _rule_version_rows(rule_versions),
        "settings": _setting_rows(settings),
        "rulebooks": _rulebook_rows(rulebooks),
        "rulebook_topics": _topic_rows(topics),
        "rules": _rule_rows(rules),
        "canonical_systems": _canonical_system_rows(canonical_systems),
        "rule_systems": _rule_system_rows(rule_system_rows),
        "rule_relations": _rule_relation_rows(rule_relations),
        "systems": _system_rows(
            systems, {c.id: c.slug for c in canonical_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),
        "rule_usage_events": _rule_usage_event_rows(rule_usage_events),
        "retrieval_tuning_events": _retrieval_tuning_event_rows(
            retrieval_tuning_events
        ),
        "repo_bindings": _repo_binding_rows(repo_bindings),
        "note_supersessions": _note_supersession_rows(supersessions),
        "code_shapes": _code_shape_rows(code_shapes),
        "code_shape_events": _code_shape_event_rows(code_shape_events),
        "code_shape_uses": _code_shape_use_rows(code_shape_uses),
    }


# ---------------------------------------------------------------------------
# Per-table builders — the import side's twin of the row helpers above (#4197)
# ---------------------------------------------------------------------------
#
# The column guard (#3182) makes a dropped column unexpressible on the way OUT
# because export goes through per-table pure helpers: hand one a stand-in and
# read which KEYS came back. Nothing watched the way back IN, and that is the
# worse half — an export gap leaves an obviously thin backup, an import gap
# means holding a complete, correct file and restoring an incomplete database
# from it, with a success message.
#
# The guard was one-sided because the code was. `_restore_v2` was one long
# procedural function with the model construction inlined in each loop, so
# there was no per-table unit to hand a stand-in to. These builders are that
# unit. Each takes an exported row plus the id maps built so far and returns
# either the model instance to add or None for a row that cannot be restored.
#
# WHAT DELIBERATELY DID NOT MOVE. The loops, the flushes and the id-map
# bookkeeping stay in `_restore_v2`: that is the order-dependent part, where a
# mistake is a restore that half-works, and it gains nothing from being split.
# Only the construction moved, because the construction is the only thing the
# guard has to see.
#
# RETURNING None IS "SKIP", A None FIELD IS "DEGRADE", and the difference is
# per table by design. `code_shape_events` skips a row whose project will not
# map — a shape event without its project says nothing. `note_usage_events`
# degrades to None — the project is optional, null already means "not
# reported", and dropping the row would lose a real pull and deflate the
# pull-through the table exists to report. Both are right for their table, so
# each builder states its own policy rather than a rule being imposed on all.


class _Maps:
    """The source-id to new-id maps a restore accumulates as it goes.

    One object rather than nine locals so a builder's signature says what it
    depends on, and so the guard can hand every builder the same fully
    populated stand-in without knowing which maps that table happens to read.
    """

    __slots__ = (
        "users", "projects", "milestones", "notes", "rulebooks", "topics",
        "rules", "systems", "design_systems", "shapes", "canonical_by_slug",
    )

    def __init__(self) -> None:
        self.users: dict[int, int] = {}
        self.projects: dict[int, int] = {}
        self.milestones: dict[int, int] = {}
        self.notes: dict[int, int] = {}
        self.rulebooks: dict[int, int] = {}
        self.topics: dict[int, int] = {}
        self.rules: dict[int, int] = {}
        self.systems: dict[int, int] = {}
        self.design_systems: dict[int, int] = {}
        self.shapes: dict[int, int] = {}
        self.canonical_by_slug: dict[str, int] = {}


def _build_user(row: dict, maps: _Maps) -> User:
    """Users are the root of every map, so nothing can fail to resolve."""
    return User(
        username=row["username"],
        email=row.get("email"),
        password_hash=row.get("password_hash"),
        oauth_sub=row.get("oauth_sub"),
        role=row.get("role", "user"),
        session_version=row.get("session_version", 1),
        created_at=_dt(row.get("created_at")),
    )


def _build_project(row: dict, maps: _Maps) -> Project | None:
    uid = maps.users.get(row.get("user_id", 0))
    if uid is None:
        return None
    return Project(
        user_id=uid,
        title=row.get("title", ""),
        description=row.get("description", ""),
        goal=row.get("goal", ""),
        status=row.get("status", "active"),
        color=row.get("color"),
        created_at=_dt(row.get("created_at")),
        updated_at=_dt(row.get("updated_at")),
    )


def _build_milestone(row: dict, maps: _Maps) -> Milestone | None:
    uid = maps.users.get(row.get("user_id", 0))
    pid = maps.projects.get(row.get("project_id", 0))
    if uid is None or pid is None:
        return None
    return Milestone(
        user_id=uid,
        project_id=pid,
        title=row.get("title", ""),
        description=row.get("description"),
        body=row.get("body"),
        status=row.get("status", "active"),
        order_index=row.get("order_index", 0),
        created_at=_dt(row.get("created_at")),
        updated_at=_dt(row.get("updated_at")),
    )


def _build_note(row: dict, maps: _Maps) -> Note | None:
    """`parent_id` and `arose_from_id` land NULL and are patched afterwards.

    Both are ids in the SOURCE database (#3182), and the target note may not
    have been created yet when this row is built.
    """
    uid = maps.users.get(row.get("user_id", 0))
    if uid is None:
        return None
    return Note(
        user_id=uid,
        title=row.get("title", ""),
        body=row.get("body", ""),
        tags=row.get("tags", []),
        parent_id=None,
        arose_from_id=None,
        description=row.get("description"),
        note_type=row.get("note_type") or "note",
        task_kind=row.get("task_kind") or "work",
        data=row.get("data"),
        started_at=_dt_or_none(row.get("started_at")),
        completed_at=_dt_or_none(row.get("completed_at")),
        recurrence_rule=row.get("recurrence_rule"),
        recurrence_next_spawn_at=_dt_or_none(row.get("recurrence_next_spawn_at")),
        project_id=maps.projects.get(row["project_id"]) if row.get("project_id") else None,
        milestone_id=maps.milestones.get(row["milestone_id"]) if row.get("milestone_id") else None,
        status=row.get("status"),
        priority=row.get("priority"),
        due_date=_d(row.get("due_date")),
        created_at=_dt(row.get("created_at")),
        updated_at=_dt(row.get("updated_at")),
        verify_with=row.get("verify_with"),
        expires_when=row.get("expires_when"),
        # _dt_or_none — see the note on the other restore path.
        verified_at=_dt_or_none(row.get("verified_at")),
    )


def _build_task_log(row: dict, maps: _Maps) -> TaskLog | None:
    uid = maps.users.get(row.get("user_id", 0))
    tid = maps.notes.get(row.get("task_id", 0))
    if uid is None or tid is None:
        return None
    return TaskLog(
        user_id=uid,
        task_id=tid,
        content=row.get("content", ""),
        duration_minutes=row.get("duration_minutes"),
        created_at=_dt(row.get("created_at")),
        updated_at=_dt(row.get("updated_at")),
    )


def _build_note_draft(row: dict, maps: _Maps) -> NoteDraft | None:
    uid = maps.users.get(row.get("user_id", 0))
    nid = maps.notes.get(row.get("note_id", 0))
    if uid is None or nid is None:
        return None
    return NoteDraft(
        user_id=uid,
        note_id=nid,
        proposed_body=row.get("proposed_body", ""),
        original_body=row.get("original_body", ""),
        instruction=row.get("instruction", ""),
        scope=row.get("scope", "document"),
        created_at=_dt(row.get("created_at")),
        updated_at=_dt(row.get("updated_at")),
    )


def _build_note_version(row: dict, maps: _Maps) -> NoteVersion | None:
    uid = maps.users.get(row.get("user_id", 0))
    nid = maps.notes.get(row.get("note_id", 0))
    if uid is None or nid is None:
        return None
    return NoteVersion(
        user_id=uid,
        note_id=nid,
        title=row.get("title", ""),
        body=row.get("body", ""),
        tags=row.get("tags", []),
        pin_kind=row.get("pin_kind"),
        pin_label=row.get("pin_label"),
        created_at=_dt(row.get("created_at")),
    )


def _build_setting(row: dict, maps: _Maps) -> Setting | None:
    uid = maps.users.get(row.get("user_id", 0))
    if uid is None:
        return None
    return Setting(user_id=uid, key=row["key"], value=row.get("value", ""))


def _build_retrieval_tuning_event(row: dict, maps: _Maps) -> RetrievalTuningEvent | None:
    """No id remapping beyond the user: `surface` is a registry NAME, not a
    foreign key, which is what lets this history survive a restore into an
    install whose row ids all differ."""
    uid = maps.users.get(row.get("user_id") or 0)
    if uid is None:
        return None
    return RetrievalTuningEvent(
        user_id=uid,
        surface=row.get("surface", ""),
        dial=row.get("dial", ""),
        old_value=row.get("old_value"),
        new_value=row.get("new_value"),
        actor=row.get("actor") or "model",
        reason=row.get("reason", ""),
        # .get with no default, deliberately (v17): an archive written before
        # the stamp existed has no key here, and None is the right answer for
        # it — the same "unstamped" a pre-#4104 row carries in place.
        embedding_model=row.get("embedding_model"),
        shape_version=row.get("shape_version"),
        created_at=_dt(row.get("created_at")),
    )


def _build_rulebook(row: dict, maps: _Maps) -> Rulebook | None:
    uid = maps.users.get(row.get("owner_user_id", 0))
    if uid is None:
        return None
    return Rulebook(
        owner_user_id=uid,
        title=row.get("title", ""),
        description=row.get("description", ""),
        created_at=_dt(row.get("created_at")),
        updated_at=_dt(row.get("updated_at")),
    )


def _build_topic(row: dict, maps: _Maps) -> RulebookTopic | None:
    rbid = maps.rulebooks.get(row.get("rulebook_id", 0))
    if rbid is None:
        return None
    return RulebookTopic(
        rulebook_id=rbid,
        title=row.get("title", ""),
        description=row.get("description"),
        order_index=row.get("order_index", 0),
        created_at=_dt(row.get("created_at")),
        updated_at=_dt(row.get("updated_at")),
    )


def _build_rule(row: dict, maps: _Maps) -> Rule | None:
    """A topic-rule (topic_id) XOR a project-rule (project_id). Neither
    mapping surviving means the rule is orphaned and is not restored."""
    topic = maps.topics.get(row["topic_id"]) if row.get("topic_id") else None
    proj = maps.projects.get(row["project_id"]) if row.get("project_id") else None
    if topic is None and proj is None:
        return None
    return Rule(
        topic_id=topic,
        project_id=proj,
        title=row.get("title", ""),
        statement=row.get("statement", ""),
        why=row.get("why") or None,
        how_to_apply=row.get("how_to_apply") or None,
        when_to_apply=row.get("when_to_apply") or None,
        # A file written before milestone 394 carries `tier` and `always_on`;
        # neither is read. Dropping a field the schema no longer has is the
        # tolerant direction — an archive records what WAS, and refusing it
        # because it remembers a deleted column would make every pre-394
        # backup unrestorable.
        #
        # Same shape, same reason: a file written before 0098 has no kind, and
        # every rule in it was a rule. Defaulting the other way would restore
        # an old backup with things that had always bound quietly no longer
        # binding.
        kind=row.get("kind") or "rule",
        verify_with=row.get("verify_with") or None,
        expires_when=row.get("expires_when") or None,
        # Restored as-is, NOT reset to null. `verified_at` records when someone
        # last ran the check; a restore does not make that untrue, and clearing
        # it would put every constraint at the top of the sweep with nothing
        # having actually changed.
        verified_at=_dt_or_none(row.get("verified_at")),
        # Remapped through the note map like every other note edge. Exported
        # since 0088 but dropped on the way back in until milestone 312 — a
        # restore silently lost every rule's provenance link. SET NULL
        # semantics apply here too: a source note that didn't restore leaves
        # the rule intact.
        arose_from_id=maps.notes.get(row.get("arose_from_id") or 0),
        order_index=row.get("order_index", 0),
        created_at=_dt(row.get("created_at")),
        updated_at=_dt(row.get("updated_at")),
    )


def _build_canonical_system(row: dict, maps: _Maps) -> CanonicalSystem | None:
    """Matched on SLUG, not id. This install already has the standard
    vocabulary from its migrations, so the common case creates nothing and the
    restore simply learns which local id each slug is; only an entry an admin
    added on the source instance is built here."""
    slug = row.get("slug") or ""
    if not slug or slug in maps.canonical_by_slug:
        return None
    return CanonicalSystem(
        name=row.get("name", ""),
        slug=slug,
        description=row.get("description"),
        order_index=row.get("order_index", 0),
    )


def _build_rule_relation(row: dict, maps: _Maps) -> RuleRelation | None:
    """Both ends must map, and a self-edge is not a relation."""
    src = maps.rules.get(row.get("from_rule_id", 0))
    dst = maps.rules.get(row.get("to_rule_id", 0))
    if src is None or dst is None or src == dst:
        return None
    return RuleRelation(
        from_rule_id=src,
        to_rule_id=dst,
        kind=row.get("kind", "co_surfaces"),
        note=row.get("note") or None,
    )


def _build_rule_version(row: dict, maps: _Maps) -> RuleVersion | None:
    rid = maps.rules.get(row.get("rule_id", 0))
    if rid is None:
        return None
    return RuleVersion(
        rule_id=rid,
        # Unlike NoteVersion, an unmappable user does NOT drop the row.
        # user_id is the ACTOR and is nullable by design: the column is SET
        # NULL precisely so history outlives the account that wrote it.
        # Skipping here would delete the record the FK preserves.
        user_id=maps.users.get(row.get("user_id") or 0),
        title=row.get("title", ""),
        statement=row.get("statement", ""),
        why=row.get("why"),
        how_to_apply=row.get("how_to_apply"),
        when_to_apply=row.get("when_to_apply"),
        # NOT defaulted, unlike the rule itself. A version records what was;
        # absent means nobody wrote it down, and inventing "rule" here would
        # put an artifact where a measurement belongs.
        kind=row.get("kind"),
        verify_with=row.get("verify_with"),
        expires_when=row.get("expires_when"),
        created_at=_dt(row.get("created_at")),
    )


def _build_system(row: dict, maps: _Maps) -> System | None:
    uid = maps.users.get(row.get("user_id", 0))
    pid = maps.projects.get(row.get("project_id", 0))
    if uid is None or pid is None:
        return None
    return System(
        user_id=uid,
        project_id=pid,
        name=row.get("name", ""),
        description=row.get("description"),
        color=row.get("color"),
        status=row.get("status", "active"),
        order_index=row.get("order_index", 0),
        # An unknown slug restores UNMAPPED rather than failing: the System and
        # its records are the payload, the mapping is an aid.
        canonical_id=maps.canonical_by_slug.get(row.get("canonical_slug") or ""),
    )


def _build_record_system(row: dict, maps: _Maps) -> RecordSystem | None:
    nid = maps.notes.get(row.get("note_id", 0))
    sid = maps.systems.get(row.get("system_id", 0))
    if nid is None or sid is None:
        return None
    return RecordSystem(note_id=nid, system_id=sid)


def _build_note_supersession(row: dict, maps: _Maps) -> NoteSupersession | None:
    """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."""
    new_id = maps.notes.get(row.get("superseder_id", 0))
    old_id = maps.notes.get(row.get("superseded_id", 0))
    if new_id is None or old_id is None or new_id == old_id:
        return None
    return NoteSupersession(superseder_id=new_id, superseded_id=old_id)


def _build_design_system(row: dict, maps: _Maps) -> DesignSystem | None:
    """The export orders these parent-first, so a parent's new id is always in
    the map by the time a child needs it — and a child whose parent is missing
    lands as a root rather than failing the whole restore."""
    uid = maps.users.get(row.get("owner_user_id", 0))
    if uid is None:
        return None
    return DesignSystem(
        owner_user_id=uid,
        title=row.get("title", ""),
        description=row.get("description"),
        guidance=row.get("guidance"),
        parent_id=maps.design_systems.get(row.get("parent_id") or 0),
    )


def _build_design_token(row: dict, maps: _Maps) -> DesignToken | None:
    dsid = maps.design_systems.get(row.get("design_system_id", 0))
    if dsid is None:
        return None
    return DesignToken(
        design_system_id=dsid,
        name=row.get("name", ""),
        value_by_mode=row.get("value_by_mode") or {},
        group_name=row.get("group_name"),
        purpose=row.get("purpose"),
        rationale=row.get("rationale"),
        supersedes=row.get("supersedes") or [],
        order_index=row.get("order_index", 0),
    )


def _build_usage_event(row: dict, maps: _Maps) -> NoteUsageEvent | None:
    """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."""
    nid = maps.notes.get(row.get("note_id", 0))
    if nid is None:
        return None
    return NoteUsageEvent(
        user_id=maps.users.get(row.get("user_id") or 0),
        note_id=nid,
        event=row.get("event", ""),
        source=row.get("source", ""),
        # DEGRADES to None rather than dropping the row: unlike a shape event,
        # a usage event without a project is still a real pull, and discarding
        # it would deflate the pull-through this table exists to report. Null
        # already means "not reported".
        project_id=(maps.projects.get(row["project_id"])
                    if row.get("project_id") else None),
        created_at=_dt(row.get("created_at")),
    )


def _build_rule_usage_event(row: dict, maps: _Maps) -> RuleUsageEvent | None:
    """Resolved through the RULE map, not the note map. A rule id run through
    the note map would either drop (best case) or land on whatever note took
    that number, producing telemetry that is wrong rather than missing and
    that nothing downstream could detect (milestone 333)."""
    rid = maps.rules.get(row.get("rule_id", 0))
    if rid is None:
        return None
    return RuleUsageEvent(
        user_id=maps.users.get(row.get("user_id") or 0),
        rule_id=rid,
        event=row.get("event", ""),
        source=row.get("source", ""),
        # `.get(...) or None` rather than a bare default: an archive written
        # before 0106 has no key at all, and one written after may carry "" for
        # a non-departure row. Both mean "no reason", and both must land as
        # NULL so the readout does not have to tell an empty string from an
        # absent one.
        detail=(row.get("detail") or None),
        created_at=_dt(row.get("created_at")),
    )


def _build_repo_binding(row: dict, maps: _Maps) -> RepoBinding | None:
    """Small, but losing these means every bound repo quietly stops loading
    its project at session start."""
    uid = maps.users.get(row.get("user_id", 0))
    pid = maps.projects.get(row.get("project_id", 0))
    if uid is None or pid is None:
        return None
    return RepoBinding(
        user_id=uid,
        project_id=pid,
        repo_key=row.get("repo_key", ""),
        ref=row.get("ref"),
    )


def _build_code_shape(row: dict, maps: _Maps) -> CodeShape | None:
    """The ledger's classifications are judgments worth carrying. A judgment
    whose snippet target didn't survive the re-mapping (canonical/instance/
    variant with a gone snippet) is downgraded to unclassified so it rejoins
    the todo honestly instead of dangling; exempt needs no target and keeps."""
    pid = maps.projects.get(row.get("project_id", 0))
    if pid is None:
        return None
    status = row.get("status", "unclassified")
    snippet_id = maps.notes.get(row.get("snippet_id") or 0)
    classified_by = row.get("classified_by")
    classified_at = row.get("classified_at")
    if status in ("canonical", "instance", "variant") and snippet_id is None:
        status = "unclassified"
        classified_by = None
        classified_at = None
    return CodeShape(
        project_id=pid,
        repo_key=row.get("repo_key", ""),
        path=row.get("path", ""),
        symbol=row.get("symbol", ""),
        kind=row.get("kind", "sym"),
        status=status,
        snippet_id=snippet_id,
        reason=row.get("reason"),
        # Exported since #2874 and dropped on the way back in until #4197 —
        # every judged row restored with its verdict and without the code for
        # WHY, which is the column the accounting reads to tell a scoped-css
        # exemption from a convention-plumbing one. Kept on a downgrade for
        # the same reason `reason` is: the free text and its code are one
        # statement, and the existing downgrade clears the judgment's author
        # and date, not its argument.
        reason_code=row.get("reason_code"),
        classified_by=classified_by,
        classified_at=_dt(classified_at) if classified_at else None,
        first_seen_commit=row.get("first_seen_commit", ""),
        last_seen_commit=row.get("last_seen_commit", ""),
        vanished_at=_dt(row["vanished_at"]) if row.get("vanished_at") else None,
        # Fingerprints restore; proposals (#2792) deliberately do not — they
        # are mechanical, and the next refresh recomputes them against the
        # restored snippet ids.
        signature=row.get("signature", ""),
        body_sha=row.get("body_sha", ""),
        classified_sha=row.get("classified_sha", ""),
        # Both exported and both dropped until #4197. `recheck_at` is the
        # standing "this judgment asks to be confirmed again" flag, so losing
        # it restores a tree that looks settled and is not.
        recheck_at=_dt(row["recheck_at"]) if row.get("recheck_at") else None,
        # RE-MAPPED, not carried: it is a FK to notes.id like `snippet_id`,
        # and a raw source id would point at whatever snippet took that
        # number in the destination — wrong rather than missing, which is the
        # milestone 333 trap one table over.
        diverges_from=maps.notes.get(row.get("diverges_from") or 0),
        created_at=_dt(row.get("created_at")),
        updated_at=_dt(row.get("updated_at")),
    )


def _build_code_shape_event(row: dict, maps: _Maps) -> CodeShapeEvent | None:
    """The ledger's history rides its shapes. `snippet_id` is kept as the
    history's own claim (FK-free by design) but re-mapped when the snippet
    survived, so a restored timeline points at restored records.

    SKIPS a row whose project will not map, unlike its usage-event sibling: a
    shape event without its project says nothing."""
    shape_id = maps.shapes.get(row.get("shape_id") or 0)
    pid = maps.projects.get(row.get("project_id", 0))
    if shape_id is None or pid is None:
        return None
    old_sid = row.get("snippet_id")
    return CodeShapeEvent(
        shape_id=shape_id,
        project_id=pid,
        path=row.get("path", ""),
        symbol=row.get("symbol", ""),
        kind=row.get("kind", "sym"),
        event=row.get("event", "classified"),
        status=row.get("status"),
        snippet_id=maps.notes.get(old_sid, old_sid) if old_sid else None,
        classified_by=row.get("classified_by"),
        reason=row.get("reason"),
        commit=row.get("commit", ""),
        at=_dt(row.get("at")),
    )


def _build_code_shape_use(row: dict, maps: _Maps) -> CodeShapeUse | None:
    """Consumption edges (#2870) ride their shape AND their snippet — both
    ends must have survived, or the edge is no longer a fact."""
    shape_id = maps.shapes.get(row.get("shape_id") or 0)
    snippet_id = maps.notes.get(row.get("snippet_id") or 0)
    if shape_id is None or snippet_id is None:
        return None
    return CodeShapeUse(
        shape_id=shape_id,
        snippet_id=snippet_id,
        basis=row.get("basis", "import"),
        evidence=row.get("evidence"),
        created_at=_dt(row.get("created_at")),
    )


# ---------------------------------------------------------------------------
# Restore
# ---------------------------------------------------------------------------

async def restore_full_backup(data: dict) -> dict:
    """Restore from backup JSON. Dispatches by version."""
    version = data.get("version", 1)
    if version == 1:
        return await _restore_v1(data)
    # v2 and v3 share one path; v3-only sections are guarded by data.get so a
    # v2 payload (without them) restores cleanly.
    return await _restore_v2(data)


async def _restore_v1(data: dict) -> dict:
    """Restore legacy v1 backup (original format).

    Pre-pivot v1 backups included conversations + messages; those are
    skipped during restore now that the chat subsystem is gone.
    """
    stats = {"users": 0, "notes": 0, "settings": 0}

    async with async_session() as session:
        user_id_map: dict[int, int] = {}
        for u_data in data.get("users", []):
            old_id = u_data["id"]
            user = User(
                username=u_data["username"],
                email=u_data.get("email"),
                password_hash=u_data["password_hash"],
                role=u_data.get("role", "user"),
                created_at=_dt(u_data.get("created_at")),
            )
            session.add(user)
            await session.flush()
            user_id_map[old_id] = user.id
            stats["users"] += 1

        note_id_map: dict[int, int] = {}
        for n_data in data.get("notes", []):
            old_id = n_data.get("id")
            mapped_user_id = user_id_map.get(n_data.get("user_id", 0))
            if mapped_user_id is None:
                continue
            note = Note(
                user_id=mapped_user_id,
                title=n_data.get("title", ""),
                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")),
                created_at=_dt(n_data.get("created_at")),
                updated_at=_dt(n_data.get("updated_at")),
                verify_with=n_data.get("verify_with"),
                expires_when=n_data.get("expires_when"),
                # _dt_or_none, NOT _dt: an absent stamp must stay absent. _dt
                # substitutes now(), which would restore every never-checked
                # note as checked at the moment of the restore — inverting the
                # one signal the sweep reads.
                verified_at=_dt_or_none(n_data.get("verified_at")),
            )
            session.add(note)
            await session.flush()
            if old_id is not None:
                note_id_map[old_id] = note.id
            stats["notes"] += 1

        # 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_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))
            if mapped_user_id is None:
                continue
            session.add(Setting(user_id=mapped_user_id, key=s_data["key"], value=s_data.get("value", "")))
            stats["settings"] += 1

        await session.commit()

    logger.info("Restored v1 backup: %s", stats)
    return stats


async def _restore_v2(data: dict) -> dict:
    """Restore v2/v3 backup with full FK re-mapping.

    Conversations, push subscriptions, and (as of v4) events in older backups
    are silently skipped — those subsystems were removed. v3+ sections
    (rulebooks/topics/rules/join-tables) are guarded by data.get so a v2
    payload restores without them.

    Every model is constructed by a `_build_*` helper above, never inline.
    That is what gives the import side a per-table unit the column guard can
    hand a stand-in to (#4197); what stays here is the ordering, the flushes
    and the id-map bookkeeping, which is the part a split would only make
    harder to read.
    """
    stats: dict[str, int] = {
        "users": 0, "projects": 0, "milestones": 0, "notes": 0,
        "task_logs": 0, "note_drafts": 0, "note_versions": 0,
        "settings": 0, "rulebooks": 0, "rulebook_topics": 0, "rules": 0,
        "systems": 0, "record_systems": 0, "design_systems": 0,
        "design_tokens": 0, "note_usage_events": 0, "rule_usage_events": 0,
        "repo_bindings": 0,
        "note_supersessions": 0, "code_shapes": 0, "code_shape_events": 0,
        "code_shape_uses": 0, "canonical_systems": 0,
        "rule_systems": 0, "rule_relations": 0, "rule_versions": 0,
        "retrieval_tuning_events": 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

        # 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

        # 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

        # 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

        # 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 = 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

        # 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

        # 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. From milestone 416 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

        # 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

        # 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

        # 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

        # 12-14. Rulebook subscriptions, rule and topic suppressions (v3-v14)
        # `rulebook_subscriptions`, `rule_suppressions` and `topic_suppressions`
        # are READ BY NOBODY since milestone 414 dropped their tables. An
        # archive carrying them still imports, for the reason 14b gives.

        # 14b. Always-on rulebook exclusions (v10, milestone 297)
        # `rulebook_exclusions` was a v10 section and is READ BY NOBODY since
        # milestone 394 removed the table. An archive carrying it still
        # imports — the key is simply not looked at — because refusing a
        # backup for remembering something we deleted would make every v10-v13
        # 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.

        # 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

        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

        # A rule's edit history (milestone 323). Must come after the rules
        # themselves — the rule map is only populated above — and both ids are
        # ids in the SOURCE database, which is #3182's arose_from_id trap.
        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

        # 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

        # 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

        # 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

        # 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

        # 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

        # 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

        # 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

        # 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, #2787)
        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

        # 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 (#2870).
        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 (milestone 394). A pre-394 archive
                # carries 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 subscribe_rulebooks since milestone 414.
                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}

        await session.commit()

    logger.info("Restored v2/v3 backup: %s", stats)
    return stats