feat(523): the backup as an uncompressed tar of per-table JSON Lines, ordered so restic dedupes it

export_full_backup orders every query by primary key: an UPDATE moves a
row to the end of the heap, so an unordered export shifted every byte
after an edited row. services/backup_file.py writes the payload as one
dated tar per run - manifest.json plus <section>.jsonl, sorted keys,
fixed member headers - atomically via a dot-prefixed .part file, and
reads it back into the dict restore_full_backup takes, refusing a
missing or short section.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/scribe/services/backup.py

@@ -982,42 +982,53 @@ def _rule_rows(rows) -> list[dict]:
 # ---------------------------------------------------------------------------
 
 async def export_full_backup() -> dict:
-    """Export all data as a version-5 JSON backup."""
+    """Export all data as a full backup payload.
+
+    Every query is ORDERED, by primary key unless a restore needs another
+    order. Postgres returns an unordered select in heap order, and an UPDATE
+    moves a row to the end of the heap: an edited note would jump to the
+    bottom of `notes`, shifting every byte after its old place. The scheduled
+    backup file (services/backup_file.py) is deduplicated by restic between
+    runs, which only works when unchanged rows serialise to the same bytes in
+    the same place.
+    """
     async with async_session() as session:
-        users = (await session.execute(select(User))).scalars().all()
+        users =(await session.execute(select(User).order_by(User.id))).scalars().all()
-        projects = (await session.execute(select(Project))).scalars().all()
+        projects = (await session.execute(select(Project).order_by(Project.id))).scalars().all()
-        milestones = (await session.execute(select(Milestone))).scalars().all()
+        milestones = (await session.execute(select(Milestone).order_by(Milestone.id))).scalars().all()
-        notes = (await session.execute(select(Note))).scalars().all()
+        notes = (await session.execute(select(Note).order_by(Note.id))).scalars().all()
-        task_logs = (await session.execute(select(TaskLog))).scalars().all()
+        task_logs = (await session.execute(select(TaskLog).order_by(TaskLog.id))).scalars().all()
-        note_drafts = (await session.execute(select(NoteDraft))).scalars().all()
+        note_drafts = (await session.execute(select(NoteDraft).order_by(NoteDraft.id))).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()
+        settings = (await session.execute(select(Setting).order_by(Setting.user_id, Setting.key))).scalars().all()
-        systems = (await session.execute(select(System))).scalars().all()
+        systems = (await session.execute(select(System).order_by(System.id))).scalars().all()
         canonical_systems = (await session.execute(
             select(CanonicalSystem).where(CanonicalSystem.deleted_at.is_(None))
-            .order_by(CanonicalSystem.order_index)
+            .order_by(CanonicalSystem.order_index, CanonicalSystem.id)
         )).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)
+            .order_by(rule_systems_t.c.rule_id, CanonicalSystem.slug)
         )).all()
         rule_moment_rows = (await session.execute(
             select(rule_moments_t.c.rule_id, rule_moments_t.c.moment)
+            .order_by(rule_moments_t.c.rule_id, rule_moments_t.c.moment)
         )).all()
         rule_moment_judgments = (await session.execute(
-            select(RuleMomentJudgment)
+            select(RuleMomentJudgment).order_by(RuleMomentJudgment.id)
         )).scalars().all()
-        rule_relations = (await session.execute(select(RuleRelation))).scalars().all()
+        rule_relations = (await session.execute(select(RuleRelation).order_by(RuleRelation.id))).scalars().all()
-        lesson_rule_links = (await session.execute(select(LessonRuleLink))).scalars().all()
+        lesson_rule_links = (await session.execute(select(LessonRuleLink).order_by(LessonRuleLink.id))).scalars().all()
-        lesson_no_rule = (await session.execute(select(LessonNoRule))).scalars().all()
+        lesson_no_rule = (await session.execute(select(LessonNoRule).order_by(LessonNoRule.lesson_id))).scalars().all()
-        record_systems = (await session.execute(select(RecordSystem))).scalars().all()
+        record_systems = (await session.execute(select(RecordSystem).order_by(RecordSystem.id))).scalars().all()
         supersessions = (
-            await session.execute(select(NoteSupersession))
+            await session.execute(select(NoteSupersession).order_by(NoteSupersession.id))
         ).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
