diff --git a/modules/repos/test/test_config.py b/modules/repos/test/test_config.py
index b6b98974..4d1c8863 100644
--- a/modules/repos/test/test_config.py
+++ b/modules/repos/test/test_config.py
@@ -3,6 +3,7 @@ import pytest
 from modules.repos import lib
 from modules.repos.models import Repository
 from wrolpi.collections.models import Collection
+from wrolpi.conftest import config_versions_lost
 from wrolpi.downloader import Download, DownloadFrequency
 
 
@@ -31,9 +32,12 @@ async def test_dump_and_import(test_session, test_directory, async_client, test_
     for repo in test_session.query(Repository).all():
         lib._delete_repository(test_session, repo)
     test_session.commit()
+    config_versions_lost(test_session)
     assert test_session.query(Collection).count() == 0
 
     config.import_config()
+    # Imported without deletes; the database is saved, which records its version.
+    await await_switches()
     test_session.expire_all()
     repos = {i.name: i for i in test_session.query(Repository)}
     assert repos['kiwix-tools'].directory == test_directory / 'repos/software/kiwix-tools'
diff --git a/modules/repos/test/test_never.py b/modules/repos/test/test_never.py
index 75f219f0..e3922cb2 100644
--- a/modules/repos/test/test_never.py
+++ b/modules/repos/test/test_never.py
@@ -9,6 +9,7 @@ import yaml
 
 from modules.repos import lib
 from modules.repos.models import Repository
+from wrolpi.conftest import config_versions_lost
 from wrolpi.downloader import Download, DownloadFrequency, DownloadStatus
 
 
@@ -146,6 +147,7 @@ async def test_never_config(test_session, test_directory, async_client, test_wro
     for download in test_session.query(Download).all():
         test_session.delete(download)
     test_session.commit()
+    config_versions_lost(test_session)
     subprocess.run(['rm', '-rf', str(test_directory / 'repos/missing')], check=True)
 
     config.import_config()
diff --git a/modules/repos/test/test_ownership.py b/modules/repos/test/test_ownership.py
index 0ee5d871..e573d795 100644
--- a/modules/repos/test/test_ownership.py
+++ b/modules/repos/test/test_ownership.py
@@ -5,6 +5,7 @@ import pytest
 
 from modules.repos import lib
 from modules.repos.conftest import TEST_GIT_ENV
+from wrolpi.conftest import config_versions_lost
 from wrolpi.downloader import Download, DownloadStatus
 
 
@@ -395,6 +396,7 @@ async def test_clone_token_survives_the_config(test_session, async_client, git_r
     # Lose the database, restore it from the config.
     lib._delete_repository(test_session, repo)
     test_session.commit()
+    config_versions_lost(test_session)
     config.import_config()
     test_session.expire_all()
     restored = test_session.query(lib.Repository).one()
diff --git a/modules/videos/test/test_channels_config.py b/modules/videos/test/test_channels_config.py
index 8da741b5..fc1a401c 100644
--- a/modules/videos/test/test_channels_config.py
+++ b/modules/videos/test/test_channels_config.py
@@ -14,7 +14,7 @@ import yaml
 
 from modules.videos.lib import get_channels_config
 from modules.videos.models import Channel
-from wrolpi.conftest import config_was_imported
+from wrolpi.conftest import config_was_imported, config_versions_lost
 
 
 @pytest.mark.asyncio
@@ -319,6 +319,7 @@ class TestChannelsConfigEdgeCases:
         # Clear the file_format to simulate factory reset
         channel.collection.file_format = None
         test_session.commit()
+        config_versions_lost(test_session)
 
         # Re-import
         config.initialize()
diff --git a/wrolpi/collections/config.py b/wrolpi/collections/config.py
index 50258fd5..91beaa86 100644
--- a/wrolpi/collections/config.py
+++ b/wrolpi/collections/config.py
@@ -14,6 +14,11 @@ logger = logger.getChild(__name__)
 __all__ = ['PlaylistsConfig', 'playlists_config', 'save_playlists_config', 'get_playlists_config']
 
 
+def _item_key(item) -> tuple:
+    """What a playlist item points to; a playlist has at most one item per key (see CollectionItem's constraints)."""
+    return item.item_kind, item.file_group_id, item.zim_id, item.zim_entry, item.url
+
+
 @dataclass
 class PlaylistsConfigValidator:
     """Validator for the playlists config file."""
@@ -97,7 +102,8 @@ class PlaylistsConfig(ConfigFile):
 
                     collection = session.query(Collection).filter_by(
                         name=name, kind='playlist').one_or_none()
-                    if not collection:
+                    new = not collection
+                    if new:
                         collection = Collection(name=name, kind='playlist')
                         session.add(collection)
                     collection.description = data.get('description')
@@ -110,12 +116,7 @@ class PlaylistsConfig(ConfigFile):
                     collection.tag = Tag.get_by_name(session, tag_name) if tag_name else None
                     session.flush([collection])
 
-                    # Rebuild items to match the config's order exactly.
-                    for item in list(collection.items):
-                        session.delete(item)
-                    session.flush()
-
-                    position = 1
+                    file_items = []
                     for item_data in (data.get('items') or []):
                         item = None
                         if 'file' in item_data:
@@ -143,9 +144,30 @@ class PlaylistsConfig(ConfigFile):
                                                   url=item_data['url'])
                         if item is not None:
                             item.title = item_data.get('title')
-                            item.position = position
-                            position += 1
-                            session.add(item)
+                            file_items.append(item)
+
+                    if decision == IMPORT or new:
+                        # Rebuild items to match the config's order exactly.
+                        for item in list(collection.items):
+                            session.delete(item)
+                        session.flush()
+                        items = file_items
+                    else:
+                        # Without deletes, keep the database's items in their order; add the file's other items
+                        # after the item that precedes them in the file.
+                        items = sorted(collection.items, key=lambda i: i.position)
+                        existing = {_item_key(i): i for i in items}
+                        previous = None
+                        for item in file_items:
+                            if found := existing.get(_item_key(item)):
+                                previous = found
+                                continue
+                            items.insert(items.index(previous) + 1 if previous else 0, item)
+                            previous = item
+
+                    for position, item in enumerate(items, start=1):
+                        item.position = position
+                        session.add(item)
 
                     session.flush()
                     imported_names.add(name)
diff --git a/wrolpi/collections/models.py b/wrolpi/collections/models.py
index 2e69b42d..df04a034 100644
--- a/wrolpi/collections/models.py
+++ b/wrolpi/collections/models.py
@@ -12,7 +12,7 @@ from sqlalchemy.orm import Mapped, relationship, Session, joinedload, selectinlo
 from wrolpi import flags
 from wrolpi.common import loaded, Base, ModelHelper, logger, get_media_directory, get_relative_to_media_directory, \
     unique_by_predicate, TRACE_LEVEL
-from wrolpi.config_version import register_ownership
+from wrolpi.config_version import collection_kinds, count_config_change, register_ownership
 from wrolpi.db import serializer
 from wrolpi.downloader import Download, save_downloads_config
 from wrolpi.errors import ValidationError
@@ -735,6 +735,9 @@ class Collection(ModelHelper, Base):
         ).delete(synchronize_session=False)
 
         logger.debug(f'Removed {deleted} FileGroups from Collection {self.id}')
+        if deleted and self.kind == 'playlist':
+            # A bulk delete is not flushed: count it.
+            count_config_change(session, 'playlists.yaml')
 
         # Resequence positions to close gaps (optional, but keeps positions clean)
         items = session.query(CollectionItem).filter_by(
@@ -1317,5 +1320,5 @@ class CollectionItem(ModelHelper, Base):
 for _kind, _config in (('channel', 'channels.yaml'), ('domain', 'domains.yaml'), ('playlist', 'playlists.yaml'),
                        ('repo', 'repos.yaml')):
     register_ownership(_config, Collection, row=lambda collection, kind=_kind: collection.kind == kind)
-# Only playlists have items.
-register_ownership('playlists.yaml', CollectionItem)
+# Only a playlist's items are saved (author and subject Collections have items too).
+register_ownership('playlists.yaml', CollectionItem, row=lambda item: 'playlist' in collection_kinds(item))
diff --git a/wrolpi/collections/test/test_playlists_config.py b/wrolpi/collections/test/test_playlists_config.py
index ef769049..474c0507 100644
--- a/wrolpi/collections/test/test_playlists_config.py
+++ b/wrolpi/collections/test/test_playlists_config.py
@@ -4,6 +4,7 @@ from copy import deepcopy
 from http import HTTPStatus
 
 import pytest
+import yaml
 
 from wrolpi.collections.config import get_playlists_config
 from wrolpi.collections.models import Collection
@@ -370,3 +371,24 @@ async def test_playlist_changes_wait_for_config_import(test_session, test_direct
     playlists_config.import_config()
     request, response = await async_client.post('/api/collections', content=json.dumps(dict(name='New')))
     assert response.status_code == HTTPStatus.CREATED, response.json
+
+
+@pytest.mark.asyncio
+async def test_import_without_deletes_keeps_database_items(test_session, test_directory, playlists_config):
+    """Without a database version (e.g. the first startup after an upgrade) a playlist's items in the database are
+    kept in their order, and the file's other items are added after the item that precedes them in the file."""
+    collection = Collection(name='Mix', kind='playlist')
+    test_session.add(collection)
+    test_session.flush([collection])
+    collection.add_url(test_session, '/a', title='A')
+    collection.add_url(test_session, '/c', title='C')  # Added after the file was saved.
+    test_session.commit()
+    playlists_config.get_file().write_text(yaml.dump(dict(version=1, playlists=[
+        dict(name='Mix', items=[dict(url='/b', title='B'), dict(url='/a', title='A')]),
+    ])))
+
+    playlists_config.import_config()
+
+    test_session.expire_all()
+    collection = test_session.query(Collection).filter_by(name='Mix', kind='playlist').one()
+    assert [i.url for i in collection.items] == ['/b', '/a', '/c']
diff --git a/wrolpi/common.py b/wrolpi/common.py
index d2f6e136..c04d60ff 100644
--- a/wrolpi/common.py
+++ b/wrolpi/common.py
@@ -741,6 +741,9 @@ class ConfigFile:
                     Events.send_config_save_failed(message)
                 raise e
 
+        if self.is_versioned_by_database() and file == self.get_file():
+            self.record_saved_version(config['version'])
+
     def write_config_data(self, config: dict, config_file: pathlib.Path):
         write_config_data(config, config_file, width=self.width)
 
@@ -802,18 +805,29 @@ class ConfigFile:
         from wrolpi.config_version import versioned_config_names
         return self.file_name in versioned_config_names()
 
+    # What the last `import_decision` allowed; `record_import_version` records the version that follows from it.
+    last_import_decision: str = None
+
     def import_decision(self) -> str:
         """After reading the file (`ConfigFile.import_config`), decide what importing it may do: IMPORT,
-        IMPORT_WITHOUT_DELETES, or SAVE (the database is newer; call `save_database_instead` and import nothing)."""
-        from wrolpi.config_version import IMPORT, decide, get_config_version
+        IMPORT_WITHOUT_DELETES, or SAVE (the database is newer; call `save_database_instead` and import nothing).
+
+        On IMPORT the database's version is removed until `record_import_version`: the import's own writes must not
+        count, or an import that fails partway would leave the database "newer" than the file it did not finish
+        applying, and the next startup would save that partial database over the file."""
+        from wrolpi.config_version import IMPORT, decide, delete_config_version, get_config_version
         from wrolpi.db import get_db_session
-        if self.force_import:
-            self.force_import = False
-            return IMPORT
-        with get_db_session() as session:
+        with get_db_session(commit=True) as session:
             db_version = get_config_version(session, self.file_name)
-        decision = decide(self.version or 0, db_version)
+            if self.force_import:
+                self.force_import = False
+                decision = IMPORT
+            else:
+                decision = decide(self.version or 0, db_version)
+            if decision == IMPORT:
+                delete_config_version(session, self.file_name)
         logger_.debug(f'{self.file_name}: file version {self.version}, database version {db_version}: {decision}')
+        self.last_import_decision = decision
         return decision
 
     def save_database_instead(self):
@@ -824,19 +838,35 @@ class ConfigFile:
 
     def record_import_version(self):
         """The database now matches this config's file: record the file's version.  Call last, after every write
-        the import makes."""
-        from wrolpi.config_version import set_config_version
+        the import makes.
+
+        Not after IMPORT_WITHOUT_DELETES: the database may keep rows the file does not list (changes that were never
+        saved), so recording the file's version would let the next import delete them.  Save them into the file
+        instead; that save records the version (`record_saved_version`).  A save after a partial import is refused,
+        so the version stays unknown until an import applies the whole file."""
+        from wrolpi.config_version import IMPORT, set_config_version
         from wrolpi.db import get_db_session
+        if self.last_import_decision != IMPORT:
+            self.background_dump.activate_switch()
+            return
         with get_db_session(commit=True) as session:
             set_config_version(session, self.file_name, self.version or 0)
 
     def db_version_for_dump(self, session) -> int:
         """The version `dump_config` writes: the database's, read in the same transaction as the rows it saves.
-        Before the database has one (this config was never imported by this code), the file's next version."""
+        Before the database has one, the file's next version (`save` records it)."""
         from wrolpi.config_version import get_config_version
         db_version = get_config_version(session, self.file_name)
         return db_version if db_version is not None else (self.version or 0) + 1
 
+    def record_saved_version(self, version: int):
+        """The file was just written from the database.  If the database's version was unknown (see
+        `record_import_version`), the two are now in step: record it, so changes from here on count against it."""
+        from wrolpi.config_version import create_config_version
+        from wrolpi.db import get_db_session
+        with get_db_session(commit=True) as session:
+            create_config_version(session, self.file_name, version)
+
     def update(self, config: dict, overwrite: bool = False):
         """Update any values of this config.  Save the config to its file."""
         from wrolpi.api_utils import api_app
diff --git a/wrolpi/config_version.py b/wrolpi/config_version.py
index 0ff24c20..c56e371e 100644
--- a/wrolpi/config_version.py
+++ b/wrolpi/config_version.py
@@ -11,10 +11,13 @@ So an import can tell which side is newer by comparing two integers (see `Config
 * file > database: the file is newer (a restored or wiped database, a copied config); import it the same way.
 * file < database: the database changed and was not saved yet (e.g. WROLPi restarted first).  Save the database
   instead of importing the older file, which would undo those changes.
-* no database row: not known yet (a config never imported by this code).  Import without deleting.
+* no database row: not known yet (a config never imported by this code, or an import that did not finish).
+  Import without deleting, and leave the version unknown: the database may have rows the file does not list.  Then
+  save them into the file; that save records its version (`ConfigFile.record_saved_version`).
 
 Integers only: a Raspberry Pi has no clock that survives a reboot, so timestamps cannot order these.
 """
+import json
 from collections import defaultdict
 from dataclasses import dataclass
 from typing import Callable, Dict, FrozenSet, List, Optional, Set, Type
@@ -79,6 +82,9 @@ def _owners(obj, changed: bool) -> Set[str]:
     return names
 
 
+_COLLECTION_KINDS = 'config_version_collection_kinds'
+
+
 @event.listens_for(Session, 'after_flush')
 def _count_config_changes(session: Session, flush_context):
     """Add one to the database version of each config whose rows this flush changed, in the same transaction.
@@ -88,12 +94,16 @@ def _count_config_changes(session: Session, flush_context):
     if not OWNERSHIPS:
         return
     names = set()
-    for obj in session.new:
-        names |= _owners(obj, changed=False)
-    for obj in session.deleted:
-        names |= _owners(obj, changed=False)
-    for obj in session.dirty:
-        names |= _owners(obj, changed=True)
+    session.info[_COLLECTION_KINDS] = dict()
+    try:
+        for obj in session.new:
+            names |= _owners(obj, changed=False)
+        for obj in session.deleted:
+            names |= _owners(obj, changed=False)
+        for obj in session.dirty:
+            names |= _owners(obj, changed=True)
+    finally:
+        del session.info[_COLLECTION_KINDS]
     if names:
         count_config_change(session, *names)
 
@@ -117,14 +127,24 @@ def count_config_change_curs(curs, *names: str):
             raise
 
 
-def collection_kind(obj) -> Optional[str]:
-    """The kind of the Collection `obj` (e.g. a Download) belongs to, read on its flush's connection."""
-    if not getattr(obj, 'collection_id', None):
-        return None
+def collection_kinds(obj) -> Set[str]:
+    """The kinds of the Collections `obj` (e.g. a Download) belongs to, and belonged to before this flush (moving
+    it out of a Collection changes that Collection's config too), read on its flush's connection."""
+    ids = {i for i in inspect(obj).attrs.collection_id.history.deleted if i}
+    if obj.collection_id:
+        ids.add(obj.collection_id)
+    if not ids:
+        return set()
     session = Session.object_session(obj)
-    row = session.connection().execute(
-        text('SELECT kind FROM collection WHERE id = :id'), {'id': obj.collection_id}).first()
-    return row[0] if row else None
+    # A Collection's kind never changes: look each up once per flush (a playlist import writes thousands of items).
+    kinds = session.info.get(_COLLECTION_KINDS, dict())
+    if missing := ids - kinds.keys():
+        rows = session.connection().execute(
+            text('SELECT id, kind FROM collection WHERE id IN (SELECT value FROM json_each(:ids))'),
+            {'ids': json.dumps(sorted(missing))})
+        kinds.update(dict.fromkeys(missing))
+        kinds.update(rows.all())
+    return {kinds[i] for i in ids} - {None}
 
 
 def get_config_version(session: Session, name: str) -> Optional[int]:
@@ -133,6 +153,17 @@ def get_config_version(session: Session, name: str) -> Optional[int]:
     return row[0] if row else None
 
 
+def delete_config_version(session: Session, name: str):
+    session.connection().execute(text('DELETE FROM config_version WHERE name = :name'), {'name': name})
+
+
+def create_config_version(session: Session, name: str, version: int):
+    """Record `version` unless the database already has one."""
+    session.connection().execute(
+        text('INSERT INTO config_version (name, version) VALUES (:name, :version) ON CONFLICT (name) DO NOTHING'),
+        {'name': name, 'version': version})
+
+
 def set_config_version(session: Session, name: str, version: int):
     session.connection().execute(
         text('INSERT INTO config_version (name, version) VALUES (:name, :version) '
diff --git a/wrolpi/conftest.py b/wrolpi/conftest.py
index e6da5d0b..906b7ed0 100644
--- a/wrolpi/conftest.py
+++ b/wrolpi/conftest.py
@@ -141,6 +141,14 @@ def config_was_imported(session, *file_names: str):
     session.commit()
 
 
+def config_versions_lost(session):
+    """A test that deletes rows to simulate a lost database must lose the database's config versions too (a real
+    lost database has none); otherwise deleting the rows counts as a database change newer than the file."""
+    from wrolpi.config_version import ConfigVersion
+    session.query(ConfigVersion).delete()
+    session.commit()
+
+
 @pytest.fixture(autouse=True)
 def no_deferred_request_writes(request):
     """Fail any test during which an API request wrote outside `request.ctx.db.write()`."""
diff --git a/wrolpi/downloader.py b/wrolpi/downloader.py
index 95f26b37..e3df6e44 100644
--- a/wrolpi/downloader.py
+++ b/wrolpi/downloader.py
@@ -24,9 +24,9 @@ from zoneinfo import ZoneInfo
 import feedparser
 import pytz
 from feedparser import FeedParserDict
-from sqlalchemy import Column, Integer, String, Text, ForeignKey, Index, JSON
+from sqlalchemy import Column, Integer, String, Text, ForeignKey, Index, JSON, text
 from sqlalchemy.exc import OperationalError
-from sqlalchemy.orm import Mapped, Session, relationship
+from sqlalchemy.orm import Mapped, Session, column_property, relationship
 
 from wrolpi import flags
 from wrolpi.api_utils import api_app, perpetual_signal
@@ -34,7 +34,7 @@ from wrolpi.cmd import which, run_command, CommandResult
 from wrolpi.common import loaded, Base, ModelHelper, logger, wrol_mode_check, zig_zag, ConfigFile, \
     wrol_mode_enabled, background_task, get_absolute_media_path, timer, aiohttp_get, \
     get_download_info, trim_file_name, get_wrolpi_config, TRACE_LEVEL, normalize_domain
-from wrolpi.config_version import IMPORT, SAVE, collection_kind, count_config_change, register_ownership
+from wrolpi.config_version import IMPORT, SAVE, collection_kinds, count_config_change, register_ownership
 from wrolpi.dates import TZDateTime, now, Seconds
 from wrolpi.db import get_db_session, session_curs, serializer
 from wrolpi.errors import InvalidDownload, UnrecoverableDownloadError, BotBlockedDownloadError, UnknownDownload, \
@@ -269,7 +269,9 @@ class Download(ModelHelper, Base):  # noqa
     tag_names = Column(JSON)  # a list of Tag names.
 
     # A Download may be associated with a Collection (downloads for channels, domains, etc.).
-    collection_id = Column(Integer, ForeignKey('collection.id', ondelete='SET NULL'))
+    # The previous value is loaded on change: leaving a Collection changes that Collection's config too.
+    collection_id = column_property(Column(Integer, ForeignKey('collection.id', ondelete='SET NULL')),
+                                    active_history=True)
     collection = relationship('Collection', primaryjoin='Download.collection_id==Collection.id',
                               back_populates='downloads')
 
@@ -1318,9 +1320,12 @@ class DownloadManager:
         @param reset_attempts: Will set `download.attempts` to 0 if True.
         """
         values = dict(status=DownloadStatus.new, attempts=0) if reset_attempts else dict(status=DownloadStatus.new)
-        session.query(Download) \
+        retried = session.query(Download) \
             .filter(Download.status.in_((DownloadStatus.pending, DownloadStatus.deferred, DownloadStatus.failed))) \
             .update(values, synchronize_session=False)
+        if retried:
+            # A bulk update is not flushed: count it.  `status` is saved only by download_manager.yaml.
+            count_config_change(session, DownloadManagerConfig.file_name)
 
     def retry_downloads_on_startup(self):
         """Startup is not a request: retry incomplete Downloads in a short write transaction of its own."""
@@ -1784,9 +1789,9 @@ class DownloadManager:
         if not download_ids:
             return []
         deleted_ids = [i for i, in session.query(Download.id).filter(Download.id.in_(download_ids)).all()]
+        # A Core delete is not flushed: count it.
+        count_download_changes(session, deleted_ids)
         session.execute(Download.__table__.delete().where(Download.id.in_(deleted_ids)))
-        # A Core delete is not flushed.  These may include a Channel's or a Repo's Download.
-        count_config_change(session, DownloadManagerConfig.file_name, 'channels.yaml', 'repos.yaml')
         session.commit()
         save_downloads_config.activate_switch()
         return deleted_ids
@@ -2309,9 +2314,23 @@ register_ownership(DownloadManagerConfig.file_name, Download)
 # channels.yaml and repos.yaml each save their own Collections' Downloads (the URL and how often it runs).  Only
 # the config of that Collection's kind owns one: counting another config's change would make that config's next
 # import save instead of import.
-for _kind, _config in (('channel', 'channels.yaml'), ('repo', 'repos.yaml')):
+DOWNLOAD_COLLECTION_CONFIGS = (('channel', 'channels.yaml'), ('repo', 'repos.yaml'))
+for _kind, _config in DOWNLOAD_COLLECTION_CONFIGS:
     register_ownership(_config, Download, columns={'url', 'frequency', 'collection_id'},
-                       row=lambda download, kind=_kind: collection_kind(download) == kind)
+                       row=lambda download, kind=_kind: kind in collection_kinds(download))
+
+
+def count_download_changes(session: Session, download_ids: List[int]):
+    """Count a write that bypasses the flush (Core or bulk) to these Downloads, before it runs: for
+    download_manager.yaml, and for the config of each Collection kind they belong to."""
+    if not download_ids:
+        return
+    kinds = {i for i, in session.execute(
+        text('SELECT DISTINCT c.kind FROM download d JOIN collection c ON c.id = d.collection_id '
+             'WHERE d.id IN (SELECT value FROM json_each(:ids))'),
+        {'ids': json.dumps(list(download_ids))})}
+    count_config_change(session, DownloadManagerConfig.file_name,
+                        *(config for kind, config in DOWNLOAD_COLLECTION_CONFIGS if kind in kinds))
 TEST_DOWNLOAD_MANAGER_CONFIG: DownloadManagerConfig = None
 
 
diff --git a/wrolpi/files/models.py b/wrolpi/files/models.py
index 6284c8cf..6c3635c1 100644
--- a/wrolpi/files/models.py
+++ b/wrolpi/files/models.py
@@ -13,7 +13,7 @@ from sqlalchemy.orm import Mapped, deferred, relationship, Session, selectinload
 from wrolpi.common import loaded, Base, ModelHelper, logger, recursive_map, get_media_directory, \
     get_relative_to_media_directory, unique_by_predicate, replace_file
 from wrolpi.dates import TZDateTime, now, from_timestamp, strptime_ms, strftime
-from wrolpi.config_version import register_ownership
+from wrolpi.config_version import count_config_change, register_ownership
 from wrolpi.db import get_db_session, serializer
 from wrolpi.downloader import Download
 from wrolpi.errors import FileGroupIsTagged, UnknownFile
@@ -252,6 +252,8 @@ class FileGroup(ModelHelper, Base):
             session.flush(self.tag_files)
 
         if deleted_tags := (existing_tag_ids - tag_ids):
+            # A bulk delete is not flushed: count it.
+            count_config_change(session, 'tags.yaml')
             session.query(TagFile).filter(
                 TagFile.file_group_id == self.id,
                 TagFile.tag_id.in_(deleted_tags),
diff --git a/wrolpi/files/worker.py b/wrolpi/files/worker.py
index 98e54b0c..941ee525 100644
--- a/wrolpi/files/worker.py
+++ b/wrolpi/files/worker.py
@@ -802,7 +802,7 @@ class FileWorker(FileMoveMixin, FileReorganizeMixin):
             operation_percent=0,
         )
 
-        from wrolpi.downloader import Download, download_manager
+        from wrolpi.downloader import Download, count_download_changes, download_manager
         from wrolpi.files.models import FileGroup
         from wrolpi.tags import TagFile, save_tags_config, sync_tags_directory
 
@@ -830,7 +830,9 @@ class FileWorker(FileMoveMixin, FileReorganizeMixin):
                 for fg in file_groups:
                     if fg.url:
                         all_urls.append(fg.url)
-                        if Download.get_by_url(session, fg.url):
+                        if download := Download.get_by_url(session, fg.url):
+                            # A bulk delete is not flushed: count it.
+                            count_download_changes(session, [download.id])
                             session.query(Download).filter(
                                 Download.url == fg.url
                             ).delete(synchronize_session=False)
diff --git a/wrolpi/test/test_config_version.py b/wrolpi/test/test_config_version.py
index 521fe219..d8bf9bc7 100644
--- a/wrolpi/test/test_config_version.py
+++ b/wrolpi/test/test_config_version.py
@@ -91,8 +91,9 @@ async def test_unsaved_database_change_is_saved_not_undone(test_session, test_ta
     assert set(saved['tags']) == {'old', 'new'} and saved['version'] == 4
 
 
+@pytest.mark.asyncio
 @pytest.mark.parametrize('file_version', [3, 9])
-def test_current_file_imports_with_deletes(test_session, test_tags_config, file_version):
+async def test_current_file_imports_with_deletes(test_session, test_tags_config, file_version, async_client):
     """A file as new as the database (or newer) is the truth: what it no longer lists is deleted.  The database
     records the file's version."""
     test_session.add_all([Tag(name='kept', color='#000000'), Tag(name='removed', color='#000000')])
@@ -107,19 +108,50 @@ def test_current_file_imports_with_deletes(test_session, test_tags_config, file_
     assert db_version(test_session) == file_version
 
 
-def test_unknown_database_version_imports_without_deletes(test_session, test_tags_config):
-    """A config this code never imported (no database row) imports without deleting, then records its version."""
+@pytest.mark.asyncio
+async def test_unknown_database_version_imports_without_deletes(test_session, test_tags_config, await_switches):
+    """A config this code never imported (no database row) imports without deleting.  The database may keep rows
+    the file does not list, so the file's version is not recorded (the next import would delete them); they are
+    saved into the file instead, and that save records the version."""
     test_session.add_all([Tag(name='kept', color='#000000'), Tag(name='unlisted', color='#000000')])
     test_session.commit()
     write_tags_config(test_tags_config, 7, ['kept', 'added'])
 
     import_tags_config()
-
     assert {i.name for i in test_session.query(Tag)} == {'kept', 'unlisted', 'added'}
-    assert db_version(test_session) == 7
+    assert db_version(test_session) is None
+
+    await await_switches()
+    saved = yaml.safe_load(test_tags_config.read_text())
+    assert set(saved['tags']) == {'kept', 'unlisted', 'added'} and saved['version'] == 8
+    assert db_version(test_session) == 8
 
 
-def test_import_button_imports_an_older_file(test_session, test_tags_config):
+@pytest.mark.asyncio
+async def test_failed_import_leaves_version_unknown(test_session, test_tags_config, monkeypatch, async_client):
+    """An import's own writes do not count: one that fails partway leaves the version unknown, not newer than the
+    file it did not finish applying (which would save that partial database over the file)."""
+    from wrolpi.tags import TagsConfig
+
+    test_session.add(Tag(name='kept', color='#000000'))
+    test_session.commit()
+    set_config_version(test_session, 'tags.yaml', 3)
+    test_session.commit()
+    write_tags_config(test_tags_config, 3, ['kept', 'added'])
+
+    def fail(self):
+        raise RuntimeError('Import failed')
+
+    monkeypatch.setattr(TagsConfig, 'record_import_version', fail)
+    with pytest.raises(RuntimeError):
+        import_tags_config()
+
+    assert {i.name for i in test_session.query(Tag)} == {'kept', 'added'}
+    assert db_version(test_session) is None
+
+
+@pytest.mark.asyncio
+async def test_import_button_imports_an_older_file(test_session, test_tags_config, async_client):
     """The Import button applies the file whatever its version: the user chose it."""
     test_session.add_all([Tag(name='kept', color='#000000'), Tag(name='removed', color='#000000')])
     test_session.commit()
@@ -136,6 +168,7 @@ def test_import_button_imports_an_older_file(test_session, test_tags_config):
 @pytest.mark.asyncio
 async def test_raw_writes_count(test_session, test_directory, make_files_structure, async_client):
     """Writes that bypass the ORM flush count their changes explicitly."""
+    from wrolpi.downloader import Download, DownloadManager, DownloadStatus
     from wrolpi.files.lib import _bulk_delete_tag_files, _bulk_insert_tag_files
 
     make_files_structure(['a.txt'])
@@ -151,6 +184,23 @@ async def test_raw_writes_count(test_session, test_directory, make_files_structu
     _bulk_delete_tag_files([(tag.id, file_group.id)])
     assert db_version(test_session) == 3
 
+    # Untagging deletes in bulk.
+    file_group.add_tag(test_session, tag.id)
+    test_session.commit()
+    assert db_version(test_session) == 4
+    file_group.untag(test_session, tag.id)
+    test_session.commit()
+    assert db_version(test_session) == 5
+
+    # Retrying Downloads updates their status in bulk.
+    test_session.add(Download(url='https://example.com/1', status=DownloadStatus.failed))
+    test_session.commit()
+    set_config_version(test_session, 'download_manager.yaml', 1)
+    test_session.commit()
+    DownloadManager.retry_downloads(test_session)
+    test_session.commit()
+    assert db_version(test_session, 'download_manager.yaml') == 2
+
 
 def test_downloads_count_for_their_collections_config_only(test_session):
     """A Channel's Download is saved by channels.yaml, not repos.yaml: counting it for repos.yaml would make the repos
@@ -170,3 +220,59 @@ def test_downloads_count_for_their_collections_config_only(test_session):
     assert db_version(test_session, 'channels.yaml') == 2
     assert db_version(test_session, 'download_manager.yaml') == 2
     assert db_version(test_session, 'repos.yaml') == 1
+
+
+def test_download_leaving_a_collection_counts_for_its_config(test_session):
+    """Moving a Download out of a Channel changes channels.yaml: otherwise its next import would put it back."""
+    from wrolpi.downloader import Download
+
+    channel = Collection(name='Channel', kind='channel')
+    test_session.add(channel)
+    test_session.commit()
+    download = Download(url='https://example.com/channel', frequency=86400, collection_id=channel.id)
+    test_session.add(download)
+    test_session.commit()
+    set_config_version(test_session, 'channels.yaml', 1)
+    test_session.commit()
+
+    download.collection_id = None
+    test_session.commit()
+
+    assert db_version(test_session, 'channels.yaml') == 2
+
+
+def test_deleting_once_downloads_counts_for_download_manager_only(test_session, test_download_manager):
+    """Deleting a once-download (no Collection) does not change channels.yaml or repos.yaml: counting it would make
+    their next import save instead of import."""
+    from wrolpi.downloader import Download
+
+    download = Download(url='https://example.com/once')
+    test_session.add(download)
+    test_session.commit()
+    for name in ('channels.yaml', 'repos.yaml', 'download_manager.yaml'):
+        set_config_version(test_session, name, 1)
+    test_session.commit()
+
+    test_download_manager.delete_downloads_by_ids(test_session, [download.id])
+
+    assert db_version(test_session, 'download_manager.yaml') == 2
+    assert db_version(test_session, 'channels.yaml') == 1
+    assert db_version(test_session, 'repos.yaml') == 1
+
+
+def test_only_playlist_items_count_for_playlists(test_session):
+    """Author and subject Collections have items too; only a playlist's are saved by playlists.yaml."""
+    author = Collection(name='Author', kind='author')
+    playlist = Collection(name='Mix', kind='playlist')
+    test_session.add_all([author, playlist])
+    test_session.commit()
+    set_config_version(test_session, 'playlists.yaml', 1)
+    test_session.commit()
+
+    author.add_url(test_session, '/map?x=1', title='Spot')
+    test_session.commit()
+    assert db_version(test_session, 'playlists.yaml') == 1
+
+    playlist.add_url(test_session, '/map?x=1', title='Spot')
+    test_session.commit()
+    assert db_version(test_session, 'playlists.yaml') == 2