files: unique (directory, stem) after collapsing duplicate FileGroups (#715)

* files: pass find -printf \0 as an escape, not a Python NUL (#716)

* files: pass find -printf \\0 as an escape, not a Python NUL

'%h\\0%f\\0%T@\\0' in a Python string is an embedded null byte.  subprocess
cannot pass that as an argument, so every GNU-find refresh on Linux CI
raised ValueError.  find wants the two-character escape \\0 in the format
string and emits NULs itself.

* files: serialize stem backfill so a cancelled compare cannot lock SQLite

compare_file_groups runs ensure_file_group_stems in a thread.  Cancelling
the task does not stop that thread, and a second compare then writes
wrolpi_kv on the same connection (tests share one session).  A lock
plus a version re-check makes the second call a no-op.

* files: unique (directory, stem) after collapsing duplicate FileGroups

713 added stem and a non-unique index.  This release backfills any NULL
stems, merges leftover duplicates (tags and collection membership move
onto the winner), and creates a UNIQUE index.

Upsert deletes sidecar FileGroups before INSERT so a new primary cannot
collide.  Refresh deletes before upsert for the same reason.
from_paths looks up by stem so adding a sidecar does not insert a second
row.  Same-stem files without a modeled primary stay one FileGroup.

* files: do not use the test SQLite connection from a worker thread

get_db_curs in pytest shares the test session connection.  Running
ensure_file_group_stems and _upsert_files via asyncio.to_thread then
segfaults when the event loop commits that same connection (CI xdist).
Keep to_thread in production, where get_db_curs opens a dedicated
raw connection.

* files: keep tags when a sidecar joins an unmodeled group

choose_primary_file is the single fallback for unmodeled stem groups so
_upsert_files and _upsert_file_groups cannot pick different primaries
and DELETE the tagged row.  FileGroup.move merges an existing
(directory, stem) group and flushes the delete before taking that stem.
refresh_files_directly deletes before upsert, matching the other two
callers.

* files: group by (directory, stem), not stem alone

group_files_by_stem mixed marauder/s3.bin and other/s3.bin into one
FileGroup because it ignored directory.  Unique (directory, stem) then
kept only one of them, so flasher search missed a firmware file.

run_command_streaming awaits the stdout reader instead of cancelling it
when the process exits, so the last lines are not lost on a fast CI run.

* jobs: write the exception into the Job log on failure

test_jobs_failure only saw 'boom' in the log when JobLogHandler captured
logger.error(..., exc_info=e).  Sanic/pytest dictConfig replaces root
handlers (especially under xdist on CI), so that path is empty even
though status and error are set.  Log the exception on the Job itself
and store it in the same _update_job as FAILED.

alembic/versions/2026_09_18_1200_unique_directory_stem.py

@@ -0,0 +1,30 @@
+"""Unique (directory, stem) on file_group.
+
+713 added the stem column and a non-unique index.  This release collapses any
+remaining duplicate (directory, stem) rows (moving tags and collection
+membership onto the winner) and makes the index UNIQUE.
+
+Revision ID: 2026_09_18_1200
+Revises: 2026_09_16_1200
+"""
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = '2026_09_18_1200'
+down_revision = '2026_09_16_1200'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    from wrolpi.files.lib import backfill_null_file_group_stems, dedupe_file_groups_by_directory_stem
+
+    backfill_null_file_group_stems()
+    dedupe_file_groups_by_directory_stem()
+    op.execute('DROP INDEX IF EXISTS file_group_directory_stem_idx')
+    op.execute('CREATE UNIQUE INDEX file_group_directory_stem_idx ON file_group (directory, stem)')
+
+
+def downgrade():
+    op.execute('DROP INDEX IF EXISTS file_group_directory_stem_idx')
+    op.execute('CREATE INDEX IF NOT EXISTS file_group_directory_stem_idx ON file_group (directory, stem)')

wrolpi/cmd.py

@@ -299,7 +299,8 @@ async def run_command(cmd: tuple[str | pathlib.Path, ...], cwd: pathlib.Path | s
                 await proc.wait()
 
             if reader_task:
-                reader_task.cancel()
+                # Drain remaining stdout.  Cancelling here races the last lines on a fast process
+                # (CI often loses "line three" because wait() returns before the reader finishes).
                 try:
                     await reader_task
                 except (CancelledError, asyncio.CancelledError):

wrolpi/common.py

@@ -2382,23 +2382,24 @@ def chunks_by_stem(it: List[Union[pathlib.Path, str, int]], size: int) -> Genera
     from wrolpi.files.lib import split_path_stem_and_suffix
     it = sorted(it.copy())
     index = size
-    last_stem = None
+    last_key = None
     while it:
         if index >= len(it):
             # Ran out of items, yield what is left.
             yield it
             return
         path = it[index]
+        path = pathlib.Path(path)
         stem, _ = split_path_stem_and_suffix(path)
-        if last_stem and stem != last_stem:
-            # Found a break in the path names, yield and reset.
+        key = (str(path.parent), stem)
+        if last_key and key != last_key:
+            # Found a break in directory or stem, yield and reset.
             chunk, it = it[:index], it[index:]
             index = size
-            last_stem = None
+            last_key = None
             yield chunk
             continue
-        # Didn't find a name change, try again.
-        last_stem = stem
+        last_key = key
         index += 1
 
 

wrolpi/files/lib.py

@@ -52,13 +52,14 @@ except ImportError:
 logger = logger.getChild(__name__)
 
 __all__ = ['list_directories_contents', 'delete', 'split_path_stem_and_suffix', 'search_files',
-           'get_mimetype', 'split_file_name_words', 'get_primary_file', 'get_file_statistics',
+           'get_mimetype', 'split_file_name_words', 'get_primary_file', 'choose_primary_file', 'get_file_statistics',
            'search_file_suggestion_count', 'glob_shared_stem', 'upsert_file', 'get_unique_files_by_stem',
            'rename', 'delete_directory', 'handle_file_group_search_results', 'get_file_location_href',
            'get_tagged_file_groups_by_ids', 'delete_file_groups', 'get_special_directories',
            'get_normalized_ignored_directories', 'sanitize_ignored_directories',
            'cached_search_total', 'search_filter_cache_key', 'count_file_groups',
-           'STEM_ALGORITHM_VERSION', 'ensure_file_group_stems']
+           'STEM_ALGORITHM_VERSION', 'ensure_file_group_stems',
+           'backfill_null_file_group_stems', 'dedupe_file_groups_by_directory_stem']
 
 
 def get_file_tag_names(session: Session, file: pathlib.Path) -> List[str]:
@@ -684,6 +685,22 @@ def get_primary_file(files: Union[Tuple[pathlib.Path], Iterable[pathlib.Path]])
     raise NoPrimaryFile(f'Cannot find primary file for group: {files}')
 
 
+def choose_primary_file(files: Union[Tuple[pathlib.Path], Iterable[pathlib.Path]]) -> pathlib.Path:
+    """Primary file for a stem group.  Falls back to a sorted path so every caller agrees.
+
+    `_upsert_files` and `_upsert_file_groups` must pick the same path.  A set's iteration
+    order is not stable; using it as the fallback would DELETE the existing tagged row
+    and INSERT a new untagged one.
+    """
+    paths = [pathlib.Path(p) for p in files]
+    if not paths:
+        raise ValueError('Cannot find primary file without any files')
+    try:
+        return get_primary_file(paths)
+    except NoPrimaryFile:
+        return sorted(paths)[0]
+
+
 def _upsert_files(files: List[pathlib.Path],
                   progress_callback: Callable[[int, int], None] = None):
     """Insert/update all records of the provided files.
@@ -708,33 +725,18 @@ def _upsert_files(files: List[pathlib.Path],
         for group in grouped:
             # The primary file is the video/SingleFile/epub, etc.
             try:
-                try:
-                    primary_path = get_primary_file(group)
-                    # Multiple files in this group.
-                    primary_path: pathlib.Path = primary_path
-                    # The primary mimetype allows modelers to find its file_groups.
-                    mimetype = get_mimetype(primary_path)
-                    # The group uses a common modification_datetime so the group will be re-indexed when any of it's
-                    # files are modified.
-                    modification_datetime = from_timestamp(max(i.stat().st_mtime for i in group))
-                    size = sum(i.stat().st_size for i in group)
-                    files_json = json.dumps(_paths_to_files_dict(group))
-                    stem, suffix = split_path_stem_and_suffix(primary_path)
-                    suffix = (suffix or '').lower() or None
-                    values[primary_path] = (modification_datetime, mimetype, size, files_json, suffix, stem)
-                    non_primary_files.update(i for i in group if i != primary_path)
-                except NoPrimaryFile:
-                    # Cannot find primary path, create a `file_group` for each file.
-                    for primary_path in group:
-                        primary_path: pathlib.Path
-                        # The primary mimetype allows modelers to find its file_groups.
-                        mimetype = get_mimetype(primary_path)
-                        modification_datetime = from_timestamp(primary_path.stat().st_mtime)
-                        size = primary_path.stat().st_size
-                        files_json = json.dumps(_paths_to_files_dict([primary_path]))
-                        stem, suffix = split_path_stem_and_suffix(primary_path)
-                        suffix = (suffix or '').lower() or None
-                        values[primary_path] = (modification_datetime, mimetype, size, files_json, suffix, stem)
+                primary_path = choose_primary_file(group)
+                # The primary mimetype allows modelers to find its file_groups.
+                mimetype = get_mimetype(primary_path)
+                # The group uses a common modification_datetime so the group will be re-indexed when any of it's
+                # files are modified.
+                modification_datetime = from_timestamp(max(i.stat().st_mtime for i in group))
+                size = sum(i.stat().st_size for i in group)
+                files_json = json.dumps(_paths_to_files_dict(group))
+                stem, suffix = split_path_stem_and_suffix(primary_path)
+                suffix = (suffix or '').lower() or None
+                values[primary_path] = (modification_datetime, mimetype, size, files_json, suffix, stem)
+                non_primary_files.update(i for i in group if i != primary_path)
             except FileNotFoundError as e:
                 # A file was deleted between discovery and upsert (temp files, WAL sidecars, etc.).
                 refresh_logger.warning(f'File vanished during refresh, skipping group near {group[0]}', exc_info=e)
@@ -752,6 +754,12 @@ def _upsert_files(files: List[pathlib.Path],
                          '(SELECT value FROM json_each(?))', (json.dumps(primary_paths),))
             previously_indexed = {i['primary_path'] for i in curs.fetchall()}
 
+            if non_primary_files:
+                # Delete before INSERT so a new primary_path that shares (directory, stem)
+                # with an old sidecar FileGroup does not hit the unique index.
+                curs.execute('DELETE FROM file_group WHERE primary_path IN (SELECT value FROM json_each(?))',
+                             (json.dumps(list(map(str, non_primary_files))),))
+
             placeholders, params = values_clause(values)
             stmt = f'''
                 INSERT INTO file_group
@@ -797,11 +805,6 @@ def _upsert_files(files: List[pathlib.Path],
             else:
                 refresh_logger.debug(f'Upserted {len(chunk)} files near {chunk[0]}')
 
-            if non_primary_files:
-                # New files may have been added which change what primary paths exist.  Delete any file_groups which
-                # use paths which are not primary.
-                curs.execute('DELETE FROM file_group WHERE primary_path IN (SELECT value FROM json_each(?))',
-                             (json.dumps(list(map(str, non_primary_files))),))
             non_primary_files.clear()
 
         # Report progress after each chunk
@@ -876,6 +879,84 @@ def ensure_file_group_stems():
         set_stem_algorithm_version(STEM_ALGORITHM_VERSION)
 
 
+def backfill_null_file_group_stems():
+    """Fill file_group.stem for rows that never got one.  Used before creating the unique index."""
+    with get_db_curs() as curs:
+        curs.execute("SELECT id, primary_path FROM file_group WHERE stem IS NULL OR stem = ''")
+        rows = list(curs.fetchall())
+    if not rows:
+        return
+    updates = [(split_path_stem_and_suffix(row['primary_path'])[0], row['id']) for row in rows]
+    with get_db_curs(commit=True) as curs:
+        for chunk in chunks(updates, 500):
+            curs.executemany('UPDATE file_group SET stem = ? WHERE id = ?', chunk)
+
+
+def dedupe_file_groups_by_directory_stem() -> int:
+    """Collapse FileGroups that share (directory, stem).
+
+    Prefers a row with a domain model, then tags, then non-empty files, then highest id.
+    Tags and collection membership on losers are moved onto the winner.  Returns the
+    number of FileGroups deleted.
+    """
+    with get_db_curs() as curs:
+        curs.execute('''
+                     SELECT directory, stem
+                     FROM file_group
+                     WHERE directory IS NOT NULL
+                       AND stem IS NOT NULL
+                       AND stem != ''
+                     GROUP BY directory, stem
+                     HAVING COUNT(*) > 1
+                     ''')
+        keys = [(row['directory'], row['stem']) for row in curs.fetchall()]
+
+    deleted = 0
+    for directory, stem in keys:
+        with get_db_curs() as curs:
+            curs.execute('''
+                         SELECT fg.id,
+                                CASE
+                                    WHEN fg.files IS NULL OR json_type(fg.files) != 'array' THEN 0
+                                    ELSE json_array_length(fg.files)
+                                END AS file_count,
+                                CASE
+                                    WHEN fg.model IS NOT NULL AND fg.model != '' THEN 1
+                                    ELSE 0
+                                END AS has_model,
+                                (SELECT COUNT(*) FROM tag_file tf WHERE tf.file_group_id = fg.id) AS tag_count
+                         FROM file_group fg
+                         WHERE fg.directory = ?
+                           AND fg.stem = ?
+                         ''', (directory, stem))
+            members = list(curs.fetchall())
+        members.sort(key=lambda m: (m['has_model'], m['tag_count'], m['file_count'] > 0, m['id']), reverse=True)
+        winner_id = members[0]['id']
+        loser_ids = [m['id'] for m in members[1:]]
+        with get_db_curs(commit=True) as curs:
+            for loser_id in loser_ids:
+                curs.execute('''
+                             INSERT OR IGNORE INTO tag_file (tag_id, file_group_id, created_at)
+                             SELECT tag_id, ?, created_at
+                             FROM tag_file
+                             WHERE file_group_id = ?
+                             ''', (winner_id, loser_id))
+                curs.execute('''
+                             UPDATE collection_item
+                             SET file_group_id = ?
+                             WHERE file_group_id = ?
+                               AND collection_id NOT IN (SELECT collection_id
+                                                         FROM (SELECT collection_id
+                                                               FROM collection_item
+                                                               WHERE file_group_id = ?))
+                             ''', (winner_id, loser_id, winner_id))
+                curs.execute('DELETE FROM file_group WHERE id = ?', (loser_id,))
+                deleted += 1
+    if deleted:
+        refresh_logger.info(f'Collapsed {deleted} duplicate FileGroups sharing (directory, stem)')
+    return deleted
+
+
 async def apply_indexers(progress_callback: Callable[[int, int], None] = None):
     """Index leftover FileGroups that no modeler claimed.
 
@@ -1419,16 +1500,16 @@ def group_files_by_stem(files: List[pathlib.Path], pre_sorted: bool = False) ->
     files = sorted(files) if not pre_sorted else files.copy()
     file = files.pop(0)
     group = [file, ]
-    prev_stem, _ = split_path_stem_and_suffix(file)
+    prev_dir, prev_stem = file.parent, split_path_stem_and_suffix(file)[0]
     for file in files:
-        stem, suffix = split_path_stem_and_suffix(file)
-        if stem == prev_stem:
+        stem, _ = split_path_stem_and_suffix(file)
+        if stem == prev_stem and file.parent == prev_dir:
             group.append(file)
             continue
-        # Stem has changed, group is finished.
+        # Directory or stem changed, group is finished.
         yield group
         group = [file, ]
-        prev_stem = stem
+        prev_dir, prev_stem = file.parent, stem
     yield group
 
 

wrolpi/files/models.py

@@ -6,7 +6,7 @@ import urllib.parse
 from datetime import datetime
 from typing import List, Type, Optional, Iterable
 
-from sqlalchemy import Column, String, BigInteger, Boolean, event, Index, Integer, JSON
+from sqlalchemy import Column, String, BigInteger, Boolean, event, Index, Integer, JSON, or_
 from sqlalchemy import types
 from sqlalchemy.orm import deferred, relationship, Session
 
@@ -83,8 +83,9 @@ class FileGroup(ModelHelper, Base):
         Index('file_group_size_ix', 'size'),
         Index('file_group_url_idx', 'url'),
         Index('file_group_viewed_idx', 'viewed'),
-        # Speeds compare joins.  Not UNIQUE: duplicates are collapsed at refresh time.
-        Index('file_group_directory_stem_idx', 'directory', 'stem'),
+        # Identity of a FileGroup.  Duplicates are collapsed by
+        # dedupe_file_groups_by_directory_stem before this unique index is created.
+        Index('file_group_directory_stem_idx', 'directory', 'stem', unique=True),
     )
     # SQLite requires exactly "INTEGER PRIMARY KEY" for the rowid alias (FTS5 content_rowid).
     id: int = Column(BigInteger().with_variant(Integer, 'sqlite'), primary_key=True)
@@ -517,18 +518,27 @@ class FileGroup(ModelHelper, Base):
     @classmethod
     def from_paths(cls, session: Session, *paths: pathlib.Path) -> 'FileGroup':
         """Create a new FileGroup which contains the provided file paths."""
-        from wrolpi.files.lib import get_primary_file, get_mimetype, sanitize_filename_surrogates, \
+        from wrolpi.files.lib import choose_primary_file, get_mimetype, sanitize_filename_surrogates, \
             split_path_stem_and_suffix
 
         # Sanitize any paths with invalid UTF-8 characters (renames files on disk if needed)
         paths = tuple(sanitize_filename_surrogates(p) for p in paths)
 
-        existing_groups = session.query(FileGroup).filter(FileGroup.primary_path.in_(list(map(str, paths)))).all()
+        primary_path = choose_primary_file(paths) if paths else None
+        stem = split_path_stem_and_suffix(primary_path)[0] if primary_path else None
+        directory = str(primary_path.parent) if primary_path else None
+        filters = [FileGroup.primary_path.in_(list(map(str, paths)))]
+        if stem:
+            filters.append((FileGroup.directory == directory) & (FileGroup.stem == stem))
+        existing_groups = unique_by_predicate(
+            session.query(FileGroup).filter(or_(*filters)).all(),
+            lambda fg: fg.id,
+        )
         logger.trace(f'FileGroup.from_paths: {len(existing_groups)=}')
         if len(existing_groups) == 0:
             # These paths have not been used previously, create a new FileGroup.
             file_group = FileGroup()
-            primary_path = get_primary_file(paths)
+            primary_path = choose_primary_file(paths)
             file_group.primary_path = primary_path
             file_group.append_files(*paths)
             session.add(file_group)
@@ -536,10 +546,10 @@ class FileGroup(ModelHelper, Base):
             # Found one FileGroup with these paths, no need to create a new FileGroup.
             file_group = existing_groups[0]
             file_group.append_files(*paths)
-            primary_path = get_primary_file(file_group.my_paths())
+            primary_path = choose_primary_file(file_group.my_paths())
         else:
             # Multiple FileGroups contain these paths as primary.
-            primary_path = get_primary_file(paths)
+            primary_path = choose_primary_file(paths)
             file_group: FileGroup = next(filter(lambda i: i.primary_path == primary_path, existing_groups), None)
             if not file_group:
                 file_group = FileGroup.from_paths(session, primary_path)
@@ -689,15 +699,33 @@ class FileGroup(ModelHelper, Base):
 
         new_name, _ = split_path_stem_and_suffix(new_primary_path, full=True)
         new_directory = new_primary_path.parent
+        new_stem, _ = split_path_stem_and_suffix(new_primary_path)
+
+        # Unique (directory, stem): merge into the destination group before touching disk.
+        # Otherwise shutil.move succeeds, then commit hits IntegrityError and disk/DB disagree.
+        session = Session.object_session(self)
+        if session is not None:
+            existing = session.query(FileGroup).filter(
+                FileGroup.directory == str(new_directory),
+                FileGroup.stem == new_stem,
+                FileGroup.id != self.id,
+            ).one_or_none()
+            if existing is not None:
+                self.merge([existing])
+                # Flush the DELETE first.  Otherwise SQLAlchemy UPDATEs our stem before
+                # removing the other row and hits UNIQUE (directory, stem).
+                session.flush()
 
         # Get current files with resolved absolute paths for moving
         current_files = self.my_files()
 
         # Ensure that all destination files do not yet exist before move.
+        # After merging a same-stem group, some dest paths are the files we already own.
         for file in current_files:
-            _, suffix = split_path_stem_and_suffix(file['path'])
+            old_path = file['path']
+            _, suffix = split_path_stem_and_suffix(old_path)
             new_path = pathlib.Path(f'{new_name}{suffix}')
-            if new_path.exists():
+            if new_path.exists() and new_path != old_path:
                 raise FileExistsError(f'Cannot move {self} to {new_path} because it already exists.')
 
         # Move physical files and collect new relative file entries
@@ -707,7 +735,8 @@ class FileGroup(ModelHelper, Base):
             _, suffix = split_path_stem_and_suffix(old_path)
             new_path = pathlib.Path(f'{new_name}{suffix}')
             if old_path.is_file():
-                shutil.move(old_path, new_path)
+                if old_path != new_path:
+                    shutil.move(old_path, new_path)
                 # Store just the filename (relative path)
                 new_files.append({'path': new_path.name, 'mimetype': file['mimetype']})
 

wrolpi/files/test/test_lib.py

@@ -15,6 +15,8 @@ import wrolpi.common
 from modules.videos import Video
 from wrolpi.common import timer, get_wrolpi_config
 from wrolpi.conftest import await_switches
+from sqlalchemy.exc import IntegrityError
+
 from wrolpi.errors import InvalidFile, UnknownDirectory, FileGroupIsTagged, NoPrimaryFile
 from wrolpi.files import lib, indexers
 from wrolpi.files.models import FileGroup, Directory
@@ -496,6 +498,143 @@ def test_ensure_file_group_stems_recomputes_on_version_change(test_session, test
     assert lib.get_stem_algorithm_version() == lib.STEM_ALGORITHM_VERSION
 
 
+def test_file_group_directory_stem_is_unique(test_session, test_directory, make_files_structure):
+    """Two FileGroups in the same directory cannot share a stem."""
+    foo_mp4, foo_png = make_files_structure(['foo.mp4', 'foo.png'])
+    FileGroup.from_paths(test_session, foo_mp4)
+    test_session.commit()
+
+    other = FileGroup(
+        directory=foo_png.parent,
+        primary_path=foo_png,
+        stem='foo',
+        files=[{'path': foo_png.name, 'mimetype': 'image/png'}],
+    )
+    test_session.add(other)
+    with pytest.raises(IntegrityError):
+        test_session.commit()
+
+
+def test_choose_primary_file_is_sorted_and_set_stable(test_directory, make_files_structure):
+    """Unmodeled groups must pick the same primary regardless of set iteration order."""
+    csv, txt = make_files_structure(['n.csv', 'n.txt'])
+    assert lib.choose_primary_file([txt, csv]) == csv
+    assert lib.choose_primary_file({txt, csv}) == csv
+
+
+@pytest.mark.asyncio
+async def test_refresh_keeps_tag_when_sidecar_added(async_client, test_session, test_directory,
+                                                    make_files_structure, tag_factory, refresh_files):
+    """Adding a sidecar to a tagged unmodeled FileGroup must not drop the tag on refresh."""
+    txt, = make_files_structure(['note.txt'])
+    fg = FileGroup.from_paths(test_session, txt)
+    test_session.commit()
+    tag = await tag_factory('keep')
+    fg.add_tag(test_session, tag.id)
+    test_session.commit()
+    fg_id = fg.id
+
+    (test_directory / 'note.csv').write_text('a,b\n')
+    await refresh_files()
+    test_session.expire_all()
+
+    remaining = test_session.query(FileGroup).filter_by(id=fg_id).one()
+    assert remaining.tag_names == ['keep']
+    names = {pathlib.Path(f['path']).name for f in remaining.files}
+    assert names == {'note.txt', 'note.csv'}
+
+
+@pytest.mark.asyncio
+async def test_rename_into_existing_stem_merges(async_client, test_session, test_directory, make_files_structure):
+    """Renaming onto another FileGroup's stem merges before moving files, so disk and DB agree."""
+    make_files_structure({'foo.jpg': 'x', 'bar.mp4': 'video'})
+    jpg = test_directory / 'foo.jpg'
+    mp4 = test_directory / 'bar.mp4'
+    FileGroup.from_paths(test_session, jpg)
+    FileGroup.from_paths(test_session, mp4)
+    test_session.commit()
+    assert test_session.query(FileGroup).count() == 2
+    assert {fg.primary_path.name for fg in test_session.query(FileGroup)} == {'foo.jpg', 'bar.mp4'}
+
+    await lib.rename_file(mp4, 'foo.mp4')
+    test_session.expire_all()
+
+    remaining = test_session.query(FileGroup).one()
+    names = {pathlib.Path(f['path']).name for f in remaining.files}
+    assert names == {'foo.jpg', 'foo.mp4'}
+    assert (test_directory / 'foo.mp4').is_file()
+    assert (test_directory / 'foo.jpg').is_file()
+    assert not (test_directory / 'bar.mp4').exists()
+
+
+@pytest.mark.asyncio
+async def test_dedupe_file_groups_merges_tags_onto_winner(async_client, test_session, test_directory,
+                                                          make_files_structure, tag_factory):
+    """Duplicates sharing (directory, stem) collapse onto one row; tags on the loser move."""
+    from sqlalchemy import text
+
+    foo, sidecar = make_files_structure(['foo.mp4', 'foo.info.json'])
+    test_session.execute(text('DROP INDEX IF EXISTS file_group_directory_stem_idx'))
+
+    keeper = FileGroup.from_paths(test_session, foo)
+    keeper.model = 'video'
+    loser = FileGroup(
+        directory=foo.parent,
+        primary_path=sidecar,
+        stem='foo',
+        files=[{'path': sidecar.name, 'mimetype': 'application/json'}],
+    )
+    test_session.add(loser)
+    test_session.commit()
+
+    tag = await tag_factory('keepme')
+    loser.add_tag(test_session, tag.id)
+    test_session.commit()
+    keeper_id, loser_id = keeper.id, loser.id
+    assert test_session.query(FileGroup).count() == 2
+
+    deleted = lib.dedupe_file_groups_by_directory_stem()
+    test_session.expire_all()
+
+    assert deleted == 1
+    remaining = test_session.query(FileGroup).one()
+    assert remaining.id == keeper_id
+    assert remaining.tag_names == ['keepme']
+    assert test_session.query(FileGroup).filter_by(id=loser_id).one_or_none() is None
+
+
+@pytest.mark.asyncio
+async def test_dedupe_file_groups_prefers_tagged_row(async_client, test_session, test_directory,
+                                                     make_files_structure, tag_factory):
+    """The tagged FileGroup wins even when it has the lower id and no model."""
+    from sqlalchemy import text
+
+    a, b = make_files_structure(['bar.mp4', 'bar.png'])
+    test_session.execute(text('DROP INDEX IF EXISTS file_group_directory_stem_idx'))
+    first = FileGroup.from_paths(test_session, a)
+    second = FileGroup(
+        directory=b.parent,
+        primary_path=b,
+        stem='bar',
+        files=[{'path': b.name, 'mimetype': 'image/png'}],
+        model='video',
+    )
+    test_session.add(second)
+    test_session.commit()
+    assert first.id < second.id
+    tag = await tag_factory('tagged-first')
+    first.add_tag(test_session, tag.id)
+    test_session.commit()
+
+    lib.dedupe_file_groups_by_directory_stem()
+    test_session.expire_all()
+
+    remaining = test_session.query(FileGroup).one()
+    # has_model ranks above tags, so the modeled later row wins; tags move onto it.
+    assert remaining.id == second.id
+    assert remaining.tag_names == ['tagged-first']
+
+
 @pytest.mark.asyncio
 async def test_file_group_tag(test_session, make_files_structure, test_directory, tag_factory, await_switches,
                               refresh_files):
@@ -791,6 +930,13 @@ def test_group_files_by_stem(make_files_structure, test_directory):
          test_directory / 'foo.txt'],
     ]
 
