diff --git a/backend/app/api/cleanup.py b/backend/app/api/cleanup.py
index 68e0f00..44cf6d3 100644
--- a/backend/app/api/cleanup.py
+++ b/backend/app/api/cleanup.py
@@ -30,7 +30,7 @@ from sqlalchemy import select
 
 from ..extensions import get_session
 from ..models import LibraryAuditRun
-from ..services import cleanup_service
+from ..services import cleanup_service, library_layout
 from ._responses import error_response as _bad
 
 cleanup_bp = Blueprint("cleanup", __name__, url_prefix="/api/cleanup")
@@ -196,3 +196,25 @@ async def audit_cancel(audit_id: int):
         )
         await session.commit()
     return jsonify({"cancelled": True})
+
+
+@cleanup_bp.route("/layout", methods=["GET"])
+async def layout_survey():
+    """Milestone #421 blast radius: which ImageRecord rows sit outside their
+    artist's canonical slug directory, per artist.
+
+    Read-only. `?check_disk=1` additionally stats every destination to find
+    collisions with a file already there and sources that have gone missing —
+    the numbers the apply refuses on, at the cost of one stat per misplaced
+    row over NFS. It is OFF by default because a count-only pass answers "how
+    big is this" in seconds where the disk pass can run for minutes and time
+    the request out.
+    """
+    check_disk = request.args.get("check_disk", "").lower() in ("1", "true", "yes")
+    async with get_session() as session:
+        report = await session.run_sync(
+            lambda s: library_layout.survey_layout(
+                s, IMAGES_ROOT, check_disk=check_disk,
+            )
+        )
+    return jsonify({**report.as_dict(), "checked_disk": check_disk})
diff --git a/backend/app/services/library_layout.py b/backend/app/services/library_layout.py
new file mode 100644
index 0000000..4e00dc3
--- /dev/null
+++ b/backend/app/services/library_layout.py
@@ -0,0 +1,225 @@
+"""Milestone #421: where an image file BELONGS, and which rows are not there.
+
+The library is keyed on the Artist row's `slug` — one directory per artist.
+It grew a second (and third, and fourth) home for many of them because
+`Importer._copy_to_library` used to name the destination after the IMPORT
+folder while the downloader wrote under the slug. That writer is fixed
+(`utils.paths.canonical_subdir`, task #4244); this module is the other half —
+finding the rows whose files are still in the old places, and saying where
+each one goes.
+
+## Preview and apply share these predicates, they do not re-derive them
+
+`_misplaced_conditions` and `destination_for` are the whole decision. The
+report (task #4245) and the move (task #4246) both spread them rather than
+writing their own — the house shape for rule 93, snippet #3087. A preview
+that computes its set differently from the apply is a preview that can lie,
+and here the apply RENAMES the operator's art.
+
+Nothing in this module writes. It reads rows, and it stats files when asked.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from ..models import Artist, ImageRecord
+from ..utils.paths import canonical_subdir
+
+# Top-level directories under the images root that are STORES, not artists.
+# A sweep that treats these as misplaced artwork would relocate the
+# thumbnail cache, the attachment blobs, or the credential key.
+#   thumbs/      sha-addressed thumbnail cache (NOT path-keyed — see below)
+#   attachments/ sha-addressed non-media blobs
+#   cookies/, secrets/   credential material
+#   _backups/, _quarantine/  backup artifacts; files pulled out of the library
+RESERVED_TOP_LEVEL = frozenset({
+    "thumbs", "attachments", "cookies", "secrets", "_backups", "_quarantine",
+})
+
+
+def canonical_dir(images_root: Path, slug: str) -> Path:
+    """The one directory an artist's files belong under."""
+    return images_root / slug
+
+
+def _misplaced_conditions(images_root: Path, artist_id: int, slug: str) -> list:
+    """Rows of `artist_id` whose file is NOT under that artist's canonical
+    directory. Spread into both halves — never restated.
+
+    The prefix carries a trailing separator on purpose: without it, artist
+    `ara` would match every path under `arbuzbudesh/`, and the sweep would
+    report one artist's whole library as correctly placed while quietly
+    skipping another's.
+    """
+    prefix = f"{canonical_dir(images_root, slug)}/"
+    return [
+        ImageRecord.artist_id == artist_id,
+        ImageRecord.path.is_not(None),
+        ~ImageRecord.path.startswith(prefix),
+    ]
+
+
+def destination_for(path: str, images_root: Path, slug: str) -> Path | None:
+    """Where `path`'s file belongs, or None when this row must not be moved.
+
+    None means: the path is outside the images root, or its top-level segment
+    is a reserved store. Both are refusals rather than errors — a row pointing
+    somewhere unexpected is exactly what should NOT be relocated automatically.
+
+    ## The one place this diverges from `canonical_subdir`
+
+    A file sitting at the images ROOT with a known artist moves under that
+    artist's directory here, where `canonical_subdir` would leave it alone.
+    The two answer different questions. At import time an empty subdir means
+    no artist was resolved, so there is nothing to canonicalise against. Here
+    the row already CARRIES an artist_id, so a file at the root is an anomaly
+    with a known correct home — which is the whole point of the sweep.
+
+    (The 660 unattributed files at the root have no artist_id at all and are
+    not reachable from these predicates; task #4247 decides those.)
+    """
+    p = Path(path)
+    try:
+        rel_dir = p.parent.relative_to(images_root)
+    except ValueError:
+        return None
+    parts = rel_dir.parts
+    if parts and parts[0] in RESERVED_TOP_LEVEL:
+        return None
+    sub = canonical_subdir(str(rel_dir) if str(rel_dir) != "." else "", slug)
+    if not sub:
+        # At the root, with an artist — see the docstring above.
+        return canonical_dir(images_root, slug) / p.name
+    return images_root / sub / p.name
+
+
+@dataclass
+class ArtistLayout:
+    """One artist's verdict."""
+
+    artist_id: int
+    name: str
+    slug: str
+    canonical_rows: int = 0
+    misplaced_rows: int = 0
+    # The non-canonical top-level directories this artist's files sit in —
+    # "Conto", "StickySpoodge", … This is what makes the report readable as
+    # the family list the disk survey found.
+    stray_dirs: list[str] = field(default_factory=list)
+    collisions: list[str] = field(default_factory=list)
+    missing_files: int = 0
+    unmovable: int = 0
+
+
+@dataclass
+class LayoutReport:
+    artists: list[ArtistLayout] = field(default_factory=list)
+    total_rows: int = 0
+    canonical_rows: int = 0
+    misplaced_rows: int = 0
+    collision_count: int = 0
+    missing_files: int = 0
+    unmovable: int = 0
+    unattributed_rows: int = 0
+
+    def as_dict(self) -> dict:
+        return {
+            "total_rows": self.total_rows,
+            "canonical_rows": self.canonical_rows,
+            "misplaced_rows": self.misplaced_rows,
+            "collision_count": self.collision_count,
+            "missing_files": self.missing_files,
+            "unmovable": self.unmovable,
+            "unattributed_rows": self.unattributed_rows,
+            "artists": [
+                {
+                    "artist_id": a.artist_id,
+                    "name": a.name,
+                    "slug": a.slug,
+                    "canonical_rows": a.canonical_rows,
+                    "misplaced_rows": a.misplaced_rows,
+                    "stray_dirs": a.stray_dirs,
+                    "collisions": a.collisions,
+                    "missing_files": a.missing_files,
+                    "unmovable": a.unmovable,
+                }
+                for a in self.artists
+                if a.misplaced_rows or a.collisions
+            ],
+        }
+
+
+def survey_layout(
+    session: Session, images_root: Path, *, check_disk: bool = True,
+) -> LayoutReport:
+    """Read-only blast radius for the consolidation.
+
+    `check_disk` stats every destination to find rows that would collide with
+    a file already there, and sources that have already gone missing. It is
+    the honest number and it is what the apply will refuse on, but it costs
+    one stat per misplaced row over NFS — turn it off when you only want
+    counts.
+    """
+    report = LayoutReport()
+    report.total_rows = session.execute(
+        select(func.count(ImageRecord.id))
+    ).scalar_one()
+    report.unattributed_rows = session.execute(
+        select(func.count(ImageRecord.id)).where(ImageRecord.artist_id.is_(None))
+    ).scalar_one()
+
+    artists = session.execute(
+        select(Artist).order_by(Artist.slug)
+    ).scalars().all()
+
+    for artist in artists:
+        layout = ArtistLayout(
+            artist_id=artist.id, name=artist.name, slug=artist.slug,
+        )
+        conds = _misplaced_conditions(images_root, artist.id, artist.slug)
+        owned = session.execute(
+            select(func.count(ImageRecord.id))
+            .where(ImageRecord.artist_id == artist.id)
+        ).scalar_one()
+        rows = session.execute(
+            select(ImageRecord.id, ImageRecord.path).where(*conds)
+        ).all()
+        layout.misplaced_rows = len(rows)
+        layout.canonical_rows = owned - len(rows)
+
+        strays: set[str] = set()
+        destinations: dict[str, int] = {}
+        for row_id, path in rows:
+            dest = destination_for(path, images_root, artist.slug)
+            if dest is None:
+                layout.unmovable += 1
+                continue
+            try:
+                top = Path(path).parent.relative_to(images_root).parts
+                strays.add(top[0] if top else "<root>")
+            except ValueError:
+                strays.add("<outside>")
+            key = str(dest)
+            if key in destinations:
+                layout.collisions.append(key)
+            else:
+                destinations[key] = row_id
+            if check_disk:
+                if not Path(path).exists():
+                    layout.missing_files += 1
+                elif dest.exists():
+                    layout.collisions.append(key)
+
+        layout.stray_dirs = sorted(strays)
+        report.artists.append(layout)
+        report.canonical_rows += layout.canonical_rows
+        report.misplaced_rows += layout.misplaced_rows
+        report.collision_count += len(layout.collisions)
+        report.missing_files += layout.missing_files
+        report.unmovable += layout.unmovable
+
+    return report
diff --git a/tests/test_library_layout.py b/tests/test_library_layout.py
new file mode 100644
index 0000000..52cd88d
--- /dev/null
+++ b/tests/test_library_layout.py
@@ -0,0 +1,215 @@
+"""Milestone #421 — the shared predicate behind the consolidation.
+
+`destination_for` is pure and tested without a database; `survey_layout` gets
+the integration treatment because the counts are the number the apply is
+checked against.
+"""
+
+from pathlib import Path
+
+import pytest
+from sqlalchemy import select
+
+from backend.app.models import Artist, ImageRecord
+from backend.app.services.library_layout import (
+    RESERVED_TOP_LEVEL,
+    _misplaced_conditions,
+    canonical_dir,
+    destination_for,
+    survey_layout,
+)
+
+ROOT = Path("/images")
+
+
+# --- destination_for (pure) -------------------------------------------------
+
+
+def test_destination_rewrites_only_the_artist_segment():
+    assert destination_for(
+        "/images/Conto/patreon/2026-01_a_Post/x.png", ROOT, "conto"
+    ) == Path("/images/conto/patreon/2026-01_a_Post/x.png")
+
+
+def test_destination_is_identity_for_a_row_already_in_place():
+    p = "/images/conto/patreon/x.png"
+    assert destination_for(p, ROOT, "conto") == Path(p)
+
+
+def test_destination_pulls_a_root_level_row_under_its_artist():
+    """Diverges from canonical_subdir deliberately: the row CARRIES an
+    artist_id, so a file at the root is an anomaly with a known home."""
+    assert destination_for("/images/loose.png", ROOT, "conto") == Path(
+        "/images/conto/loose.png"
+    )
+
+
+def test_destination_refuses_paths_outside_the_images_root():
+    assert destination_for("/srv/elsewhere/x.png", ROOT, "conto") is None
+
+
+@pytest.mark.parametrize("reserved", sorted(RESERVED_TOP_LEVEL))
+def test_destination_refuses_the_reserved_stores(reserved):
+    """Relocating these would move the thumbnail cache, the attachment blobs
+    or the credential key into an artist folder."""
+    assert destination_for(f"/images/{reserved}/aa/x.png", ROOT, "conto") is None
+
+
+def test_destination_is_idempotent():
+    once = destination_for("/images/Conto/patreon/x.png", ROOT, "conto")
+    assert destination_for(str(once), ROOT, "conto") == once
+
+
+# --- the predicate ----------------------------------------------------------
+
+
+def test_canonical_prefix_carries_a_separator():
+    """Without the trailing slash, artist `ara` matches every path under
+    `arbuzbudesh/` — one artist reads as fully placed while another's rows
+    are silently skipped."""
+    conds = _misplaced_conditions(ROOT, 1, "ara")
+    rendered = str(conds[-1].compile(compile_kwargs={"literal_binds": True}))
+    assert "/images/ara/" in rendered
+
+
+# --- survey_layout (integration) --------------------------------------------
+#
+# Marked per-test rather than with a module-level `pytestmark`: the
+# destination_for cases above are pure and belong in the fast unit lane.
+
+
+def _artist(db, name, slug):
+    a = Artist(name=name, slug=slug)
+    db.add(a)
+    db.flush()
+    return a
+
+
+def _image(db, path, artist=None, n=0):
+    rec = ImageRecord(
+        path=path, sha256=f"{n:064d}", size_bytes=1, mime="image/png",
+        width=10, height=10, origin="imported_filesystem",
+        integrity_status="unknown",
+        artist_id=artist.id if artist else None,
+    )
+    db.add(rec)
+    db.flush()
+    return rec
+
+
+@pytest.mark.integration
+def test_survey_splits_canonical_from_misplaced(db_sync):
+    conto = _artist(db_sync, "Conto", "conto")
+    _image(db_sync, "/images/conto/patreon/a.png", conto, 1)
+    _image(db_sync, "/images/Conto/patreon/b.png", conto, 2)
+    _image(db_sync, "/images/Conto/patreon/c.png", conto, 3)
+
+    report = survey_layout(db_sync, ROOT, check_disk=False)
+
+    assert report.misplaced_rows == 2
+    assert report.canonical_rows == 1
+    row = next(a for a in report.artists if a.slug == "conto")
+    assert row.stray_dirs == ["Conto"]
+
+
+@pytest.mark.integration
+def test_survey_does_not_confuse_a_prefix_sharing_artist(db_sync):
+    """`ara` vs `arbuzbudesh` — the reason the predicate anchors on a
+    separator. Both are real artists in the operator's library."""
+    ara = _artist(db_sync, "Ara", "ara")
+    arbuz = _artist(db_sync, "ArbuzBudesh", "arbuzbudesh")
+    _image(db_sync, "/images/ara/x.png", ara, 4)
+    _image(db_sync, "/images/arbuzbudesh/y.png", arbuz, 5)
+
+    report = survey_layout(db_sync, ROOT, check_disk=False)
+
+    assert report.misplaced_rows == 0
+    assert report.canonical_rows == 2
+
+
+@pytest.mark.integration
+def test_survey_counts_two_rows_landing_on_one_destination(db_sync):
+    """A collision is the case the apply must refuse, so the report has to
+    surface it rather than promise a move that cannot happen."""
+    sticky = _artist(db_sync, "StickySpoodge", "stickyspoodge")
+    _image(db_sync, "/images/StickySpoodge/p/dup.png", sticky, 6)
+    _image(db_sync, "/images/Stickyspoodge/p/dup.png", sticky, 7)
+
+    report = survey_layout(db_sync, ROOT, check_disk=False)
+
+    assert report.collision_count == 1
+    row = next(a for a in report.artists if a.slug == "stickyspoodge")
+    assert row.collisions == ["/images/stickyspoodge/p/dup.png"]
+    assert row.stray_dirs == ["StickySpoodge", "Stickyspoodge"]
+
+
+@pytest.mark.integration
+def test_survey_reports_unattributed_rows_without_moving_them(db_sync):
+    """The 660 loose root files have no artist_id, so no predicate reaches
+    them. They are counted, and left for task #4247."""
+    _image(db_sync, "/images/orphan.png", None, 8)
+
+    report = survey_layout(db_sync, ROOT, check_disk=False)
+
+    assert report.unattributed_rows == 1
+    assert report.misplaced_rows == 0
+
+
+@pytest.mark.integration
+def test_survey_refuses_a_row_under_a_reserved_store(db_sync):
+    thumbs = _artist(db_sync, "Thumbsy", "thumbsy")
+    _image(db_sync, "/images/thumbs/aa/weird.png", thumbs, 9)
+
+    report = survey_layout(db_sync, ROOT, check_disk=False)
+
+    assert report.unmovable == 1
+    row = next(a for a in report.artists if a.slug == "thumbsy")
+    assert row.misplaced_rows == 1
+    assert row.collisions == []
+
+
+@pytest.mark.integration
+def test_survey_counts_a_missing_source_file(db_sync, tmp_path):
+    """check_disk is what separates "would move" from "can move"."""
+    gone = _artist(db_sync, "Gone", "gone")
+    _image(db_sync, str(tmp_path / "Gone" / "missing.png"), gone, 10)
+
+    report = survey_layout(db_sync, tmp_path, check_disk=True)
+
+    assert report.missing_files == 1
+
+
+@pytest.mark.integration
+def test_survey_flags_a_destination_that_already_exists(db_sync, tmp_path):
+    occupied = _artist(db_sync, "Occupied", "occupied")
+    src = tmp_path / "Occupied" / "x.png"
+    src.parent.mkdir(parents=True)
+    src.write_bytes(b"src")
+    dest = canonical_dir(tmp_path, "occupied") / "x.png"
+    dest.parent.mkdir(parents=True)
+    dest.write_bytes(b"already here")
+    _image(db_sync, str(src), occupied, 11)
+
+    report = survey_layout(db_sync, tmp_path, check_disk=True)
+
+    assert report.collision_count == 1
+    assert dest.read_bytes() == b"already here"  # read-only: nothing moved
+
+
+@pytest.mark.integration
+def test_survey_is_read_only(db_sync, tmp_path):
+    a = _artist(db_sync, "Reader", "reader")
+    src = tmp_path / "Reader" / "x.png"
+    src.parent.mkdir(parents=True)
+    src.write_bytes(b"x")
+    rec = _image(db_sync, str(src), a, 12)
+    before = rec.path
+
+    survey_layout(db_sync, tmp_path, check_disk=True)
+
+    db_sync.expire_all()
+    assert db_sync.get(ImageRecord, rec.id).path == before
+    assert src.exists()
+    assert db_sync.execute(
+        select(ImageRecord.path).where(ImageRecord.id == rec.id)
+    ).scalar_one() == before