From 73fe9fc75d004e50b491acbc357a8ee05be4d8a5 Mon Sep 17 00:00:00 2001
From: Alastair McFarlane <arm@apache.org>
Date: Wed, 8 Jul 2026 17:29:47 +0100
Subject: [PATCH] #1370 - support import and export of PMC CSVs

---
 atr/admin/__init__.py                         | 233 ++++++++
 atr/admin/templates/catalog-committee.html    |   1 +
 .../templates/catalog-import-preview.html     |  12 +
 atr/admin/templates/catalog-import.html       |  59 ++
 atr/shared/catalogue_diff.py                  | 491 +++++++++++++++++
 atr/shared/catalogue_import.py                | 312 +++++++++++
 atr/shared/catalogue_rows.py                  | 138 +++++
 atr/shared/repoint.py                         |   6 +-
 atr/storage/writers/catalogue.py              | 299 ++++++++--
 atr/templates/includes/topnav.html            |   5 +
 tests/unit/test_admin_catalog.py              |  63 ++-
 tests/unit/test_catalogue_diff.py             | 458 ++++++++++++++++
 tests/unit/test_catalogue_import_apply.py     | 509 ++++++++++++++++++
 13 files changed, 2512 insertions(+), 74 deletions(-)
 create mode 100644 atr/admin/templates/catalog-import-preview.html
 create mode 100644 atr/admin/templates/catalog-import.html
 create mode 100644 atr/shared/catalogue_diff.py
 create mode 100644 atr/shared/catalogue_import.py
 create mode 100644 atr/shared/catalogue_rows.py
 create mode 100644 tests/unit/test_catalogue_diff.py
 create mode 100644 tests/unit/test_catalogue_import_apply.py

diff --git a/atr/admin/__init__.py b/atr/admin/__init__.py
index 25eb4b74..15f4144a 100644
--- a/atr/admin/__init__.py
+++ b/atr/admin/__init__.py
@@ -17,6 +17,7 @@
 
 import asyncio
 import collections
+import dataclasses
 import datetime
 import hashlib
 import ipaddress
@@ -36,6 +37,7 @@
 import jwt
 import pydantic
 import quart
+import quart.datastructures as datastructures
 import sqlalchemy
 import sqlmodel
 import werkzeug.exceptions as exceptions
@@ -61,6 +63,9 @@
 import atr.paths as paths
 import atr.principal as principal
 import atr.shared as shared
+import atr.shared.catalogue_diff as catalogue_diff
+import atr.shared.catalogue_import as catalogue_import
+import atr.shared.catalogue_rows as catalogue_rows
 import atr.storage as storage
 import atr.storage.outcome as outcome
 import atr.tasks as tasks