@@ -1026,13 +1037,13 @@ async def export_full_backup() -> dict:
             select(DesignSystem).order_by(DesignSystem.parent_id.nullsfirst(),
                                           DesignSystem.id)
         )).scalars().all()
-        design_tokens = (await session.execute(select(DesignToken))).scalars().all()
+        design_tokens = (await session.execute(select(DesignToken).order_by(DesignToken.id))).scalars().all()
-        usage_events = (await session.execute(select(NoteUsageEvent))).scalars().all()
+        usage_events = (await session.execute(select(NoteUsageEvent).order_by(NoteUsageEvent.id))).scalars().all()
         rule_usage_events = (
-            await session.execute(select(RuleUsageEvent))
+            await session.execute(select(RuleUsageEvent).order_by(RuleUsageEvent.id))
         ).scalars().all()
         system_usage_events = (
-            await session.execute(select(SystemUsageEvent))
+            await session.execute(select(SystemUsageEvent).order_by(SystemUsageEvent.id))
         ).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
@@ -1043,30 +1054,33 @@ async def export_full_backup() -> dict:
         moment_mappings = (await session.execute(
             select(MomentMapping).order_by(MomentMapping.id)
         )).scalars().all()
-        repo_bindings = (await session.execute(select(RepoBinding))).scalars().all()
+        repo_bindings = (await session.execute(select(RepoBinding).order_by(RepoBinding.id))).scalars().all()
-        code_shapes = (await session.execute(select(CodeShape))).scalars().all()
+        code_shapes = (await session.execute(select(CodeShape).order_by(CodeShape.id))).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)
+            select(CodeShapeUse).order_by(CodeShapeUse.shape_id, CodeShapeUse.snippet_id, CodeShapeUse.id)
         )).scalars().all()
-        rulebooks = (await session.execute(select(Rulebook))).scalars().all()
+        rulebooks = (await session.execute(select(Rulebook).order_by(Rulebook.id))).scalars().all()
-        topics = (await session.execute(select(RulebookTopic))).scalars().all()
+        topics = (await session.execute(select(RulebookTopic).order_by(RulebookTopic.id))).scalars().all()
-        rules = (await session.execute(select(Rule))).scalars().all()
+        rules = (await session.execute(select(Rule).order_by(Rule.id))).scalars().all()
         platforms = (await session.execute(
             select(Platform).where(Platform.deleted_at.is_(None))
-            .order_by(Platform.order_index)
+            .order_by(Platform.order_index, Platform.id)
         )).scalars().all()
-        project_platforms = (await session.execute(select(ProjectPlatform))).scalars().all()
+        project_platforms = (await session.execute(select(ProjectPlatform).order_by(
-        family_ideas = (await session.execute(select(FamilyIdea))).scalars().all()
+            ProjectPlatform.project_id, ProjectPlatform.platform_id))).scalars().all()
+        family_ideas = (await session.execute(select(FamilyIdea).order_by(FamilyIdea.note_id))).scalars().all()
         family_idea_platforms = (await session.execute(
-            select(FamilyIdeaPlatform)
+            select(FamilyIdeaPlatform).order_by(
+                FamilyIdeaPlatform.note_id, FamilyIdeaPlatform.platform_id)
         )).scalars().all()
         family_idea_references = (await session.execute(
-            select(FamilyIdeaReference)
+            select(FamilyIdeaReference).order_by(
+                FamilyIdeaReference.idea_id, FamilyIdeaReference.snippet_id)
         )).scalars().all()