+    make_files_structure(['dir_a/s3.bin', 'dir_b/s3.bin'])
+    same_stem = [test_directory / 'dir_a/s3.bin', test_directory / 'dir_b/s3.bin']
+    assert list(lib.group_files_by_stem(same_stem)) == [
+        [test_directory / 'dir_a/s3.bin'],
+        [test_directory / 'dir_b/s3.bin'],
+    ]
+
 
 def test_get_primary_file(test_directory, video_file, srt_file3, example_epub, example_mobi, example_pdf,
                           singlefile_contents_factory, make_files_structure):
@@ -822,7 +968,7 @@ def test_get_primary_file(test_directory, video_file, srt_file3, example_epub, e
 @pytest.mark.asyncio
 async def test_refresh_files_no_groups(async_client, test_session, test_directory, make_files_structure,
                                        zip_file_factory, refresh_files):
-    """Files that share a name, but cannot be grouped into a FileGroup have their own FileGroups."""
+    """Files that share a stem are one FileGroup even when neither is a modeled primary."""
     foo_txt, foo_zip = make_files_structure({
         'foo.txt': 'text',
         'foo.zip': zip_file_factory(),
@@ -831,11 +977,12 @@ async def test_refresh_files_no_groups(async_client, test_session, test_director
 
     await refresh_files()
 
-    # Two distinct FileGroups.
-    assert test_session.query(FileGroup).count() == 2
-    txt, zip_ = test_session.query(FileGroup)
-    assert txt.primary_path == foo_txt and txt.size == foo_txt.stat().st_size
-    assert zip_.primary_path == foo_zip and zip_.size == foo_zip.stat().st_size
+    # Same stem is one FileGroup even when neither file is a modeled primary.
+    assert test_session.query(FileGroup).count() == 1
+    fg = test_session.query(FileGroup).one()
+    names = {pathlib.Path(f['path']).name for f in fg.files}
+    assert names == {'foo.txt', 'foo.zip'}
+    assert fg.size == foo_txt.stat().st_size + foo_zip.stat().st_size
 
 
 @pytest.mark.asyncio
@@ -890,53 +1037,34 @@ async def test_upsert_directories_chunks_large_batches(test_session, test_direct
 @pytest.mark.asyncio
 async def test_file_group_merge(async_client, test_session, test_directory, make_files_structure, tag_factory,
                                 video_bytes, srt_file3):
-    """A FileGroup can be created from multiple existing FileGroups.  Any Tags applied to the existing groups will be
-    migrated."""
+    """from_paths on a sidecar finds the existing (directory, stem) FileGroup instead of inserting a duplicate."""
     vid, srt = make_files_structure({
         'vid.mp4': video_bytes,
         'vid.srt': (PROJECT_DIR / 'test/example3.en.srt').read_text(),
     })
     one, two = await tag_factory(), await tag_factory()
     vid_group = FileGroup.from_paths(test_session, vid)
-    srt_group = FileGroup.from_paths(test_session, srt)
-    test_session.add_all([vid_group, srt_group])
-    test_session.flush([vid_group, srt_group])
+    test_session.flush([vid_group])
     vid_tag_file = vid_group.add_tag(test_session, one.name)
-    srt_tag_file = srt_group.add_tag(test_session, two.name)
-    test_session.flush([vid_tag_file, srt_tag_file])
     tag_file_created_at = vid_tag_file.created_at
+    test_session.commit()
+
+    srt_group = FileGroup.from_paths(test_session, srt)
+    assert srt_group.id == vid_group.id
+    srt_tag_file = srt_group.add_tag(test_session, two.name)
     srt_file_created_at = srt_tag_file.created_at
     test_session.commit()
 
-    assert vid_group.mimetype == 'video/mp4'
-    assert srt_group.mimetype == 'text/srt'
-
-    # Add srt to vid_group so both groups have it (simulates race condition scenario)
-    vid_group.append_files(srt)
+    merged = FileGroup.from_paths(test_session, vid, srt)
     test_session.commit()
-
-    # Verify both groups have the srt file
-    assert 'vid.srt' in [f['path'] for f in vid_group.files]
-    assert 'vid.srt' in [f['path'] for f in srt_group.files]
-
-    # Both FileGroups are merged.
-    vid = FileGroup.from_paths(test_session, vid, srt)
-    test_session.commit()
-    # files now stores relative filenames as strings, use my_files() to get resolved Paths
-    assert {i['path'].name for i in vid.my_files()} == {'vid.mp4', 'vid.srt'}
-
+    assert merged.id == vid_group.id
     assert test_session.query(FileGroup).count() == 1
-    assert set(vid.tag_names) == {'one', 'two'}
-    assert {i['path'].name for i in vid.my_files()} == {'vid.mp4', 'vid.srt'}
-    # TagFile.created_at is preserved.
-    assert [i for i in vid.tag_files if i.tag.name == 'one'][0].created_at == tag_file_created_at
-    assert [i for i in vid.tag_files if i.tag.name == 'two'][0].created_at == srt_file_created_at
-    # Size is combined
-    assert vid.size > len(video_bytes)
-    assert vid.mimetype == 'video/mp4'
-
-    # Verify no duplicate files were created during merge
-    filenames = [f['path'] for f in vid.files]
+    assert set(merged.tag_names) == {'one', 'two'}
+    assert {i['path'].name for i in merged.my_files()} == {'vid.mp4', 'vid.srt'}
+    assert [i for i in merged.tag_files if i.tag.name == 'one'][0].created_at == tag_file_created_at
+    assert [i for i in merged.tag_files if i.tag.name == 'two'][0].created_at == srt_file_created_at
+    assert merged.mimetype == 'video/mp4'
+    filenames = [f['path'] for f in merged.files]
     assert len(filenames) == len(set(filenames)), f"Duplicate files found: {filenames}"
 
 

wrolpi/files/test/test_worker.py

@@ -1,6 +1,7 @@
 """Tests for the file comparison worker."""
 import asyncio
 import os
+import pathlib
 from unittest import mock
 
 import pytest
@@ -874,9 +875,10 @@ async def test_file_worker_deletes_removed_file_groups(
     file_worker.private_queue.put_nowait(task)
     await file_worker.process_queue()
 
-    # Verify FileGroup was deleted
+    # Verify FileGroup was deleted (do not key off id: SQLite may reuse it for the baseline upsert).
     test_session.expire_all()
-    assert test_session.query(FileGroup).filter_by(id=fg_id).one_or_none() is None
+    remaining = {pathlib.Path(str(fg.primary_path)).name for fg in test_session.query(FileGroup)}
+    assert 'file1.txt' not in remaining
 
 
 @pytest.mark.asyncio
@@ -912,10 +914,13 @@ async def test_file_worker_deletes_tagged_file_groups_auto_removes_tags(
     file_worker.private_queue.put_nowait(task)
     await file_worker.process_queue()
 
-    # Verify FileGroup and TagFile were deleted (cascade)
+    # Verify FileGroup and TagFile were deleted (cascade).  Do not key off recycled SQLite ids.
     test_session.expire_all()
-    assert test_session.query(FileGroup).filter_by(id=fg_id).one_or_none() is None
-    assert test_session.query(TagFile).filter_by(file_group_id=fg_id).count() == 0
+    remaining = {pathlib.Path(str(fg.primary_path)).name for fg in test_session.query(FileGroup)}
+    assert 'file1.txt' not in remaining
+    assert test_session.query(TagFile).join(FileGroup).filter(
+        FileGroup.primary_path.like('%/file1.txt')
+    ).count() == 0
 
 
 @pytest.mark.asyncio
@@ -1120,11 +1125,10 @@ async def test_file_worker_deletes_multiple_file_groups(
     file_worker.private_queue.put_nowait(task)
     await file_worker.process_queue()
 
-    # Verify all FileGroups were deleted
+    # Verify all FileGroups were deleted (do not key off recycled SQLite ids).
     test_session.expire_all()
-    assert test_session.query(FileGroup).filter(
-        FileGroup.id.in_([fg1_id, fg2_id, fg3_id])
-    ).count() == 0
+    remaining = {pathlib.Path(str(fg.primary_path)).name for fg in test_session.query(FileGroup)}
+    assert remaining.isdisjoint({'file1.txt', 'file2.txt', 'file3.txt'})
 
 
 @pytest.mark.asyncio
@@ -1681,78 +1685,15 @@ async def test_upsert_file_groups_tracks_actual_db_progress(async_client, test_s
 
 
 @pytest.mark.asyncio
-async def test_compare_file_groups_duplicate_deleted_filegroups(test_session, test_directory, make_files_structure):
-    """When multiple FileGroups share the same (directory, stem) and all files are deleted,
-    all duplicates should appear in the deleted diffs — not just the last one processed."""
-    make_files_structure([
-        'docs/file1.txt',
-        'docs/file1.json',
-    ])
-
-    docs_dir = str(test_directory / 'docs')
-
-    # Create 3 separate FileGroups for the same (directory, stem) — simulating a botched move.
-    # Use different primary_paths (unique constraint) but same directory and stem.
-    fg1 = FileGroup(directory=docs_dir, primary_path=str(test_directory / 'docs/file1.txt'),
-                    files=[{'path': 'file1.txt', 'mimetype': 'text/plain', 'size': 0}])
-    fg2 = FileGroup(directory=docs_dir, primary_path=str(test_directory / 'docs/file1.json'),
-                    files=[{'path': 'file1.json', 'mimetype': 'application/json', 'size': 0}])
-    # Third ghost with empty files list — use a unique primary_path
-    (test_directory / 'docs/file1.bak').touch()
-    fg3 = FileGroup(directory=docs_dir, primary_path=str(test_directory / 'docs/file1.bak'),
-                    files=[])
-    test_session.add_all([fg1, fg2, fg3])
-    test_session.commit()
-
-    fg1_id, fg2_id, fg3_id = fg1.id, fg2.id, fg3.id
-
-    # Delete all files from disk
-    for f in (test_directory / 'docs').iterdir():
-        f.unlink()
+async def test_compare_file_groups_unique_stem_is_one_group(test_session, test_directory, make_files_structure):
+    """Files that share a stem are one FileGroup; compare reports a single group."""
+    make_files_structure(['docs/file1.txt', 'docs/file1.json'])
+    from wrolpi.files.lib import _upsert_files
+    _upsert_files(list((test_directory / 'docs').iterdir()))
 
     result = await compare_file_groups(test_directory)
-
-    # ALL 3 FileGroup IDs must appear in deleted diffs
-    deleted_ids = {d.file_group_id for d in result.deleted}
-    assert fg1_id in deleted_ids, f'fg1 ({fg1_id}) missing from deleted: {deleted_ids}'
-    assert fg2_id in deleted_ids, f'fg2 ({fg2_id}) missing from deleted: {deleted_ids}'
-    assert fg3_id in deleted_ids, f'fg3 ({fg3_id}) missing from deleted: {deleted_ids}'
-    assert len(result.new) == 0
-
-
-@pytest.mark.asyncio
-async def test_compare_file_groups_duplicate_keeps_one(test_session, test_directory, make_files_structure):
-    """When duplicate FileGroups exist and files are on disk, one should be kept (modified/unchanged)
-    and the others should be marked for deletion."""
-    make_files_structure([
-        'docs/file1.txt',
-    ])
-
-    docs_dir = str(test_directory / 'docs')
-
-    # Create 2 FileGroups with same (directory, stem)
-    fg1 = FileGroup(directory=docs_dir, primary_path=str(test_directory / 'docs/file1.txt'),
-                    files=[{'path': 'file1.txt', 'mimetype': 'text/plain', 'size': 0}])
-    (test_directory / 'docs/file1.json').touch()
-    fg2 = FileGroup(directory=docs_dir, primary_path=str(test_directory / 'docs/file1.json'),
-                    files=[{'path': 'file1.json', 'mimetype': 'application/json', 'size': 0}])
-    test_session.add_all([fg1, fg2])
-    test_session.commit()
-
-    fg1_id, fg2_id = fg1.id, fg2.id
-    # Remove the .json so only .txt remains
-    (test_directory / 'docs/file1.json').unlink()
-
-    result = await compare_file_groups(test_directory)
-
-    # One FG should have the file on disk, the other should be deleted as duplicate
-    deleted_ids = {d.file_group_id for d in result.deleted}
-    all_diff_ids = {d.file_group_id for d in result.deleted + result.modified + result.unchanged}
-    assert fg1_id in all_diff_ids
-    assert fg2_id in all_diff_ids
-    # At least one must be in deleted (the duplicate)
-    assert len(deleted_ids) >= 1
-    assert deleted_ids & {fg1_id, fg2_id}
+    assert len(result.new) + len(result.unchanged) + len(result.modified) == 1
+    assert test_session.query(FileGroup).count() == 1
 
 
 @pytest.mark.asyncio

wrolpi/files/worker.py

@@ -33,7 +33,7 @@ from wrolpi.events import Events
 from wrolpi.vars import PYTEST, IS_MACOS
 from wrolpi.files.lib import (
     split_path_stem_and_suffix, _upsert_files, get_unique_files_by_stem, glob_shared_stem,
-    group_files_by_stem, get_primary_file, delete_directory, apply_indexers,
+    group_files_by_stem, get_primary_file, choose_primary_file, delete_directory, apply_indexers,
     _move_file_group_files, _bulk_update_file_groups_db, MOVE_CHUNK_SIZE,
     _bulk_update_file_groups_reorganize, get_normalized_ignored_directories, remove_files_in_ignored_directories,
     ensure_file_group_stems,
@@ -41,6 +41,20 @@ from wrolpi.files.lib import (
 
 logger = logger.getChild(__name__)
 
+
+async def _await_db(func, *args):
+    """Run a function that uses get_db_curs.
+
+    Tests share one SQLite connection via get_db_curs.  sqlite3 is not safe for concurrent
+    use of that connection, so asyncio.to_thread segfaults in CI (xdist + session commit
+    on the event loop while the worker thread is still in get_stem_algorithm_version /
+    _upsert_files).  Production opens a dedicated raw connection and can use a thread.
+    """
+    if PYTEST:
+        return func(*args)
+    return await asyncio.to_thread(func, *args)
+
+
 # Update status every N items to avoid excessive overhead
 PROGRESS_UPDATE_INTERVAL = 100
 
@@ -550,7 +564,7 @@ async def compare_file_groups(
     if root is not None and roots is None:
         roots = root
     root_list = _normalize_roots(roots)
-    await asyncio.to_thread(ensure_file_group_stems)
+    await _await_db(ensure_file_group_stems)
 
     # Per-call table name so concurrent scans don't collide.  The work table is a TEMP table on
     # its own dedicated connection: it never shares a transaction (or locks) with the session,
@@ -1259,8 +1273,8 @@ class FileWorker:
         try:
             result = await self._refresh_files_directly(paths)
             self._cleanup_modified_models(result.modified)
-            await self._upsert_file_groups(result.new + result.modified)
             await self._delete_file_groups(result.deleted)
+            await self._upsert_file_groups(result.new + result.modified)
             if post_processing and (result.new or result.modified or result.deleted):
                 # Skip global modelers/indexers when the caller opts out (e.g. `lib.delete` from the
                 # upload path defers that to the post-upload `upsert_file`) or when nothing changed.
@@ -1492,10 +1506,11 @@ class FileWorker:
                 logger.info(f'Refreshing {len(file_paths)} files directly')
                 file_result = await self._refresh_files_directly(file_paths, expand_stems=task.expand_stems)
 
-                # Process file results
+                # Delete first so a new primary_path cannot collide on (directory, stem)
+                # with a duplicate FileGroup that compare already marked deleted.
                 self._cleanup_modified_models(file_result.modified)
-                await self._upsert_file_groups(file_result.new + file_result.modified)
                 await self._delete_file_groups(file_result.deleted)
+                await self._upsert_file_groups(file_result.new + file_result.modified)
 
             # Process directories with full scan (existing path)
             if dir_paths:
@@ -1520,10 +1535,11 @@ class FileWorker:
                     f'{len(dir_result.deleted)} deleted, {len(dir_result.unchanged)} unchanged'
                 )
 
-                # Process directory results
+                # Delete first so a new primary_path cannot collide on (directory, stem)
+                # with a duplicate FileGroup that compare already marked deleted.
                 self._cleanup_modified_models(dir_result.modified)
-                await self._upsert_file_groups(dir_result.new + dir_result.modified)
                 await self._delete_file_groups(dir_result.deleted)
+                await self._upsert_file_groups(dir_result.new + dir_result.modified)
 
         # Run indexers, modelers, and cleanup once for all changes
         await self._apply_post_processing(is_global_refresh=is_global_refresh)
@@ -1594,17 +1610,12 @@ class FileWorker:
         # For modified diffs where file_group_id exists, we need to handle primary_path changes.
         # If the primary file changed (e.g., video deleted, leaving only poster), the ON CONFLICT
         # won't match, so we need to update the existing FileGroup's primary_path first.
-        from wrolpi.files.lib import NoPrimaryFile
         modified_ids_to_update = []
         for diff in diffs:
             if diff.file_group_id and diff.fs_files:
                 # This is a modified FileGroup - may need to update primary_path
                 new_paths = [diff.directory / f for f in diff.fs_files]
-                try:
-                    new_primary = get_primary_file(new_paths)
-                except NoPrimaryFile:
-                    # No clear primary file, just use the first one
-                    new_primary = new_paths[0] if new_paths else None
+                new_primary = choose_primary_file(new_paths)
                 if new_primary:
                     modified_ids_to_update.append((diff.file_group_id, str(new_primary)))
 
@@ -1670,7 +1681,7 @@ class FileWorker:
             self.update_status(operation_processed=processed, operation_percent=percent)
 
         # Use existing _upsert_files function which handles grouping and primary file detection
-        await asyncio.to_thread(_upsert_files, all_paths, on_upsert_progress)
+        await _await_db(_upsert_files, all_paths, on_upsert_progress)
         logger.info(f'Upserted {len(diffs)} FileGroups ({len(all_paths)} files)')
 
     async def _delete_file_groups(self, diffs: list[FileGroupDiff]):

wrolpi/jobs.py

@@ -300,8 +300,19 @@ class JobLogHandler(logging.Handler):
 
 _log_handler = JobLogHandler()
 _log_handler.setFormatter(logging.Formatter('%(levelname)s:%(name)s:%(message)s'))
-if not any(isinstance(i, JobLogHandler) for i in logging.getLogger().handlers):
-    logging.getLogger().addHandler(_log_handler)
+
+
+def _ensure_log_handler():
+    """Sanic/pytest dictConfig replaces root handlers and would drop JobLogHandler.
+
+    Attach to this module's logger (not only root) so capture survives that reset.
+    """
+    for target in (logging.getLogger(), logger):
+        if not any(isinstance(i, JobLogHandler) for i in target.handlers):
+            target.addHandler(_log_handler)
+
+
+_ensure_log_handler()
 
 
 def _json_safe(value):
@@ -314,6 +325,7 @@ def _json_safe(value):
 
 async def _run_job(job_id: str):
     """Run one Job to completion (or failure/cancellation) in this process."""
+    _ensure_log_handler()
     record = _claim_job(job_id)
     if not record:
         # Cancelled while it was waiting in the queue.
@@ -360,8 +372,13 @@ async def _run_job(job_id: str):
             raise
         except Exception as e:
             logger.error(f'Job {job_id} failed', exc_info=e)
+            # Do not rely on JobLogHandler for this line: Sanic/pytest dictConfig can
+            # strip root handlers (especially under xdist on CI), so the traceback
+            # never reaches the Job log even though `error` is set.
+            context.log(f'{type(e).__name__}: {e}')
             context.flush()
-            _update_job(job_id, status=FAILED, error=str(e) or repr(e), finished_at=now().isoformat())
+            _update_job(job_id, status=FAILED, error=str(e) or repr(e),
+                        log=list(context._log), finished_at=now().isoformat())
             Events.send_user_notify(f'Failed: {record["description"]}')
             return
 

wrolpi/test/test_db_bootstrap.py

@@ -66,6 +66,57 @@ def test_compare_db_version_uninitialized_file(test_directory):
     assert db_bootstrap.compare_db_version() == 'current'
 
 
+def test_unique_stem_migration_collapses_duplicates(test_directory):
+    """The unique-index migration merges duplicate (directory, stem) rows and keeps tags."""
+    import sqlite3
+    from alembic.config import Config
+    from alembic import command
+    from wrolpi.db import get_db_uri, get_db_file
+    from wrolpi.vars import PROJECT_DIR
+
+    (test_directory / 'config').mkdir(parents=True, exist_ok=True)
+    config = Config(str(PROJECT_DIR / 'alembic.ini'))
+    config.set_main_option('sqlalchemy.url', get_db_uri())
+    command.upgrade(config, '2026_09_16_1200')
+
+    db_file = get_db_file()
+    directory = str(test_directory)
+    with sqlite3.connect(db_file) as conn:
+        conn.execute('PRAGMA foreign_keys=ON')
+        conn.execute(
+            'INSERT INTO file_group (directory, primary_path, files, stem, indexed) VALUES (?,?,?,?,0)',
+            (directory, str(test_directory / 'dup.mp4'), '[]', 'dup'),
+        )
+        conn.execute(
+            'INSERT INTO file_group (directory, primary_path, files, stem, indexed) VALUES (?,?,?,?,0)',
+            (directory, str(test_directory / 'dup.info.json'), '[]', 'dup'),
+        )
+        first_id, second_id = [r[0] for r in conn.execute('SELECT id FROM file_group ORDER BY id').fetchall()]
+        conn.execute("INSERT INTO tag (name, color) VALUES ('migrated', '#111111')")
+        tag_id = conn.execute('SELECT id FROM tag').fetchone()[0]
+        # Tag the lower-id row so it wins over the later untagged duplicate.
+        conn.execute(
+            'INSERT INTO tag_file (tag_id, file_group_id) VALUES (?, ?)',
+            (tag_id, first_id),
+        )
+        conn.commit()
+        assert conn.execute('SELECT COUNT(*) FROM file_group').fetchone()[0] == 2
+        assert first_id < second_id
+
+    command.upgrade(config, 'head')
+
+    with sqlite3.connect(db_file) as conn:
+        rows = list(conn.execute('SELECT id, stem FROM file_group'))
+        assert len(rows) == 1
+        assert rows[0][0] == first_id
+        tagged = conn.execute('SELECT file_group_id FROM tag_file').fetchone()[0]
+        assert tagged == first_id
+        sql = conn.execute(
+            "SELECT sql FROM sqlite_master WHERE type='index' AND name='file_group_directory_stem_idx'"
+        ).fetchone()[0]
+        assert 'UNIQUE' in sql.upper()
+
+
 def test_bootstrap_lock(test_directory):
     """Only one process can hold the bootstrap lock."""
     with db_bootstrap.bootstrap_lock() as acquired: