Merge pull request 'Near-duplicate dedup rebuilt on three gates, backup credential leak closed, library consolidation started' (#255) from dev into main

alembic/versions/0098_phash_256_bit_rehash.py

@@ -0,0 +1,90 @@
+"""Widen image_record.phash to 256-bit and re-hash the library (issue #4223).
+
+The operator reported a 15-image variant pack landing as 3 records, and then
+that variants were STILL being dropped with `phash_threshold` at 0. Zero was
+already the floor of the dial, so no setting could have fixed it: at
+`hash_size=8` a pHash is 64 bits of coarse light/dark layout, and variant
+artwork sharing a composition produces the SAME 64 bits. Distance 0 meant
+"identical hash", never "identical image".
+
+`utils/phash.py` moves to `hash_size=16` (256 bits, what ImageRepo always
+used) and adds an aspect-ratio gate plus a pixel-level confirm, so a merge is
+accepted on the files rather than on the hash.
+
+## Why this NULLs every phash
+
+Widening the column does not correct the values already in it. Every stored
+hash is a 64-bit hash of an image the app will now hash at 256 bits, and the
+two cannot be compared — `find_similar` skips a mismatched-length candidate
+rather than guessing, so leaving them would silently mean "no dedup, forever,
+for everything imported before today". NULL is the state `backfill_phash`
+already knows how to repair: it is NULL-only, keyset-paginated and
+restart-safe, and the beat schedule runs it daily.
+
+Until that backfill finishes, image dedup degrades to sha256 only —
+duplicates may be kept. That is the safe direction, and the only one
+available: the alternative is comparing hashes of different widths, which
+would drop artwork. NOTHING here deletes or supersedes a file.
+
+## Why the threshold is reset rather than carried over
+
+`phash_threshold` counts bits, and the denominator went from 64 to 256. The
+stored number would keep its value while meaning something four times
+tighter. There is no honest carry-over, so every row goes to the new default
+of 24 — including the operator's 0, which was a workaround for the bug this
+revision fixes.
+
+Revision ID: 0098
+Revises: 0097
+Create Date: 2026-09-21
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0098"
+down_revision: Union[str, None] = "0097"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # varchar(32) -> varchar(64): widening a length limit is a catalog-only
+    # change in Postgres, so this does not rewrite the table or its index.
+    op.alter_column(
+        "image_record", "phash",
+        existing_type=sa.String(32),
+        type_=sa.String(64),
+        existing_nullable=True,
+    )
+    op.execute("UPDATE image_record SET phash = NULL WHERE phash IS NOT NULL")
+    op.alter_column(
+        "import_settings", "phash_threshold",
+        existing_type=sa.Integer(),
+        server_default="24",
+        existing_nullable=False,
+    )
+    op.execute("UPDATE import_settings SET phash_threshold = 24")
+
+
+def downgrade() -> None:
+    # The 64-bit hashes this replaced are gone, and a 64-char value does not
+    # fit back into varchar(32) — so the column is cleared again on the way
+    # down and left for backfill_phash to refill at whatever HASH_SIZE the
+    # code is running. Rule #22: no legacy to preserve.
+    op.execute("UPDATE image_record SET phash = NULL WHERE phash IS NOT NULL")
+    op.alter_column(
+        "image_record", "phash",
+        existing_type=sa.String(64),
+        type_=sa.String(32),
+        existing_nullable=True,
+    )
+    op.alter_column(
+        "import_settings", "phash_threshold",
+        existing_type=sa.Integer(),
+        server_default="10",
+        existing_nullable=False,
+    )
+    op.execute("UPDATE import_settings SET phash_threshold = 10")

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})

backend/app/celery_app.py

@@ -120,6 +120,14 @@ def make_celery() -> Celery:
                 "schedule": 86400.0,  # daily — sweep .part/.partial left by a
                 # download/import killed mid-write (graceful-shutdown fallout)
             },
+            "backfill-phash-daily": {
+                "task": "backend.app.tasks.maintenance.backfill_phash",
+                "schedule": 86400.0,  # daily — NULL-only, so a no-op once the
+                # library is hashed. This is what makes migration 0098's
+                # re-hash happen on its own: 0098 NULLs every phash, and
+                # without a scheduled refill the library would sit
+                # dedup-disabled until someone ran a deep scan (#4223).
+            },
             "train-heads-nightly": {
                 "task": "backend.app.tasks.ml.scheduled_train_heads",
                 "schedule": 86400.0,  # passive cadence; manual retrain stays available

backend/app/models/image_record.py

@@ -64,7 +64,11 @@ class ImageRecord(Base):
     # that 0001 also built was an exact duplicate of it — dropped in 0089
     # (#3301). Lookups by sha256 use the constraint's index.
     sha256: Mapped[str] = mapped_column(String(64), nullable=False)
-    phash: Mapped[str | None] = mapped_column(String(32), nullable=True, index=True)
+    # 64 hex chars = the 256-bit hash utils.phash emits at hash_size=16. Was
+    # String(32) (64-bit) until migration 0098; the narrow column was the
+    # reason for the undersized hash, and the undersized hash was collapsing
+    # variant artwork into one record (#4223).
+    phash: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
     size_bytes: Mapped[int] = mapped_column(BigInteger, nullable=False)
     mime: Mapped[str] = mapped_column(String(64), nullable=False)
     width: Mapped[int | None] = mapped_column(Integer, nullable=True)

backend/app/models/import_settings.py

@@ -42,7 +42,12 @@ class ImportSettings(Base):
     single_color_threshold: Mapped[float] = mapped_column(Float, nullable=False, default=0.95, server_default="0.95")
     single_color_tolerance: Mapped[int] = mapped_column(Integer, nullable=False, default=30, server_default="30")
 
-    phash_threshold: Mapped[int] = mapped_column(Integer, nullable=False, default=10, server_default="10")
+    # Hamming distance over a 256-bit pHash (utils.phash, hash_size=16). The
+    # unit CHANGED in migration 0098 — it used to be bits out of 64 — so the
+    # old default of 10 is not this scale's 10, and 0098 resets every row.
+    # This is now the cheap PRE-FILTER: the aspect + pixel gates decide, which
+    # is what lets it be generous enough to catch a re-encoded rescale.
+    phash_threshold: Mapped[int] = mapped_column(Integer, nullable=False, default=24, server_default="24")
 
     # FC-3c downloader knobs
     download_rate_limit_seconds: Mapped[float] = mapped_column(

backend/app/services/backup_service.py

@@ -24,6 +24,25 @@ from pathlib import Path
 
 _BACKUPS_DIRNAME = "_backups"
 
+# Excluded from the images tarball, and each for its own reason (#4233, #4234):
+#
+#   _backups    — the archive would otherwise contain every previous archive.
+#                 This is not hypothetical: the 2026-05-23/24 runs, taken before
+#                 this exclude existed, grew 43G -> 107G -> ... -> 2123G as each
+#                 swallowed its predecessors, and cost 4.3T of the images
+#                 filesystem until they were reclaimed on 2026-09-21.
+#   _quarantine — holds files deliberately pulled OUT of the library.
+#   secrets     — `credential_key.b64`, the key that decrypts the stored
+#                 Patreon/SubscribeStar session credentials.
+#   cookies     — those session cookies themselves.
+#
+# The last two are the ones worth stating plainly: an images tarball is a media
+# archive, and a media archive that carries the key to the operator's accounts
+# is a credential leak wearing a backup's name. Encryption at rest buys nothing
+# when the key rides along in the same file. A restore therefore does NOT
+# re-establish credentials — you sign in again, which is the correct outcome.
+_IMAGES_EXCLUDED_DIRNAMES = ("_backups", "_quarantine", "secrets", "cookies")
+
 # Subprocess-level guardrails BEYOND the Celery soft_time_limit. The Celery
 # soft limit signals the Python process; subprocess.Popen in a blocking syscall
 # ignores that signal, so these bound the worst case directly. Each sits just
@@ -173,8 +192,10 @@ def backup_images(
         [
             "tar", "--zstd", "-cf", str(tar_path),
             "-C", str(images_root.parent), images_root.name,
-            f"--exclude={images_root.name}/_backups",
-            f"--exclude={images_root.name}/_quarantine",
+            *(
+                f"--exclude={images_root.name}/{name}"
+                for name in _IMAGES_EXCLUDED_DIRNAMES
+            ),
         ],
         _IMAGES_SUBPROCESS_TIMEOUT_S,
     )

backend/app/services/gallery_service.py

@@ -322,7 +322,7 @@ def _gallery_images(rows, artists: dict[int, dict]) -> list[GalleryImage]:
     ]
 
 
-def _diversify_similar(src, rows, limit, *, dup_threshold=8, lam=0.40):
+def _diversify_similar(src, rows, limit, *, dup_threshold=32, lam=0.40):
     """Trim a nearest-cosine candidate pool down to `limit` diverse picks.
 
     1. pHash collapse: drop any candidate whose perceptual hash is within
@@ -338,6 +338,11 @@ def _diversify_similar(src, rows, limit, *, dup_threshold=8, lam=0.40):
     2026-07-01 — dropped 0.55→0.40, dup 6→8, paired with a wider pool in
     `similar()`).
 