-        family_adoptions = (await session.execute(select(FamilyAdoption))).scalars().all()
+        family_adoptions = (await session.execute(select(FamilyAdoption).order_by(FamilyAdoption.id))).scalars().all()
         # Oldest first: a precedent is always an earlier decision, so a restore
         # in this order has every precedent mapped before anything cites it.
         family_decisions = (await session.execute(

src/scribe/services/backup_file.py

@@ -0,0 +1,214 @@
+"""The full backup as a file on disk: one uncompressed tar per run.
+
+The scheduled backup lands somewhere that keeps a history of it — a folder a
+backup service watches, an S3 bucket — and from there it is usually copied
+off-site by a deduplicating tool such as restic. Everything about the layout
+serves that copy: restic cuts a file into chunks at boundaries chosen by the
+content, so a chunk it has seen before costs nothing to store again. The file
+is therefore laid out so that what did not change between two runs is the
+same bytes, in the same place, both times.
+
+- **Never compressed.** Compression rewrites every byte after the first
+  change, so no chunk would ever match. restic compresses its own repository
+  after deduplicating.
+- **One member per payload section**, `<section>.jsonl`, one row per line.
+  Rows arrive in primary-key order (export_full_backup orders every query)
+  and each is written with sorted keys and no spacing, so an unchanged row is
+  identical bytes, an edit changes one line, and rows appended to a growing
+  table land at the end of its member. A busy table never shifts the bytes of
+  a quiet one.
+- **The parts that change every run live in `manifest.json` only**: the
+  export time, the app version and the row counts. The member headers are
+  fixed (mtime 0, owner 0, mode 0644) and the members come in one order: the
+  manifest, then the sections by name.
+- **One file per run, under a new dated name**, so a service that keeps one
+  file per backup can thin the history on its own schedule.
+
+The payload inside is exactly `export_full_backup()`'s, so the coverage
+guards on that payload (tests/test_services_backup.py) cover this file too,
+and `read_archive` hands `restore_full_backup` the same dict it always took.
+"""
+from __future__ import annotations
+
+import io
+import json
+import os
+import tarfile
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import IO
+
+ARCHIVE_FORMAT = "scribe-backup-tar"
+ARCHIVE_FORMAT_VERSION = 1
+MANIFEST_NAME = "manifest.json"
+SECTION_SUFFIX = ".jsonl"
+NAME_PREFIX = "scribe-"
+NAME_SUFFIX = ".tar"
+
+
+def archive_name(at: datetime) -> str:
+    """`scribe-20261015T030000Z.tar` — UTC, sortable, one per run."""
+    return f"{NAME_PREFIX}{at.astimezone(timezone.utc):%Y%m%dT%H%M%SZ}{NAME_SUFFIX}"
+
+
+def _encode(value) -> str:
+    # default=str matches the admin download's json.dumps, so a datetime that
+    # reaches the payload unconverted reads back the same way on restore.
+    return json.dumps(
+        value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=str,
+    )
+
+
+def _split(payload: dict) -> tuple[dict, dict[str, list]]:
+    """Separate the row sections from the header (version, scope, notices).
+
+    A section is a list of rows under a plain name. `_not_included` is a list
+    too, but of table names: it is a notice, so it stays in the header.
+    """
+    header: dict = {}
+    sections: dict[str, list] = {}
+    for key, value in payload.items():
+        if isinstance(value, list) and not key.startswith("_"):
+            sections[key] = value
+        else:
+            header[key] = value
+    return header, sections
+
+
+def _member(name: str, data: bytes) -> tuple[tarfile.TarInfo, io.BytesIO]:
+    info = tarfile.TarInfo(name)
+    info.size = len(data)
+    info.mtime = 0
+    info.mode = 0o644
+    info.uid = info.gid = 0
+    info.uname = info.gname = ""
+    return info, io.BytesIO(data)
+
+
+def write_archive_to(payload: dict, out: IO[bytes], *, app_version: str) -> dict:
+    """Write `payload` as the backup tar to an open binary file.
+
+    Returns the manifest, so the caller can report row counts without
+    reading the file back.
+    """
+    header, sections = _split(payload)
+    names = sorted(sections)
+    manifest = {
+        "format": ARCHIVE_FORMAT,
+        "format_version": ARCHIVE_FORMAT_VERSION,
+        "app_version": app_version,
+        "header": header,
+        "sections": {name: len(sections[name]) for name in names},
+    }
+    with tarfile.open(fileobj=out, mode="w", format=tarfile.USTAR_FORMAT) as tar:
+        tar.addfile(*_member(
+            MANIFEST_NAME, (json.dumps(manifest, sort_keys=True, indent=2,
+                                       default=str) + "\n").encode(),
+        ))
+        for name in names:
+            data = b"".join(
+                _encode(row).encode() + b"\n" for row in sections[name]
+            )
+            tar.addfile(*_member(name + SECTION_SUFFIX, data))
+    return manifest
+
+
+def write_archive(
+    payload: dict, directory: Path, at: datetime, *, app_version: str,
+) -> Path:
+    """Write the backup tar into `directory` under its dated name, atomically.
+
+    The bytes go to a dot-prefixed `.part` file first, are fsync'd, and only
+    then renamed to the final name. A watcher of the folder (or anything else
+    listing it) never sees half a backup under a backup's name, and a run that
+    fails leaves nothing behind but the error.
+    """
+    directory.mkdir(parents=True, exist_ok=True)
+    final = directory / archive_name(at)
+    part = directory / f".{final.name}.part"
+    try:
+        with open(part, "wb") as f:
+            write_archive_to(payload, f, app_version=app_version)
+            f.flush()
+            os.fsync(f.fileno())
+        os.replace(part, final)
+    except BaseException:
+        part.unlink(missing_ok=True)
+        raise
+    _fsync_dir(directory)
+    return final
+
+
+def _fsync_dir(directory: Path) -> None:
+    """Make the rename itself durable. Some filesystems (and NFS mounts)
+    refuse a directory fsync; the file's own fsync already happened, so that
+    refusal is not a failure of the backup."""
+    try:
+        fd = os.open(directory, os.O_RDONLY)
+    except OSError:
+        return
+    try:
+        os.fsync(fd)
+    except OSError:
+        pass
+    finally:
+        os.close(fd)
+
+
+class ArchiveError(ValueError):
+    """The file is not a backup tar this version can read, or is damaged."""
+
+
+def is_archive(fileobj: IO[bytes]) -> bool:
+    """True when `fileobj` (seekable, at its start) holds a tar. Leaves the
+    position where it was, so the caller can fall back to reading JSON."""
+    pos = fileobj.tell()
+    try:
+        head = fileobj.read(tarfile.BLOCKSIZE)
+    finally:
+        fileobj.seek(pos)
+    return len(head) == tarfile.BLOCKSIZE and head[257:262] == b"ustar"
+
+
+def read_archive(fileobj: IO[bytes]) -> dict:
+    """Read a backup tar back into the payload `restore_full_backup` takes.
+
+    Every section the manifest names must be present with the row count it
+    names: a member that went missing or was cut short is an error here, not
+    a restore that reports success with less than it was given.
+    """
+    try:
+        tar = tarfile.open(fileobj=fileobj, mode="r:")
+    except tarfile.TarError as exc:
+        raise ArchiveError(f"not a backup archive: {exc}") from exc
+    with tar:
+        manifest = json.loads(_read_member(tar, MANIFEST_NAME))
+        if manifest.get("format") != ARCHIVE_FORMAT:
+            raise ArchiveError("not a Scribe backup archive (no Scribe manifest)")
+        if manifest.get("format_version", 0) > ARCHIVE_FORMAT_VERSION:
+            raise ArchiveError(
+                f"archive format {manifest['format_version']} is newer than "
+                f"this Scribe reads ({ARCHIVE_FORMAT_VERSION}); upgrade first"
+            )
+        payload = dict(manifest.get("header") or {})
+        for name, count in (manifest.get("sections") or {}).items():
+            raw = _read_member(tar, name + SECTION_SUFFIX)
+            rows = [json.loads(line) for line in raw.splitlines() if line]
+            if len(rows) != count:
+                raise ArchiveError(
+                    f"{name}{SECTION_SUFFIX} holds {len(rows)} rows; "
+                    f"the manifest says {count}"
+                )
+            payload[name] = rows
+    return payload
+
+
+def _read_member(tar: tarfile.TarFile, name: str) -> bytes:
+    try:
+        member = tar.getmember(name)
+    except KeyError as exc:
+        raise ArchiveError(f"the archive has no {name}") from exc
+    f = tar.extractfile(member)
+    if f is None:
+        raise ArchiveError(f"{name} in the archive is not a file")
+    return f.read()

tests/test_integration_backup_file_ordering.py

@@ -0,0 +1,94 @@
+"""Real-Postgres check that the full export is ORDERED, so the backup file
+deduplicates between runs (milestone 523).
+
+Only Postgres shows this. An unordered select returns rows in heap order, and
+an UPDATE writes the row's new version at the end of the heap — so before
+export_full_backup ordered its queries, editing one old note moved it to the
+bottom of `notes`. Every byte after its old place shifted, and a restic run
+over the next backup stored most of the table again. A mocked session returns
+whatever order it was handed and could never catch it.
+"""
+import io
+import tarfile
+
+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 scribe.services.backup_file import write_archive_to
+from tests.helpers import ensure_user
+
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
+
+USERNAME = "backup_file_ordering_owner"
+
+
+async def _purge() -> None:
+    async with async_session() as s:
+        for user in (await s.execute(
+            select(User).where(User.username == USERNAME)
+        )).scalars().all():
+            for note in (await s.execute(
+                select(Note).where(Note.user_id == user.id)
+            )).scalars().all():
+                await s.delete(note)
+            await s.delete(user)
+        await s.commit()
+
+
+@pytest_asyncio.fixture(autouse=True)
+async def _no_leftovers():
+    """Setup only — see the rule_version roundtrip module for why a teardown
+    here would break the next test's event loop."""
+    await _purge()
+
+
+@pytest_asyncio.fixture
+async def notes():
+    async with async_session() as s:
+        user = await ensure_user(s, USERNAME)
+        rows = [Note(user_id=user.id, title=f"ordering {n}", body="b" * 100)
+                for n in range(5)]
+        s.add_all(rows)
+        await s.commit()
+        ids = [n.id for n in rows]
+    yield ids
+    await _purge()
+
+
+def _sections(payload: dict) -> dict[str, bytes]:
+    buf = io.BytesIO()
+    write_archive_to(payload, buf, app_version="test")
+    buf.seek(0)
+    with tarfile.open(fileobj=buf, mode="r:") as tar:
+        return {m.name: tar.extractfile(m).read() for m in tar.getmembers()
+                if m.name != "manifest.json"}
+
+
+async def _edit_first(ids: list[int]) -> None:
+    async with async_session() as s:
+        note = await s.get(Note, ids[0])
+        note.body = "edited after the first export"
+        await s.commit()
+
+
+async def test_an_edited_row_keeps_its_place(notes):
+    await _edit_first(notes)
+    payload = await backup.export_full_backup()
+    exported = [row["id"] for row in payload["notes"]]
+    assert exported == sorted(exported)
+    mine = [i for i in exported if i in set(notes)]
+    assert mine == notes
+
+
+async def test_two_exports_of_unchanged_data_are_the_same_bytes(notes):
+    await _edit_first(notes)
+    first = _sections(await backup.export_full_backup())
+    second = _sections(await backup.export_full_backup())
+    assert first.keys() == second.keys()
+    for name in first:
+        assert first[name] == second[name], name

tests/test_services_backup_file.py

@@ -0,0 +1,201 @@
+"""Unit tests for the backup file (services/backup_file.py).
+
+The file exists to be deduplicated between runs by a tool like restic, so
+most of what is tested here is a property of BYTES: what stays identical
+when nothing changed, and how little moves when one row does. A round trip
+alone would pass for a gzip of one big JSON document, which is exactly the
+shape this file replaced.
+"""
+import io
+import tarfile
+from datetime import datetime, timezone
+from unittest.mock import patch
+
+import pytest
+
+from scribe.services import backup_file
+from scribe.services.backup_file import (
+    ArchiveError, archive_name, is_archive, read_archive, write_archive,
+    write_archive_to,
+)
+
+AT = datetime(2026, 10, 15, 3, 0, 0, tzinfo=timezone.utc)
+
+
+def _payload(**over) -> dict:
+    payload = {
+        "version": 25,
+        "scope": "full",
+        "exported_at": "2026-10-15T03:00:00+00:00",
+        "_security_notice": "hashed passwords inside",
+        "_not_included": ["api_keys", "note_embeddings"],
+        "users": [{"id": 1, "username": "ada", "role": "admin"}],
+        "notes": [
+            {"id": n, "user_id": 1, "title": f"note {n}", "body": "x" * 200,
+             "tags": ["a", "b"]}
+            for n in range(1, 51)
+        ],
+        "settings": [],
+    }
+    payload.update(over)
+    return payload
+
+
+def _members(payload: dict) -> dict[str, bytes]:
+    buf = io.BytesIO()
+    write_archive_to(payload, buf, app_version="2026.10.15.0300")
+    buf.seek(0)
+    with tarfile.open(fileobj=buf, mode="r:") as tar:
+        return {m.name: tar.extractfile(m).read() for m in tar.getmembers()}
+
+
+def test_the_name_is_dated_in_utc_and_sorts():
+    assert archive_name(AT) == "scribe-20261015T030000Z.tar"
+    local = datetime(2026, 10, 14, 23, 0, 0).astimezone()
+    assert archive_name(local).endswith("Z.tar")
+
+
+def test_round_trip_returns_the_payload_restore_takes():
+    payload = _payload()
+    buf = io.BytesIO()
+    write_archive_to(payload, buf, app_version="v")
+    buf.seek(0)
+    assert read_archive(buf) == payload
+
+
+def test_notices_stay_in_the_manifest_not_as_sections():
+    """`_not_included` is a list, but of table names; it must not become a
+    member that a restore would read as rows."""
+    names = set(_members(_payload()))
+    assert names == {"manifest.json", "notes.jsonl", "settings.jsonl", "users.jsonl"}
+
+
+def test_members_come_in_one_order_with_fixed_headers():
+    buf = io.BytesIO()
+    write_archive_to(_payload(), buf, app_version="v")
+    buf.seek(0)
+    with tarfile.open(fileobj=buf, mode="r:") as tar:
+        members = tar.getmembers()
+    assert [m.name for m in members] == [
+        "manifest.json", "notes.jsonl", "settings.jsonl", "users.jsonl",
+    ]
+    for m in members:
+        assert (m.mtime, m.uid, m.gid, m.uname, m.gname, m.mode) == (
+            0, 0, 0, "", "", 0o644,
+        )
+
+
+def test_an_unchanged_payload_gives_identical_sections_on_another_run():
+    """The run's own time and the app version live in the manifest and
+    nowhere else, so a second run over unchanged data repeats every section
+    byte for byte."""
+    first = _members(_payload())
+    second = _members(_payload(exported_at="2026-10-16T03:00:00+00:00"))
+    assert first["manifest.json"] != second["manifest.json"]
+    for name in first:
+        if name != "manifest.json":
+            assert first[name] == second[name], name
+
+
+def test_key_order_in_a_row_does_not_change_its_bytes():
+    rows = _payload()["notes"]
+    shuffled = [dict(reversed(list(r.items()))) for r in rows]
+    assert _members(_payload(notes=shuffled))["notes.jsonl"] == \
+        _members(_payload())["notes.jsonl"]
+
+
+def test_editing_one_row_changes_one_line_of_one_member():
+    before = _members(_payload())
+    notes = _payload()["notes"]
+    notes[24] = {**notes[24], "body": "edited"}
+    after = _members(_payload(notes=notes))
+
+    changed = [n for n in before if before[n] != after[n]]
+    assert changed == ["notes.jsonl"]
+    old_lines = before["notes.jsonl"].splitlines()
+    new_lines = after["notes.jsonl"].splitlines()
+    assert len(old_lines) == len(new_lines)
+    assert [i for i, (a, b) in enumerate(zip(old_lines, new_lines)) if a != b] == [24]
+
+
+def test_a_new_row_appends_to_the_end_of_its_member():
+    """Rows come in primary-key order, so a table that grows grows at its
+    tail: everything already sent is a prefix of the new member."""
+    before = _members(_payload())["notes.jsonl"]
+    notes = _payload()["notes"] + [{"id": 51, "user_id": 1, "title": "new"}]
+    after = _members(_payload(notes=notes))["notes.jsonl"]
+    assert after.startswith(before)
+    assert after.count(b"\n") == before.count(b"\n") + 1
+
+
+def test_the_file_is_an_uncompressed_tar():
+    buf = io.BytesIO()
+    write_archive_to(_payload(), buf, app_version="v")
+    raw = buf.getvalue()
+    assert raw[257:262] == b"ustar"
+    assert b'"title":"note 7"' in raw
+
+
+def test_is_archive_tells_a_tar_from_the_json_download():
+    buf = io.BytesIO()
+    write_archive_to(_payload(), buf, app_version="v")
+    buf.seek(0)
+    assert is_archive(buf)
+    assert buf.tell() == 0
+    assert not is_archive(io.BytesIO(b'{"version": 25, "users": []}'))
+
+
+def test_write_archive_lands_under_the_dated_name_only(tmp_path):
+    path = write_archive(_payload(), tmp_path, AT, app_version="v")
+    assert path == tmp_path / "scribe-20261015T030000Z.tar"
+    assert [p.name for p in tmp_path.iterdir()] == [path.name]
+    with open(path, "rb") as f:
+        assert read_archive(f) == _payload()
+
+
+def test_a_failed_write_leaves_nothing_behind(tmp_path):
+    """Neither a half backup under the final name (a watcher would count it
+    as arrived) nor a stray part file."""
+    with patch.object(backup_file, "write_archive_to", side_effect=OSError("disk full")):
+        with pytest.raises(OSError):
+            write_archive(_payload(), tmp_path, AT, app_version="v")
+    assert list(tmp_path.iterdir()) == []
+
+
+def _rewrite(members: dict[str, bytes]) -> io.BytesIO:
+    buf = io.BytesIO()
+    with tarfile.open(fileobj=buf, mode="w", format=tarfile.USTAR_FORMAT) as tar:
+        for name, data in members.items():
+            info = tarfile.TarInfo(name)
+            info.size = len(data)
+            tar.addfile(info, io.BytesIO(data))
+    buf.seek(0)
+    return buf
+
+
+def test_a_missing_section_is_an_error_not_a_thinner_restore():
+    members = _members(_payload())
+    del members["notes.jsonl"]
+    with pytest.raises(ArchiveError, match="notes.jsonl"):
+        read_archive(_rewrite(members))
+
+
+def test_a_section_cut_short_is_an_error():
+    members = _members(_payload())
+    members["notes.jsonl"] = b"\n".join(members["notes.jsonl"].splitlines()[:10]) + b"\n"
+    with pytest.raises(ArchiveError, match="10 rows"):
+        read_archive(_rewrite(members))
+
+
+def test_a_newer_format_is_refused_by_name():
+    members = _members(_payload())
+    members["manifest.json"] = members["manifest.json"].replace(
+        b'"format_version": 1', b'"format_version": 99',
+    )
+    with pytest.raises(ArchiveError, match="upgrade"):
+        read_archive(_rewrite(members))
+
+
+def test_garbage_is_not_an_archive():
+    with pytest.raises(ArchiveError):
+        read_archive(io.BytesIO(b"not a tar at all" * 100))