fix(maintenance): protect one file reached through two mounts

Re-review of #313, round 2:

- R1: a removal on another device than its kept copy, with the same
  size and mtimes within 2 s, is protected when its inode is equal or its
  path within its source is the same ("May be the same file reached
  through another mount or share"), in the preview and again right before
  each unlink. The job holds the kept copy open during the unlink, so on
  Windows deleting the kept copy itself through another path fails.
- R2: a file's spellings are found by containment: every configured,
  monitored, ComfyUI-output or listed folder whose stat identity is a
  folder above the file. Parent spellings of nested sources and case
  variants match; favorites move under every listed root holding the
  file. Lookups stay index searches.
- m1: a failed unlink's row says the links now name the kept copy.
- m2: a volume with no file ids reads "Can't confirm this is a separate
  file on this drive".

TESTING.md

@@ -2609,7 +2609,15 @@ find lists a physical file once, by `(st_dev, st_ino)`: a source listed twice un
 spellings, a symlinked source inside another one and a hardlink are one file, never a
 copy of themselves (`test_a_folder_listed_twice_never_loses_its_only_copy`,
 `test_a_symlinked_source_inside_another_is_one_file`,
-`test_a_hardlink_is_never_a_copy_to_remove`). The server holds the groups (content
+`test_a_hardlink_is_never_a_copy_to_remove`). One file reached through two mounts is
+protected, never guessed at: a removal on another `st_dev` than its kept copy, with the
+same size and mtimes within 2 s, is kept as "may be the same file reached through another
+mount or share" when its `st_ino` is equal or its path within its source is the same
+(NFS and SMB, a loopback mount, a FUSE pool: `TestAnotherMount`). Some true copies on
+another disk are therefore never offered. A volume that reports no file id protects
+every pair ("Can't confirm this is a separate file on this drive"). The job checks both
+again right before each unlink and holds the kept copy open meanwhile, so on Windows
+deleting the kept copy itself through another path fails instead. The server holds the groups (content
 hash, root-relative paths, source index, size, mtime and inode) for the last three
 finds; `GET /prompt_manager/maintenance/duplicates/{job_id}?offset&limit` pages
 them ten at a time (`not-found` 404, `job-running` or `no-groups` 409). The preview
@@ -2636,14 +2644,19 @@ move leaves the file and its records alone: no row ever names a deleted file
 (`test_a_failure_between_two_files_leaves_no_row_on_a_deleted_file`). The status line
 counts the files removed and not removed. A changed file is kept (`test_changed_content_skips_file`); the
 first failed unlink stops the job, a locked or read-only file with "The file is open in
-another program or read-only; it was not deleted." and never the OS text, and the
+another program or read-only; it was not deleted, and its links now name the kept copy."
+and never the OS text, and the
 files after it are listed as not attempted (`test_partial_failure_reports_each_path`).
 Records move for the reviewed files only (`database/duplicate_copies.py`): each image
-record, looked up under every spelling of its source (the real path, the configured and
-monitored folders and ComfyUI's output folder that resolve to it:
-`test_links_under_a_symlinked_source_spelling_move`), moves to the kept copy under the
-same spelling, or, when its prompt already links the kept copy, hands its saved-pair
-links to that record and is deleted; a favorite moves with it
+record, looked up under every spelling of the file by containment
+(`py/api/duplicate_spellings.py`: each configured, monitored, ComfyUI output or listed
+folder whose `os.stat` identity is a folder above the file, plus the path below it, so a
+parent spelling of a nested source and a case variant match:
+`test_links_under_a_symlinked_source_spelling_move`,
+`test_a_parent_spelling_of_a_nested_source`), moves to the kept copy under the same
+spelling, or, when its prompt already links the kept copy, hands its saved-pair
+links to that record and is deleted; a favorite under any listed root that holds the
+file moves with it (`test_nested_sources_subfolder_first_with_a_parent_favorite`)
 (`test_db_links_reconciled_only_for_removed_files`); the copy's image tags join the kept
 copy's as their union; and its pending Needs review items (still `pending` at revision
 0 with no decision row) are settled (`test_tags_move_as_a_union_and_pending_items_settle`).

database/duplicate_copies.py