+    `dup_threshold` counts Hamming bits, so it moved 8→32 when the pHash went
+    from 64 to 256 bits (#4223, migration 0098) — the same fraction of the
+    hash, i.e. the tuning the operator chose, unchanged. This collapse is
+    DISPLAY-only: it hides a near-dup from one rail, it never drops a record.
+
     Falls back to nearest-order (`rows[:limit]`) on any failure or a small pool.
     """
     if len(rows) <= 1:

backend/app/services/importer.py

@@ -33,13 +33,14 @@ from ..models import (
 )
 from ..utils import safe_probe
 from ..utils.paths import (
+    canonical_subdir,
     derive_subdir,
     derive_top_level_artist,
     filehash_from_url,
     hash_suffixed_name,
     safe_ext,
 )
-from ..utils.phash import compute_phash, find_similar
+from ..utils.phash import compute_phash, find_similar, fingerprint_path, fingerprints_match
 from ..utils.sidecar import find_sidecar, parse_sidecar
 from ..utils.slug import slugify
 from .archive_extractor import extract_archive, is_archive
@@ -234,6 +235,38 @@ class Importer:
                 (phash, width or 0, height or 0, image_id)
             )
 
+    def _pixel_confirmer(self, source: Path):
+        """Build `find_similar`'s gate-3 callback for an incoming file.
+
+        pHash proposes; this accepts. A candidate is a duplicate only if its
+        file really is the same picture as `source` at a different size —
+        which is the only merge the operator asked for (#4223). Everything
+        else (a missing file, an unreadable one, a deleted row) returns
+        False: the destructive outcomes here are dropping a download and
+        overwriting a kept file, so an unanswerable question must not read
+        as "yes".
+
+        Both sides' fingerprints are computed lazily and cached, so an
+        import that matches nothing costs no I/O at all and an archive
+        member that keeps hitting the same candidate pays for it once.
+        """
+        new_fp: list = []
+        cand_fps: dict[int, object] = {}
+
+        def confirm(candidate_id: int) -> bool:
+            if not new_fp:
+                new_fp.append(fingerprint_path(source))
+            if new_fp[0] is None:
+                return False
+            if candidate_id not in cand_fps:
+                rec = self.session.get(ImageRecord, candidate_id)
+                cand_fps[candidate_id] = (
+                    fingerprint_path(Path(rec.path)) if rec and rec.path else None
+                )
+            return fingerprints_match(new_fp[0], cand_fps[candidate_id])
+
+        return confirm
+
     def _get_or_create(self, stmt, factory):
         """Race-safe find-or-create. Run `stmt` (scalar_one_or_none); if a
         row exists, return it. Otherwise open a savepoint and INSERT
@@ -862,6 +895,7 @@ class Importer:
                 rel, match_id = find_similar(
                     phash, width or 0, height or 0,
                     candidates, self.settings.phash_threshold,
+                    confirm=self._pixel_confirmer(source),
                 )
                 if rel == "larger_exists":
                     # Enrich-on-duplicate (parity with attach_in_place).
@@ -911,7 +945,7 @@ class Importer:
                 )
                 return ImportResult(status="superseded", image_id=match_id)
 
-        dest = self._copy_to_library(source, sha, attribution_path)
+        dest = self._copy_to_library(source, sha, attribution_path, path_artist)
 
         record = ImageRecord(
             path=str(dest),
@@ -1241,6 +1275,7 @@ class Importer:
                 rel, match_id = find_similar(
                     phash, width or 0, height or 0,
                     candidates, self.settings.phash_threshold,
+                    confirm=self._pixel_confirmer(path),
                 )
                 if rel == "larger_exists":
                     # Enrich-on-duplicate: link the near-dup's post to the
@@ -1566,7 +1601,8 @@ class Importer:
         )
 
     def _copy_to_library(
-        self, source: Path, sha: str, attribution_path: Path
+        self, source: Path, sha: str, attribution_path: Path,
+        artist: Artist | None = None,
     ) -> Path:
         """Copy `source` to its final library path. Returns the destination.
 
@@ -1574,8 +1610,18 @@ class Importer:
         _import_media (filesystem scan) and _supersede (when new_path is
         not passed). FC-3c's attach_in_place skips this helper entirely
         — the file is already at its final home.
+
+        `artist`, when resolved, decides the top-level directory: the
+        library is keyed on the Artist row's slug, NOT on however the
+        import folder happened to be capitalised. Without that, an import
+        from `/import/Conto/` and a download for the same artist write to
+        `Conto/` and `conto/` respectively and the library grows a second
+        home for one artist (milestone #421).
         """
-        subdir = derive_subdir(attribution_path, self.import_root)
+        subdir = canonical_subdir(
+            derive_subdir(attribution_path, self.import_root),
+            artist.slug if artist else None,
+        )
         dest_dir = self.images_root / subdir if subdir else self.images_root
         dest_dir.mkdir(parents=True, exist_ok=True)
         dest_name = hash_suffixed_name(source.stem, sha, source.suffix)
@@ -1610,7 +1656,16 @@ class Importer:
         that path (FC-3c attach_in_place case) — skip the copy step.
         Otherwise the file is copied via _copy_to_library."""
         if new_path is None:
-            dest = self._copy_to_library(source, sha, source)
+            # The KEPT row's artist decides the destination, not the incoming
+            # file's folder — a supersede rewrites `existing.path`, so writing
+            # it anywhere but that artist's canonical directory would move a
+            # row OUT of the tree milestone #421 is consolidating. ImageRecord
+            # carries `artist_id` with no relationship attribute, so this is a
+            # lookup rather than `existing.artist`.
+            kept_artist = artist
+            if kept_artist is None and existing.artist_id is not None:
+                kept_artist = self.session.get(Artist, existing.artist_id)
+            dest = self._copy_to_library(source, sha, source, kept_artist)
         else:
             dest = new_path
 

backend/app/services/library_layout.py

@@ -0,0 +1,231 @@
+"""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.
+
+    `startswith` compiles to LIKE, where `_` and `%` are wildcards, and this
+    does not escape them. That is safe ONLY because `utils.slug.slugify`
+    reduces a slug to `[a-z0-9-]` — neither character can reach the pattern.
+    Widen that charset and this needs `autoescape=True`, or `poch4n_art`
+    starts matching `poch4nXart` too.
+    """
+    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

backend/app/tasks/maintenance.py

@@ -502,10 +502,16 @@ def prune_task_runs() -> dict:
     soft_time_limit=1800, time_limit=2100,
 )
 def backfill_phash() -> int:
-    """Recompute phash for stored images that have none (imported before
-    FC-2d-i+ii). Keyset-paginated by id (restart-safe), NULL-only fill,
-    idempotent. Videos legitimately keep phash NULL. A missing/unreadable
-    file is logged and left NULL — never fails the task."""
+    """Recompute phash for stored images that have none. Keyset-paginated by
+    id (restart-safe), NULL-only fill, idempotent. Videos legitimately keep
+    phash NULL. A missing/unreadable file is logged and left NULL — never
+    fails the task.
+
+    Two sources of NULLs: images imported before FC-2d-i+ii, and migration
+    0098, which cleared every phash so the library could be re-hashed at
+    hash_size=16 (#4223). The daily beat entry exists for the second — until
+    a row is refilled it takes no part in dedup, which is why this runs on a
+    schedule rather than waiting for a deep scan."""
     SessionLocal = _sync_session_factory()
     updated = 0
     last_id = 0

backend/app/utils/paths.py

@@ -60,6 +60,37 @@ def derive_subdir(source_path: Path, import_root: Path) -> str:
     return str(rel) if str(rel) != "." else ""
 
 
+def canonical_subdir(subdir: str, artist_slug: str | None) -> str:
+    """`subdir` with its TOP-LEVEL segment replaced by the artist's slug.
+
+        canonical_subdir("Conto/patreon", "conto")  -> "conto/patreon"
+        canonical_subdir("Conto", "conto")          -> "conto"
+        canonical_subdir("Conto/patreon", None)     -> "Conto/patreon"
+        canonical_subdir("", "conto")               -> ""
+
+    `derive_subdir` mirrors the IMPORT tree's folder names verbatim, so a
+    filesystem import out of `/import/Conto/...` used to write
+    `<images_root>/Conto/...` while the download path wrote `<root>/conto/...`
+    for the very same Artist row. One artist, two directories, forever — 57
+    such families had accumulated by 2026-09-21, and the database never had
+    duplicate artists at all (milestone #421).
+
+    The slug is the canonical name because it is the Artist row's own
+    identifier: it is what `/api/artists` reports, what the ingesters already
+    write, and the one spelling that cannot vary with how a folder happened to
+    be capitalised on the way in.
+
+    Two deliberate pass-throughs. NO artist resolved means there is nothing
+    authoritative to canonicalise against, and an EMPTY subdir is a file
+    landing at the images root — those have no artist folder to correct, and
+    what becomes of them is its own decision (task #4247), not a side effect
+    of this helper.
+    """
+    if not artist_slug or not subdir:
+        return subdir
+    return str(Path(artist_slug, *Path(subdir).parts[1:]))
+
+
 def hash_suffixed_name(stem: str, sha256_hex: str, ext: str) -> str:
     """Builds 'stem__<first10ofhash><ext>'.
 

backend/app/utils/phash.py

@@ -1,55 +1,220 @@
 """Perceptual-hash dedup helpers (ported from ImageRepo).
 
-hash_size=8 -> 64-bit hash -> 16-hex-char string, which fits the existing
-ImageRecord.phash String(32) column (no image_record migration). IR uses
-hash_size=16; the deliberate FC deviation keeps the schema unchanged. The
-Hamming threshold is the operator-exposed dial (ImportSettings).
+hash_size=16 -> 256-bit hash -> 64-hex-char string, which is why
+`ImageRecord.phash` is String(64) (widened in migration 0098).
+
+## Why 16, after running at 8 from FC-2d until 2026-09-21
+
+`hash_size=8` keeps only the top-left 8x8 block of the DCT — 64 bits
+describing an image's coarse light/dark layout and nothing else. Variant
+artwork that shares a composition (same pose and framing, a different
+outfit / expression / overlay) collided OUTRIGHT at that size, so the
+operator's near-duplicate dial could not separate variants from rescales
+even at its floor: `phash_threshold=0` means "the same 64 bits", not "the
+same image", and packs of 15 variants were landing as 3 records
+(issue #4223). IR always used 16; FC's deviation to 8 was made to fit the
+old String(32) column without a migration — a schema convenience that cost
+the operator artwork.
+
+## The hash no longer decides a merge on its own
+
+Dropping a file or superseding one is destructive, so `find_similar` now
+runs three gates, cheapest first:
+
+  1. Hamming distance within the operator's `phash_threshold` — the cheap
+     indexed pre-filter that PROPOSES candidates.
+  2. Aspect ratio within ASPECT_TOL. A crop or a re-canvas is not a
+     rescale, and this is the same identity test the tier-1 video near-dup
+     path already uses (`_VIDEO_DUP_ASPECT_TOL`).
+  3. A pixel-level confirm on the candidate's actual file, supplied by the
+     caller (the importer opens the files; this module stays I/O-free apart
+     from `fingerprint_path`).
+
+Because the confirm is what ACCEPTS a match, the threshold can stay
+generous enough to tolerate a re-encode without putting variants at risk.
+
+Every gate fails CLOSED: unknown dimensions, an unreadable candidate, a
+hash that won't parse — all mean "not a duplicate". The two failure
+directions are not symmetrical. Too strict keeps a redundant lower-res copy,
+which the operator can see and delete; too loose deletes artwork that only
+a re-walk of the source can bring back.
 """
 
 import imagehash
+from PIL import Image, ImageChops, ImageStat
 
-HASH_SIZE = 8
+HASH_SIZE = 16
+
+# Gate 2. Matching `importer._VIDEO_DUP_ASPECT_TOL` — a rescale preserves
+# aspect ratio to within rounding, so this only has to absorb off-by-one
+# pixel dimensions.
+ASPECT_TOL = 0.02
+
+# Gate 3. Both images are reduced to one FINGERPRINT_SIZE-square grayscale
+# thumbnail and compared directly, which is the question actually being
+# asked: "are these the same picture at different resolutions?"
+#
+# Two criteria, because they catch different things. MEAN absolute
+# difference catches a global change (a recolour, a filter, a different
+# shading pass) that leaves the composition intact. The CHANGED-PIXEL
+# fraction catches a LOCAL one — an added overlay, a different expression,
+# an alternate outfit on part of the figure — which a mean over 4096 pixels
+# would otherwise dilute into noise.
+#
+# Tuned to fail closed (see the module docstring). A true rescale lands near
+# zero on both; these ceilings sit well above that and well below a variant.
+FINGERPRINT_SIZE = 64
+FINGERPRINT_MAX_MEAN_DIFF = 6.0
+FINGERPRINT_CHANGED_LEVEL = 64
+FINGERPRINT_MAX_CHANGED_FRACTION = 0.02
+
+
+def _seek_first_frame(pil_image) -> None:
+    """Animated images (multi-frame WebP/GIF/APNG) are hashed and
+    fingerprinted on frame 0 — the conventional choice for animated
+    content, and the one that keeps PIL from iterating every frame."""
+    if getattr(pil_image, "is_animated", False):
+        try:
+            pil_image.seek(0)
+        except Exception:
+            pass
 
 
 def compute_phash(pil_image) -> str | None:
     """Perceptual hash of an opened PIL image, as a hex string. None on any
     failure (videos/unreadable/non-image).
 
-    For animated images (multi-frame WebP/GIF/APNG), explicitly seek to
-    frame 0 first. Without this, some PIL operations downstream of
-    imagehash.phash (convert("L"), resize) can iterate all frames and
-    blow past Celery's hard time limit on large animations
-    (operator-flagged 2026-05-26 against animated WebPs). The pHash of
-    frame 0 is the conventional choice for animated content.
+    Frame 0 for animated images: without the seek, PIL operations
+    downstream of imagehash.phash (convert("L"), resize) can iterate all
+    frames and blow past Celery's hard time limit on large animations
+    (operator-flagged 2026-05-26 against animated WebPs).
     """
     try:
-        if getattr(pil_image, "is_animated", False):
-            try:
-                pil_image.seek(0)
-            except Exception:
-                pass
+        _seek_first_frame(pil_image)
         return str(imagehash.phash(pil_image, hash_size=HASH_SIZE))
     except Exception:
         return None
 
 
+def fingerprint(pil_image):
+    """A small grayscale thumbnail of an opened PIL image, for gate 3.
+
+    Returns a detached PIL image (so the caller may close the original) or
+    None on any failure. PIL-only on purpose: numpy is an imagehash
+    transitive dependency, not a declared one for this path.
+    """
+    try:
+        _seek_first_frame(pil_image)
+        return pil_image.convert("L").resize(
+            (FINGERPRINT_SIZE, FINGERPRINT_SIZE), Image.LANCZOS
+        )
+    except Exception:
+        return None
+
+
+def fingerprint_path(path) -> Image.Image | None:
+    """`fingerprint` for a file on disk. None if it cannot be read — which
+    the gate treats as "not a duplicate"."""
+    try:
+        with Image.open(path) as im:
+            return fingerprint(im)
+    except Exception:
+        return None
+
+
+def fingerprint_diff(
+    a, b, *, changed_level: int = FINGERPRINT_CHANGED_LEVEL,
+) -> tuple[float, float] | None:
+    """(mean absolute difference, fraction of pixels past `changed_level`) for
+    two fingerprints. None if either is missing or the comparison fails.
+
+    This is the MEASUREMENT behind `fingerprints_match`, split out so the
+    calibration report (scripts/phash_gate_report.py) can show how far a pair
+    sat from the limits instead of only which side of them it fell on. The
+    constants were chosen without a real-library sample; the numbers this
+    returns are what moves them.
+    """
+    if a is None or b is None:
+        return None
+    try:
+        diff = ImageChops.difference(a, b)
+        hist = diff.histogram()
+        total = sum(hist)
+        if not total:
+            return None
+        return (ImageStat.Stat(diff).mean[0], sum(hist[changed_level:]) / total)
+    except Exception:
+        return None
+
+
+def fingerprints_match(
+    a, b,
+    *,
+    max_mean_diff: float = FINGERPRINT_MAX_MEAN_DIFF,
+    changed_level: int = FINGERPRINT_CHANGED_LEVEL,
+    max_changed_fraction: float = FINGERPRINT_MAX_CHANGED_FRACTION,
+) -> bool:
+    """True when two fingerprints are the same picture: no large global
+    drift AND no meaningful local region that differs. False on any
+    failure."""
+    measured = fingerprint_diff(a, b, changed_level=changed_level)
+    if measured is None:
+        return False
+    mean, changed_fraction = measured
+    return mean <= max_mean_diff and changed_fraction <= max_changed_fraction
+
+
+def aspect_matches(
+    width: int | None, height: int | None,
+    cand_width: int | None, cand_height: int | None,
+    *, tol: float = ASPECT_TOL,
+) -> bool:
+    """Gate 2. False when either side's dimensions are unknown — an
+    unmeasurable candidate is not a proven duplicate."""
+    if not width or not height or not cand_width or not cand_height:
+        return False
+    a, b = width / height, cand_width / cand_height
+    if a <= 0 or b <= 0:
+        return False
+    return abs(a - b) / max(a, b) <= tol
+
+
 def find_similar(
     phash_hex: str,
     width: int,
     height: int,
     candidates: list[tuple[str, int, int, int]],
     threshold: int,
+    *,
+    confirm=None,
 ) -> tuple[str, int | None]:
     """candidates: (phash_hex, width, height, image_id). Returns one of
     ("none", None) / ("larger_exists", id) / ("smaller_exists", id).
-    First qualifying candidate wins (IR loop order)."""
+    First candidate to pass EVERY gate wins (IR loop order).
+
+    `confirm` is gate 3: an optional callable(image_id) -> bool, called only
+    for a candidate that already passed the hash and aspect gates, and
+    expected to compare the two files' pixels. A rejected candidate does not
+    end the search — the loop moves on, so a false pre-filter hit cannot
+    mask a real duplicate further down the list. Omitting it leaves the
+    hash+aspect behaviour, which is what the unit tests exercise.
+    """
     new_h = imagehash.hex_to_hash(phash_hex)
     for cand_hex, cw, ch, cid in candidates:
         try:
             dist = new_h - imagehash.hex_to_hash(cand_hex)
         except Exception:
+            # Includes the mismatched-length case while a library re-hash
+            # (migration 0098) is still in flight: an old 64-bit hash cannot
+            # be compared to a new 256-bit one, and skipping it degrades to
+            # "no dedup yet" rather than to a wrong merge.
+            continue
+        if dist > threshold:
+            continue
+        if not aspect_matches(width, height, cw, ch):
+            continue
+        if confirm is not None and not confirm(cid):
             continue
-        if dist <= threshold:
         if cw >= width and ch >= height:
             return ("larger_exists", cid)
         if width > cw or height > ch:

frontend/src/components/settings/ImportFiltersForm.vue

@@ -10,16 +10,20 @@
       <div class="fc-phash">
         <div class="fc-phash__title">Near-duplicate sensitivity</div>
         <div class="fc-help mb-1">
-          How aggressively imports merge look-alike images (perceptual-hash
-          distance). <strong>Lower it if edits/variants of the same image are
-          being dropped as duplicates;</strong> raise it to collapse more
-          look-alikes. Applies to new imports.
+          How wide a net imports cast when looking for the same image at a
+          different resolution (perceptual-hash distance, out of 256 bits).
+          A match is only merged if the two files also share an aspect ratio
+          <em>and</em> their pixels agree, so variant artwork — a different
+          outfit, expression or overlay on the same pose — is kept even at a
+          generous setting. <strong>Raise it if higher-resolution re-uploads
+          are landing as separate copies;</strong> lower it to merge less.
+          Applies to new imports.
         </div>
         <v-row no-gutters class="align-center">
           <v-col cols="12" sm="9">
             <v-slider
               v-model="local.phash_threshold"
-              :min="0" :max="16" :step="1"
+              :min="0" :max="64" :step="1"
               :ticks="PHASH_TICKS" show-ticks="always" tick-size="4"
               thumb-label color="accent" hide-details
               class="fc-phash__slider"
@@ -139,8 +143,10 @@ const store = useImportStore()
 // instead of relying on the old tab's mount hook.
 onMounted(() => { if (!store.settings) store.loadSettings() })
 // Labelled stops so the less-initiated get the gist without knowing what a
-// Hamming distance is. 0 = byte-for-byte only; 10 = the shipped default.
-const PHASH_TICKS = { 0: 'Exact', 4: 'Strict', 10: 'Default', 16: 'Loose' }
+// Hamming distance is. 0 = an identical hash only; 24 = the shipped default.
+// Bits out of 256 (utils/phash hash_size=16). The scale changed with
+// migration 0098 — these are NOT the old 0-16 stops renamed (#4223).
+const PHASH_TICKS = { 0: 'Exact', 12: 'Strict', 24: 'Default', 48: 'Loose' }
 // Downloader + schedule-defaults fields moved to
 // /subscriptions?tab=settings (operator decision 2026-05-27). This form
 // now only owns image-import filters.
@@ -148,7 +154,7 @@ const local = reactive({
   min_width: 0, min_height: 0,
   skip_transparent: false, transparency_threshold: 0.9,
   skip_single_color: false, single_color_threshold: 0.95,
-  phash_threshold: 10,
+  phash_threshold: 24,
   wip_title_tagging_enabled: true,
   wip_soft_title_tagging_enabled: false,
 })

scripts/phash_gate_report.py

@@ -0,0 +1,197 @@
+"""What the near-duplicate gates would decide about a folder of real artwork.
+
+Read-only. Opens image files, touches no database, imports nothing from the
+app but `utils/phash.py` itself — so what it reports is what the importer
+would actually do, not a re-implementation that could drift from it.
+
+## Why this exists
+
+Issue #4223 replaced a single 64-bit pHash comparison with three gates
+(threshold -> aspect ratio -> pixel confirm). The threshold is the operator's
+dial and the aspect tolerance mirrors the video path, but the three pixel
+constants — FINGERPRINT_MAX_MEAN_DIFF, FINGERPRINT_CHANGED_LEVEL,
+FINGERPRINT_MAX_CHANGED_FRACTION — were chosen without ever measuring real
+artwork, because FC verifies in CI and CI has only synthetic fixtures. This
+prints the measurements those constants should have been chosen from.
+
+Point it at a folder whose right answer you already know — a variant set that
+should stay whole, or an image you have at two resolutions that should
+collapse — and read the MARGIN column. A pair that lands just inside a limit
+is the one that will flip on the next slightly-different file.
+
+## Usage
+
+    python scripts/phash_gate_report.py <dir> [--threshold N] [--recursive]
+    python scripts/phash_gate_report.py <file_a> <file_b> [...]
+
+No local Python environment needed — run it inside the published image, which
+already has PIL and imagehash, with the folder mounted read-only:
+
+    docker run --rm --entrypoint python -e PYTHONPATH=/app \
+      -v /path/to/art:/data:ro -v "$PWD/scripts":/scripts:ro \
+      git.fabledsword.com/bvandeusen/fabledcurator:dev \
+      /scripts/phash_gate_report.py /data
+
+(The image ships `backend/` at /app but not `scripts/`, hence the second
+mount and PYTHONPATH. The `:ro` on the art folder is the point — this reads.)
+
+`:dev` is the rolling channel this branch publishes to (rule 147) — the same
+bytes that would run in the app, without merging anything to main.
+"""
+
+import argparse
+import itertools
+import sys
+from pathlib import Path
+
+import imagehash
+from PIL import Image
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from backend.app.utils.phash import (  # noqa: E402
+    ASPECT_TOL,
+    FINGERPRINT_MAX_CHANGED_FRACTION,
+    FINGERPRINT_MAX_MEAN_DIFF,
+    HASH_SIZE,
+    aspect_matches,
+    compute_phash,
+    find_similar,
+    fingerprint,
+    fingerprint_diff,
+    fingerprints_match,
+)
+
+IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tif", ".tiff"}
+DEFAULT_THRESHOLD = 24
+
+
+class Shot:
+    """One image, measured once."""
+
+    def __init__(self, path: Path):
+        self.path = path
+        self.phash = None
+        self.width = 0
+        self.height = 0
+        self.fingerprint = None
+        self.error = None
+        try:
+            with Image.open(path) as im:
+                self.width, self.height = im.size
+                self.phash = compute_phash(im)
+                self.fingerprint = fingerprint(im)
+        except Exception as exc:
+            self.error = str(exc)
+
+    @property
+    def ok(self) -> bool:
+        return self.phash is not None and self.fingerprint is not None
+
+    @property
+    def label(self) -> str:
+        return f"{self.path.name} ({self.width}x{self.height})"
+
+
+def collect(paths, recursive: bool) -> list[Path]:
+    found: list[Path] = []
+    for p in paths:
+        p = Path(p)
+        if p.is_dir():
+            walk = p.rglob("*") if recursive else p.glob("*")
+            found += sorted(f for f in walk if f.suffix.lower() in IMAGE_EXTS)
+        elif p.is_file():
+            found.append(p)
+    return found
+
+
+def verdict(a: Shot, b: Shot, threshold: int) -> tuple[str, str]:
+    """Run the REAL find_similar for "a is being imported, b is in the
+    library". Returns (verdict, the gate that decided it)."""
+    rel, _ = find_similar(
+        a.phash, a.width, a.height,
+        [(b.phash, b.width, b.height, 1)],
+        threshold,
+        confirm=lambda _: fingerprints_match(a.fingerprint, b.fingerprint),
+    )
+    if rel == "larger_exists":
+        return ("DROP", f"{a.path.name} dropped; {b.path.name} kept (>= in both)")
+    if rel == "smaller_exists":
+        return ("SUPERSEDE", f"{a.path.name} replaces {b.path.name}'s file")
+    # Not a match — say which gate refused, cheapest first, since that is the
+    # constant to move if the answer is wrong.
+    dist = imagehash.hex_to_hash(a.phash) - imagehash.hex_to_hash(b.phash)
+    if dist > threshold:
+        return ("KEEP BOTH", f"hash distance {dist} > threshold {threshold}")
+    if not aspect_matches(a.width, a.height, b.width, b.height):
+        return ("KEEP BOTH", "aspect ratios differ")
+    return ("KEEP BOTH", "pixels differ")
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
+    ap.add_argument("paths", nargs="+", help="a directory, or two or more files")
+    ap.add_argument("--threshold", type=int, default=DEFAULT_THRESHOLD,
+                    help=f"phash_threshold to simulate (default {DEFAULT_THRESHOLD})")
+    ap.add_argument("--recursive", action="store_true", help="walk subdirectories")
+    ap.add_argument("--limit", type=int, default=60,
+                    help="refuse more than this many images (pairs grow as N^2)")
+    args = ap.parse_args()
+
+    files = collect(args.paths, args.recursive)
+    if len(files) < 2:
+        print(f"Need at least 2 images; found {len(files)}.", file=sys.stderr)
+        return 2
+    if len(files) > args.limit:
+        print(
+            f"{len(files)} images would be {len(files) * (len(files) - 1) // 2} "
+            f"pairs. Narrow the folder or raise --limit.", file=sys.stderr
+        )
+        return 2
+
+    print(f"hash_size={HASH_SIZE} ({HASH_SIZE * HASH_SIZE} bits)  "
+          f"threshold={args.threshold}  aspect_tol={ASPECT_TOL}")
+    print(f"pixel limits: mean <= {FINGERPRINT_MAX_MEAN_DIFF}, "
+          f"changed <= {FINGERPRINT_MAX_CHANGED_FRACTION:.1%}\n")
+
+    shots = [Shot(f) for f in files]
+    for s in shots:
+        if not s.ok:
+            print(f"  ! unreadable, excluded: {s.path.name} — {s.error}")
+    shots = [s for s in shots if s.ok]
+    if len(shots) < 2:
+        print("Not enough readable images.", file=sys.stderr)
+        return 2
+
+    print(f"{'verdict':<10} {'dist':>5} {'mean':>7} {'changed':>8}  pair")
+    print("-" * 78)
+    counts: dict[str, int] = {}
+    rows = []
+    for a, b in itertools.combinations(shots, 2):
+        v, why = verdict(a, b, args.threshold)
+        counts[v] = counts.get(v, 0) + 1
+        dist = imagehash.hex_to_hash(a.phash) - imagehash.hex_to_hash(b.phash)
+        measured = fingerprint_diff(a.fingerprint, b.fingerprint)
+        mean, changed = measured if measured else (float("nan"), float("nan"))
+        rows.append((v, dist, mean, changed, a, b, why))
+
+    # Closest pairs first: the interesting decisions are the near-misses at
+    # both limits, not the obvious strangers at the bottom of the list.
+    for v, dist, mean, changed, a, b, why in sorted(rows, key=lambda r: r[1]):
+        print(f"{v:<10} {dist:>5} {mean:>7.2f} {changed:>7.2%}  "
+              f"{a.label} vs {b.label}")
+        print(f"{'':<10} {'':>5} {'':>7} {'':>8}  -> {why}")
+
+    print("\n" + "  ".join(f"{k}: {n}" for k, n in sorted(counts.items())))
+    print(
+        "\nRead the margin, not just the verdict. A pair you consider the SAME "
+        f"image should sit far under mean {FINGERPRINT_MAX_MEAN_DIFF} / changed "
+        f"{FINGERPRINT_MAX_CHANGED_FRACTION:.0%}; a pair you consider DIFFERENT "
+        "artwork\nshould sit far over. Anything that only just cleared a limit "
+        "is what will flip on the next file, and is the reason to move a constant."
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_api_settings.py

@@ -258,7 +258,7 @@ async def test_patch_rejects_non_object(client):
 async def test_import_settings_phash_threshold_default(client):
     resp = await client.get("/api/settings/import")
     assert resp.status_code == 200
-    assert (await resp.get_json())["phash_threshold"] == 10
+    assert (await resp.get_json())["phash_threshold"] == 24
 
 
 @pytest.mark.asyncio

tests/test_backup_service.py

@@ -134,6 +134,29 @@ def test_backup_images_excludes_backups_and_quarantine(tmp_path, fake_subprocess
     assert any("_quarantine" in e for e in excludes)
 
 
+def test_backup_images_excludes_credentials(tmp_path, fake_subprocess):
+    """#4234: the images tarball carried `secrets/credential_key.b64` — the key
+    that decrypts the stored session cookies — and `cookies/` itself. A media
+    archive must not be a credential leak; encryption at rest is worth nothing
+    if the key travels with the data."""
+    backup_service.backup_images(images_root=tmp_path)
+    excludes = [a for a in fake_subprocess[0] if a.startswith("--exclude=")]
+    assert any(e.endswith("/secrets") for e in excludes)
+    assert any(e.endswith("/cookies") for e in excludes)
+
+
+def test_backup_images_excludes_are_root_relative(tmp_path, fake_subprocess):
+    """tar matches --exclude against the archived path, which is prefixed with
+    the root's own directory name (`-C <parent> <name>`). A bare `secrets`
+    would also match an ARTIST folder called secrets; the prefix is what keeps
+    the exclusion to the top level."""
+    backup_service.backup_images(images_root=tmp_path)
+    excludes = [a for a in fake_subprocess[0] if a.startswith("--exclude=")]
+    assert excludes and all(
+        e.startswith(f"--exclude={tmp_path.name}/") for e in excludes
+    )
+
+
 # --- restore_db ------------------------------------------------------
 
 

tests/test_gallery_similar.py

@@ -146,12 +146,17 @@ async def test_similar_collapses_near_duplicate_phashes(db):
     dupes = []
     for n in range(2, 7):                       # 5 near-identical reposts
         r = await _img(db, n, _vec(1, 0.01 * n))
-        r.phash = "ffffffffffffffff"            # identical perceptual hash
+        r.phash = "f" * 64                      # identical perceptual hash
         dupes.append(r)
+    # 64 hex chars = the 256-bit hash utils.phash emits (#4223, migration
+    # 0098). The widths must match the real ones: _diversify_similar's
+    # dup_threshold moved 8 -> 32 with the hash, so 64-bit fixtures would
+    # now read as near-duplicates of each other and collapse the very rows
+    # this asserts come through.
     distinct_a = await _img(db, 7, _vec(1, 1))
-    distinct_a.phash = "0000000000000000"
+    distinct_a.phash = "0" * 64                 # 256 bits away from the dupes
     distinct_b = await _img(db, 8, _vec(0, 1))
-    distinct_b.phash = "0f0f0f0f0f0f0f0f"
+    distinct_b.phash = "0f" * 32                # 128 bits away
     await db.flush()
 
     res = await GalleryService(db).similar(src.id, limit=10)

tests/test_importer.py

@@ -58,10 +58,36 @@ def test_import_one_happy_path(importer, import_layout):
     result = importer.import_one(src)
     assert result.status == "imported"
     record = importer.session.get(ImageRecord, result.image_id)
-    assert record.path.startswith(str(images_root / "Alice"))
+    # Under the artist's SLUG, not the import folder's "Alice" (milestone
+    # #421) — the import tree's capitalisation is an accident of how the
+    # operator named a folder, and honouring it gave one artist two homes.
+    assert record.path.startswith(str(images_root / "alice"))
     assert record.path.endswith(".jpg")
 
 
+def test_library_path_uses_the_artist_slug_not_the_folder_name(
+    importer, import_layout,
+):
+    """The regression that grew 57 duplicate artist directories: an import
+    from `/import/Conto/...` wrote `<images>/Conto/...` while the downloader
+    wrote `<images>/conto/...` for the same Artist row."""
+    import_root, images_root = import_layout
+    src = import_root / "StickySpoodge" / "patreon" / "a.jpg"
+    _make_jpeg(src)
+    result = importer.import_one(src)
+
+    record = importer.session.get(ImageRecord, result.image_id)
+    artist = importer.session.execute(
+        select(Artist).where(Artist.slug == "stickyspoodge")
+    ).scalar_one()
+    assert artist.name == "StickySpoodge"
+    # Artist segment canonicalised; the post hierarchy below it untouched.
+    assert record.path.startswith(
+        str(images_root / "stickyspoodge" / "patreon")
+    )
+    assert not Path(record.path).is_relative_to(images_root / "StickySpoodge")
+
+
 def test_folder_creates_artist_and_links_artist_id(importer, import_layout):
     # FC-2d-vii-c: folder import creates the Artist and sets the canonical
     # image_record.artist_id — no artist-kind Tag (that path was retired;

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

tests/test_paths.py

@@ -1,6 +1,7 @@
 from pathlib import Path
 
 from backend.app.utils.paths import (
+    canonical_subdir,
     derive_subdir,
     derive_top_level_artist,
     filehash_from_url,
@@ -59,3 +60,38 @@ def test_derive_top_level_artist_nested():
 
 def test_derive_top_level_artist_root():
     assert derive_top_level_artist(Path("/import/x.png"), IMPORT) is None
+
+
+# --- canonical_subdir (milestone #421) --------------------------------------
+
+
+def test_canonical_subdir_replaces_the_top_segment():
+    assert canonical_subdir("Conto/patreon", "conto") == "conto/patreon"
+    assert canonical_subdir("Conto", "conto") == "conto"
+
+
+def test_canonical_subdir_keeps_everything_below_the_artist():
+    """Only the artist segment is authoritative — post folders below it are
+    the downloader's business and must survive untouched."""
+    assert canonical_subdir(
+        "Big Bang/patreon/2026-01-02_123_A Post", "big-bang"
+    ) == "big-bang/patreon/2026-01-02_123_A Post"
+
+
+def test_canonical_subdir_passes_through_without_an_artist():
+    """No resolved artist means nothing authoritative to canonicalise against,
+    so the import folder's own name stands."""
+    assert canonical_subdir("Conto/patreon", None) == "Conto/patreon"
+    assert canonical_subdir("Conto", "") == "Conto"
+
+
+def test_canonical_subdir_leaves_the_images_root_alone():
+    """An empty subdir is a file landing at the root — there is no artist
+    folder to correct, and what becomes of those is task #4247's call."""
+    assert canonical_subdir("", "conto") == ""
+    assert canonical_subdir("", None) == ""
+
+
+def test_canonical_subdir_is_idempotent():
+    once = canonical_subdir("Conto/patreon", "conto")
+    assert canonical_subdir(once, "conto") == once

tests/test_phash_dedup.py

@@ -81,7 +81,7 @@ def test_non_similar_imports_with_phash(importer, import_layout):
     r = importer.import_one(src)
     assert r.status == "imported"
     row = importer.session.get(ImageRecord, r.image_id)
-    assert row.phash is not None and len(row.phash) == 16
+    assert row.phash is not None and len(row.phash) == 64  # 256-bit
 
 
 def test_larger_existing_skips_new_phash_dup(importer, import_layout):
@@ -223,6 +223,74 @@ def test_threshold_controls_match(importer, import_layout):
     assert r.status == "imported"  # threshold 0 + far → independent import
 
 
+# --- The three gates (#4223) ------------------------------------------------
+#
+# Each of these opens the hash gate all the way (threshold 256 = every
+# candidate passes) so the test is about the gate named in its title, and not
+# about whether two fixtures happen to hash apart. That is the regression
+# being guarded: the operator ran the dial down to 0 and STILL lost variants,
+# because the hash was never the thing that could tell them apart.
+
+
+def _wide_open(importer):
+    _set_threshold(importer, 256)
+
+
+def test_variant_survives_a_wide_open_threshold(importer, import_layout):
+    """Same aspect, same size, different picture — only the pixel confirm can
+    save it, and it must."""
+    import_root, _ = import_layout
+    a = import_root / "v.png"
+    _write_split(a, "v", (400, 400))
+    _wide_open(importer)
+    assert importer.import_one(a).status == "imported"
+
+    b = import_root / "h.png"
+    _write_split(b, "h", (400, 400))
+    assert importer.import_one(b).status == "imported"
+    assert importer.session.execute(
+        select(func.count()).select_from(ImageRecord)
+    ).scalar_one() == 2
+
+
+def test_rescale_still_supersedes_at_a_wide_open_threshold(importer, import_layout):
+    """The other half of the deal: the merge the operator DOES want still
+    happens, and keeps the higher resolution."""
+    import_root, _ = import_layout
+    small = import_root / "small.png"
+    _write_split(small, "v", (200, 200))
+    _wide_open(importer)
+    r1 = importer.import_one(small)
+    assert r1.status == "imported"
+
+    big = import_root / "big.png"
+    _write_split(big, "v", (900, 900))
+    r2 = importer.import_one(big)
+    assert r2.status == "superseded"
+    assert r2.image_id == r1.image_id
+
+    importer.session.expire_all()
+    row = importer.session.get(ImageRecord, r1.image_id)
+    assert row.width == 900 and row.height == 900
+
+
+def test_different_aspect_is_never_a_duplicate(importer, import_layout):
+    """Solid colours are pixel-identical once fingerprinted, so the aspect
+    gate is the only thing standing between a crop and a supersede."""
+    import_root, _ = import_layout
+    square = import_root / "square.png"
+    _write(square, (90, 40, 180), (400, 400))
+    _wide_open(importer)
+    assert importer.import_one(square).status == "imported"
+
+    wide = import_root / "wide.png"
+    _write(wide, (90, 40, 180), (800, 400))
+    assert importer.import_one(wide).status == "imported"
+    assert importer.session.execute(
+        select(func.count()).select_from(ImageRecord)
+    ).scalar_one() == 2
+
+
 def test_import_task_maps_superseded_to_complete_and_requeues():
     from backend.app.services.importer import ImportResult
     from backend.app.tasks.import_file import _map_result_to_status

tests/test_phash_util.py

@@ -3,7 +3,13 @@ import io
 import imagehash
 from PIL import Image
 
-from backend.app.utils.phash import compute_phash, find_similar
+from backend.app.utils.phash import (
+    aspect_matches,
+    compute_phash,
+    find_similar,
+    fingerprint,
+    fingerprints_match,
+)
 
 
 def _img(color, size=(64, 64)):
@@ -25,7 +31,7 @@ def _split(orient, size=64):
 def test_compute_phash_stable_and_hex():
     h1 = compute_phash(_img((10, 120, 200)))
     h2 = compute_phash(_img((10, 120, 200)))
-    assert isinstance(h1, str) and len(h1) == 16  # hash_size=8 -> 64-bit -> 16 hex
+    assert isinstance(h1, str) and len(h1) == 64  # hash_size=16 -> 256-bit -> 64 hex
     assert h1 == h2
 
 
@@ -62,3 +68,73 @@ def test_find_similar_threshold_boundary_inclusive_and_first_match():
     assert rel == "larger_exists"
     rel2, _ = find_similar(h, 10, 10, [(far, 999, 999, 1)], threshold=d - 1)
     assert rel2 == "none"
+
+
+# --- Gate 2: aspect ratio (#4223) -------------------------------------------
+
+def test_aspect_matches_tolerates_rounding_but_not_a_crop():
+    assert aspect_matches(1000, 1000, 250, 250)
+    assert aspect_matches(1999, 1000, 1000, 500)     # off-by-one on a rescale
+    assert not aspect_matches(1000, 1000, 1000, 500)  # 1:1 vs 2:1
+    # Unmeasurable is not a proven duplicate — the gate fails closed.
+    assert not aspect_matches(1000, 1000, None, None)
+    assert not aspect_matches(0, 0, 100, 100)
+
+
+def test_find_similar_skips_a_hash_twin_with_a_different_aspect():
+    """A crop or re-canvas keeps the composition, so it can land inside the
+    threshold. Dimensions are what say it is not a rescale."""
+    h = compute_phash(_img((123, 50, 7)))
+    rel, mid = find_similar(h, 100, 100, [(h, 400, 200, 9)], threshold=64)
+    assert rel == "none" and mid is None
+
+
+# --- Gate 3: the pixel confirm (#4223) --------------------------------------
+
+def test_fingerprints_match_across_a_rescale():
+    assert fingerprints_match(
+        fingerprint(_split("v", 64)), fingerprint(_split("v", 512))
+    )
+
+
+def test_fingerprints_reject_a_local_change():
+    """The case the hash cannot see: same composition, one region redrawn."""
+    base = _split("v", 128)
+    variant = base.copy()
+    for y in range(20, 60):
+        for x in range(80, 120):        # a patch on the white half
+            variant.putpixel((x, y), (0, 0, 0))
+    assert not fingerprints_match(fingerprint(base), fingerprint(variant))
+
+
+def test_fingerprints_match_is_false_when_either_side_is_missing():
+    assert not fingerprints_match(fingerprint(_split("v")), None)
+    assert not fingerprints_match(None, None)
+
+
+def test_find_similar_confirm_can_veto_a_hash_and_aspect_match():
+    h = compute_phash(_img((123, 50, 7)))
+    cand = [(h, 400, 400, 9)]
+    rel, mid = find_similar(h, 100, 100, cand, threshold=64, confirm=lambda _: True)
+    assert rel == "larger_exists" and mid == 9
+    rel2, mid2 = find_similar(h, 100, 100, cand, threshold=64, confirm=lambda _: False)
+    assert rel2 == "none" and mid2 is None
+
+
+def test_find_similar_keeps_looking_after_a_veto():
+    """A vetoed candidate must not end the search, or one false pre-filter hit
+    would hide the real duplicate sitting behind it."""
+    h = compute_phash(_img((123, 50, 7)))
+    candidates = [(h, 400, 400, 1), (h, 400, 400, 2)]
+    rel, mid = find_similar(
+        h, 100, 100, candidates, threshold=64, confirm=lambda cid: cid == 2
+    )
+    assert rel == "larger_exists" and mid == 2
+
+
+def test_find_similar_skips_a_hash_of_the_wrong_width():
+    """Mid-re-hash (migration 0098): a leftover 64-bit hash cannot be compared
+    to a 256-bit one. Skipping it degrades to no dedup, never to a merge."""
+    h = compute_phash(_img((123, 50, 7)))
+    rel, mid = find_similar(h, 100, 100, [("ffffffffffffffff", 400, 400, 3)], threshold=64)
+    assert rel == "none" and mid is None