@@ -680,6 +685,234 @@ async def catalog_remove_post(
     return await session.redirect(catalog_committee_get, committee_key=str(committee_key))
 
 
+_CATALOG_PREVIEW_LIMIT: Final[int] = 50
+
+
+class CatalogImportForm(form.Form):
+    projects: form.File = form.label("Projects CSV", "Optional.")
+    releases: form.File = form.label("Releases CSV", "Optional.")
+    artifacts: form.File = form.label("Artifacts CSV", "Optional.")
+    overwrite: form.Bool = form.label(
+        "Overwrite everything for this PMC",
+        "Delete anything the files do not list, so the files become the whole catalogue for this PMC."
+        " Left unticked, rows are matched against what is already there and nothing is deleted",
+        default=False,
+    )
+
+
+class CatalogImportApplyForm(form.Form):
+    token: str = form.label("Token", widget=form.Widget.HIDDEN)
+
+
+@dataclasses.dataclass
+class CatalogExportArgs:
+    table: str = ""
+
+
+@admin.typed
+async def catalog_export_get(
+    _session: web.Committer,
+    _catalog: Literal["catalog"],
+    committee_key: safe.CommitteeKey,
+    _export: Literal["export"],
+    query_args: CatalogExportArgs,
+) -> web.QuartResponse:
+    """
+    URL: GET /catalog/<committee_key>/export
+
+    Download the committee's current entries for one table as a CSV.
+    """
+    table = query_args.table
+    async with db.session() as data:
+        try:
+            content = await catalogue_import.export_table(data, str(committee_key), table)
+        except KeyError as error:
+            raise exceptions.NotFound(f"Unknown table '{table}'.") from error
+    headers = {"Content-Disposition": f'attachment; filename="{committee_key}-{table}.csv"'}
+    return quart.Response(content, mimetype="text/csv", headers=headers)
+
+
+@admin.typed
+async def catalog_import_get(
+    _session: web.Committer,
+    _catalog: Literal["catalog"],
+    committee_key: safe.CommitteeKey,
+    _import: Literal["import"],
+) -> str:
+    """
+    URL: GET /catalog/<committee_key>/import
+
+    Upload catalogue CSVs to preview against the committee.
+    """
+    rendered_form = await form.render(
+        model_cls=CatalogImportForm,
+        submit_label="Preview import",
+        submit_classes="btn-primary",
+    )
+    return await template.render("catalog-import.html", committee_key=str(committee_key), form=rendered_form)
+
+
+@admin.typed
+async def catalog_import_post(
+    _session: web.Committer,
+    _catalog: Literal["catalog"],
+    committee_key: safe.CommitteeKey,
+    _import: Literal["import"],
+    import_form: CatalogImportForm,
+) -> str:
+    """
+    URL: POST /catalog/<committee_key>/import
+
+    Store the uploads and show the classified diff, grouped by table.
+    """
+    files = _catalog_import_files(import_form)
+    if not files:
+        raise exceptions.BadRequest("Upload at least one CSV.")
+    mode = catalogue_diff.Mode.REPLACE if import_form.overwrite else catalogue_diff.Mode.ADDITIVE
+    token = await catalogue_import.store_uploads(files, str(committee_key), mode)
+    rows = await asyncio.to_thread(catalogue_import.read_uploads, token)
+    async with db.session() as data:
+        diff = await _catalog_import_diff(data, str(committee_key), rows, mode)
+    # The apply recomputes the diff, and refuses if the catalogue has moved since this preview
+    await asyncio.to_thread(catalogue_import.record_fingerprint, token, catalogue_diff.fingerprint(diff))
+    apply_form = await form.render(
+        model_cls=CatalogImportApplyForm,
+        action=util.as_url(catalog_import_apply_post, committee_key=str(committee_key)),
+        submit_label="Apply import",
+        submit_classes="btn-danger",
+        defaults={"token": token},
+    )
+    return await template.render(
+        "catalog-import-preview.html",
+        committee_key=str(committee_key),
+        preview=_catalog_import_preview(diff),
+        form=apply_form,
+    )
+
+
+@admin.typed
+async def catalog_import_apply_post(
+    session: web.Committer,
+    _catalog: Literal["catalog"],
+    committee_key: safe.CommitteeKey,
+    _import: Literal["import"],
+    _apply: Literal["apply"],
+    apply_form: CatalogImportApplyForm,
+) -> web.WerkzeugResponse:
+    """
+    URL: POST /catalog/<committee_key>/import/apply
+
+    Re-read the stored uploads, recompute the diff under the write lock, and apply it.
+    """
+    token = apply_form.token.strip()
+    try:
+        rows = await asyncio.to_thread(catalogue_import.read_uploads, token)
+    except KeyError as error:
+        raise exceptions.BadRequest("The upload has expired; please upload again.") from error
+    if await asyncio.to_thread(catalogue_import.committee_of, token) != str(committee_key):
+        raise exceptions.BadRequest("That upload was previewed against a different committee.")
+    mode = await asyncio.to_thread(catalogue_import.mode_of, token)
+    previewed = await asyncio.to_thread(catalogue_import.fingerprint_of, token)
+    try:
+        async with storage.write(session) as write:
+            catalogue_writer = write.as_foundation_admin().catalogue
+            diff = await catalogue_writer.import_catalogue_csvs(rows, str(committee_key), mode, previewed)
+    except catalogue_import.StaleError as error:
+        raise exceptions.Conflict(
+            "The catalogue changed since this import was previewed, so nothing was written."
+            " Upload the files again to see what would happen now."
+        ) from error
+    finally:
+        await asyncio.to_thread(catalogue_import.discard, token)
+    counts = diff.counts
+    if diff.refused:
+        raise exceptions.BadRequest(f"Nothing was imported: {counts['conflict']} conflicts. Upload again to see them.")
+    return await session.redirect(
+        catalog_committee_get,
+        committee_key=str(committee_key),
+        success=f"Imported: {_catalog_import_summary(diff)}",
+    )
+
+
+def _catalog_import_files(import_form: CatalogImportForm) -> dict[str, datastructures.FileStorage]:
+    # Each table has its own optional picker, so which CSV is which comes from the field, not
+    # the uploaded file's name
+    files: dict[str, datastructures.FileStorage] = {}
+    for name in ("projects", "releases", "artifacts"):
+        upload = getattr(import_form, name)
+        if (upload is not None) and upload.filename:
+            files[name] = upload
+    return files
+
+
+async def _catalog_import_diff(
+    data: db.Session, committee_key: str, rows: dict[str, list[dict[str, str]]], mode: catalogue_diff.Mode
+) -> catalogue_diff.CatalogueDiff:
+    snapshot = await catalogue_import.build_snapshot(data, committee_key)
+    return catalogue_diff.classify(rows, snapshot, committee_key, mode)
+
+
+def _catalog_capped(items: list[str]) -> list[htm.Element]:
+    shown = [htpy.li[item] for item in items[:_CATALOG_PREVIEW_LIMIT]]
+    if len(items) > _CATALOG_PREVIEW_LIMIT:
+        shown.append(htpy.li[f"... and {len(items) - _CATALOG_PREVIEW_LIMIT} more"])
+    return shown
+
+
+def _catalog_import_summary(diff: catalogue_diff.CatalogueDiff) -> str:
+    counts = diff.counts
+    parts = [f"{counts['add']} adds", f"{counts['delete']} deletes"]
+    if counts["release_repoint"]:
+        parts.append(f"{counts['release_repoint']} release repoints")
+    if counts["artifact_repoint"]:
+        parts.append(f"{counts['artifact_repoint']} artifact repoints")
+    if counts["unchanged"]:
+        parts.append(f"{counts['unchanged']} unchanged")
+    parts.append(f"{counts['conflict']} conflicts")
+    return ", ".join(parts) + "."
+
+
+def _catalog_import_preview(diff: catalogue_diff.CatalogueDiff) -> htm.Element:
+    sections: list[htm.Element] = [htpy.p[_catalog_import_summary(diff)]]
+    for warning in diff.warnings:
+        sections.append(htm.div(".alert.alert-warning")[warning])
+    if diff.project_deletions or diff.release_deletions or diff.artifact_deletions:
+        sections.append(htpy.h2["Deletions"])
+        sections.append(
+            htm.div(".alert.alert-danger")[
+                "Your files replace this committee's catalogue, so these are removed. Check you have"
+                " all the data to import, or that you are happy for anything unspecified to be deleted."
+            ]
+        )
+        deletions = [f"project {key}, and its releases and artifacts" for key in diff.project_deletions]
+        deletions += [f"release {key}, and its artifacts" for key in diff.release_deletions]
+        deletions += [f"artifact {pk[2]} from {pk[0]} {pk[1]}" for pk in diff.artifact_deletions]
+        sections.append(htpy.ul[_catalog_capped(deletions)])
+    if diff.conflicts:
+        sections.append(htpy.h2["Conflicts"])
+        sections.append(htpy.ul[_catalog_capped([f"{c.table} {c.key}: {c.reason}" for c in diff.conflicts])])
+    project_adds = [add for add in diff.adds if isinstance(add, catalogue_rows.ProjectRow)]
+    release_adds = [add for add in diff.adds if isinstance(add, catalogue_rows.ReleaseRow)]
+    artifact_adds = [add for add in diff.adds if isinstance(add, catalogue_rows.ArtifactRow)]
+    if project_adds or release_adds:
+        sections.append(htpy.h2["Projects and releases to add"])
+        sections.append(htpy.ul[_catalog_capped([str(add.key) for add in project_adds + release_adds])])
+    if diff.release_repoints:
+        sections.append(htpy.h2["Releases to repoint"])
+        sections.append(
+            htpy.ul[_catalog_capped([f"{repoint.key} to {repoint.to_project}" for repoint in diff.release_repoints])]
+        )
+    if artifact_adds or diff.artifact_repoints:
+        sections.append(htpy.h2["Artifacts"])
+        rows = [htpy.li[f"add {add.artifact_path} to {add.project_key} {add.version}"] for add in artifact_adds]
+        rows += [
+            htpy.li[f"repoint {repoint.dist[1]} to {repoint.to_row.project_key} {repoint.to_row.version}"]
+            for repoint in diff.artifact_repoints
+        ]
+        sections.append(htpy.ul[rows])
+    return htpy.div[sections]
+
+
 @admin.typed
 async def configuration(_session: web.Committer, _configuration: Literal["configuration"]) -> web.QuartResponse:
     """
diff --git a/atr/admin/templates/catalog-committee.html b/atr/admin/templates/catalog-committee.html
index 03dc51e8..8535bd26 100644
--- a/atr/admin/templates/catalog-committee.html
+++ b/atr/admin/templates/catalog-committee.html
@@ -7,5 +7,6 @@
 {% block content %}
     <h1>Catalogue: {{ committee_key }}</h1>
     <p><a href="/admin/catalog">Back to committees</a></p>
+    <p><a href="/admin/catalog/{{ committee_key }}/import" class="btn btn-primary">Import CSVs</a></p>
     {{ projects }}
 {% endblock content %}
diff --git a/atr/admin/templates/catalog-import-preview.html b/atr/admin/templates/catalog-import-preview.html
new file mode 100644
index 00000000..8dc4ac41
--- /dev/null
+++ b/atr/admin/templates/catalog-import-preview.html
@@ -0,0 +1,12 @@
+{% extends "layouts/base-admin.html" %}
+
+{%- block title -%}Import preview: {{ committee_key }} ~ ATR{%- endblock title -%}
+
+{%- block description -%}Review the catalogue import for {{ committee_key }}.{%- endblock description -%}
+
+{% block content %}
+    <h1>Import preview: {{ committee_key }}</h1>
+    <p><a href="/admin/catalog/{{ committee_key }}">Cancel</a></p>
+    {{ preview }}
+    {{ form }}
+{% endblock content %}
diff --git a/atr/admin/templates/catalog-import.html b/atr/admin/templates/catalog-import.html
new file mode 100644
index 00000000..e63d4ccf
--- /dev/null
+++ b/atr/admin/templates/catalog-import.html
@@ -0,0 +1,59 @@
+{% extends "layouts/base-admin.html" %}
+
+{%- block title -%}Import catalogue: {{ committee_key }} ~ ATR{%- endblock title -%}
+
+{%- block description -%}Upload catalogue CSVs to correct {{ committee_key }}.{%- endblock description -%}
+
+{% block content %}
+    <h1>Import catalogue: {{ committee_key }}</h1>
+    <p><a href="/admin/catalog/{{ committee_key }}">Back to {{ committee_key }}</a></p>
+    <h2>Download current entries</h2>
+    <p>
+        Each file holds this committee's current entries, in the same columns the upload expects, so a
+        file you download can be edited and uploaded straight back.
+    </p>
+    <ul>
+        <li><a href="/admin/catalog/{{ committee_key }}/export?table=projects">projects.csv</a></li>
+        <li><a href="/admin/catalog/{{ committee_key }}/export?table=releases">releases.csv</a></li>
+        <li><a href="/admin/catalog/{{ committee_key }}/export?table=artifacts">artifacts.csv</a></li>
+    </ul>
+    <h2>Upload</h2>
+    <p>
+        Each file is optional, and nothing is written until you confirm the preview on the next page.
+    </p>
+    <p>
+        By default the rows are matched against what is already catalogued: a row that matches nothing
+        is added, one that names a different project moves there, and one that matches where it says it
+        is does nothing. Nothing is deleted, a row that cannot be applied is skipped and the rest still
+        go ahead, and columns other than the keys are not written back. Use this to top a committee up
+        rather than to correct it.
+    </p>
+    <p>
+        Tick <strong>Overwrite everything for this PMC</strong> and the files you upload
+        <strong>are</strong> this committee's catalogue instead. Download, edit, upload: what the files
+        list is what the committee ends up with, so an added row is an entry added, a removed row is an
+        entry deleted, and an edited row is an entry corrected.
+    </p>
+    <p>
+        On an overwrite, the deepest file you upload decides how far the replacement reaches.
+        artifacts.csv replaces the artifacts; releases.csv replaces the releases and their artifacts;
+        projects.csv replaces the projects, their releases and their artifacts. Upload the levels below
+        the one you are correcting, or the preview will warn you what is about to be left empty.
+    </p>
+    <p>
+        An overwrite applies as a whole or not at all. If any row cannot be applied it is reported as a
+        conflict, nothing is changed, and you can fix the file and upload again. This is deliberate: a
+        row that fell out would be deleted rather than corrected.
+    </p>
+    <p>
+        Because an overwrite rewrites the rows, columns the CSVs do not carry are reset. Run
+        <strong>Update projects</strong> afterwards to restore project metadata, and re-import
+        signatures and lifecycle events if you have them.
+    </p>
+    <p>
+        Projects carrying live release-workflow data are never changed or deleted by this page, in
+        either mode. To move a project to another committee, or to rename or merge one, use the actions
+        on the committee page instead.
+    </p>
+    {{ form }}
+{% endblock content %}
diff --git a/atr/shared/catalogue_diff.py b/atr/shared/catalogue_diff.py
new file mode 100644
index 00000000..426c9606
--- /dev/null
+++ b/atr/shared/catalogue_diff.py
@@ -0,0 +1,491 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import dataclasses
+import enum
+import hashlib
+import json
+
+import pydantic
+
+import atr.shared.catalogue_rows as catalogue_rows
+
+
+class Mode(enum.StrEnum):
+    # REPLACE clears the committee and re-adds from the files, so every row is an add. ADDITIVE
+    # matches each row against what is already there, so a row can also repoint a release or an
+    # artifact, or say nothing new. The two share their types and their parsing, but not a code path
+    REPLACE = "replace"
+    ADDITIVE = "additive"
+
+
+@dataclasses.dataclass(frozen=True)
+class DbSnapshot:
+    project_committee: dict[str, str]
+    release_project: dict[str, str]
+    artifact_by_dist: dict[tuple[str, str], tuple[str, str, str]]
+    managed_project_keys: frozenset[str]
+    # Every release key foundation-wide, unlike the workspace-scoped release_project
+    release_keys: frozenset[str] = dataclasses.field(default_factory=frozenset)
+    # The workspace committee's artifact primary keys, (project_key, version, artifact_path)
+    artifact_pks: frozenset[tuple[str, str, str]] = dataclasses.field(default_factory=frozenset)
+
+
+@dataclasses.dataclass(frozen=True)
+class Conflict:
+    table: str
+    key: str
+    reason: str
+
+
+@dataclasses.dataclass(frozen=True)
+class ArtifactRepoint:
+    dist: tuple[str, str]
+    from_pk: tuple[str, str, str]
+    to_row: catalogue_rows.ArtifactRow
+
+
+@dataclasses.dataclass(frozen=True)
+class ReleaseRepoint:
+    key: str
+    from_project: str
+    to_project: str
+    row: catalogue_rows.ReleaseRow
+
+
+@dataclasses.dataclass
+class CatalogueDiff:
+    adds: list[catalogue_rows.Row] = dataclasses.field(default_factory=list)
+    artifact_repoints: list[ArtifactRepoint] = dataclasses.field(default_factory=list)
+    release_repoints: list[ReleaseRepoint] = dataclasses.field(default_factory=list)
+    conflicts: list[Conflict] = dataclasses.field(default_factory=list)
+    project_deletions: list[str] = dataclasses.field(default_factory=list)
+    release_deletions: list[str] = dataclasses.field(default_factory=list)
+    artifact_deletions: list[tuple[str, str, str]] = dataclasses.field(default_factory=list)
+    warnings: list[str] = dataclasses.field(default_factory=list)
+    unchanged: int = 0
+    # A replace applies as a whole or not at all, so a conflict leaves the committee untouched
+    refused: bool = False
+
+    @property
+    def counts(self) -> dict[str, int]:
+        return {
+            "add": len(self.adds),
+            "artifact_repoint": len(self.artifact_repoints),
+            "release_repoint": len(self.release_repoints),
+            "delete": len(self.project_deletions) + len(self.release_deletions) + len(self.artifact_deletions),
+            "conflict": len(self.conflicts),
+            "unchanged": self.unchanged,
+        }
+
+
+def classify(
+    rows_by_table: dict[str, list[dict[str, str]]], snapshot: DbSnapshot, committee_key: str, mode: Mode
+) -> CatalogueDiff:
+    match mode:
+        case Mode.REPLACE:
+            return classify_replace(rows_by_table, snapshot, committee_key)
+        case Mode.ADDITIVE:
+            return classify_additive(rows_by_table, snapshot, committee_key)
+
+
+def classify_additive(
+    rows_by_table: dict[str, list[dict[str, str]]], snapshot: DbSnapshot, committee_key: str
+) -> CatalogueDiff:
+    # Rows are matched against what the committee already holds. A row that matches nothing is an
+    # add, one that matches somewhere else repoints it, and one that matches where it says it is is
+    # unchanged. Nothing is deleted, so a bad row is skipped and the rest still apply
+    diff = CatalogueDiff()
+    projects, releases, artifacts = _parse_all(rows_by_table, diff)
+    _classify_projects(projects, snapshot, diff, committee_key)
+    _classify_releases(releases, snapshot, diff, committee_key)
+    _classify_artifacts(artifacts, snapshot, diff, committee_key)
+    return diff
+
+
+def classify_replace(
+    rows_by_table: dict[str, list[dict[str, str]]], snapshot: DbSnapshot, committee_key: str
+) -> CatalogueDiff:
+    # The files *are* the committee's catalogue. The deepest table uploaded decides what is deleted,
+    # which cascades downwards. Against a cleared committee nothing matches, so the same
+    # classification leaves every uploaded row an add, and a surviving row is a managed one
+    diff = CatalogueDiff()
+    snapshot = _clear(rows_by_table, snapshot, diff, committee_key)
+    projects, releases, artifacts = _parse_all(rows_by_table, diff)
+    _classify_projects(projects, snapshot, diff, committee_key)
+    _classify_releases(releases, snapshot, diff, committee_key)
+    _classify_artifacts(artifacts, snapshot, diff, committee_key)
+    if diff.conflicts:
+        # A row that cannot be applied would be deleted and not restored, so the files only describe
+        # a complete catalogue if every row in them is good
+        return _refused(diff)
+    return diff
+
+
+def fingerprint(diff: CatalogueDiff) -> str:
+    # The outcome the preview showed. The files decide the rows, but the catalogue underneath them
+    # decides what those rows do, so two classifications of one upload only agree while it holds
+    # still. An apply compares this against the preview rather than trusting the files alone
+    outcome = {
+        "adds": sorted(_row_identity(add) for add in diff.adds),
+        "artifact_deletions": sorted(diff.artifact_deletions),
+        "artifact_repoints": sorted(
+            [list(repoint.dist), list(repoint.from_pk), _row_identity(repoint.to_row)]
+            for repoint in diff.artifact_repoints
+        ),
+        "conflicts": sorted([conflict.table, conflict.key, conflict.reason] for conflict in diff.conflicts),
+        "project_deletions": sorted(diff.project_deletions),
+        "refused": diff.refused,
+        "release_deletions": sorted(diff.release_deletions),
+        "release_repoints": sorted(
+            [repoint.key, repoint.from_project, repoint.to_project] for repoint in diff.release_repoints
+        ),
+        "unchanged": diff.unchanged,
+    }
+    return hashlib.sha256(json.dumps(outcome, sort_keys=True).encode()).hexdigest()
+
+
+def _classify_artifacts(
+    rows: list[catalogue_rows.ArtifactRow], snapshot: DbSnapshot, diff: CatalogueDiff, committee_key: str
+) -> None:
+    # An artifact stays within its own committee. The releases are classified first, so a repointed
+    # release has already taken its artifacts' primary keys to the target project, and a row can
+    # name the artifact at either end of that: where the database still has it, or where it lands
+    repointed = {
+        (repoint.from_project, str(repoint.row.version)): repoint.to_project for repoint in diff.release_repoints
+    }
+    snapshot = _after_release_repoints(snapshot, diff)
+    known_projects = _known_projects(snapshot, diff, committee_key)
+    known_releases = set(snapshot.release_project) | {
+        str(add.key) for add in diff.adds if isinstance(add, catalogue_rows.ReleaseRow)
+    }
+    # Keys the database holds now, plus the keys this diff adds to them. A repoint away from a key
+    # never frees it, because the writer inserts the adds before it repoints anything
+    claimed = set(snapshot.artifact_pks) | {_repointed_pk(pk, repointed) for pk in snapshot.artifact_pks}
+    seen_dists: set[tuple[str, str]] = set()
+    seen_pks: set[tuple[str, str, str]] = set()
+    for row in rows:
+        path = str(row.artifact_path)
+        pk = (str(row.project_key), str(row.version), path)
+        reference_conflict = _artifact_reference_conflict(row, snapshot, known_projects, known_releases)
+        if reference_conflict is not None:
+            diff.conflicts.append(reference_conflict)
+            continue
+        suffix = str(row.download_path_suffix) if row.download_path_suffix else ""
+        if ((suffix, path) in seen_dists) or (pk in seen_pks):
+            diff.conflicts.append(Conflict(table="artifacts", key=path, reason="listed more than once in this file"))
+            continue
+        seen_dists.add((suffix, path))
+        seen_pks.add(pk)
+        # The database still holds the artifact where it was, which is where the writer has to find it
+        matched = snapshot.artifact_by_dist.get((suffix, path)) if suffix else None
+        if (matched is not None) and (pk in (matched, _repointed_pk(matched, repointed))):
+            # The row names the artifact where it is, or where its release is about to take it
+            diff.unchanged += 1
+            continue
+        if matched is not None:
+            if matched[0] in snapshot.managed_project_keys:
+                diff.conflicts.append(
+                    Conflict(table="artifacts", key=path, reason="source project has live workflow data")
+                )
+                continue
+            if pk in claimed:
+                diff.conflicts.append(Conflict(table="artifacts", key=path, reason="target already exists"))
+                continue
+            claimed.add(pk)
+            diff.artifact_repoints.append(ArtifactRepoint(dist=(suffix, path), from_pk=matched, to_row=row))
+            continue
+        # Without a dist path there is nothing to match on, so the row can only describe a new
+        # artifact. The dist path is the artifact's identity, so an edited one does too
+        if pk in claimed:
+            diff.conflicts.append(
+                Conflict(table="artifacts", key=path, reason="artifact already exists at that project and version")
+            )
+            continue
+        claimed.add(pk)
+        diff.adds.append(row)
+
+
+def _classify_projects(
+    rows: list[catalogue_rows.ProjectRow], snapshot: DbSnapshot, diff: CatalogueDiff, committee_key: str
+) -> None:
+    seen: set[str] = set()
+    for row in rows:
+        key = str(row.key)
+        if key in seen:
+            diff.conflicts.append(Conflict(table="projects", key=key, reason="listed more than once in this file"))
+            continue
+        seen.add(key)
+        conflict = _project_reference_conflict(row, snapshot, committee_key)
+        if conflict is not None:
+            diff.conflicts.append(conflict)
+            continue
+        if key in snapshot.project_committee:
+            diff.unchanged += 1
+            continue
+        diff.adds.append(row)
+
+
+def _classify_releases(
+    rows: list[catalogue_rows.ReleaseRow], snapshot: DbSnapshot, diff: CatalogueDiff, committee_key: str
+) -> None:
+    known_projects = _known_projects(snapshot, diff, committee_key)
+    seen: set[str] = set()
+    # Keys this diff will end up holding, so two rows cannot converge on one release
+    claimed: set[str] = set()
+    for row in rows:
+        key = str(row.key)
+        if key in seen:
+            diff.conflicts.append(Conflict(table="releases", key=key, reason="listed more than once in this file"))
+            continue
+        seen.add(key)
+        target_project = str(row.project_key)
+        current_project = snapshot.release_project.get(key)
+        reference_conflict = _release_reference_conflict(row, snapshot, known_projects, current_project)
+        if reference_conflict is not None:
+            diff.conflicts.append(reference_conflict)
+            continue
+        if current_project is None:
+            if (key in snapshot.release_keys) or (key in claimed):
+                diff.conflicts.append(Conflict(table="releases", key=key, reason=f"release '{key}' already exists"))
+                continue
+            claimed.add(key)
+            diff.adds.append(row)
+            continue
+        if current_project == target_project:
+            diff.unchanged += 1
+            continue
+        version = str(row.version)
+        target_release_key = f"{target_project}-{version}"
+        if (target_release_key in snapshot.release_keys) or (target_release_key in claimed):
+            diff.conflicts.append(
+                Conflict(table="releases", key=key, reason=f"target already has release '{target_release_key}'")
+            )
+            continue
+        collision = _repoint_artifact_collision(snapshot, current_project, target_project, version)
+        if collision is not None:
+            diff.conflicts.append(Conflict(table="releases", key=key, reason=collision))
+            continue
+        claimed.add(target_release_key)
+        diff.release_repoints.append(
+            ReleaseRepoint(key=key, from_project=current_project, to_project=target_project, row=row)
+        )
+
+
+def _after_release_repoints(snapshot: DbSnapshot, diff: CatalogueDiff) -> DbSnapshot:
+    if not diff.release_repoints:
+        return snapshot
+    # A release key is "{owning project}-{version}", so a repoint rekeys the release. Both keys have
+    # to resolve: an artifact row edited to follow the release names the new one, and an untouched
+    # row from the same download still names the old one
+    new_keys = {
+        f"{release_repoint.to_project}-{release_repoint.row.version}": release_repoint.to_project
+        for release_repoint in diff.release_repoints
+    }
+    return dataclasses.replace(
+        snapshot,
+        release_project=snapshot.release_project | new_keys,
+        release_keys=snapshot.release_keys | frozenset(new_keys),
+    )
+
+
+def _artifact_reference_conflict(
+    row: catalogue_rows.ArtifactRow, snapshot: DbSnapshot, known_projects: set[str], known_releases: set[str]
+) -> Conflict | None:
+    path = str(row.artifact_path)
+    project_key = str(row.project_key)
+    if project_key in snapshot.managed_project_keys:
+        return Conflict(table="artifacts", key=path, reason="target project has live workflow data")
+    if project_key not in known_projects:
+        return Conflict(table="artifacts", key=path, reason=f"project '{project_key}' is not in this committee")
+    if row.release_key not in known_releases:
+        return Conflict(table="artifacts", key=path, reason=f"release '{row.release_key}' does not exist")
+    return None
+
+
+def _clear(
+    rows_by_table: dict[str, list[dict[str, str]]], snapshot: DbSnapshot, diff: CatalogueDiff, committee_key: str
+) -> DbSnapshot:
+    # Projects with live workflow data stay in the snapshot, and are refused as usual
+    catalogued = {
+        key
+        for key, committee in snapshot.project_committee.items()
+        if (committee == committee_key) and (key not in snapshot.managed_project_keys)
+    }
+    clear_projects = "projects" in rows_by_table
+    clear_releases = clear_projects or ("releases" in rows_by_table)
+    clear_artifacts = clear_releases or ("artifacts" in rows_by_table)
+
+    releases = {key for key, project in snapshot.release_project.items() if project in catalogued}
+    artifacts = {dist for dist, pk in snapshot.artifact_by_dist.items() if pk[0] in catalogued}
+    # Keyed on the primary keys, so an artifact with no dist path is cleared too
+    artifact_pks = {pk for pk in snapshot.artifact_pks if pk[0] in catalogued}
+
+    if clear_projects:
+        diff.project_deletions.extend(sorted(catalogued))
+    if clear_releases:
+        diff.release_deletions.extend(sorted(releases))
+    if clear_artifacts:
+        diff.artifact_deletions.extend(sorted(artifact_pks))
+    _warn_about_empty_levels(rows_by_table, diff, len(releases), len(artifact_pks))
+
+    return dataclasses.replace(
+        snapshot,
+        project_committee={
+            key: committee
+            for key, committee in snapshot.project_committee.items()
+            if not (clear_projects and (key in catalogued))
+        },
+        release_project={
+            key: project
+            for key, project in snapshot.release_project.items()
+            if not (clear_releases and (key in releases))
+        },
+        release_keys=snapshot.release_keys - (releases if clear_releases else set()),
+        artifact_by_dist={
+            dist: pk for dist, pk in snapshot.artifact_by_dist.items() if not (clear_artifacts and (dist in artifacts))
+        },
+        artifact_pks=snapshot.artifact_pks - (artifact_pks if clear_artifacts else set()),
+    )
+
+
+def _first_error(error: pydantic.ValidationError) -> str:
+    detail = error.errors()[0]
+    field = ".".join(str(part) for part in detail["loc"])
+    message = detail["msg"].removeprefix("Value error, ")
+    return f"invalid {field}: {message}"
+
+
+def _known_projects(snapshot: DbSnapshot, diff: CatalogueDiff, committee_key: str) -> set[str]:
+    # A project added earlier in this diff counts as present
+    return {key for key, committee in snapshot.project_committee.items() if committee == committee_key} | {
+        str(add.key) for add in diff.adds if isinstance(add, catalogue_rows.ProjectRow)
+    }
+
+
+def _repoint_artifact_collision(snapshot: DbSnapshot, from_project: str, to_project: str, version: str) -> str | None:
+    # A repointed release rewrites its artifacts' project key, which is part of their primary key
+    for _, _, path in (pk for pk in snapshot.artifact_pks if (pk[0] == from_project) and (pk[1] == version)):
+        if (to_project, version, path) in snapshot.artifact_pks:
+            return f"target already has artifact '{path}' at version {version}"
+    return None
+
+
+def _row_identity(row: catalogue_rows.Row) -> list[str]:
+    match row:
+        case catalogue_rows.ArtifactRow():
+            return ["artifacts", str(row.project_key), str(row.version), str(row.artifact_path)]
+        case catalogue_rows.ProjectRow():
+            return ["projects", str(row.key)]
+        case catalogue_rows.ReleaseRow():
+            return ["releases", str(row.key)]
+
+
+def _repointed_pk(pk: tuple[str, str, str], repointed: dict[tuple[str, str], str]) -> tuple[str, str, str]:
+    # Where an artifact's primary key ends up once the release repoints have applied
+    project = repointed.get((pk[0], pk[1]))
+    if project is None:
+        return pk
+    return (project, pk[1], pk[2])
+
+
+def _parse[R: pydantic.BaseModel](
+    diff: CatalogueDiff, table: str, identity: str, model_cls: type[R], rows: list[dict[str, str]]
+) -> list[R]:
+    parsed: list[R] = []
+    for raw in rows:
+        try:
+            parsed.append(model_cls.model_validate(raw))
+        except pydantic.ValidationError as error:
+            diff.conflicts.append(Conflict(table=table, key=raw.get(identity, "?"), reason=_first_error(error)))
+    return parsed
+
+
+def _parse_all(
+    rows_by_table: dict[str, list[dict[str, str]]], diff: CatalogueDiff
+) -> tuple[list[catalogue_rows.ProjectRow], list[catalogue_rows.ReleaseRow], list[catalogue_rows.ArtifactRow]]:
+    return (
+        _parse(diff, "projects", "key", catalogue_rows.ProjectRow, rows_by_table.get("projects", [])),
+        _parse(diff, "releases", "key", catalogue_rows.ReleaseRow, rows_by_table.get("releases", [])),
+        _parse(diff, "artifacts", "artifact_path", catalogue_rows.ArtifactRow, rows_by_table.get("artifacts", [])),
+    )
+
+
+def _project_reference_conflict(
+    row: catalogue_rows.ProjectRow, snapshot: DbSnapshot, committee_key: str
+) -> Conflict | None:
+    key = str(row.key)
+    if key in snapshot.managed_project_keys:
+        return Conflict(table="projects", key=key, reason="project has live workflow data")
+    committee = str(row.committee_key) if row.committee_key else ""
+    if committee != committee_key:
+        return Conflict(table="projects", key=key, reason=f"committee '{committee}' is not this committee")
+    # A project key is unique foundation-wide, so one already held elsewhere cannot be taken
+    holder = snapshot.project_committee.get(key)
+    if (holder is not None) and (holder != committee_key):
+        return Conflict(table="projects", key=key, reason=f"project '{key}' belongs to committee '{holder}'")
+    return None
+
+
+def _refused(diff: CatalogueDiff) -> CatalogueDiff:
+    return CatalogueDiff(
+        conflicts=diff.conflicts,
+        refused=True,
+        warnings=[
+            "Nothing was changed. Your files replace this committee's catalogue, so a row that cannot"
+            " be applied would be deleted and not restored. Fix the conflicts below and upload again."
+        ],
+    )
+
+
+def _release_reference_conflict(
+    row: catalogue_rows.ReleaseRow,
+    snapshot: DbSnapshot,
+    known_projects: set[str],
+    current_project: str | None,
+) -> Conflict | None:
+    key = str(row.key)
+    target_project = str(row.project_key)
+    # A release key is "{owning project}-{version}", so the three columns have one degree of freedom
+    # between them; an edit that breaks the derivation is refused rather than applied
+    owner = current_project if (current_project is not None) else target_project
+    if key != f"{owner}-{row.version}":
+        return Conflict(table="releases", key=key, reason=f"key does not match version '{row.version}'")
+    if target_project in snapshot.managed_project_keys:
+        return Conflict(table="releases", key=key, reason="target project has live workflow data")
+    if target_project not in known_projects:
+        if target_project in snapshot.project_committee:
+            return Conflict(table="releases", key=key, reason=f"project '{target_project}' is not in this committee")
+        return Conflict(table="releases", key=key, reason=f"target project '{target_project}' does not exist")
+    if (current_project is not None) and (current_project in snapshot.managed_project_keys):
+        return Conflict(table="releases", key=key, reason="source project has live workflow data")
+    return None
+
+
+def _warn_about_empty_levels(
+    rows_by_table: dict[str, list[dict[str, str]]], diff: CatalogueDiff, releases: int, artifacts: int
+) -> None:
+    if ("projects" in rows_by_table) and ("releases" not in rows_by_table):
+        diff.warnings.append(
+            f"releases.csv was not uploaded, so {releases} releases and {artifacts} artifacts will be"
+            " deleted and not restored, leaving the projects empty. Import releases.csv next to restore them."
+        )
+    if ("releases" in rows_by_table) and ("artifacts" not in rows_by_table):
+        diff.warnings.append(
+            f"artifacts.csv was not uploaded, so {artifacts} artifacts will be deleted and not restored,"
+            " leaving the releases with no artifacts. Import artifacts.csv next to restore them."
+        )
diff --git a/atr/shared/catalogue_import.py b/atr/shared/catalogue_import.py
new file mode 100644
index 00000000..b172f9c6
--- /dev/null
+++ b/atr/shared/catalogue_import.py
@@ -0,0 +1,312 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import asyncio
+import csv
+import datetime
+import io
+import pathlib
+import re
+import shutil
+import time
+import uuid
+from collections.abc import Iterable
+from typing import Final
+
+import aiofiles.os
+import quart.datastructures as datastructures
+import sqlmodel
+
+import atr.config as config
+import atr.db as db
+import atr.models.sql as sql
+import atr.shared.catalogue_diff as catalogue_diff
+
+# CSV columns per table, matching the fields the catalogue_rows models parse, so a downloaded
+# file re-uploads cleanly
+_EXPORT_COLUMNS: Final[dict[str, tuple[str, ...]]] = {
+    "projects": ("key", "name", "status", "committee_key", "version_method"),
+    "releases": ("key", "project_key", "version", "phase", "created", "released", "archived", "release_status"),
+    "artifacts": (
+        "project_key",
+        "version",
+        "artifact_path",
+        "signature_path",
+        "checksum_path",
+        "sbom_path",
+        "classification",
+        "download_path_suffix",
+        "managed",
+        "dated",
+    ),
+}
+
+# The uploaded CSVs live here, one directory per token
+IMPORT_ROOT: Final[pathlib.Path] = pathlib.Path(config.get().STATE_DIR) / "temporary" / "catalogue-imports"
+_TABLES: Final[tuple[str, ...]] = ("projects", "releases", "artifacts")
+# A token is a uuid4 hex string (no path separators or dots)
+_TOKEN_RE: Final = re.compile(r"^[0-9a-f]{32}$")
+_UPLOAD_TTL_SECONDS: Final = 24 * 60 * 60
+# Stored beside the uploads rather than carried through the form, so the intent recorded at
+# upload cannot differ from the one the preview showed
+_MODE_MARKER: Final = "mode"
+_COMMITTEE_MARKER: Final = "committee"
+_FINGERPRINT_MARKER: Final = "fingerprint"
+
+
+class StaleError(Exception):
+    # The catalogue changed between the preview and the apply, so the diff the admin approved is not
+    # the diff that would be written
+    pass
+
+
+async def build_snapshot(data: db.Session, committee_key: str) -> catalogue_diff.DbSnapshot:
+    via = sql.validate_instrumented_attribute
+    # Projects and the managed set are foundation-wide; releases and artifacts are scoped to
+    # the workspace committee
+    all_projects = (await data.execute(sqlmodel.select(via(sql.Project.key), via(sql.Project.committee_key)))).all()
+    project_committee = {row[0]: row[1] for row in all_projects if row[1] is not None}
+    workspace_keys = [key for key, committee in project_committee.items() if committee == committee_key]
+    release_project = await _release_project(data, workspace_keys)
+    artifact_by_dist, artifact_pks = await _artifacts(data, workspace_keys)
+    release_keys = (await data.execute(sqlmodel.select(via(sql.Release.key)))).scalars().all()
+    return catalogue_diff.DbSnapshot(
+        project_committee=project_committee,
+        release_project=release_project,
+        artifact_by_dist=artifact_by_dist,
+        managed_project_keys=await _managed_project_keys(data),
+        release_keys=frozenset(release_keys),
+        artifact_pks=artifact_pks,
+    )
+
+
+def discard(token: str) -> None:
+    target = _token_dir(token)
+    if target.is_dir():
+        shutil.rmtree(target)
+
+
+async def export_table(data: db.Session, committee_key: str, table: str) -> str:
+    # The import refuses to touch a project with live workflow data, so the export leaves it out.
+    # Otherwise a downloaded file could not be uploaded again without editing those rows back out
+    managed = await _managed_project_keys(data)
+    projects = [project for project in await _committee_projects(data, committee_key) if project.key not in managed]
+    if table == "projects":
+        return _to_csv(table, (_project_export(project) for project in projects))
+    project_keys = [project.key for project in projects]
+    if table == "releases":
+        releases = await _committee_releases(data, project_keys)
+        return _to_csv(table, (_release_export(release) for release in releases))
+    if table == "artifacts":
+        artifacts = await _committee_artifacts(data, project_keys)
+        return _to_csv(table, (_artifact_export(artifact) for artifact in artifacts))
+    raise KeyError(table)
+
+
+def mode_of(token: str) -> catalogue_diff.Mode:
+    marker = _token_dir(token) / _MODE_MARKER
+    if not marker.is_file():
+        raise KeyError(token)
+    return catalogue_diff.Mode(marker.read_text())
+
+
+def committee_of(token: str) -> str:
+    # The upload belongs to the committee it was previewed against, so it cannot be applied to another
+    marker = _token_dir(token) / _COMMITTEE_MARKER
+    if not marker.is_file():
+        raise KeyError(token)
+    return marker.read_text()
+
+
+def fingerprint_of(token: str) -> str:
+    marker = _token_dir(token) / _FINGERPRINT_MARKER
+    if not marker.is_file():
+        raise KeyError(token)
+    return marker.read_text()
+
+
+def record_fingerprint(token: str, value: str) -> None:
+    target = _token_dir(token)
+    if not target.is_dir():
+        raise KeyError(token)
+    (target / _FINGERPRINT_MARKER).write_text(value)
+
+
+def read_uploads(token: str) -> dict[str, list[dict[str, str]]]:
+    target = _token_dir(token)
+    if not target.is_dir():
+        raise KeyError(token)
+    rows: dict[str, list[dict[str, str]]] = {}
+    for name in _TABLES:
+        path = target / f"{name}.csv"
+        if path.is_file():
+            with path.open(newline="") as stream:
+                rows[name] = list(csv.DictReader(stream))
+    return rows
+
+
+async def store_uploads(
+    files: dict[str, datastructures.FileStorage], committee_key: str, mode: catalogue_diff.Mode
+) -> str:
+    await asyncio.to_thread(_sweep_stale)
+    token = uuid.uuid4().hex
+    target = _token_dir(token)
+    await aiofiles.os.makedirs(target, exist_ok=True)
+    for name, upload in files.items():
+        if name in _TABLES:
+            await upload.save(target / f"{name}.csv")
+    await asyncio.to_thread((target / _COMMITTEE_MARKER).write_text, committee_key)
+    await asyncio.to_thread((target / _MODE_MARKER).write_text, str(mode))
+    return token
+
+
+def _artifact_export(artifact: sql.Artifact) -> dict[str, str]:
+    return {
+        "project_key": artifact.project_key,
+        "version": artifact.version,
+        "artifact_path": artifact.artifact_path,
+        "signature_path": artifact.signature_path or "",
+        "checksum_path": artifact.checksum_path or "",
+        "sbom_path": artifact.sbom_path or "",
+        "classification": artifact.classification or "",
+        "download_path_suffix": artifact.download_path_suffix or "",
+        "managed": "true" if artifact.managed else "false",
+        "dated": _date_str(artifact.dated),
+    }
+
+
+async def _artifacts(
+    data: db.Session, workspace_keys: list[str]
+) -> tuple[dict[tuple[str, str], tuple[str, str, str]], frozenset[tuple[str, str, str]]]:
+    # An artifact is identified by its dist path, but only artifacts that have one can be matched
+    # that way; the primary keys cover every artifact, including any with no dist path
+    if not workspace_keys:
+        return {}, frozenset()
+    via = sql.validate_instrumented_attribute
+    rows = (
+        await data.execute(
+            sqlmodel.select(
+                via(sql.Artifact.project_key),
+                via(sql.Artifact.version),
+                via(sql.Artifact.artifact_path),
+                via(sql.Artifact.download_path_suffix),
+            ).where(via(sql.Artifact.project_key).in_(workspace_keys))
+        )
+    ).all()
+    by_dist = {(row[3], row[2]): (row[0], row[1], row[2]) for row in rows if row[3]}
+    pks = frozenset((row[0], row[1], row[2]) for row in rows)
+    return by_dist, pks
+
+
+async def _committee_artifacts(data: db.Session, project_keys: list[str]) -> list[sql.Artifact]:
+    if not project_keys:
+        return []
+    via = sql.validate_instrumented_attribute
+    result = await data.execute(sqlmodel.select(sql.Artifact).where(via(sql.Artifact.project_key).in_(project_keys)))
+    return list(result.scalars().all())
+
+
+async def _committee_projects(data: db.Session, committee_key: str) -> list[sql.Project]:
+    via = sql.validate_instrumented_attribute
+    result = await data.execute(sqlmodel.select(sql.Project).where(via(sql.Project.committee_key) == committee_key))
+    return list(result.scalars().all())
+
+
+async def _committee_releases(data: db.Session, project_keys: list[str]) -> list[sql.Release]:
+    if not project_keys:
+        return []
+    via = sql.validate_instrumented_attribute
+    result = await data.execute(sqlmodel.select(sql.Release).where(via(sql.Release.project_key).in_(project_keys)))
+    return list(result.scalars().all())
+
+
+async def _managed_project_keys(data: db.Session) -> frozenset[str]:
+    # A project is managed once one of its releases has a revision, which is where ATR keeps the
+    # files it holds. Foundation-wide, since a row can name a project in any committee
+    via = sql.validate_instrumented_attribute
+    rows = (
+        await data.execute(
+            sqlmodel.select(via(sql.Release.project_key))
+            .join(sql.Revision, via(sql.Revision.release_key) == via(sql.Release.key))
+            .distinct()
+        )
+    ).all()
+    return frozenset(row[0] for row in rows)
+
+
+def _date_str(value: datetime.datetime | None) -> str:
+    return value.strftime("%Y-%m-%d") if value else ""
+
+
+def _project_export(project: sql.Project) -> dict[str, str]:
+    return {
+        "key": project.key,
+        "name": project.name or "",
+        "status": str(project.status),
+        "committee_key": project.committee_key or "",
+        "version_method": str(project.version_method),
+    }
+
+
+def _release_export(release: sql.Release) -> dict[str, str]:
+    return {
+        "key": release.key,
+        "project_key": release.project_key,
+        "version": release.version,
+        "phase": str(release.phase),
+        "created": _date_str(release.created),
+        "released": _date_str(release.released),
+        "archived": _date_str(release.archived),
+        "release_status": "retired" if release.is_archived else "active",
+    }
+
+
+async def _release_project(data: db.Session, workspace_keys: list[str]) -> dict[str, str]:
+    if not workspace_keys:
+        return {}
+    via = sql.validate_instrumented_attribute
+    rows = (
+        await data.execute(
+            sqlmodel.select(via(sql.Release.key), via(sql.Release.project_key)).where(
+                via(sql.Release.project_key).in_(workspace_keys)
+            )
+        )
+    ).all()
+    return {row[0]: row[1] for row in rows}
+
+
+def _sweep_stale() -> None:
+    if not IMPORT_ROOT.is_dir():
+        return
+    cutoff = time.time() - _UPLOAD_TTL_SECONDS
+    for entry in IMPORT_ROOT.iterdir():
+        if entry.is_dir() and (entry.stat().st_mtime < cutoff):
+            shutil.rmtree(entry, ignore_errors=True)
+
+
+def _to_csv(table: str, rows: Iterable[dict[str, str]]) -> str:
+    buffer = io.StringIO()
+    writer = csv.DictWriter(buffer, fieldnames=list(_EXPORT_COLUMNS[table]))
+    writer.writeheader()
+    writer.writerows(rows)
+    return buffer.getvalue()
+
+
+def _token_dir(token: str) -> pathlib.Path:
+    if not _TOKEN_RE.fullmatch(token):
+        raise KeyError(token)
+    return IMPORT_ROOT / token
diff --git a/atr/shared/catalogue_rows.py b/atr/shared/catalogue_rows.py
new file mode 100644
index 00000000..2f81c3ea
--- /dev/null
+++ b/atr/shared/catalogue_rows.py
@@ -0,0 +1,138 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import datetime
+from typing import Annotated
+
+import pydantic
+
+import atr.models.safe as safe
+import atr.models.schema as schema
+import atr.models.sql as sql
+
+
+def _blank_to_none(value: object) -> object:
+    if isinstance(value, str) and (not value.strip()):
+        return None
+    return value
+
+
+def _optional_date(value: object) -> object:
+    if isinstance(value, str):
+        if not value.strip():
+            return None
+        return datetime.datetime.strptime(value.strip(), "%Y-%m-%d").replace(tzinfo=datetime.UTC)
+    return value
+
+
+def _parse_bool(value: object) -> object:
+    if isinstance(value, str):
+        return value.strip().lower() in ("true", "1")
+    return value
+
+
+def _required_date(value: object) -> object:
+    if isinstance(value, str):
+        return datetime.datetime.strptime(value, "%Y-%m-%d").replace(tzinfo=datetime.UTC)
+    return value
+
+
+_Name = Annotated[str | None, pydantic.BeforeValidator(_blank_to_none)]
+_OptionalCommitteeKey = Annotated[safe.CommitteeKey | None, pydantic.BeforeValidator(_blank_to_none)]
+_OptionalRelPath = Annotated[safe.RelPath | None, pydantic.BeforeValidator(_blank_to_none)]
+_Managed = Annotated[bool, pydantic.BeforeValidator(_parse_bool)]
+_RequiredDate = Annotated[datetime.datetime, pydantic.BeforeValidator(_required_date)]
+_OptionalDate = Annotated[datetime.datetime | None, pydantic.BeforeValidator(_optional_date)]
+
+
+class ProjectRow(schema.Subset):
+    key: safe.ProjectKey
+    name: _Name = None
+    status: sql.ProjectStatus
+    committee_key: _OptionalCommitteeKey = None
+    version_method: sql.VersionMethod
+
+    def to_sql(self) -> sql.Project:
+        return sql.Project(
+            key=str(self.key),
+            name=self.name,
+            status=self.status,
+            committee_key=str(self.committee_key) if self.committee_key else None,
+            version_method=self.version_method,
+        )
+
+
+class ReleaseRow(schema.Subset):
+    key: safe.ReleaseKey
+    project_key: safe.ProjectKey
+    version: safe.VersionKey
+    phase: sql.ReleasePhase
+    created: _RequiredDate
+    released: _OptionalDate = None
+    archived: _OptionalDate = None
+    release_status: _Name = None
+
+    @property
+    def is_archived(self) -> bool:
+        return self.release_status == "retired"
+
+    def to_sql(self) -> sql.Release:
+        return sql.Release(
+            key=str(self.key),
+            project_key=str(self.project_key),
+            version=str(self.version),
+            phase=self.phase,
+            created=self.created,
+            released=self.released,
+            archived=self.archived,
+            is_archived=self.is_archived,
+        )
+
+
+class ArtifactRow(schema.Subset):
+    project_key: safe.ProjectKey
+    version: safe.VersionKey
+    artifact_path: safe.RelPath
+    signature_path: _OptionalRelPath = None
+    checksum_path: _OptionalRelPath = None
+    sbom_path: _OptionalRelPath = None
+    classification: _Name = None
+    download_path_suffix: _OptionalRelPath = None
+    managed: _Managed = False
+    dated: _OptionalDate = None
+
+    @property
+    def release_key(self) -> str:
+        return f"{self.project_key}-{self.version}"
+
+    def to_sql(self) -> sql.Artifact:
+        return sql.Artifact(
+            project_key=str(self.project_key),
+            version=str(self.version),
+            artifact_path=str(self.artifact_path),
+            release_key=self.release_key,
+            signature_path=str(self.signature_path) if self.signature_path else None,
+            checksum_path=str(self.checksum_path) if self.checksum_path else None,
+            sbom_path=str(self.sbom_path) if self.sbom_path else None,
+            classification=self.classification,
+            download_path_suffix=str(self.download_path_suffix) if self.download_path_suffix else None,
+            managed=self.managed,
+            dated=self.dated,
+        )
+
+
+type Row = ProjectRow | ReleaseRow | ArtifactRow
diff --git a/atr/shared/repoint.py b/atr/shared/repoint.py
index 135502ce..a05e78e1 100644
--- a/atr/shared/repoint.py
+++ b/atr/shared/repoint.py
@@ -33,9 +33,8 @@
     (sql.CheckResultIgnore, "project_key"),
 ]
 
-# Tables whose column holds a release key, "{project_key}-{version}". Task.version_key
-# is a logical reference, not a foreign key, but we rewrite it too so a moved project
-# leaves no stale check-task pointers
+# Tables whose column holds a release key, "{project_key}-{version}". Task.version_key holds
+# a bare version, not a release key, so it does not belong here
 RELEASE_KEY_REFS: Final[list[tuple[type[sqlmodel.SQLModel], str]]] = [
     (sql.Release, "key"),
     (sql.Artifact, "release_key"),
@@ -45,7 +44,6 @@
     (sql.Distribution, "release_key"),
     (sql.Quarantined, "release_key"),
     (sql.Revision, "release_key"),
-    (sql.Task, "version_key"),
 ]
 
 # Tables whose column holds a cycle key, "{project_key}-{cycle}"
diff --git a/atr/storage/writers/catalogue.py b/atr/storage/writers/catalogue.py
index 3eafda33..e7c114f0 100644
--- a/atr/storage/writers/catalogue.py
+++ b/atr/storage/writers/catalogue.py
@@ -18,6 +18,8 @@
 # Removing this will cause circular imports
 from __future__ import annotations
 
+from typing import Final
+
 import sqlalchemy
 import sqlmodel
 
@@ -25,9 +27,15 @@
 import atr.db as db
 import atr.models.safe as safe
 import atr.models.sql as sql
+import atr.shared.catalogue_diff as catalogue_diff
+import atr.shared.catalogue_import as catalogue_import
+import atr.shared.catalogue_rows as catalogue_rows
 import atr.shared.repoint as repoint
 import atr.storage as storage
 
+# SQLite allows 999 bound variables by default, and an artifact key is three of them
+_DELETE_CHUNK: Final[int] = 300
+
 
 class FoundationAdmin:
     def __init__(self, write: storage.Write, write_as: storage.WriteAsFoundationAdmin, data: db.Session):
@@ -126,14 +134,13 @@ async def rehome_project(self, source_key: safe.ProjectKey, target_key: safe.Pro
 
     async def _rehome_rows(self, source: str, target: str, source_release_keys: list[str]) -> list[str]:
         via = sql.validate_instrumented_attribute
-        # A sub-project of the source hangs under the target once its parent is gone
+        # Sub-projects of the source reparent onto the target
         await self.__data.execute(
             sqlmodel.update(sql.Project)
             .where(via(sql.Project.super_project_key) == source)
             .values(super_project_key=target)
         )
-        # Each release moves under the target project; its cycle is resolved later, against
-        # the target's own version scheme rather than assumed to be the default
+        # Each release moves under the target project; its cycle is resolved later
         releases = await self.__data.release(project_key=source).all()
         moved_keys = [f"{target}-{release.version}" for release in releases]
         for release in releases:
@@ -142,8 +149,7 @@ async def _rehome_rows(self, source: str, target: str, source_release_keys: list
                 .where(via(sql.Release.key) == release.key)
                 .values(key=f"{target}-{release.version}", project_key=target)
             )
-        # Release-key children rewrite their prefix (Release itself moved above); project-key
-        # children point at the target. Both scoped to the source's own rows
+        # Release-key children rewrite their prefix; project-key children point at the target
         release_child_refs = [(model, attr) for model, attr in repoint.RELEASE_KEY_REFS if model is not sql.Release]
         await self._rewrite_prefix(release_child_refs, source_release_keys, source, target)
         for model, attr in repoint.PROJECT_KEY_REFS:
@@ -155,16 +161,13 @@ async def _rehome_rows(self, source: str, target: str, source_release_keys: list
 
     async def _rehome_cycles(self, target: str, moved_keys: list[str], source_cycle_keys: list[str]) -> None:
         via = sql.validate_instrumented_attribute
-        # Re-resolve cycle membership for the moved releases against the target's scheme,
-        # creating any cycle the target lacks. The raw updates bypassed the ORM, so expire
-        # first to re-read the moved rows fresh
+        # Re-resolve cycle membership for the moved releases against the target's scheme
         self.__data.expire_all()
         target_project = await self.__data.project(key=target).get()
         if target_project is None:
             raise storage.AccessError("The rehome target could not be reloaded.", status=500)
         await cycles.reassign_release_cycles(self.__data, target_project)
-        # A lifecycle event belongs in the same cycle as its release. version_key already
-        # points at the moved release key, so align each event with the cycle just resolved
+        # Align each moved release's lifecycle events with its resolved cycle
         for moved_key in moved_keys:
             moved_release = await self.__data.get(sql.Release, moved_key)
             if moved_release is None:
@@ -174,8 +177,7 @@ async def _rehome_cycles(self, target: str, moved_keys: list[str], source_cycle_
                 .where(via(sql.LifecycleEvent.version_key) == moved_key)
                 .values(cycle_key=moved_release.cycle_key)
             )
-        # Project-level events with no release can't take a release's cycle, so the ones left
-        # on a source cycle fall back to the target default, which always exists
+        # Release-less events left on a source cycle fall back to the target default
         await self.__data.execute(
             sqlmodel.update(sql.LifecycleEvent)
             .where(via(sql.LifecycleEvent.cycle_key).in_(source_cycle_keys))
@@ -184,7 +186,6 @@ async def _rehome_cycles(self, target: str, moved_keys: list[str], source_cycle_
 
     async def _drop_project(self, source: str) -> None:
         via = sql.validate_instrumented_attribute
-        # The source's own cycles and project row go; a rehome target keeps its cycles
         await self.__data.execute(sqlmodel.delete(sql.ProjectCycle).where(via(sql.ProjectCycle.project_key) == source))
         await self.__data.execute(sqlmodel.delete(sql.Project).where(via(sql.Project.key) == source))
 
@@ -197,35 +198,248 @@ async def delete_project(self, project_key: safe.ProjectKey) -> None:
                 raise storage.AccessError(f"Project '{key}' not found.", status=404)
             await self._ensure_catalog_only(key)
             await self.__data.execute(sqlalchemy.text("PRAGMA defer_foreign_keys=ON"))
-            via = sql.validate_instrumented_attribute
-            release_keys = await self._release_keys(key)
-            cycle_keys = await self._cycle_keys(key)
+            await self._delete_project_rows(key)
+            await self._assert_fk_integrity()
+            await self.__data.commit()
+        except Exception:
+            await self.__data.rollback()
+            raise
+        self.__write_as.append_to_audit_log(asf_uid=self.__asf_uid, deleted_project=key)
+
+    async def _delete_project_rows(self, key: str) -> None:
+        via = sql.validate_instrumented_attribute
+        release_keys = await self._release_keys(key)
+        cycle_keys = await self._cycle_keys(key)
+
+        # Sub-projects survive with their super pointer cleared
+        await self.__data.execute(
+            sqlmodel.update(sql.Project).where(via(sql.Project.super_project_key) == key).values(super_project_key=None)
+        )
+        for model, attr in repoint.RELEASE_KEY_REFS:
+            column = via(getattr(model, attr))
+            await self.__data.execute(sqlmodel.delete(model).where(column.in_(release_keys)))
+        for model, attr in repoint.CYCLE_KEY_REFS:
+            column = via(getattr(model, attr))
+            await self.__data.execute(sqlmodel.delete(model).where(column.in_(cycle_keys)))
+        for model, attr in repoint.PROJECT_KEY_REFS:
+            if (model, attr) == (sql.Project, "super_project_key"):
+                continue
+            column = via(getattr(model, attr))
+            await self.__data.execute(sqlmodel.delete(model).where(column == key))
+        await self.__data.execute(sqlmodel.delete(sql.Project).where(via(sql.Project.key) == key))
+
+    async def _delete_release_rows(self, release_keys: list[str]) -> None:
+        if not release_keys:
+            return
+        via = sql.validate_instrumented_attribute
+        for model, attr in repoint.RELEASE_KEY_REFS:
+            if model is sql.Release:
+                continue
+            column = via(getattr(model, attr))
+            await self.__data.execute(sqlmodel.delete(model).where(column.in_(release_keys)))
+        await self.__data.execute(sqlmodel.delete(sql.Release).where(via(sql.Release.key).in_(release_keys)))
+
+    async def _delete_artifact_rows(self, pks: list[tuple[str, str, str]]) -> None:
+        if not pks:
+            return
+        via = sql.validate_instrumented_attribute
+        identity = sqlalchemy.tuple_(
+            via(sql.Artifact.project_key), via(sql.Artifact.version), via(sql.Artifact.artifact_path)
+        )
+        # Chunked to stay inside SQLite's bound-variable limit
+        for start in range(0, len(pks), _DELETE_CHUNK):
+            chunk = pks[start : start + _DELETE_CHUNK]
+            await self.__data.execute(sqlmodel.delete(sql.Artifact).where(identity.in_(chunk)))
+
+    async def _resolve_super_projects(self, project_keys: list[str]) -> None:
+        # A sub-project hangs under the longest project key in its committee that prefixes its own,
+        # which is the rule the project creation path applies. The CSV does not carry the column
+        if not project_keys:
+            return
+        self.__data.expire_all()
+        candidates_by_committee: dict[str, list[str]] = {}
+        for project_key in project_keys:
+            project = await self.__data.project(key=project_key).get()
+            if project is None:
+                raise storage.AccessError(f"Project '{project_key}' could not be reloaded.", status=500)
+            committee_key = project.committee_key
+            if committee_key is None:
+                continue
+            if committee_key not in candidates_by_committee:
+                siblings = await self.__data.project(committee_key=committee_key).all()
+                candidates_by_committee[committee_key] = sorted(
+                    (str(sibling.key) for sibling in siblings), key=len, reverse=True
+                )
+            project.super_project_key = next(
+                (
+                    key
+                    for key in candidates_by_committee[committee_key]
+                    if (key != project_key) and project_key.startswith(f"{key}-")
+                ),
+                None,
+            )
+
+    async def _reassign_cycles_for_added_releases(self, diff: catalogue_diff.CatalogueDiff) -> None:
+        # An added release lands in the project's default cycle, so resolve it against the version
+        # scheme, which also creates any cycle the project lacks
+        project_keys = sorted({str(add.project_key) for add in diff.adds if isinstance(add, catalogue_rows.ReleaseRow)})
+        if not project_keys:
+            return
+        await self.__data.flush()
+        self.__data.expire_all()
+        for project_key in project_keys:
+            project = await self.__data.project(key=project_key).get()
+            if project is None:
+                raise storage.AccessError(f"Project '{project_key}' could not be reloaded.", status=500)
+            await cycles.reassign_release_cycles(self.__data, project)
 
-            # A sub-project survives with its super pointer cleared; we delete this project,
-            # not the ones that hang beneath it
+    async def _apply_deletions(self, diff: catalogue_diff.CatalogueDiff) -> list[str]:
+        # Children first: a release cascades its artifacts, a project needs its releases gone. Each
+        # level goes in one statement per table, the way _delete_project_rows already does it
+        await self._delete_artifact_rows(diff.artifact_deletions)
+        await self._delete_release_rows(diff.release_deletions)
+        orphaned = await self._surviving_sub_projects(diff.project_deletions)
+        for project_key in diff.project_deletions:
+            await self._delete_project_rows(project_key)
+        return orphaned
+
+    async def _surviving_sub_projects(self, project_keys: list[str]) -> list[str]:
+        # Deleting a project clears the super pointer of every project beneath it, including the
+        # ones that are not being deleted, so they have to be given a new one afterwards
+        if not project_keys:
+            return []
+        via = sql.validate_instrumented_attribute
+        rows = (
             await self.__data.execute(
-                sqlmodel.update(sql.Project)
-                .where(via(sql.Project.super_project_key) == key)
-                .values(super_project_key=None)
+                sqlmodel.select(via(sql.Project.key)).where(via(sql.Project.super_project_key).in_(project_keys))
             )
-            for model, attr in repoint.RELEASE_KEY_REFS:
-                column = via(getattr(model, attr))
-                await self.__data.execute(sqlmodel.delete(model).where(column.in_(release_keys)))
-            for model, attr in repoint.CYCLE_KEY_REFS:
-                column = via(getattr(model, attr))
-                await self.__data.execute(sqlmodel.delete(model).where(column.in_(cycle_keys)))
-            for model, attr in repoint.PROJECT_KEY_REFS:
-                if (model, attr) == (sql.Project, "super_project_key"):
-                    continue
-                column = via(getattr(model, attr))
-                await self.__data.execute(sqlmodel.delete(model).where(column == key))
-            await self.__data.execute(sqlmodel.delete(sql.Project).where(via(sql.Project.key) == key))
-            await self._assert_fk_integrity()
+        ).all()
+        return [row[0] for row in rows if row[0] not in set(project_keys)]
+
+    async def import_catalogue_csvs(
+        self,
+        rows_by_table: dict[str, list[dict[str, str]]],
+        committee_key: str,
+        mode: catalogue_diff.Mode,
+        previewed: str | None = None,
+    ) -> catalogue_diff.CatalogueDiff:
+        # Recompute the diff against a snapshot taken inside this transaction, under the write lock
+        await self.__data.begin_immediate()
+        self.__data.expire_all()
+        try:
+            snapshot = await catalogue_import.build_snapshot(self.__data, committee_key)
+            diff = catalogue_diff.classify(rows_by_table, snapshot, committee_key, mode)
+            if (previewed is not None) and (catalogue_diff.fingerprint(diff) != previewed):
+                # The catalogue moved since the preview, so this would write something the admin
+                # has not seen. The watcher catalogues releases while the page is open
+                raise catalogue_import.StaleError(committee_key)
+            await self._apply_diff(diff)
             await self.__data.commit()
         except Exception:
             await self.__data.rollback()
             raise
-        self.__write_as.append_to_audit_log(asf_uid=self.__asf_uid, deleted_project=key)
+        self._audit_import(diff, committee_key, mode)
+        return diff
+
+    async def _apply_diff(self, diff: catalogue_diff.CatalogueDiff) -> None:
+        await self.__data.execute(sqlalchemy.text("PRAGMA defer_foreign_keys=ON"))
+        # Deletions run first, so the committee is cleared before the files are re-added and a row
+        # can reuse a key it has just released. A deleted project leaves its surviving sub-projects
+        # with no super pointer, and they hang under whatever the files put back in its place
+        orphaned = await self._apply_deletions(diff)
+        # Adds in FK order: projects, then releases, then artifacts
+        for add in diff.adds:
+            if isinstance(add, catalogue_rows.ProjectRow):
+                self.__data.add(add.to_sql())
+        await self.__data.flush()
+        added = [str(add.key) for add in diff.adds if isinstance(add, catalogue_rows.ProjectRow)]
+        await self._resolve_super_projects(added + orphaned)
+        for add in diff.adds:
+            if isinstance(add, catalogue_rows.ReleaseRow):
+                self.__data.add(add.to_sql())
+        await self.__data.flush()
+        for add in diff.adds:
+            if isinstance(add, catalogue_rows.ArtifactRow):
+                self.__data.add(add.to_sql())
+        await self._reassign_cycles_for_added_releases(diff)
+        # Artifacts repoint first: a repointed release takes its artifacts by project and version,
+        # so an artifact the file sends elsewhere has to leave before the release takes it along
+        for artifact_repoint in diff.artifact_repoints:
+            await self._repoint_one_artifact(artifact_repoint)
+        for release_repoint in diff.release_repoints:
+            await self._repoint_one_release(release_repoint)
+        await self._assert_fk_integrity()
+
+    def _audit_import(self, diff: catalogue_diff.CatalogueDiff, committee_key: str, mode: catalogue_diff.Mode) -> None:
+        counts = diff.counts
+        self.__write_as.append_to_audit_log(
+            asf_uid=self.__asf_uid,
+            committee_key=committee_key,
+            mode=str(mode),
+            added=counts["add"],
+            releases_repointed=counts["release_repoint"],
+            artifacts_repointed=counts["artifact_repoint"],
+            deleted=counts["delete"],
+            conflicts=counts["conflict"],
+        )
+
+    async def _repoint_one_release(self, release_repoint: catalogue_diff.ReleaseRepoint) -> None:
+        via = sql.validate_instrumented_attribute
+        old_key = release_repoint.key
+        version = str(release_repoint.row.version)
+        new_key = f"{release_repoint.to_project}-{version}"
+        await self.__data.execute(
+            sqlmodel.update(sql.Artifact)
+            .where(via(sql.Artifact.project_key) == release_repoint.from_project)
+            .where(via(sql.Artifact.version) == version)
+            .values(release_key=new_key, project_key=release_repoint.to_project)
+        )
+        await self.__data.execute(
+            sqlmodel.update(sql.LifecycleEvent)
+            .where(via(sql.LifecycleEvent.version_key) == old_key)
+            .values(version_key=new_key, project_key=release_repoint.to_project)
+        )
+        # Other release-key children re-point to the new key
+        for model, attr in repoint.RELEASE_KEY_REFS:
+            if model in (sql.Release, sql.Artifact, sql.LifecycleEvent):
+                continue
+            column = via(getattr(model, attr))
+            await self.__data.execute(sqlmodel.update(model).where(column == old_key).values(**{attr: new_key}))
+        await self.__data.execute(
+            sqlmodel.update(sql.Release)
+            .where(via(sql.Release.key) == old_key)
+            .values(key=new_key, project_key=release_repoint.to_project)
+        )
+        # Resolve the repointed release's cycle against the target's scheme, then align its events
+        self.__data.expire_all()
+        target_project = await self.__data.project(key=release_repoint.to_project).get()
+        if target_project is None:
+            raise storage.AccessError("The repoint target could not be reloaded.", status=500)
+        await cycles.reassign_release_cycles(self.__data, target_project)
+        repointed_release = await self.__data.get(sql.Release, new_key)
+        if repointed_release is None:
+            raise storage.AccessError("The repointed release could not be reloaded.", status=500)
+        await self.__data.execute(
+            sqlmodel.update(sql.LifecycleEvent)
+            .where(via(sql.LifecycleEvent.version_key) == new_key)
+            .values(cycle_key=repointed_release.cycle_key)
+        )
+
+    async def _repoint_one_artifact(self, artifact_repoint: catalogue_diff.ArtifactRepoint) -> None:
+        via = sql.validate_instrumented_attribute
+        old_project, old_version, old_path = artifact_repoint.from_pk
+        row = artifact_repoint.to_row
+        await self.__data.execute(
+            sqlmodel.update(sql.Artifact)
+            .where(via(sql.Artifact.project_key) == old_project)
+            .where(via(sql.Artifact.version) == old_version)
+            .where(via(sql.Artifact.artifact_path) == old_path)
+            .values(
+                project_key=str(row.project_key),
+                version=str(row.version),
+                release_key=row.release_key,
+            )
+        )
 
     async def _collisions(self, source_key: str, target_key: str) -> list[repoint.Collision]:
         via = sql.validate_instrumented_attribute
@@ -251,8 +465,7 @@ async def _collisions(self, source_key: str, target_key: str) -> list[repoint.Co
         return collisions
 
     async def _repoint(self, old_key: str, new_key: str) -> None:
-        # Defer foreign key checks to commit, so we can rewrite parent primary keys
-        # and their children in any order within this one transaction
+        # Foreign key checks deferred to commit while parent and child keys are rewritten
         await self.__data.execute(sqlalchemy.text("PRAGMA defer_foreign_keys=ON"))
         via = sql.validate_instrumented_attribute
         release_keys = await self._release_keys(old_key)
@@ -266,8 +479,8 @@ async def _repoint(self, old_key: str, new_key: str) -> None:
         for model, attr in repoint.PROJECT_KEY_REFS:
             column = via(getattr(model, attr))
             await self.__data.execute(sqlmodel.update(model).where(column == old_key).values(**{attr: new_key}))
-        # Columns that hold a release key; scoped to this project's own release keys, so a
-        # sibling project whose key shares this prefix is never swept in
+        # Columns that hold a release key, scoped to this project's own keys (never a sibling
+        # sharing the prefix)
         await self._rewrite_prefix(repoint.RELEASE_KEY_REFS, release_keys, old_key, new_key)
         # Columns that hold a cycle key
         await self._rewrite_prefix(repoint.CYCLE_KEY_REFS, cycle_keys, old_key, new_key)
@@ -300,9 +513,8 @@ async def _cycle_keys(self, project_key: str) -> list[str]:
         return list(result.scalars().all())
 
     async def _ensure_catalog_only(self, project_key: str) -> None:
-        # A catalogued project carries only project, cycle, release, artifact and lifecycle
-        # rows. A live ATR project also has revisions and release file-state, whose keys this
-        # page does not yet re-point, so refuse those rather than corrupt them
+        # A live project also carries revisions and release file-state, whose keys this page
+        # does not re-point
         via = sql.validate_instrumented_attribute
         release_keys = await self._release_keys(project_key)
         if not release_keys:
@@ -320,8 +532,7 @@ async def _ensure_catalog_only(self, project_key: str) -> None:
             )
 
     async def _assert_fk_integrity(self) -> None:
-        # Flush pending ORM work so the check sees the full picture, then refuse to commit if
-        # a correction has left any dangling reference behind
+        # Flush pending ORM work, then check for dangling references before commit
         await self.__data.flush()
         result = await self.__data.execute(sqlalchemy.text("PRAGMA foreign_key_check"))
         violations = result.fetchall()
diff --git a/atr/templates/includes/topnav.html b/atr/templates/includes/topnav.html
index 90ff15f7..0e34b0f8 100644
--- a/atr/templates/includes/topnav.html
+++ b/atr/templates/includes/topnav.html
@@ -275,6 +275,11 @@
                    href="{{ as_url(admin.projects_update_get) }}"
                    {% if request.endpoint == 'atr_admin_projects_update_get' %}class="active"{% endif %}><i class="bi bi-arrow-repeat"></i> Update projects</a>
               </li>
+              <li>
+                <a class="dropdown-item"
+                   href="{{ as_url(admin.catalog_get) }}"
+                   {% if request.endpoint == 'atr_admin_catalog_get' %}class="active"{% endif %}><i class="bi bi-diagram-3"></i> Catalog admin</a>
+              </li>
               <li>
                 <hr class="dropdown-divider" />
               </li>
diff --git a/tests/unit/test_admin_catalog.py b/tests/unit/test_admin_catalog.py
index d1f16698..8fad537b 100644
--- a/tests/unit/test_admin_catalog.py
+++ b/tests/unit/test_admin_catalog.py
@@ -16,29 +16,40 @@
 # under the License.
 
 import atr.admin as admin_module
-
-
-def test_catalog_handlers_exist_and_import() -> None:
-    # Importing atr.admin registers the handlers via the @admin.typed decorator; if the
-    # type annotations are malformed the import itself raises, so this both smoke-tests
-    # import and confirms the two workspace handlers are present
-    assert hasattr(admin_module, "catalog_get")
-    assert hasattr(admin_module, "catalog_committee_get")
-
-
-def test_move_handlers_exist() -> None:
-    assert hasattr(admin_module, "catalog_move_get")
-    assert hasattr(admin_module, "catalog_move_post")
-    assert hasattr(admin_module, "CatalogMoveForm")
-
-
-def test_rename_handlers_exist() -> None:
-    assert hasattr(admin_module, "catalog_rename_get")
-    assert hasattr(admin_module, "catalog_rename_post")
-    assert hasattr(admin_module, "CatalogRenameForm")
-
-
-def test_remove_handlers_exist() -> None:
-    assert hasattr(admin_module, "catalog_remove_get")
-    assert hasattr(admin_module, "catalog_remove_post")
-    assert hasattr(admin_module, "CatalogRemoveForm")
+import atr.shared.catalogue_diff as catalogue_diff
+import atr.shared.catalogue_rows as catalogue_rows
+
+
+def _release_row(key: str, project: str, version: str) -> catalogue_rows.ReleaseRow:
+    return catalogue_rows.ReleaseRow.model_validate(
+        {"key": key, "project_key": project, "version": version, "phase": "release", "created": "2020-01-01"}
+    )
+
+
+def test_import_preview_lists_conflicts_and_repoints() -> None:
+    diff = catalogue_diff.CatalogueDiff(
+        release_repoints=[
+            catalogue_diff.ReleaseRepoint(
+                key="alpha-one-4.0.0",
+                from_project="alpha-one",
+                to_project="alpha-two",
+                row=_release_row("alpha-one-4.0.0", "alpha-two", "4.0.0"),
+            )
+        ],
+        conflicts=[catalogue_diff.Conflict(table="artifacts", key="c.tar.gz", reason="unresolvable identity")],
+    )
+    html = str(admin_module._catalog_import_preview(diff))
+    assert "unresolvable identity" in html
+    assert "alpha-one-4.0.0 to alpha-two" in html
+
+
+def test_import_preview_warns_about_deletions_and_empty_levels() -> None:
+    diff = catalogue_diff.CatalogueDiff(
+        release_deletions=["alpha-one-1.0.0"],
+        artifact_deletions=[("alpha-one", "1.0.0", "a.tar.gz")],
+        warnings=["artifacts.csv was not uploaded, so 1 artifacts will be deleted and not restored."],
+    )
+    html = str(admin_module._catalog_import_preview(diff))
+    assert "artifacts.csv was not uploaded" in html
+    assert "release alpha-one-1.0.0" in html
+    assert "anything unspecified to be deleted" in html
diff --git a/tests/unit/test_catalogue_diff.py b/tests/unit/test_catalogue_diff.py
new file mode 100644
index 00000000..0590f789
--- /dev/null
+++ b/tests/unit/test_catalogue_diff.py
@@ -0,0 +1,458 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import atr.shared.catalogue_diff as catalogue_diff
+import atr.shared.catalogue_rows as catalogue_rows
+
+
+def _project(key: str, committee: str = "alpha") -> dict[str, str]:
+    return {"key": key, "committee_key": committee, "status": "active", "version_method": "simple"}
+
+
+def _release(key: str, project: str, version: str) -> dict[str, str]:
+    return {"key": key, "project_key": project, "version": version, "phase": "release", "created": "2020-01-01"}
+
+
+def _artifact(project: str, version: str, path: str, suffix: str) -> dict[str, str]:
+    return {"project_key": project, "version": version, "artifact_path": path, "download_path_suffix": suffix}
+
+
+def _snapshot(**kw) -> catalogue_diff.DbSnapshot:
+    base = dict(
+        project_committee={"alpha-one": "alpha"},
+        release_project={"alpha-one-1.0.0": "alpha-one"},
+        artifact_by_dist={},
+        managed_project_keys=frozenset(),
+    )
+    base.update(kw)
+    return catalogue_diff.DbSnapshot(**base)
+
+
+def test_diff_counts_an_add() -> None:
+    snap = _snapshot()
+    assert "alpha-one" in snap.project_committee
+    d = catalogue_diff.classify_additive({"projects": [_project("alpha-two")]}, snap, "alpha")
+    assert d.counts["add"] == 1
+    assert d.counts["conflict"] == 0
+
+
+def test_classify_projects_adds_new_conflicts_missing_committee_and_managed() -> None:
+    snap = _snapshot(managed_project_keys=frozenset({"alpha-managed"}))
+    rows = [
+        _project("alpha-two"),  # add
+        _project("alpha-one"),  # present -> unchanged
+        _project("beta-one", committee="beta"),  # missing committee -> conflict
+        _project("alpha-managed"),  # managed -> conflict
+    ]
+    d = catalogue_diff.classify_additive({"projects": rows}, snap, "alpha")
+    assert [str(a.key) for a in d.adds] == ["alpha-two"]
+    assert d.unchanged == 1
+    reasons = {c.key: c.reason for c in d.conflicts}
+    assert "beta-one" in reasons and "committee" in reasons["beta-one"].lower()
+    assert "alpha-managed" in reasons
+
+
+def test_classify_releases_repoints_within_the_committee_and_refuses_to_leave_it() -> None:
+    snap = _snapshot(
+        project_committee={"alpha-one": "alpha", "alpha-two": "alpha", "beta-one": "beta"},
+        release_project={
+            "alpha-one-1.0.0": "alpha-one",
+            "alpha-one-2.0.0": "alpha-one",
+            "alpha-one-4.0.0": "alpha-one",
+        },
+    )
+    rows = [
+        _release("alpha-one-3.0.0", "alpha-one", "3.0.0"),  # add
+        _release("alpha-one-1.0.0", "alpha-one", "1.0.0"),  # unchanged
+        _release("alpha-one-2.0.0", "alpha-two", "2.0.0"),  # repoint within committee
+        _release("alpha-one-4.0.0", "beta-one", "4.0.0"),  # another committee -> conflict
+        _release("ghost-5.0.0", "ghost", "5.0.0"),  # target missing
+    ]
+    d = catalogue_diff.classify_additive({"releases": rows}, snap, "alpha")
+    assert [str(a.key) for a in d.adds] == ["alpha-one-3.0.0"]
+    assert d.unchanged == 1
+    assert [m.key for m in d.release_repoints] == ["alpha-one-2.0.0"]
+    reasons = {c.key: c.reason for c in d.conflicts}
+    assert reasons["alpha-one-4.0.0"] == "project 'beta-one' is not in this committee"
+    assert "does not exist" in reasons["ghost-5.0.0"]
+
+
+def test_classify_releases_accepts_a_release_under_a_project_added_in_the_same_diff() -> None:
+    # Combined upload: projects.csv adds alpha-new, releases.csv adds a release under it
+    snap = _snapshot(project_committee={"alpha-one": "alpha"}, release_project={})
+    d = catalogue_diff.classify_additive(
+        {"projects": [_project("alpha-new")], "releases": [_release("alpha-new-1.0.0", "alpha-new", "1.0.0")]},
+        snap,
+        "alpha",
+    )
+    release_adds = [a for a in d.adds if isinstance(a, catalogue_rows.ReleaseRow)]
+    assert [str(a.key) for a in release_adds] == ["alpha-new-1.0.0"]
+    assert not d.conflicts
+
+
+def test_classify_releases_flags_a_repoint_onto_an_existing_target_release() -> None:
+    snap = _snapshot(
+        project_committee={"alpha-one": "alpha", "alpha-two": "alpha"},
+        release_project={"alpha-one-1.0.0": "alpha-one"},
+        release_keys=frozenset({"alpha-one-1.0.0", "alpha-two-1.0.0"}),
+    )
+    d = catalogue_diff.classify_additive(
+        {"releases": [_release("alpha-one-1.0.0", "alpha-two", "1.0.0")]}, snap, "alpha"
+    )
+    assert not d.release_repoints
+    assert [c.reason for c in d.conflicts] == ["target already has release 'alpha-two-1.0.0'"]
+
+
+def test_classify_releases_refuses_a_row_whose_key_disagrees_with_its_version() -> None:
+    # A release key is derived from its project and version, so a row that breaks the derivation
+    # is refused rather than silently re-keying the release
+    snap = _snapshot(project_committee={"alpha-one": "alpha", "alpha-two": "alpha"})
+    d = catalogue_diff.classify_additive(
+        {"releases": [_release("alpha-one-1.0.0", "alpha-two", "2.0.0")]}, snap, "alpha"
+    )
+    assert not d.release_repoints
+    assert [c.reason for c in d.conflicts] == ["key does not match version '2.0.0'"]
+
+
+def test_classify_releases_refuses_an_add_whose_key_already_exists() -> None:
+    # The release belongs to another committee, so it is absent from the workspace-scoped map
+    snap = _snapshot(
+        project_committee={"alpha-one": "alpha"},
+        release_project={},
+        release_keys=frozenset({"alpha-one-1.0.0"}),
+    )
+    d = catalogue_diff.classify_additive(
+        {"releases": [_release("alpha-one-1.0.0", "alpha-one", "1.0.0")]}, snap, "alpha"
+    )
+    assert not d.adds
+    assert [c.reason for c in d.conflicts] == ["release 'alpha-one-1.0.0' already exists"]
+
+
+def test_classify_artifacts_refuses_an_add_that_would_duplicate_an_existing_artifact() -> None:
+    # Editing the dist suffix describes a new artifact, but its primary key is already taken
+    snap = _snapshot(
+        artifact_by_dist={("alpha/1.0.0", "a.tar.gz"): ("alpha-one", "1.0.0", "a.tar.gz")},
+        artifact_pks=frozenset({("alpha-one", "1.0.0", "a.tar.gz")}),
+    )
+    rows = [_artifact("alpha-one", "1.0.0", "a.tar.gz", "alpha/edited")]
+    d = catalogue_diff.classify_additive({"artifacts": rows}, snap, "alpha")
+    assert not d.adds
+    assert not d.artifact_repoints
+    assert [c.reason for c in d.conflicts] == ["artifact already exists at that project and version"]
+
+
+def test_classify_artifacts_refuses_a_repoint_out_of_a_managed_project() -> None:
+    snap = _snapshot(
+        project_committee={"alpha-one": "alpha", "alpha-live": "alpha"},
+        artifact_by_dist={("alpha/1.0.0", "a.tar.gz"): ("alpha-live", "1.0.0", "a.tar.gz")},
+        managed_project_keys=frozenset({"alpha-live"}),
+    )
+    rows = [_artifact("alpha-one", "1.0.0", "a.tar.gz", "alpha/1.0.0")]
+    d = catalogue_diff.classify_additive({"artifacts": rows}, snap, "alpha")
+    assert not d.artifact_repoints
+    assert [c.reason for c in d.conflicts] == ["source project has live workflow data"]
+
+
+def test_classify_artifacts_refuses_a_project_outside_the_committee() -> None:
+    snap = _snapshot(project_committee={"alpha-one": "alpha", "beta-one": "beta"})
+    rows = [_artifact("beta-one", "1.0.0", "a.tar.gz", "beta/1.0.0")]
+    d = catalogue_diff.classify_additive({"artifacts": rows}, snap, "alpha")
+    assert [c.reason for c in d.conflicts] == ["project 'beta-one' is not in this committee"]
+
+
+def test_replace_prunes_only_the_tables_that_were_uploaded() -> None:
+    snap = _snapshot(
+        project_committee={"alpha-one": "alpha", "alpha-two": "alpha"},
+        artifact_by_dist={("alpha/1.0.0", "a.tar.gz"): ("alpha-one", "1.0.0", "a.tar.gz")},
+        artifact_pks=frozenset({("alpha-one", "1.0.0", "a.tar.gz")}),
+    )
+    d = catalogue_diff.classify_replace({"artifacts": []}, snap, "alpha")
+    assert d.artifact_deletions == [("alpha-one", "1.0.0", "a.tar.gz")]
+    assert not d.project_deletions
+    assert not d.release_deletions
+
+
+def test_replace_never_deletes_anything_under_a_managed_project() -> None:
+    snap = _snapshot(
+        project_committee={"alpha-live": "alpha"},
+        release_project={"alpha-live-1.0.0": "alpha-live"},
+        artifact_by_dist={("alpha/1.0.0", "a.tar.gz"): ("alpha-live", "1.0.0", "a.tar.gz")},
+        managed_project_keys=frozenset({"alpha-live"}),
+    )
+    d = catalogue_diff.classify_replace({"projects": [], "releases": [], "artifacts": []}, snap, "alpha")
+    assert d.counts["delete"] == 0
+
+
+def test_replace_clears_downwards_from_the_deepest_table_uploaded() -> None:
+    # releases.csv clears releases and, by cascade, artifacts; projects are left alone
+    snap = _snapshot(
+        project_committee={"alpha-one": "alpha"},
+        release_project={"alpha-one-1.0.0": "alpha-one", "alpha-one-2.0.0": "alpha-one"},
+        artifact_by_dist={("alpha/1.0.0", "a.tar.gz"): ("alpha-one", "1.0.0", "a.tar.gz")},
+        artifact_pks=frozenset({("alpha-one", "1.0.0", "a.tar.gz")}),
+    )
+    rows = {"releases": [_release("alpha-one-1.0.0", "alpha-one", "1.0.0")]}
+    d = catalogue_diff.classify_replace(rows, snap, "alpha")
+    assert not d.project_deletions
+    assert d.release_deletions == ["alpha-one-1.0.0", "alpha-one-2.0.0"]
+    assert d.artifact_deletions == [("alpha-one", "1.0.0", "a.tar.gz")]
+    # the cleared committee means the uploaded release is simply re-added
+    assert [str(a.key) for a in d.adds] == ["alpha-one-1.0.0"]
+    assert not d.conflicts
+    assert any("artifacts.csv was not uploaded" in w for w in d.warnings)
+
+
+def test_replace_warns_when_a_lower_table_is_missing() -> None:
+    snap = _snapshot(
+        project_committee={"alpha-one": "alpha"},
+        release_project={"alpha-one-1.0.0": "alpha-one"},
+        artifact_by_dist={("alpha/1.0.0", "a.tar.gz"): ("alpha-one", "1.0.0", "a.tar.gz")},
+        artifact_pks=frozenset({("alpha-one", "1.0.0", "a.tar.gz")}),
+    )
+    d = catalogue_diff.classify_replace({"projects": [_project("alpha-one")]}, snap, "alpha")
+    # projects.csv alone clears everything beneath it, and says so
+    assert d.release_deletions == ["alpha-one-1.0.0"]
+    assert d.artifact_deletions == [("alpha-one", "1.0.0", "a.tar.gz")]
+    assert any("releases.csv was not uploaded" in w for w in d.warnings)
+
+
+def test_replace_lets_an_edited_dist_suffix_replace_the_artifact() -> None:
+    # An artifact's identity is its dist path, so editing it frees the primary key for the new row
+    snap = _snapshot(
+        artifact_by_dist={("alpha/old", "a.tar.gz"): ("alpha-one", "1.0.0", "a.tar.gz")},
+        artifact_pks=frozenset({("alpha-one", "1.0.0", "a.tar.gz")}),
+    )
+    rows = {"artifacts": [_artifact("alpha-one", "1.0.0", "a.tar.gz", "alpha/new")]}
+    overwritten = catalogue_diff.classify_replace(rows, snap, "alpha")
+    assert overwritten.artifact_deletions == [("alpha-one", "1.0.0", "a.tar.gz")]
+    assert [str(add.download_path_suffix) for add in overwritten.adds] == ["alpha/new"]
+    assert not overwritten.conflicts
+
+    additive = catalogue_diff.classify_additive(rows, snap, "alpha")
+    assert [c.reason for c in additive.conflicts] == ["artifact already exists at that project and version"]
+
+
+def test_classify_refuses_a_row_listed_twice_in_the_same_file() -> None:
+    # A row is checked against the rows already accepted from its own file, not just the database,
+    # so a duplicate is a conflict rather than a failed insert
+    snap = _snapshot(project_committee={"alpha-one": "alpha"}, release_project={})
+    rows = {
+        "projects": [_project("alpha-two"), _project("alpha-two")],
+        "releases": [_release("alpha-one-1.0.0", "alpha-one", "1.0.0")] * 2,
+        "artifacts": [_artifact("alpha-one", "1.0.0", "a.tar.gz", "alpha/1.0.0")] * 2,
+    }
+    d = catalogue_diff.classify_additive(rows, snap, "alpha")
+    assert d.counts["add"] == 3
+    assert [c.reason for c in d.conflicts] == ["listed more than once in this file"] * 3
+
+
+def test_replace_is_refused_outright_when_any_row_conflicts() -> None:
+    # A replace deletes what the files do not list, so a row it cannot apply would be deleted and
+    # never restored. One bad row means the files are not a complete set, and nothing is changed
+    snap = _snapshot(
+        project_committee={"alpha-one": "alpha"},
+        release_project={"alpha-one-1.0.0": "alpha-one", "alpha-one-2.0.0": "alpha-one"},
+    )
+    rows = {
+        "releases": [
+            _release("alpha-one-1.0.0", "alpha-one", "1.0.0"),
+            _release("alpha-one-2.0.0", "alpha-one", "9.9.9"),  # key disagrees with version
+        ]
+    }
+    d = catalogue_diff.classify_replace(rows, snap, "alpha")
+    assert d.counts["delete"] == 0
+    assert d.counts["add"] == 0
+    assert d.conflicts
+    assert any("Nothing was changed" in w for w in d.warnings)
+
+
+def test_classify_releases_refuses_two_rows_converging_on_one_release() -> None:
+    # The check must cover the keys the diff itself will produce, not only the ones already in the db
+    snap = _snapshot(
+        project_committee={"alpha-one": "alpha", "alpha-two": "alpha", "alpha-three": "alpha"},
+        release_project={"alpha-one-1.0.0": "alpha-one", "alpha-two-1.0.0": "alpha-two"},
+    )
+    rows = {
+        "releases": [
+            _release("alpha-one-1.0.0", "alpha-three", "1.0.0"),
+            _release("alpha-two-1.0.0", "alpha-three", "1.0.0"),
+        ]
+    }
+    d = catalogue_diff.classify_additive(rows, snap, "alpha")
+    assert len(d.release_repoints) == 1
+    assert [c.reason for c in d.conflicts] == ["target already has release 'alpha-three-1.0.0'"]
+
+
+def test_classify_artifacts_repoints_adds_warns_and_flags_dangling_references() -> None:
+    snap = _snapshot(
+        project_committee={"alpha-one": "alpha"},
+        release_project={"alpha-one-1.0.0": "alpha-one", "alpha-one-2.0.0": "alpha-one"},
+        artifact_by_dist={
+            ("incubator/alpha/1.0.0", "a.tar.gz"): ("alpha-old", "1.0.0", "a.tar.gz"),
+        },
+    )
+    rows = [
+        # dist matches existing under a different project -> repoint
+        _artifact("alpha-one", "1.0.0", "a.tar.gz", "incubator/alpha/1.0.0"),
+        # valid refs, has suffix, no dist match -> add
+        _artifact("alpha-one", "2.0.0", "b.tar.gz", "alpha/2.0.0"),
+        # no suffix, so nothing to match on, and its key is free -> add
+        _artifact("alpha-one", "1.0.0", "c.tar.gz", ""),
+        # derived release key alpha-one-3.0.0 does not exist -> conflict, not a dangling add
+        _artifact("alpha-one", "3.0.0", "d.tar.gz", "alpha/3.0.0"),
+    ]
+    d = catalogue_diff.classify_additive({"artifacts": rows}, snap, "alpha")
+    assert [str(r.to_row.project_key) for r in d.artifact_repoints] == ["alpha-one"]
+    assert [str(a.artifact_path) for a in d.adds] == ["b.tar.gz", "c.tar.gz"]
+    reasons = [c.reason for c in d.conflicts]
+    assert any("release" in r.lower() and "does not exist" in r for r in reasons)
+
+
+def test_classify_artifacts_keeps_an_artifact_without_a_dist_path() -> None:
+    # The dist path is nullable, so an exported row can carry an empty one. A replace re-adds every
+    # row it is given, and a row with nothing to match on is simply a new artifact
+    snap = _snapshot(
+        project_committee={"alpha-one": "alpha"},
+        release_project={"alpha-one-1.0.0": "alpha-one"},
+        release_keys=frozenset({"alpha-one-1.0.0"}),
+        artifact_pks=frozenset({("alpha-one", "1.0.0", "a.tgz")}),
+    )
+    d = catalogue_diff.classify_replace({"artifacts": [_artifact("alpha-one", "1.0.0", "a.tgz", "")]}, snap, "alpha")
+    assert [str(a.artifact_path) for a in d.adds] == ["a.tgz"]
+    assert not d.conflicts
+    assert d.refused is False
+
+
+def test_classify_artifacts_refuses_an_add_onto_a_key_a_repoint_has_not_yet_freed() -> None:
+    # The writer inserts the adds before it repoints anything, so a key an artifact is leaving is
+    # still occupied when the add lands on it
+    snap = _snapshot(
+        project_committee={"alpha-one": "alpha", "alpha-two": "alpha"},
+        release_project={"alpha-one-1.0.0": "alpha-one", "alpha-two-1.0.0": "alpha-two"},
+        release_keys=frozenset({"alpha-one-1.0.0", "alpha-two-1.0.0"}),
+        artifact_by_dist={("alpha/old", "a.tgz"): ("alpha-one", "1.0.0", "a.tgz")},
+        artifact_pks=frozenset({("alpha-one", "1.0.0", "a.tgz")}),
+    )
+    rows = {
+        "artifacts": [
+            _artifact("alpha-two", "1.0.0", "a.tgz", "alpha/old"),
+            _artifact("alpha-one", "1.0.0", "a.tgz", "alpha/new"),
+        ]
+    }
+    d = catalogue_diff.classify_additive(rows, snap, "alpha")
+    assert len(d.artifact_repoints) == 1
+    assert not d.adds
+    assert [c.reason for c in d.conflicts] == ["artifact already exists at that project and version"]
+
+
+def test_classify_artifacts_left_alone_follow_the_release_that_is_repointed() -> None:
+    # Only releases.csv is edited, so every artifact row still names the project the release came
+    # from. Those rows describe artifacts the repoint carries, not dangling references
+    snap = _snapshot(
+        project_committee={"alpha-one": "alpha", "alpha-two": "alpha"},
+        release_project={"alpha-one-1.0.0": "alpha-one"},
+        release_keys=frozenset({"alpha-one-1.0.0"}),
+        artifact_by_dist={("alpha/1.0.0", "a.tgz"): ("alpha-one", "1.0.0", "a.tgz")},
+        artifact_pks=frozenset({("alpha-one", "1.0.0", "a.tgz")}),
+    )
+    rows = {
+        "releases": [_release("alpha-one-1.0.0", "alpha-two", "1.0.0")],
+        "artifacts": [_artifact("alpha-one", "1.0.0", "a.tgz", "alpha/1.0.0")],
+    }
+    d = catalogue_diff.classify_additive(rows, snap, "alpha")
+    assert [r.key for r in d.release_repoints] == ["alpha-one-1.0.0"]
+    assert not d.conflicts
+    assert not d.adds
+    assert not d.artifact_repoints
+    assert d.unchanged == 1
+
+
+def test_classify_rejects_malformed_keys_and_paths() -> None:
+    snap = _snapshot()
+    rows = {
+        "projects": [{"key": "Bad-Cap", "committee_key": "alpha", "status": "active", "version_method": "simple"}],
+        "artifacts": [_artifact("alpha-one", "1.0.0", "../escape.tgz", "alpha/1.0.0")],
+    }
+    d = catalogue_diff.classify_additive(rows, snap, "alpha")
+    reasons = " ".join(c.reason for c in d.conflicts)
+    assert "invalid key" in reasons
+    assert "invalid artifact_path" in reasons
+
+
+def test_classify_projects_refuses_a_key_another_committee_already_holds() -> None:
+    # A project key is unique foundation-wide, so a row claiming one held elsewhere would take it
+    snap = _snapshot(project_committee={"beta-one": "beta"})
+    rows = {"projects": [_project("beta-one")]}
+    for d in (
+        catalogue_diff.classify_replace(rows, snap, "alpha"),
+        catalogue_diff.classify_additive(rows, snap, "alpha"),
+    ):
+        assert not d.adds
+        assert [c.reason for c in d.conflicts] == ["project 'beta-one' belongs to committee 'beta'"]
+
+
+def test_classify_releases_refuses_a_repoint_whose_artifacts_would_collide() -> None:
+    # A move rewrites its artifacts' project key, which is part of their primary key. An artifact
+    # need not have a release, so the target can hold one without holding the release it repoints onto
+    snap = _snapshot(
+        project_committee={"alpha-one": "alpha", "alpha-two": "alpha"},
+        release_project={"alpha-one-1.0.0": "alpha-one"},
+        release_keys=frozenset({"alpha-one-1.0.0"}),
+        artifact_pks=frozenset({("alpha-one", "1.0.0", "a.tar.gz"), ("alpha-two", "1.0.0", "a.tar.gz")}),
+    )
+    d = catalogue_diff.classify_additive(
+        {"releases": [_release("alpha-one-1.0.0", "alpha-two", "1.0.0")]}, snap, "alpha"
+    )
+    assert not d.release_repoints
+    assert [c.reason for c in d.conflicts] == ["target already has artifact 'a.tar.gz' at version 1.0.0"]
+
+
+def test_classify_artifacts_refuses_an_add_a_repointed_release_already_takes() -> None:
+    # The move takes the artifact to alpha-two, so an artifact row landing on the same primary key
+    # would be a second row for it rather than an add
+    snap = _snapshot(
+        project_committee={"alpha-one": "alpha", "alpha-two": "alpha"},
+        release_project={"alpha-one-1.0.0": "alpha-one"},
+        release_keys=frozenset({"alpha-one-1.0.0"}),
+        artifact_pks=frozenset({("alpha-one", "1.0.0", "a.tgz")}),
+    )
+    rows = {
+        "releases": [_release("alpha-one-1.0.0", "alpha-two", "1.0.0")],
+        "artifacts": [_artifact("alpha-two", "1.0.0", "a.tgz", "alpha/new")],
+    }
+    d = catalogue_diff.classify_additive(rows, snap, "alpha")
+    assert [m.key for m in d.release_repoints] == ["alpha-one-1.0.0"]
+    assert not d.adds
+    assert d.counts["conflict"] == 1
+
+
+def test_replace_refuses_a_dist_path_a_surviving_artifact_still_holds() -> None:
+    # Managed projects are never cleared, so their artifacts are still catalogued afterwards. A row
+    # claiming one of their dist paths would catalogue the same file in svn a second time
+    snap = _snapshot(
+        project_committee={"alpha-managed": "alpha", "alpha-two": "alpha"},
+        release_project={"alpha-two-1.0.0": "alpha-two"},
+        release_keys=frozenset({"alpha-two-1.0.0"}),
+        managed_project_keys=frozenset({"alpha-managed"}),
+        artifact_by_dist={("alpha/dist", "a.tgz"): ("alpha-managed", "1.0.0", "a.tgz")},
+        artifact_pks=frozenset({("alpha-managed", "1.0.0", "a.tgz")}),
+    )
+    rows = {"artifacts": [_artifact("alpha-two", "1.0.0", "a.tgz", "alpha/dist")]}
+    d = catalogue_diff.classify_replace(rows, snap, "alpha")
+    assert not d.adds
+    assert d.refused is True
+    assert [c.reason for c in d.conflicts] == ["source project has live workflow data"]
diff --git a/tests/unit/test_catalogue_import_apply.py b/tests/unit/test_catalogue_import_apply.py
new file mode 100644
index 00000000..dace4b2b
--- /dev/null
+++ b/tests/unit/test_catalogue_import_apply.py
@@ -0,0 +1,509 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import csv
+import datetime
+import io
+import os
+import pathlib
+import time
+import unittest.mock as mock
+from collections.abc import AsyncIterator
+
+import pytest
+import quart.datastructures as datastructures
+import sqlalchemy
+import sqlalchemy.event
+import sqlalchemy.ext.asyncio
+import sqlalchemy.pool
+import sqlmodel
+
+import atr.db as db
+import atr.models.sql as sql
+import atr.shared.catalogue_diff as catalogue_diff
+import atr.shared.catalogue_import as catalogue_import
+import atr.shared.catalogue_rows as catalogue_rows
+import atr.storage.writers.catalogue as catalogue
+
+
+@pytest.fixture
+async def sessionmaker() -> AsyncIterator[sqlalchemy.ext.asyncio.async_sessionmaker[db.Session]]:
+    engine = sqlalchemy.ext.asyncio.create_async_engine(
+        "sqlite+aiosqlite:///:memory:",
+        connect_args={"check_same_thread": False},
+        poolclass=sqlalchemy.pool.StaticPool,
+    )
+
+    @sqlalchemy.event.listens_for(engine.sync_engine, "connect")
+    def _fk_on(dbapi_connection, _record) -> None:
+        cursor = dbapi_connection.cursor()
+        cursor.execute("PRAGMA foreign_keys=ON")
+        cursor.close()
+
+    async with engine.begin() as conn:
+        await conn.run_sync(sqlmodel.SQLModel.metadata.create_all)
+    maker = sqlalchemy.ext.asyncio.async_sessionmaker(bind=engine, class_=db.Session, expire_on_commit=False)
+    yield maker
+    await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_export_table_round_trips_through_the_row_models(sessionmaker) -> None:
+    # A downloaded CSV must re-parse cleanly, so the current state can be edited and re-uploaded
+    async with sessionmaker() as data:
+        data.add(sql.Committee(key="alpha", name="Alpha", is_podling=False))
+        data.add(
+            sql.Project(
+                key="alpha-one",
+                name="Apache One",
+                committee_key="alpha",
+                status=sql.ProjectStatus("active"),
+                version_method=sql.VersionMethod("simple"),
+            )
+        )
+        await data.commit()
+        data.add(
+            sql.Release(
+                key="alpha-one-1.0.0",
+                project_key="alpha-one",
+                cycle_key="alpha-one-default",
+                version="1.0.0",
+                phase=sql.ReleasePhase.RELEASE,
+                created=datetime.datetime(2020, 1, 1, tzinfo=datetime.UTC),
+                is_archived=True,
+            )
+        )
+        await data.commit()
+        data.add(
+            sql.Artifact(
+                project_key="alpha-one",
+                version="1.0.0",
+                artifact_path="a.tar.gz",
+                release_key="alpha-one-1.0.0",
+                download_path_suffix="alpha/1.0.0",
+                managed=False,
+            )
+        )
+        await data.commit()
+
+        projects_csv = await catalogue_import.export_table(data, "alpha", "projects")
+        releases_csv = await catalogue_import.export_table(data, "alpha", "releases")
+        artifacts_csv = await catalogue_import.export_table(data, "alpha", "artifacts")
+
+    project = catalogue_rows.ProjectRow.model_validate(next(csv.DictReader(io.StringIO(projects_csv))))
+    release = catalogue_rows.ReleaseRow.model_validate(next(csv.DictReader(io.StringIO(releases_csv))))
+    artifact = catalogue_rows.ArtifactRow.model_validate(next(csv.DictReader(io.StringIO(artifacts_csv))))
+    assert (str(project.key), project.status) == ("alpha-one", sql.ProjectStatus("active"))
+    assert (str(release.key), release.is_archived) == ("alpha-one-1.0.0", True)
+    assert (str(artifact.artifact_path), artifact.managed) == ("a.tar.gz", False)
+
+
+@pytest.mark.asyncio
+async def test_export_leaves_out_projects_with_live_workflow_data(sessionmaker) -> None:
+    # The import refuses those rows, so an export that carried them could not be uploaded again
+    async with sessionmaker() as data:
+        await _seed_two_projects_with_release(data, {"alpha-one": "1.0.0", "alpha-two": "2.0.0"})
+        data.add(
+            sql.Revision(
+                release_key="alpha-two-2.0.0",
+                number="00001",
+                seq=1,
+                asfuid="tester",
+                phase=sql.ReleasePhase.RELEASE_CANDIDATE_DRAFT,
+            )
+        )
+        await data.commit()
+
+        projects_csv = await catalogue_import.export_table(data, "alpha", "projects")
+        releases_csv = await catalogue_import.export_table(data, "alpha", "releases")
+
+    projects = [row["key"] for row in csv.DictReader(io.StringIO(projects_csv))]
+    releases = [row["key"] for row in csv.DictReader(io.StringIO(releases_csv))]
+    assert projects == ["alpha-one"]
+    assert releases == ["alpha-one-1.0.0"]
+
+
+@pytest.mark.asyncio
+async def test_replace_deletes_artifacts_the_file_does_not_list(sessionmaker) -> None:
+    async with sessionmaker() as data:
+        await _seed_two_projects_with_release(data, {"alpha-one": "1.0.0"})
+        for path in ("keep.tgz", "drop.tgz"):
+            data.add(
+                sql.Artifact(
+                    project_key="alpha-one",
+                    version="1.0.0",
+                    artifact_path=path,
+                    release_key="alpha-one-1.0.0",
+                    download_path_suffix=f"alpha/{path}",
+                )
+            )
+        await data.commit()
+
+        keep = _artifact_row("alpha-one", "1.0.0", "keep.tgz")
+        keep["download_path_suffix"] = "alpha/keep.tgz"
+        diff = await _writer(data).import_catalogue_csvs({"artifacts": [keep]}, "alpha", catalogue_diff.Mode.REPLACE)
+
+        # Both are cleared, and only the listed one is re-added
+        assert diff.counts["delete"] == 2
+        assert diff.counts["add"] == 1
+        assert diff.counts["conflict"] == 0
+        assert await data.get(sql.Artifact, ("alpha-one", "1.0.0", "keep.tgz")) is not None
+        assert await data.get(sql.Artifact, ("alpha-one", "1.0.0", "drop.tgz")) is None
+        # The release itself is untouched: releases.csv was not uploaded
+        assert await data.get(sql.Release, "alpha-one-1.0.0") is not None
+
+
+@pytest.mark.asyncio
+async def test_import_hangs_a_new_project_under_its_longest_prefix_sibling(sessionmaker) -> None:
+    # The CSV does not carry super_project_key, so it is derived the way the creation path derives it
+    async with sessionmaker() as data:
+        data.add(sql.Committee(key="alpha", name="Alpha", is_podling=False))
+        data.add(sql.Project(key="alpha", name="Apache Alpha", committee_key="alpha"))
+        await data.commit()
+
+        rows = {"projects": [_project_row("alpha-one", "alpha"), _project_row("alpha-one-extra", "alpha")]}
+        await _writer(data).import_catalogue_csvs(rows, "alpha", catalogue_diff.Mode.ADDITIVE)
+
+        data.expire_all()
+        one = await data.get(sql.Project, "alpha-one")
+        extra = await data.get(sql.Project, "alpha-one-extra")
+        assert one is not None and extra is not None
+        assert one.super_project_key == "alpha"
+        assert extra.super_project_key == "alpha-one"
+
+
+@pytest.mark.asyncio
+async def test_replace_reparents_a_sub_project_it_does_not_delete(sessionmaker) -> None:
+    # Deleting a project clears the super pointer of everything beneath it, so a sub-project the
+    # replace keeps must end up under whatever the file puts back in its parent's place
+    async with sessionmaker() as data:
+        data.add(sql.Committee(key="alpha", name="Alpha", is_podling=False))
+        data.add(sql.Project(key="alpha-one", name="Apache One", committee_key="alpha"))
+        await data.commit()
+        data.add(
+            sql.Project(
+                key="alpha-one-live",
+                name="Apache One Live",
+                committee_key="alpha",
+                super_project_key="alpha-one",
+            )
+        )
+        await data.commit()
+        # A revision makes alpha-one-live managed, so the replace leaves it alone
+        data.add(
+            sql.Release(
+                key="alpha-one-live-1.0.0",
+                project_key="alpha-one-live",
+                cycle_key="alpha-one-live-default",
+                version="1.0.0",
+                phase=sql.ReleasePhase.RELEASE,
+                created=datetime.datetime(2020, 1, 1, tzinfo=datetime.UTC),
+            )
+        )
+        await data.commit()
+        data.add(
+            sql.Revision(
+                release_key="alpha-one-live-1.0.0",
+                number="00001",
+                seq=1,
+                asfuid="tester",
+                phase=sql.ReleasePhase.RELEASE_CANDIDATE_DRAFT,
+            )
+        )
+        await data.commit()
+
+        rows = {"projects": [_project_row("alpha-one", "alpha")]}
+        await _writer(data).import_catalogue_csvs(rows, "alpha", catalogue_diff.Mode.REPLACE)
+
+        data.expire_all()
+        live = await data.get(sql.Project, "alpha-one-live")
+        assert live is not None
+        assert live.super_project_key == "alpha-one"
+
+
+@pytest.mark.asyncio
+async def test_an_artifact_repoint_survives_a_repoint_of_the_release_it_came_from(sessionmaker) -> None:
+    # A repointed release takes its artifacts, so an artifact the file sends elsewhere must leave first
+    async with sessionmaker() as data:
+        await _seed_two_projects_with_release(data, {"alpha-one": "1.0.0"})
+        data.add(sql.Project(key="alpha-three", name="Apache Three", committee_key="alpha"))
+        await data.commit()
+        data.add(
+            sql.Release(
+                key="alpha-three-1.0.0",
+                project_key="alpha-three",
+                cycle_key="alpha-three-default",
+                version="1.0.0",
+                phase=sql.ReleasePhase.RELEASE,
+                created=datetime.datetime(2020, 1, 1, tzinfo=datetime.UTC),
+            )
+        )
+        await data.commit()
+        for path in ("a.tgz", "b.tgz"):
+            data.add(
+                sql.Artifact(
+                    project_key="alpha-one",
+                    version="1.0.0",
+                    artifact_path=path,
+                    release_key="alpha-one-1.0.0",
+                    download_path_suffix=f"alpha/{path}",
+                )
+            )
+        await data.commit()
+
+        release_repoint = _release_row("alpha-one", "1.0.0")
+        release_repoint["project_key"] = "alpha-two"
+        artifact_repoint = _artifact_row("alpha-three", "1.0.0", "a.tgz")
+        artifact_repoint["download_path_suffix"] = "alpha/a.tgz"
+        diff = await _writer(data).import_catalogue_csvs(
+            {"releases": [release_repoint], "artifacts": [artifact_repoint]}, "alpha", catalogue_diff.Mode.ADDITIVE
+        )
+
+        assert (diff.counts["release_repoint"], diff.counts["artifact_repoint"], diff.counts["conflict"]) == (1, 1, 0)
+        data.expire_all()
+        # The repointed artifact goes where the file asked; the other follows its release
+        assert await data.get(sql.Artifact, ("alpha-three", "1.0.0", "a.tgz")) is not None
+        assert await data.get(sql.Artifact, ("alpha-two", "1.0.0", "b.tgz")) is not None
+
+
+@pytest.mark.asyncio
+async def test_import_catalogue_csvs_classifies_and_applies_under_one_lock(sessionmaker) -> None:
+    # The production entry point: build the snapshot, classify, and apply in one transaction
+    async with sessionmaker() as data:
+        data.add(sql.Committee(key="alpha", name="Alpha", is_podling=False))
+        data.add(sql.Project(key="alpha-one", name="Apache One", committee_key="alpha"))
+        await data.commit()
+
+        rows = {
+            "projects": [_project_row("alpha-two", "alpha")],
+            "releases": [_release_row("alpha-two", "1.0.0")],
+            "artifacts": [_artifact_row("alpha-two", "1.0.0", "a.tar.gz")],
+        }
+        diff = await _writer(data).import_catalogue_csvs(rows, "alpha", catalogue_diff.Mode.ADDITIVE)
+
+        assert diff.counts["add"] == 3
+        assert diff.counts["conflict"] == 0
+        assert await data.get(sql.Project, "alpha-two") is not None
+        assert await data.get(sql.Release, "alpha-two-1.0.0") is not None
+        assert await data.get(sql.Artifact, ("alpha-two", "1.0.0", "a.tar.gz")) is not None
+
+
+@pytest.mark.asyncio
+async def test_import_repoints_a_release_and_its_artifact_to_another_project(sessionmaker) -> None:
+    async with sessionmaker() as data:
+        await _seed_two_projects_with_release(data, {"alpha-one": "1.0.0"})
+        data.add(
+            sql.Artifact(
+                project_key="alpha-one",
+                version="1.0.0",
+                artifact_path="a.tar.gz",
+                release_key="alpha-one-1.0.0",
+                download_path_suffix="alpha/1.0.0",
+            )
+        )
+        await data.commit()
+
+        # Keep the existing release key, retarget its project -> classified as a release_repoint
+        move_row = _release_row("alpha-one", "1.0.0")
+        move_row["project_key"] = "alpha-two"
+        diff = await _writer(data).import_catalogue_csvs(
+            {"releases": [move_row]}, "alpha", catalogue_diff.Mode.ADDITIVE
+        )
+
+        assert diff.counts["release_repoint"] == 1
+        assert await data.get(sql.Release, "alpha-two-1.0.0") is not None
+        assert await data.get(sql.Release, "alpha-one-1.0.0") is None
+        artifact = await data.get(sql.Artifact, ("alpha-two", "1.0.0", "a.tar.gz"))
+        assert artifact is not None
+        assert artifact.release_key == "alpha-two-1.0.0"
+
+
+@pytest.mark.asyncio
+async def test_import_refuses_a_move_onto_an_existing_target_release(sessionmaker) -> None:
+    # The target project already holds the version, so re-keying would collide; the release_repoint is a
+    # conflict the preview shows rather than a failure the apply hits
+    async with sessionmaker() as data:
+        await _seed_two_projects_with_release(data, {"alpha-one": "1.0.0", "alpha-two": "1.0.0"})
+
+        move_row = _release_row("alpha-one", "1.0.0")
+        move_row["project_key"] = "alpha-two"
+        diff = await _writer(data).import_catalogue_csvs(
+            {"releases": [move_row]}, "alpha", catalogue_diff.Mode.ADDITIVE
+        )
+
+        assert diff.counts["release_repoint"] == 0
+        assert diff.counts["conflict"] == 1
+        assert await data.get(sql.Release, "alpha-one-1.0.0") is not None
+
+
+@pytest.mark.asyncio
+async def test_import_repoints_an_artifact(sessionmaker) -> None:
+    async with sessionmaker() as data:
+        await _seed_two_projects_with_release(data, {"alpha-one": "1.0.0", "alpha-two": "1.0.0"})
+        data.add(
+            sql.Artifact(
+                project_key="alpha-one",
+                version="1.0.0",
+                artifact_path="a.tar.gz",
+                release_key="alpha-one-1.0.0",
+                download_path_suffix="alpha/1.0.0",
+            )
+        )
+        await data.commit()
+
+        # Same dist identity, different owning project -> classified as a repoint
+        repoint_row = _artifact_row("alpha-two", "1.0.0", "a.tar.gz")
+        repoint_row["download_path_suffix"] = "alpha/1.0.0"
+        diff = await _writer(data).import_catalogue_csvs(
+            {"artifacts": [repoint_row]}, "alpha", catalogue_diff.Mode.ADDITIVE
+        )
+
+        assert diff.counts["artifact_repoint"] == 1
+        assert await data.get(sql.Artifact, ("alpha-two", "1.0.0", "a.tar.gz")) is not None
+        assert await data.get(sql.Artifact, ("alpha-one", "1.0.0", "a.tar.gz")) is None
+
+
+@pytest.mark.asyncio
+async def test_store_uploads_saves_each_picker_and_reads_back(tmp_path: pathlib.Path, monkeypatch) -> None:
+    monkeypatch.setattr(catalogue_import, "IMPORT_ROOT", tmp_path / "imports")
+    files = {
+        "projects": datastructures.FileStorage(
+            stream=io.BytesIO(b"key,committee_key\nalpha-one,alpha\n"), filename="projects.csv"
+        )
+    }
+    token = await catalogue_import.store_uploads(files, "alpha", catalogue_diff.Mode.REPLACE)
+    rows = catalogue_import.read_uploads(token)
+    assert rows["projects"] == [{"key": "alpha-one", "committee_key": "alpha"}]
+    assert catalogue_import.committee_of(token) == "alpha"
+    catalogue_import.discard(token)
+    assert not (catalogue_import.IMPORT_ROOT / token).exists()
+
+
+def test_sweep_removes_stale_upload_dirs_and_keeps_recent_ones(tmp_path: pathlib.Path, monkeypatch) -> None:
+    # A previewed-but-never-applied upload is reclaimed once it ages past the TTL
+    monkeypatch.setattr(catalogue_import, "IMPORT_ROOT", tmp_path / "imports")
+    catalogue_import.IMPORT_ROOT.mkdir()
+    stale = catalogue_import.IMPORT_ROOT / ("a" * 32)
+    recent = catalogue_import.IMPORT_ROOT / ("b" * 32)
+    stale.mkdir()
+    recent.mkdir()
+    aged = time.time() - catalogue_import._UPLOAD_TTL_SECONDS - 60
+    os.utime(stale, (aged, aged))
+    catalogue_import._sweep_stale()
+    assert not stale.exists()
+    assert recent.exists()
+
+
+def test_uploads_reject_a_path_traversal_token(tmp_path: pathlib.Path, monkeypatch) -> None:
+    # The token reaches the path from form input, so anything but a uuid hex string is refused
+    monkeypatch.setattr(catalogue_import, "IMPORT_ROOT", tmp_path / "imports")
+    for bad in ("../../etc", "..", "a/b", "deadbeef", ""):
+        with pytest.raises(KeyError):
+            catalogue_import.read_uploads(bad)
+        with pytest.raises(KeyError):
+            catalogue_import.discard(bad)
+
+
+def _artifact_row(project: str, version: str, path: str) -> dict[str, str]:
+    return {
+        "project_key": project,
+        "version": version,
+        "artifact_path": path,
+        "download_path_suffix": f"{project}/{version}",
+        "managed": "false",
+    }
+
+
+def _project_row(key: str, committee: str) -> dict[str, str]:
+    return {
+        "key": key,
+        "name": f"Apache {key}",
+        "status": "active",
+        "committee_key": committee,
+        "version_method": "simple",
+    }
+
+
+def _release_row(project: str, version: str) -> dict[str, str]:
+    return {
+        "key": f"{project}-{version}",
+        "project_key": project,
+        "version": version,
+        "phase": "release",
+        "created": "2020-01-01",
+        "release_status": "active",
+    }
+
+
+async def _seed_two_projects_with_release(data: db.Session, versioned_projects: dict[str, str]) -> None:
+    data.add(sql.Committee(key="alpha", name="Alpha", is_podling=False))
+    data.add(sql.Project(key="alpha-one", name="Apache One", committee_key="alpha"))
+    data.add(sql.Project(key="alpha-two", name="Apache Two", committee_key="alpha"))
+    await data.commit()
+    for project, version in versioned_projects.items():
+        data.add(
+            sql.Release(
+                key=f"{project}-{version}",
+                project_key=project,
+                cycle_key=f"{project}-default",
+                version=version,
+                phase=sql.ReleasePhase.RELEASE,
+                created=datetime.datetime(2020, 1, 1, tzinfo=datetime.UTC),
+            )
+        )
+    await data.commit()
+
+
+@pytest.mark.asyncio
+async def test_import_refuses_to_write_a_diff_that_was_not_the_one_previewed(sessionmaker) -> None:
+    # The watcher catalogues releases while the page is open, and a replace deletes whatever the
+    # files do not list, so the apply must write only what the admin was shown
+    async with sessionmaker() as data:
+        await _seed_two_projects_with_release(data, {"alpha-one": "1.0.0"})
+        rows = {"releases": [_release_row("alpha-one", "1.0.0")]}
+        snapshot = await catalogue_import.build_snapshot(data, "alpha")
+        previewed = catalogue_diff.fingerprint(
+            catalogue_diff.classify(rows, snapshot, "alpha", catalogue_diff.Mode.REPLACE)
+        )
+
+        # A release lands under the committee between the preview and the apply
+        data.add(
+            sql.Release(
+                key="alpha-two-9.0.0",
+                project_key="alpha-two",
+                cycle_key="alpha-two-default",
+                version="9.0.0",
+                phase=sql.ReleasePhase.RELEASE,
+                created=datetime.datetime(2020, 1, 1, tzinfo=datetime.UTC),
+            )
+        )
+        await data.commit()
+
+        with pytest.raises(catalogue_import.StaleError):
+            await _writer(data).import_catalogue_csvs(rows, "alpha", catalogue_diff.Mode.REPLACE, previewed)
+
+        # The release the preview never showed is still there
+        data.expire_all()
+        assert await data.get(sql.Release, "alpha-two-9.0.0") is not None
+
+
+def _writer(data: db.Session) -> catalogue.FoundationAdmin:
+    writer = object.__new__(catalogue.FoundationAdmin)
+    writer._FoundationAdmin__data = data
+    writer._FoundationAdmin__asf_uid = "tester"
+    writer._FoundationAdmin__write_as = mock.MagicMock()
+    return writer