@@ -132,8 +132,9 @@ class DuplicateCopyOperations:
 
         ``move``: ``removed`` and ``kept`` (the spellings each file is linked
         by), ``targets`` (``{removed spelling: kept spelling}``, the spelling
-        a moved record takes; else ``kept_path``), ``favorite`` (``(root key,
-        removed key, kept root key, kept key)`` or None), ``removed_locator``,
+        a moved record takes; else ``kept_path``), ``favorites`` (``(root key,
+        removed key, kept root key, kept key)`` per listed root holding the
+        file), ``removed_locator``,
         ``kept_locator``, ``sha256`` and the kept copy's ``kept_size`` and
         ``kept_mtime_ns``. Returns ``{moved, merged, favorites, tags,
         review}``.
@@ -167,8 +168,8 @@ class DuplicateCopyOperations:
                         ),
                     )
                     counts["moved"] += 1
-            if move.get("favorite"):
-                counts["favorites"] = _move_favorite(conn, *move["favorite"])
+            for favorite in move.get("favorites", ()):
+                counts["favorites"] += _move_favorite(conn, *favorite)
             counts["tags"] = self._move_copy_tags(conn, move)
             for (item_id,) in conn.execute(
                 _PENDING_OF_LOCATOR, (move["removed_locator"],)

py/api/duplicate_files.py

@@ -58,6 +58,8 @@ from .duplicate_review import (  # noqa: F401 (CHANGED, OUTSIDE_SOURCE: public)
     KEEPER_CHANGED,
     KEPT_COPY,
     LOCKED_FILE,
+    MAYBE_SAME_FILE,
+    NO_FILE_ID_ROW,
     NOT_ATTEMPTED,
     OUTSIDE_SOURCE,
     RECORDS_FAILED,
@@ -77,6 +79,7 @@ from .duplicate_review import (  # noqa: F401 (CHANGED, OUTSIDE_SOURCE: public)
     file_problem,
     human_bytes,
 )
+from .duplicate_spellings import Spellings
 from .images import _favorite_relative_key, _favorite_root_key, _output_image_entry
 from .job_routes import busy_response
 from .jobs import (
@@ -458,7 +461,7 @@ class DuplicateFilesMixin:
                 elif record["root_index"] is None:
                     problem = OUTSIDE_SOURCE
                 else:
-                    problem = _bound_problem(record, group, keeper["abs"])
+                    problem = _bound_problem(record, group, keeper)
                 if problem:
                     protected[problem] = protected.get(problem, 0) + 1
                     continue
@@ -472,7 +475,7 @@ class DuplicateFilesMixin:
             records = db.count_copy_records(
                 {
                     t["abs"]: {
-                        "spellings": _file_spellings(spellings, t["root"], t["rel"]),
+                        "spellings": list(spellings.of(t["abs"]).values()),
                         "locator": t["locator"],
                         "sha256": t["hash"],
                     }
@@ -487,15 +490,17 @@ class DuplicateFilesMixin:
         }
 
     def _source_spellings(self):
-        """``{source identity: [spellings]}`` the monitor and config use (blocking).
+        """The folder spellings records may name files by (blocking).
 
-        The search walks each source's real path, while the image monitor
-        links a file under the folder it watches: the configured spelling,
-        ComfyUI's output folder or a symlink to it. Each absolute candidate
-        is listed under the identity (``normcase(realpath)``) it resolves to.
+        The search walks each source's real path, while a record names a file
+        as it was seen when written: under the folder the image monitor
+        watches, the configured spelling, ComfyUI's output folder, a symlink
+        to it, or a parent of a source added later. ``Spellings`` matches
+        them to each file by containment (``py/api/duplicate_spellings.py``).
         """
         from ..config import GalleryConfig
 
+        listed = [os.path.abspath(str(d)) for d in self._get_all_output_dirs()]
         candidates = list(self._monitored_directories())
         candidates += list(GalleryConfig.MONITORING_DIRECTORIES)
         try:
@@ -505,15 +510,10 @@ class DuplicateFilesMixin:
         except Exception:  # nosec B110 - outside ComfyUI there is none
             pass
         candidates.append(self._find_comfyui_output_dir())
-        spellings = {}
-        for candidate in candidates:
-            if not candidate or not os.path.isabs(str(candidate)):
-                continue
-            spelled = os.path.abspath(str(candidate))
-            listed = spellings.setdefault(_identity(spelled), [])
-            if spelled not in listed:
-                listed.append(spelled)
-        return spellings
+        folders = [
+            os.path.abspath(str(c)) for c in candidates if c and os.path.isabs(str(c))
+        ]
+        return Spellings(folders + listed, listed)
 
     def _target(self, group, record, keeper):
         return {
@@ -673,13 +673,22 @@ class DuplicateFilesMixin:
             target["size"],
             hasher,
             keeper=target["keeper_abs"],
+            keeper_root=target["keeper_root"],
         )
         if problem:
             return "kept", problem, None
         try:
-            counts = db.move_copy_records(
-                self._copy_move(target, keepers[group_id], spellings)
-            )
+            # Held open until the unlink: on Windows, deleting the kept copy
+            # itself through another path then fails instead of removing it.
+            hold = open(target["keeper_abs"], "rb")
+        except OSError:
+            return "kept", KEEPER, None
+        with hold:
+            return self._move_then_unlink(db, target, keepers[group_id], spellings)
+
+    def _move_then_unlink(self, db, target, kept_stat, spellings):
+        try:
+            counts = db.move_copy_records(self._copy_move(target, kept_stat, spellings))
         except Exception:
             self.logger.exception(f"Could not move the records of {target['rel']}")
             return "error", RECORDS_FAILED, None
@@ -706,29 +715,40 @@ class DuplicateFilesMixin:
 
     @staticmethod
     def _copy_move(target, kept_stat, spellings):
-        """What ``move_copy_records`` moves for one removal."""
-        removed = _file_spellings(spellings, target["root"], target["rel"])
-        kept = _file_spellings(spellings, target["keeper_root"], target["keeper_rel"])
-        same_source = _identity(target["root"]) == _identity(target["keeper_root"])
-        native_rel = target["rel"].replace("/", os.sep)
-        native_kept = target["keeper_rel"].replace("/", os.sep)
-        targets = {}
-        for spelled in removed:
-            prefix = spelled[: len(spelled) - len(native_rel)]
-            targets[spelled] = (
-                prefix + native_kept if same_source else target["keeper_abs"]
+        """What ``move_copy_records`` moves for one removal.
+
+        A record found under a folder spelling moves to the kept copy under
+        the same spelling when that folder holds it too, else to its real
+        path. Its favorite moves under every listed root holding the file.
+        """
+        removed = spellings.of(target["abs"])
+        kept = spellings.of(target["keeper_abs"])
+        targets = {
+            spelled: kept.get(folder, target["keeper_abs"])
+            for folder, spelled in removed.items()
+        }
+        kept_favorites = spellings.favorites(target["keeper_abs"])
+        favorites = []
+        for root, relative in spellings.favorites(target["abs"]).items():
+            kept_root, kept_rel = (
+                (root, kept_favorites[root])
+                if root in kept_favorites
+                else (target["keeper_root"], target["keeper_rel"])
+            )
+            favorites.append(
+                (
+                    _favorite_root_key(root),
+                    _favorite_relative_key(relative),
+                    _favorite_root_key(kept_root),
+                    _favorite_relative_key(kept_rel),
+                )
             )
         return {
-            "removed": removed,
-            "kept": kept,
+            "removed": list(removed.values()),
+            "kept": list(kept.values()),
             "targets": targets,
             "kept_path": target["keeper_abs"],
-            "favorite": (
-                _favorite_root_key(target["root"]),
-                _favorite_relative_key(target["rel"]),
-                _favorite_root_key(target["keeper_root"]),
-                _favorite_relative_key(target["keeper_rel"]),
-            ),
+            "favorites": favorites,
             "removed_locator": target["locator"],
             "kept_locator": target["keeper_locator"],
             "sha256": target["hash"],
@@ -738,7 +758,10 @@ class DuplicateFilesMixin:
 
 
 def _bound_problem(record, group, keeper=None):
-    """``file_problem`` without reading the file: size, mtime and inode."""
+    """``file_problem`` without reading the file: size, mtime and inode.
+
+    ``keeper`` is the kept copy's record, never offered as a copy of itself.
+    """
     return file_problem(
         record["abs"],
         record["root"],
@@ -747,7 +770,8 @@ def _bound_problem(record, group, keeper=None):
         hash_check=False,
         mtime_ns=record.get("mtime_ns"),
         inode=record.get("inode"),
-        keeper=keeper,
+        keeper=keeper["abs"] if keeper else None,
+        keeper_root=keeper["root"] if keeper else None,
     )
 
 
@@ -766,10 +790,3 @@ def _kept_copy_state(target, hasher):
         return os.stat(target["keeper_abs"])
     except OSError:
         return None
-
-
-def _file_spellings(spellings, root, rel):
-    """Every path one file is linked by: its real path, then each spelling."""
-    native = rel.replace("/", os.sep)
-    roots = [root] + spellings.get(_identity(root), [])
-    return list(dict.fromkeys(os.path.join(spelled, native) for spelled in roots))

py/api/duplicate_review.py

@@ -31,7 +31,20 @@ GONE = "gone"
 UNREADABLE = "unreadable"
 KEEPER = "keeper"
 SAME = "same"
+MAYBE_SAME = "maybe-same"
+NO_FILE_ID = "no-file-id"
+# Two mounts of one share may round mtimes differently (SMB keeps 100 ns,
+# FAT 2 s): within this, equal sizes on another device may be one file.
+MOUNT_MTIME_SLACK_NS = 2_000_000_000
 PROTECTED_REASONS = {
+    MAYBE_SAME: (
+        "file may be the same file reached through another mount or share",
+        "files may be the same file reached through another mount or share",
+    ),
+    NO_FILE_ID: (
+        "file: can't confirm this is a separate file on this drive",
+        "files: can't confirm these are separate files on this drive",
+    ),
     SAME: (
         "file is the same file as its kept copy",
         "files are the same file as their kept copy",
@@ -56,7 +69,13 @@ PROTECTED_REASONS = {
 CHANGED = "The file changed since the review; it was kept."
 KEEPER_CHANGED = "The kept copy changed or is gone; this file was kept."
 SAME_FILE = "This is the same file as its kept copy; it was kept."
+MAYBE_SAME_FILE = (
+    "May be the same file reached through another mount or share; it was kept."
+)
+NO_FILE_ID_ROW = "Can't confirm this is a separate file on this drive; it was kept."
 ROW_DETAILS = {
+    MAYBE_SAME: MAYBE_SAME_FILE,
+    NO_FILE_ID: NO_FILE_ID_ROW,
     SAME: SAME_FILE,
     OUTSIDE_SOURCE: "The file is no longer inside its source; it was kept.",
     LINK: "The file is now a link, not a regular file; it was kept.",
@@ -65,8 +84,12 @@ ROW_DETAILS = {
     UNREADABLE: "The file could not be read; it was kept.",
     KEEPER: KEEPER_CHANGED,
 }
-LOCKED_FILE = "The file is open in another program or read-only; it was not deleted."
-DELETE_FAILED = "The file could not be deleted."
+# An unlink fails only after the file's records moved to its kept copy.
+LOCKED_FILE = (
+    "The file is open in another program or read-only; it was not deleted, "
+    "and its links now name the kept copy."
+)
+DELETE_FAILED = "The file could not be deleted; its links now name the kept copy."
 NOT_ATTEMPTED = "Not attempted: the job stopped after an error."
 KEPT_COPY = "Kept copy"
 SOURCES_CHANGED = (
@@ -119,6 +142,7 @@ def file_problem(
     mtime_ns=None,
     inode=None,
     keeper=None,
+    keeper_root=None,
 ):
     """Why ``path`` may not be treated as a copy in ``root``, or None (blocking).
 
@@ -126,8 +150,9 @@ def file_problem(
     normalized path must lie below the normalized root, outside
     ``thumbnails``, with a media extension. Only then is the file looked at:
     a regular file, not a link, whose resolved path is still inside the
-    resolved root and that is not ``keeper`` itself (``os.path.samestat``: a
-    hardlink, or one file reached through two spellings), of ``size`` bytes.
+    resolved root and that is not ``keeper`` itself (``keeper_problem``: a
+    hardlink, one file reached through two spellings or two mounts, or a
+    volume that cannot tell files apart), of ``size`` bytes.
     ``mtime_ns`` and ``inode`` (``[st_dev, st_ino]``), when given, bind it to
     what the search saw; ``hash_check`` compares its content hash with
     ``digest`` (the removal job, right before an unlink). Every handle is
@@ -158,8 +183,10 @@ def file_problem(
     try:
         if not _lexically_inside(_canonical_path(path), [_canonical_path(root)]):
             return OUTSIDE_SOURCE  # a linked folder on the way
-        if keeper is not None and os.path.samestat(info, os.stat(keeper)):
-            return SAME
+        if keeper is not None:
+            same = keeper_problem(path, root, info, keeper, keeper_root)
+            if same:
+                return same
         if info.st_size != size:
             return CHANGED_CONTENT
         if mtime_ns is not None and info.st_mtime_ns != mtime_ns:
@@ -175,6 +202,48 @@ def file_problem(
     return None
 
 
+def keeper_problem(path, root, info, keeper, keeper_root=None):
+    """Why ``path`` (its ``lstat`` is ``info``) may be ``keeper`` itself, or None.
+
+    - ``NO_FILE_ID``: a volume reports no file id (``st_ino`` 0), so the two
+      cannot be told apart;
+    - ``SAME``: ``os.path.samestat`` (a hardlink, or one file reached through
+      two spellings of one mount);
+    - ``MAYBE_SAME``: another device, the same size, mtimes within
+      ``MOUNT_MTIME_SLACK_NS``, and the same ``st_ino`` (NFS and SMB report
+      the server's file id) or the same path within their sources (one share
+      mounted twice, a FUSE pool over a listed disk). Protect, never guess:
+      a true copy on another disk at the same relative path is kept too.
+
+    Raises OSError when the kept copy cannot be read (the caller's GONE or
+    UNREADABLE).
+    """
+    kept = os.stat(keeper)
+    if not info.st_ino or not kept.st_ino:
+        return NO_FILE_ID
+    if os.path.samestat(info, kept):
+        return SAME
+    if info.st_dev == kept.st_dev or info.st_size != kept.st_size:
+        return None
+    if abs(info.st_mtime_ns - kept.st_mtime_ns) > MOUNT_MTIME_SLACK_NS:
+        return None
+    if info.st_ino == kept.st_ino:
+        return MAYBE_SAME
+    if keeper_root is not None and _relative_key(path, root) == _relative_key(
+        keeper, keeper_root
+    ):
+        return MAYBE_SAME
+    return None
+
+
+def _relative_key(path, root):
+    try:
+        relative = os.path.relpath(str(path), str(root))
+    except ValueError:  # another drive (Windows)
+        return None
+    return os.path.normcase(relative).replace("\\", "/")
+
+
 @dataclass
 class HeldFind:
     """One find: where it looked and, once it finished, what it found."""

py/api/duplicate_spellings.py

@@ -0,0 +1,88 @@
+"""Every path a duplicate copy is recorded under (#249 System S6, Task 17).
+
+The search walks each source's real path, but a record names a file the way
+it was seen when it was written: under the folder the image monitor watches,
+the configured spelling, ComfyUI's output folder, a symlink to it, or a
+parent of a source listed later. ``Spellings`` finds them by containment:
+for each known folder spelling ``S`` (resolving to the folder ``R``) and a
+file ``P`` inside ``R``, the file is also recorded as ``S + relpath(P, R)``.
+
+Folders are compared by ``os.stat`` identity (``st_dev``, ``st_ino``), not
+by text, so a case variant on a case-insensitive filesystem (macOS, Windows)
+or a symlinked spelling matches. A volume that reports no file id falls
+back to ``normcase(realpath)``. Favorites are keyed by a listed root and the
+path below it, so ``favorites`` lists every listed root that holds a file.
+Blocking: each lookup stats the file's folders once (cached).
+"""
+
+import os
+from pathlib import Path
+
+
+def _identity(path):
+    try:
+        info = os.stat(path)
+    except OSError:
+        return None
+    if info.st_ino:
+        return ("inode", info.st_dev, info.st_ino)
+    return ("path", os.path.normcase(os.path.realpath(str(path))))
+
+
+class Spellings:
+    """Folder spellings and listed roots, matched to files by containment."""
+
+    def __init__(self, folders, listed_roots):
+        self._cache = {}
+        self._folders = []
+        for spelled in dict.fromkeys(str(f) for f in folders if f):
+            identity = self._identity(spelled)
+            if identity is not None:
+                self._folders.append((spelled, identity))
+        self._listed = []
+        for root in dict.fromkeys(str(r) for r in listed_roots if r):
+            identity = self._identity(root)
+            if identity is not None:
+                self._listed.append((root, identity))
+
+    def _identity(self, path):
+        if path not in self._cache:
+            self._cache[path] = _identity(path)
+        return self._cache[path]
+
+    def _ancestors(self, path):
+        """``{identity: folder}`` of every folder above ``path``, nearest first."""
+        found = {}
+        folder = os.path.dirname(os.path.abspath(str(path)))
+        while True:
+            identity = self._identity(folder)
+            if identity is not None:
+                found.setdefault(identity, folder)
+            parent = os.path.dirname(folder)
+            if parent == folder:
+                return found
+            folder = parent
+
+    def of(self, path):
+        """``{folder spelling: path spelled under it}``; ``""`` is ``path`` itself."""
+        path = os.path.abspath(str(path))
+        spelled = {"": path}
+        ancestors = self._ancestors(path)
+        for folder, identity in self._folders:
+            inside = ancestors.get(identity)
+            if inside is not None:
+                spelled.setdefault(
+                    folder, os.path.join(folder, os.path.relpath(path, inside))
+                )
+        return spelled
+
+    def favorites(self, path):
+        """``{listed root: path below it (POSIX)}`` for each listed root holding it."""
+        path = os.path.abspath(str(path))
+        ancestors = self._ancestors(path)
+        held = {}
+        for root, identity in self._listed:
+            inside = ancestors.get(identity)
+            if inside is not None:
+                held[root] = Path(os.path.relpath(path, inside)).as_posix()
+        return held

tests/test_api_duplicate_files.py

@@ -110,8 +110,12 @@ class DuplicateFilesTestCase(ApplyTestCase):
         return job, group
 
     @staticmethod
-    def keep_of(group, path):
-        [entry] = [f for f in group["files"] if f["path"] == path]
+    def keep_of(group, path, root_index=None):
+        [entry] = [
+            f
+            for f in group["files"]
+            if f["path"] == path and root_index in (None, f["root_index"])
+        ]
         return {
             "group_id": group["group_id"],
             "root_index": entry["root_index"],

tests/test_api_duplicate_safety.py

@@ -14,13 +14,16 @@ import asyncio
 import hashlib
 import os
 import sqlite3
+import types
 from pathlib import Path
 from unittest.mock import patch
 
 try:
+    from tests.open_handles import open_handles_under
     from tests import test_api_duplicate_files as dup_tests
 except ImportError:  # discovered with tests/ as the top-level directory
     import test_api_duplicate_files as dup_tests
+    from open_handles import open_handles_under
 
 from py.api import duplicate_files  # noqa: E402
 from py.config import GalleryConfig  # noqa: E402
@@ -273,3 +276,201 @@ class TestPreviewDoesNotHash(SafetyTestCase):
         cleanup = await self.apply_plan(plan)
         self.assertEqual(cleanup.counters["removed"], 1)
         self.assertGreaterEqual(len(calls), 2)  # the job hashes keeper and file
+
+
+def _fake_stat(real, **changes):
+    """A stat result with some fields changed (st_dev, st_ino, st_mtime_ns...)."""
+    fields = {name: getattr(real, name) for name in dir(real) if name.startswith("st_")}
+    fields.update(changes)
+    return types.SimpleNamespace(**fields)
+
+
+class TestAnotherMount(SafetyTestCase):
+    """Round 2, R1: one file reached through two mounts is never removed.
+
+    The second source stands for the same folder seen through another
+    filesystem: each test reports its file on another ``st_dev``.
+    """
+
+    MAYBE = "1 file may be the same file reached through another mount or share"
+
+    def setUp(self):
+        super().setUp()
+        self.real_lstat, self.real_stat = os.lstat, os.stat
+
+    def mount(self, rel, keeper_rel, *, ino="same", mtime_shift=0):
+        """Copies at ``output/keeper_rel`` and ``renders/rel``; the patch."""
+        second = self.second_source()
+        GalleryConfig.MONITORING_DIRECTORIES = [
+            self._root(self.output_dir),
+            self._root(second),
+        ]
+        stamp = 1_700_000_000_123_456_789
+        keeper = self.write(keeper_rel, b"one file")
+        other = self.write(rel, b"one file", folder=second)
+        os.utime(keeper, ns=(stamp, stamp))
+        os.utime(other, ns=(stamp, stamp))
+        target = os.path.normcase(str(other))
+        keeper_ino = self.real_stat(keeper).st_ino
+
+        def changed(real):
+            return _fake_stat(
+                real,
+                st_dev=real.st_dev + 7,
+                st_ino=keeper_ino if ino == "same" else real.st_ino + 1000,
+                st_mtime_ns=real.st_mtime_ns + mtime_shift,
+            )
+
+        def lstat(path, *args, **kwargs):
+            real = self.real_lstat(path, *args, **kwargs)
+            return changed(real) if os.path.normcase(str(path)) == target else real
+
+        def stat(path, *args, **kwargs):
+            real = self.real_stat(path, *args, **kwargs)
+            return changed(real) if os.path.normcase(str(path)) == target else real
+
+        patches = (
+            patch("os.lstat", side_effect=lstat),
+            patch("os.stat", side_effect=stat),
+        )
+        return keeper, other, patches
+
+    async def review_with(self, patches, keeper_rel):
+        # The mount reports its device and file id to the find too.
+        with patches[0], patches[1]:
+            job, group = await self.found_group()
+            plan = await self.review(job.id, [self.keep_of(group, keeper_rel, 0)])
+        return job, group, plan
+
+    async def test_nfs_and_smb_spellings_of_one_share_are_protected(self):
+        # NFS and SMB report their own device and file id; SMB rounds mtime.
+        keeper, other, patches = self.mount(
+            "x.png", "x.png", ino="other", mtime_shift=-89
+        )
+        _job, _group, plan = await self.review_with(patches, "x.png")
+        self.assertEqual(plan["scope"]["count"], 0)
+        self.assertEqual(plan["protected"]["reasons"], [self.MAYBE])
+        self.assertTrue(other.is_file() and keeper.is_file())
+
+    async def test_a_loopback_mount_with_the_same_inode_is_protected(self):
+        keeper, other, patches = self.mount("y.png", "sub/y.png", ino="same")
+        _job, _group, plan = await self.review_with(patches, "sub/y.png")
+        self.assertEqual(plan["protected"]["reasons"], [self.MAYBE])
+
+    async def test_a_fuse_pool_path_is_protected(self):
+        # mergerfs, rclone or sshfs synthesize both the device and the inode.
+        keeper, other, patches = self.mount("x.png", "x.png", ino="other")
+        _job, _group, plan = await self.review_with(patches, "x.png")
+        self.assertEqual(plan["protected"]["reasons"], [self.MAYBE])
+
+    async def test_a_true_copy_on_another_disk_is_still_offered(self):
+        keeper, other, patches = self.mount("elsewhere.png", "x.png", ino="other")
+        _job, _group, plan = await self.review_with(patches, "x.png")
+        self.assertEqual(plan["scope"]["count"], 1)
+
+    async def test_the_job_checks_again_before_the_unlink(self):
+        keeper, other, patches = self.mount("x.png", "x.png", ino="other")
+        job, group = await self.found_group()
+        plan = await self.review(job.id, [self.keep_of(group, "x.png", 0)])
+        self.assertEqual(plan["scope"]["count"], 1)  # looked separate then
+        with patches[0], patches[1]:
+            cleanup = await self.apply_plan(plan)
+        self.assertEqual(cleanup.counters["removed"], 0)
+        rows = {row["path"]: row["detail"] for row in cleanup.outcomes["kept"]}
+        self.assertEqual(rows["x.png"], duplicate_files.MAYBE_SAME_FILE)
+        self.assertTrue(other.is_file() and keeper.is_file())
+
+    async def test_a_volume_without_file_ids_is_named_truthfully(self):
+        keeper, other, patches = self.mount("x.png", "x.png")
+
+        def zero(path, *args, **kwargs):
+            return _fake_stat(self.real_stat(path), st_dev=0, st_ino=0)
+
+        with patch("os.lstat", side_effect=zero), patch("os.stat", side_effect=zero):
+            job, group = await self.found_group()
+            plan = await self.review(job.id, [self.keep_of(group, "x.png", 0)])
+        self.assertEqual(
+            plan["protected"]["reasons"],
+            ["1 file: can't confirm this is a separate file on this drive"],
+        )
+
+    async def test_the_kept_copy_is_held_open_during_the_unlink(self):
+        self.trio()
+        job, group = await self.found_group()
+        plan = await self.review(job.id, [self.keep_of(group, "a.png")])
+        real = duplicate_files._unlink
+        held = []
+
+        def unlink(path):
+            held.append(bool(open_handles_under(self.output_dir / "a.png")))
+            real(path)
+
+        if not os.path.isdir("/proc/self/fd"):  # pragma: no cover - no /proc
+            self.skipTest("open handles are read from /proc")
+        with patch.object(duplicate_files, "_unlink", side_effect=unlink):
+            await self.apply_plan(plan)
+        self.assertEqual(held, [True, True])
+
+
+class TestNestedSpellings(SafetyTestCase):
+    """Round 2, R2: spellings and favorites by containment, not equality."""
+
+    async def test_a_parent_spelling_of_a_nested_source(self):
+        # The reviewer's probe: sources [output/sub, out-link -> output].
+        self.need_symlinks()
+        self.write("sub/a.png", b"same", mtime=1)
+        self.write("sub/b.png", b"same", mtime=2)
+        out_link = Path(self.comfy_dir) / "out-link"
+        os.symlink(str(self.output_dir), str(out_link))
+        GalleryConfig.MONITORING_DIRECTORIES = [
+            self._root(self.output_dir / "sub"),
+            str(out_link),
+        ]
+        prompt = self.save("Linked before sub was a source")
+        self.api.db.link_image_to_prompt(prompt, str(out_link / "sub" / "b.png"))
+        job, group = await self.found_group()
+        plan = await self.review(job.id, [self.keep_of(group, "a.png")])
+        self.assertIn("1 database link moves", plan["details"][0]["detail"])
+        await self.apply_plan(plan)
+        [image] = self.api.db.get_prompt_images(prompt)
+        self.assertEqual(
+            _locator(image["image_path"]), _locator(self.output_dir / "sub" / "a.png")
+        )
+        self.assertEqual(self.linked_files(), [])
+
+    async def test_nested_sources_subfolder_first_with_a_parent_favorite(self):
+        self.write("sub/a.png", b"same", mtime=1)
+        self.write("sub/b.png", b"same", mtime=2)
+        GalleryConfig.MONITORING_DIRECTORIES = [
+            self._root(self.output_dir / "sub"),
+            self._root(self.output_dir),
+        ]
+        parent_key = duplicate_files.favorite_root_key(self.output_dir)
+        self.api.db.set_image_favorite(parent_key, "sub/b.png")
+        prompt = self.save("Linked under the parent")
+        self.api.db.link_image_to_prompt(prompt, str(self.output_dir / "sub" / "b.png"))
+        job, group = await self.found_group()
+        self.assertEqual({f["root_index"] for f in group["files"]}, {0})
+        plan = await self.review(job.id, [self.keep_of(group, "a.png")])
+        await self.apply_plan(plan)
+        self.assertEqual(self.api.db.get_root_favorites(parent_key), {"sub/a.png"})
+        self.assertEqual(self.linked_files(), [])
+
+
+class TestLockedRowSaysWhereLinksAre(SafetyTestCase):
+    """Round 2, m1."""
+
+    async def test_the_locked_row_names_the_kept_copy(self):
+        self.trio()
+        job, group = await self.found_group()
+        plan = await self.review(job.id, [self.keep_of(group, "a.png")])
+        with (
+            patch.object(
+                duplicate_files, "_unlink", side_effect=PermissionError(13, "denied")
+            ),
+            self.assertLogs("prompt_manager", "WARNING"),
+        ):
+            cleanup = await self.apply_plan(plan)
+        self.assertIn(
+            "its links now name the kept copy", cleanup.outcomes["errors"][0]["detail"]
+        )