fix(recipes): validate FTS index from stored metadata instead of scanning

The browser extension's apiFetch treats any 404 as a missing endpoint and
retries the legacy non-/api/lm URL, producing two spurious
'error_middleware - WARNING - API GET ... 404' log lines per occurrence.

- complete_download_in_queue / update_download_queue_status /
  retry_download_from_history: 'not found' is a normal business outcome,
  return 200 + success:false instead of 404 (extension behavior unchanged;
  apiGet ignores the HTTP status)
- error_middleware: downgrade /api/lm/download-progress/ 404s to debug like
  previews - the 404 status itself stays (extension uses it for failure
  detection), only the log level is lowered

py/middleware/error_middleware.py

@@ -46,6 +46,16 @@ async def api_json_error(
         if request.path.startswith("/api/lm/previews") and exc.status == 404:
             logger_method = logger.debug
 
+        # Download-progress 404 is routine too: in-memory tracking is removed
+        # once a download finishes/fails, so the extension's final polls 404.
+        # The extension relies on the 404 status itself (failure detection),
+        # so only the log level is lowered.
+        if (
+            request.path.startswith("/api/lm/download-progress/")
+            and exc.status == 404
+        ):
+            logger_method = logger.debug
+
         logger_method(
             "API %s %s returned HTTP %d: %s",
             request.method,

py/routes/handlers/model_handlers.py

@@ -1998,9 +1998,11 @@ class ModelDownloadHandler:
                 item_id=item_id, download_id=download_id
             )
             if item is None:
+                # Missing or non-retryable history entry is a business
+                # outcome, not a routing error: 200 lets the extension's
+                # apiFetch 404-fallback and error middleware stay quiet.
                 return web.json_response(
-                    {"success": False, "error": "History item not found or not retryable"},
-                    status=404,
+                    {"success": False, "error": "History item not found or not retryable"}
                 )
             return web.json_response({"success": True, "item": item})
         except Exception as exc:
@@ -2051,8 +2053,12 @@ class ModelDownloadHandler:
                 completed_at=completed_at,
             )
             if item is None:
+                # A missing queue item (already completed, or never queued) is
+                # a normal business outcome, not a routing error. Return 200
+                # so the browser extension's apiFetch 404-fallback and the
+                # error middleware stay quiet.
                 return web.json_response(
-                    {"success": False, "error": "Download not found in queue"}, status=404
+                    {"success": False, "error": "Download not found in queue"}
                 )
             return web.json_response({"success": True, "item": item})
         except Exception as exc:
@@ -2094,9 +2100,10 @@ class ModelDownloadHandler:
             service = await DownloadQueueService.get_instance()
             updated = await service.update_status(download_id, status)
             if not updated:
+                # Same rationale as complete_download_in_queue: a missing
+                # queue item is a business outcome, not a routing error.
                 return web.json_response(
-                    {"success": False, "error": "Download not found in queue"},
-                    status=404,
+                    {"success": False, "error": "Download not found in queue"}
                 )
             return web.json_response({"success": True})
         except Exception as exc:

py/services/downloaded_version_history_service.py

@@ -236,24 +236,33 @@ class DownloadedVersionHistoryService:
             return
 
         async with self._lock:
-            conn = self._get_conn()
-            conn.executemany(
-                """
-                INSERT INTO downloaded_model_versions (
-                    model_type, version_id, model_id, first_seen_at, last_seen_at,
-                    source, last_file_path, last_library_name, is_deleted_override
-                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0)
-                ON CONFLICT(model_type, version_id) DO UPDATE SET
-                    model_id = COALESCE(excluded.model_id, downloaded_model_versions.model_id),
-                    last_seen_at = excluded.last_seen_at,
-                    source = excluded.source,
-                    last_file_path = COALESCE(excluded.last_file_path, downloaded_model_versions.last_file_path),
-                    last_library_name = COALESCE(excluded.last_library_name, downloaded_model_versions.last_library_name),
-                    is_deleted_override = 0
-                """,
-                payload,
-            )
-            conn.commit()
+            # The connection is created with check_same_thread=False and all
+            # access is serialized by self._lock, so the executemany upsert +
+            # commit can run in the default executor without blocking the
+            # event loop on large hydration payloads.
+            loop = asyncio.get_running_loop()
+            await loop.run_in_executor(None, self._mark_downloaded_bulk_sync, payload)
+
+    def _mark_downloaded_bulk_sync(self, payload: Sequence[tuple[object, ...]]) -> None:
+        """Synchronous executemany upsert + commit; runs in a worker thread."""
+        conn = self._get_conn()
+        conn.executemany(
+            """
+            INSERT INTO downloaded_model_versions (
+                model_type, version_id, model_id, first_seen_at, last_seen_at,
+                source, last_file_path, last_library_name, is_deleted_override
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0)
+            ON CONFLICT(model_type, version_id) DO UPDATE SET
+                model_id = COALESCE(excluded.model_id, downloaded_model_versions.model_id),
+                last_seen_at = excluded.last_seen_at,
+                source = excluded.source,
+                last_file_path = COALESCE(excluded.last_file_path, downloaded_model_versions.last_file_path),
+                last_library_name = COALESCE(excluded.last_library_name, downloaded_model_versions.last_library_name),
+                is_deleted_override = 0
+            """,
+            payload,
+        )
+        conn.commit()
 
     async def mark_as_deleted(self, model_type: str, version_id: int) -> None:
         normalized_type = _normalize_model_type(model_type)

py/services/model_scanner.py

@@ -535,16 +535,21 @@ class ModelScanner:
             self._is_initializing = False
     
     async def _load_persisted_cache(self, page_type: str) -> bool:
-        """Attempt to hydrate the in-memory cache from the SQLite snapshot."""
+        """Attempt to hydrate the in-memory cache from the SQLite snapshot.
+
+        The SQLite read and the per-model rebuild (entry adjustment, tag
+        counting, validation/repair, hash index reconstruction) run in the
+        default executor so the event loop stays responsive; only applying
+        the result to shared cache state happens on the loop.
+        """
         if not getattr(self, '_persistent_cache', None):
             return False
 
         loop = asyncio.get_event_loop()
         try:
-            persisted = await loop.run_in_executor(
+            rebuilt = await loop.run_in_executor(
                 None,
-                self._persistent_cache.load_cache,
-                self.model_type
+                self._rebuild_persisted_cache
             )
         except FileNotFoundError:
             return False
@@ -552,47 +557,14 @@ class ModelScanner:
             logger.debug("%s Scanner: Could not load persisted cache: %s", self.model_type.capitalize(), exc)
             return False
 
-        if not persisted or not persisted.raw_data:
+        if rebuilt is None:
             return False
 
-        hash_index = ModelHashIndex()
-        for sha_value, path in persisted.hash_rows:
-            if sha_value and path:
-                hash_index.add_entry(sha_value.lower(), path)
-
-        # Rebuild the AutoV3 index from the persisted autov3_index rows. These
-        # cover every known autov3 -> path mapping regardless of whether a
-        # sha256 row also exists for the same file.
-        for autov3_value, path in persisted.autov3_hash_rows:
-            if autov3_value and path:
-                hash_index.add_autov3(autov3_value.lower(), path)
-
-        tags_count: Dict[str, int] = {}
-        adjusted_raw_data: List[Dict[str, Any]] = []
-        for item in persisted.raw_data:
-            adjusted_item = self.adjust_cached_entry(dict(item))
-            adjusted_raw_data.append(adjusted_item)
-
-            for tag in adjusted_item.get('tags') or []:
-                tags_count[tag] = tags_count.get(tag, 0) + 1
-
-        # Validate cache entries and check health.
-        # Always use the validated/repaired entries — even when there are no
-        # invalid entries, auto_repair may have filled in missing optional
-        # fields (model_name, file_name, folder) with safe defaults on a copied
-        # working_entry.  Without this unconditional replacement the repaired
-        # copies are discarded and None values propagate to format_response.
-        # See issue #730.
-        valid_entries, invalid_entries = CacheEntryValidator.validate_batch(
-            adjusted_raw_data, auto_repair=True
-        )
-
-        # Always use the validated entries (repaired copies)
-        adjusted_raw_data = valid_entries
+        scan_result, invalid_entries = rebuilt
 
         if invalid_entries:
             monitor = CacheHealthMonitor()
-            report = monitor.check_health(adjusted_raw_data, auto_repair=True)
+            report = monitor.check_health(scan_result.raw_data, auto_repair=True)
 
             if report.status != CacheHealthStatus.HEALTHY:
                 # Broadcast health warning to frontend
@@ -602,31 +574,22 @@ class ModelScanner:
                     f"{report.invalid_entries} invalid entries, {report.repaired_entries} repaired"
                 )
 
-            # Use only valid entries
-            adjusted_raw_data = valid_entries
-
             # Rebuild tags count from valid entries only
             tags_count = {}
-            for item in adjusted_raw_data:
+            for item in scan_result.raw_data:
                 for tag in item.get('tags') or []:
                     tags_count[tag] = tags_count.get(tag, 0) + 1
+            scan_result.tags_count = tags_count
 
             # Remove invalid entries from hash index
             for invalid_entry in invalid_entries:
                 file_path = CacheEntryValidator.get_file_path_safe(invalid_entry)
                 sha256 = CacheEntryValidator.get_sha256_safe(invalid_entry)
                 if file_path:
-                    hash_index.remove_by_path(file_path, sha256)
-
-        scan_result = CacheBuildResult(
-            raw_data=adjusted_raw_data,
-            hash_index=hash_index,
-            tags_count=tags_count,
-            excluded_models=list(persisted.excluded_models)
-        )
+                    scan_result.hash_index.remove_by_path(file_path, sha256)
 
         await self._apply_scan_result(scan_result)
-        await self._sync_download_history(adjusted_raw_data, source='scan')
+        await self._sync_download_history(scan_result.raw_data, source='scan')
 
         await ws_manager.broadcast_init_progress({
             'stage': 'loading_cache',
@@ -651,6 +614,63 @@ class ModelScanner:
 
         return True
 
+    def _rebuild_persisted_cache(self) -> Optional[Tuple[CacheBuildResult, List[Dict[str, Any]]]]:
+        """Load the SQLite snapshot and rebuild a ready-to-apply scan result.
+
+        Runs entirely in a worker thread: it must not touch ``self._cache``,
+        the websocket manager, or any asyncio primitives. Returns ``None``
+        when no usable snapshot exists, otherwise a tuple of the scan result
+        (built from validated/repaired entries) and the invalid entries.
+        """
+        persisted = self._persistent_cache.load_cache(self.model_type)
+
+        if not persisted or not persisted.raw_data:
+            return None
+
+        hash_index = ModelHashIndex()
+        for sha_value, path in persisted.hash_rows:
+            if sha_value and path:
+                hash_index.add_entry(sha_value.lower(), path)
+
+        # Rebuild the AutoV3 index from the persisted autov3_index rows. These
+        # cover every known autov3 -> path mapping regardless of whether a
+        # sha256 row also exists for the same file.
+        for autov3_value, path in persisted.autov3_hash_rows:
+            if autov3_value and path:
+                hash_index.add_autov3(autov3_value.lower(), path)
+
+        tags_count: Dict[str, int] = {}
+        adjusted_raw_data: List[Dict[str, Any]] = []
+        for item in persisted.raw_data:
+            # load_cache builds a fresh dict per row, and validate_batch below
+            # works on its own per-entry copy when auto_repair=True, so no
+            # additional dict copy is needed here.
+            adjusted_item = self.adjust_cached_entry(item)
+            adjusted_raw_data.append(adjusted_item)
+
+            for tag in adjusted_item.get('tags') or []:
+                tags_count[tag] = tags_count.get(tag, 0) + 1
+
+        # Validate cache entries and check health.
+        # Always use the validated/repaired entries — even when there are no
+        # invalid entries, auto_repair may have filled in missing optional
+        # fields (model_name, file_name, folder) with safe defaults on a copied
+        # working_entry.  Without this unconditional replacement the repaired
+        # copies are discarded and None values propagate to format_response.
+        # See issue #730.
+        valid_entries, invalid_entries = CacheEntryValidator.validate_batch(
+            adjusted_raw_data, auto_repair=True
+        )
+
+        # Always use the validated entries (repaired copies)
+        scan_result = CacheBuildResult(
+            raw_data=valid_entries,
+            hash_index=hash_index,
+            tags_count=tags_count,
+            excluded_models=list(persisted.excluded_models)
+        )
+        return scan_result, invalid_entries
+
     async def _run_autov3_backfill(self) -> None:
         """Backfill autov3 for entries loaded from the persisted cache that lack it."""
         try:
@@ -1392,8 +1412,8 @@ class ModelScanner:
         else:
             self._cache.raw_data = list(scan_result.raw_data)
 
-        self._cache.rebuild_version_index()
-
+        # resort() rebuilds folders and the version index on every path, so a
+        # separate rebuild_version_index() call here would be redundant.
         await self._cache.resort()
 
         self._log_duplicate_filename_summary()

py/services/pending_delete_service.py

@@ -500,10 +500,36 @@ class PendingDeleteService:
         QUARANTINE them (preserving the pre-registry sweep semantics). The
         walk only descends into dirs literally named ``.lm-pending-delete``,
         so false positives are structurally limited.
+
+        The filesystem walk itself runs in a worker thread so a large or slow
+        library cannot block the event loop at startup; only the (rare) batch
+        registration awaits run on the loop.
+        """
+        roots = await self._get_all_model_roots()
+        loop = asyncio.get_event_loop()
+        staging_parents = await loop.run_in_executor(
+            None,  # Use default thread pool
+            self._collect_staging_parents,  # Run the tree walk off the loop
+            roots,
+        )
+        for staging_parent in staging_parents:
+            await self._register_batch_candidates(staging_parent)
+
+    def _collect_staging_parents(self, roots: Sequence[str]) -> List[str]:
+        """Walk every model root and return its staging-parent dirs.
+
+        Pure synchronous filesystem discovery with no awaits: walks with
+        ``followlinks=True, topdown=True``, prunes symlink cycles via a
+        per-root ``visited`` realpath set (realpath is used ONLY for this
+        dedup set - the returned paths are the unresolved business paths),
+        filters out :func:`_is_excluded_dir` dirs, and collects every dir
+        named ``.lm-pending-delete`` (including the case where a model root
+        itself is one). Results are returned in walk order.
         """
         from .model_scanner import _is_excluded_dir
 
-        for root in await self._get_all_model_roots():
+        staging_parents: List[str] = []
+        for root in roots:
             if not os.path.isdir(root):
                 continue
             visited: Set[str] = set()
@@ -518,21 +544,20 @@ class PendingDeleteService:
                 visited.add(real_dir)
                 if os.path.basename(dirpath) == PENDING_DELETE_DIR_NAME:
                     # The current dir IS a staging parent (reachable only when
-                    # a model root itself is one): register its batches.
-                    await self._register_batch_candidates(dirpath)
+                    # a model root itself is one): collect its batches.
+                    staging_parents.append(dirpath)
                     dirnames[:] = []
                     continue
                 next_dirs: List[str] = []
                 for name in dirnames:
                     if name == PENDING_DELETE_DIR_NAME:
-                        await self._register_batch_candidates(
-                            os.path.join(dirpath, name)
-                        )
+                        staging_parents.append(os.path.join(dirpath, name))
                     elif _is_excluded_dir(name):
                         continue
                     else:
                         next_dirs.append(name)
                 dirnames[:] = next_dirs
+        return staging_parents
 
     async def _register_batch_candidates(self, staging_parent: str) -> None:
         """Register every non-orphaned batch subdir of a staging parent."""

py/services/persistent_recipe_cache.py

@@ -333,6 +333,44 @@ class PersistentRecipeCache:
         except Exception as exc:
             logger.debug("Failed to persist image_id_map: %s", exc)
 
+    def get_metadata_value(self, key: str) -> Optional[str]:
+        """Return a value from cache_metadata, or None if missing."""
+        if not self.is_enabled() or not self._schema_initialized:
+            return None
+
+        try:
+            with self._db_lock:
+                conn = self._connect(readonly=True)
+                try:
+                    row = conn.execute(
+                        "SELECT value FROM cache_metadata WHERE key = ?",
+                        (key,),
+                    ).fetchone()
+                    return row["value"] if row else None
+                finally:
+                    conn.close()
+        except Exception:
+            return None
+
+    def set_metadata_value(self, key: str, value: str) -> None:
+        """Store a value in cache_metadata without rewriting the full cache."""
+        if not self.is_enabled() or not self._schema_initialized:
+            return
+
+        try:
+            with self._db_lock:
+                conn = self._connect()
+                try:
+                    conn.execute(
+                        "INSERT OR REPLACE INTO cache_metadata (key, value) VALUES (?, ?)",
+                        (key, value),
+                    )
+                    conn.commit()
+                finally:
+                    conn.close()
+        except Exception as exc:
+            logger.debug("Failed to persist cache metadata %s: %s", key, exc)
+
     def get_indexed_recipe_ids(self) -> Set[str]:
         """Return all recipe IDs in the cache.
 

py/services/recipe_fts_index.py

@@ -7,13 +7,14 @@ enabling sub-100ms search times even with 20k+ recipes.
 from __future__ import annotations
 
 import asyncio
+import hashlib
 import logging
 import os
 import re
 import sqlite3
 import threading
 import time
-from typing import Any, Dict, List, Optional, Set
+from typing import Any, Dict, List, Optional, Set, Tuple
 
 from ..utils.cache_paths import CacheType, resolve_cache_path_with_migration
 
@@ -165,6 +166,7 @@ class RecipeFTSIndex:
                     batch_size = 500
                     total = len(recipes)
                     inserted = 0
+                    indexed_ids: Set[str] = set()
 
                     for i in range(0, total, batch_size):
                         batch = recipes[i:i + batch_size]
@@ -179,6 +181,7 @@ class RecipeFTSIndex:
                             row = self._prepare_fts_row(recipe)
                             rows.append(row)
                             inserted += 1
+                            indexed_ids.add(recipe_id)
 
                         if rows:
                             # Insert into FTS table
@@ -213,7 +216,11 @@ class RecipeFTSIndex:
                     )
                     conn.execute(
                         "INSERT OR REPLACE INTO fts_metadata (key, value) VALUES (?, ?)",
-                        ('recipe_count', str(inserted))
+                        (self._COUNT_METADATA_KEY, str(inserted))
+                    )
+                    conn.execute(
+                        "INSERT OR REPLACE INTO fts_metadata (key, value) VALUES (?, ?)",
+                        (self._FINGERPRINT_METADATA_KEY, self._compute_ids_fingerprint(indexed_ids))
                     )
 
                     conn.commit()
@@ -288,6 +295,12 @@ class RecipeFTSIndex:
             with self._lock:
                 conn = self._connect()
                 try:
+                    # Check existence via the rowid mapping (fast PK lookup)
+                    existed = conn.execute(
+                        "SELECT 1 FROM recipe_rowid WHERE recipe_id = ?",
+                        (recipe_id,)
+                    ).fetchone() is not None
+
                     # Remove existing entry if present
                     self._remove_recipe_locked(conn, recipe_id)
 
@@ -312,6 +325,10 @@ class RecipeFTSIndex:
                             (recipe_id, result[0])
                         )
 
+                    # Keep validation metadata in sync (only a new id changes it)
+                    if not existed:
+                        self._update_mutation_metadata_locked(conn, recipe_id, delta=1)
+
                     conn.commit()
                     return True
                 finally:
@@ -339,7 +356,13 @@ class RecipeFTSIndex:
             with self._lock:
                 conn = self._connect()
                 try:
+                    existed = conn.execute(
+                        "SELECT 1 FROM recipe_rowid WHERE recipe_id = ?",
+                        (recipe_id,)
+                    ).fetchone() is not None
                     self._remove_recipe_locked(conn, recipe_id)
+                    if existed:
+                        self._update_mutation_metadata_locked(conn, recipe_id, delta=-1)
                     conn.commit()
                     return True
                 finally:
@@ -371,6 +394,15 @@ class RecipeFTSIndex:
                 try:
                     conn.execute("DELETE FROM recipe_fts")
                     conn.execute("DELETE FROM recipe_rowid")
+                    # Reset validation metadata to the empty index state
+                    conn.execute(
+                        "INSERT OR REPLACE INTO fts_metadata (key, value) VALUES (?, ?)",
+                        (self._COUNT_METADATA_KEY, '0')
+                    )
+                    conn.execute(
+                        "INSERT OR REPLACE INTO fts_metadata (key, value) VALUES (?, ?)",
+                        (self._FINGERPRINT_METADATA_KEY, self._compute_ids_fingerprint(set()))
+                    )
                     conn.commit()
                     self._ready.clear()
                     return True
@@ -427,10 +459,12 @@ class RecipeFTSIndex:
         """Check if the FTS index matches the expected recipes.
 
         This method validates whether the existing FTS index can be reused
-        without a full rebuild. It checks:
-        1. The index has been initialized
-        2. The count matches
-        3. The recipe IDs match
+        without a full rebuild. It compares the expected count and recipe ID
+        fingerprint against metadata recorded when the index was (re)built,
+        so it does not scan the FTS content table. Indexes built by older
+        versions lack this metadata; for those the validation falls back to
+        a one-time scan of the content table and records the metadata so
+        subsequent startups are cheap.
 
         Args:
             recipe_count: Expected number of recipes.
@@ -446,7 +480,28 @@ class RecipeFTSIndex:
             return False
 
         try:
+            metadata = self._read_validation_metadata()
+            if metadata is not None:
+                stored_count, stored_fingerprint = metadata
+                if stored_count != recipe_count:
+                    logger.debug(
+                        "FTS index count mismatch: indexed=%d, expected=%d",
+                        stored_count, recipe_count
+                    )
+                    return False
+
+                if stored_fingerprint != self._compute_ids_fingerprint(recipe_ids):
+                    logger.debug("FTS index recipe ID fingerprint mismatch")
+                    return False
+
+                return True
+
+            # Legacy fallback: no stored metadata, scan the content table once
+            # and persist the metadata so later validations are cheap.
             indexed_count = self.get_indexed_count()
+            indexed_ids = self.get_indexed_recipe_ids()
+            self._store_validation_metadata(indexed_count, indexed_ids)
+
             if indexed_count != recipe_count:
                 logger.debug(
                     "FTS index count mismatch: indexed=%d, expected=%d",
@@ -454,7 +509,6 @@ class RecipeFTSIndex:
                 )
                 return False
 
-            indexed_ids = self.get_indexed_recipe_ids()
             if indexed_ids != recipe_ids:
                 missing = recipe_ids - indexed_ids
                 extra = indexed_ids - recipe_ids
@@ -471,6 +525,112 @@ class RecipeFTSIndex:
 
     # Internal helpers
 
+    _FINGERPRINT_METADATA_KEY = 'recipe_ids_fingerprint'
+    _COUNT_METADATA_KEY = 'recipe_count'
+
+    @staticmethod
+    def _fingerprint_recipe_id(recipe_id: str) -> int:
+        """Return a stable 64-bit fingerprint contribution for a recipe ID."""
+        digest = hashlib.sha256(recipe_id.encode("utf-8")).digest()
+        return int.from_bytes(digest[:8], "big")
+
+    @classmethod
+    def _compute_ids_fingerprint(cls, recipe_ids: Set[str]) -> str:
+        """Order-independent fingerprint of a recipe ID set (XOR of per-id hashes)."""
+        fingerprint = 0
+        for recipe_id in recipe_ids:
+            fingerprint ^= cls._fingerprint_recipe_id(str(recipe_id))
+        return f"{fingerprint:016x}"
+
+    def _read_validation_metadata(self) -> Optional[Tuple[int, str]]:
+        """Return stored (recipe count, ID fingerprint), or None if absent."""
+        try:
+            with self._lock:
+                conn = self._connect(readonly=True)
+                try:
+                    rows = conn.execute(
+                        "SELECT key, value FROM fts_metadata WHERE key IN (?, ?)",
+                        (self._COUNT_METADATA_KEY, self._FINGERPRINT_METADATA_KEY)
+                    ).fetchall()
+                    values = {row[0]: row[1] for row in rows}
+                    fingerprint = values.get(self._FINGERPRINT_METADATA_KEY)
+                    if fingerprint is None:
+                        return None
+                    try:
+                        count = int(values.get(self._COUNT_METADATA_KEY) or 0)
+                    except (TypeError, ValueError):
+                        return None
+                    return count, fingerprint
+                finally:
+                    conn.close()
+        except FileNotFoundError:
+            return None
+        except Exception as exc:
+            logger.debug("Failed to read FTS validation metadata: %s", exc)
+            return None
+
+    def _store_validation_metadata(self, recipe_count: int, recipe_ids: Set[str]) -> None:
+        """Persist recipe count and ID fingerprint for cheap future validation."""
+        try:
+            with self._lock:
+                conn = self._connect()
+                try:
+                    conn.execute(
+                        "INSERT OR REPLACE INTO fts_metadata (key, value) VALUES (?, ?)",
+                        (self._COUNT_METADATA_KEY, str(recipe_count))
+                    )
+                    conn.execute(
+                        "INSERT OR REPLACE INTO fts_metadata (key, value) VALUES (?, ?)",
+                        (self._FINGERPRINT_METADATA_KEY, self._compute_ids_fingerprint(recipe_ids))
+                    )
+                    conn.commit()
+                finally:
+                    conn.close()
+        except Exception as exc:
+            logger.debug("Failed to store FTS validation metadata: %s", exc)
+
+    def _update_mutation_metadata_locked(
+        self,
+        conn: sqlite3.Connection,
+        recipe_id: str,
+        delta: int,
+    ) -> None:
+        """Incrementally maintain validation metadata after add/remove.
+
+        Caller must hold the lock. The fingerprint is only updated when it
+        already exists; without it, validation falls back to a one-time scan
+        that records fresh metadata.
+        """
+        fingerprint_row = conn.execute(
+            "SELECT value FROM fts_metadata WHERE key = ?",
+            (self._FINGERPRINT_METADATA_KEY,)
+        ).fetchone()
+        if fingerprint_row and fingerprint_row[0]:
+            try:
+                fingerprint = int(fingerprint_row[0], 16)
+            except ValueError:
+                fingerprint = None
+            if fingerprint is not None:
+                fingerprint ^= self._fingerprint_recipe_id(recipe_id)
+                conn.execute(
+                    "INSERT OR REPLACE INTO fts_metadata (key, value) VALUES (?, ?)",
+                    (self._FINGERPRINT_METADATA_KEY, f"{fingerprint & 0xFFFFFFFFFFFFFFFF:016x}")
+                )
+
+        count_row = conn.execute(
+            "SELECT value FROM fts_metadata WHERE key = ?",
+            (self._COUNT_METADATA_KEY,)
+        ).fetchone()
+        if count_row:
+            try:
+                count = max(0, int(count_row[0] or 0) + delta)
+            except (TypeError, ValueError):
+                return
+            conn.execute(
+                "INSERT OR REPLACE INTO fts_metadata (key, value) VALUES (?, ?)",
+                (self._COUNT_METADATA_KEY, str(count))
+            )
+
     def _connect(self, readonly: bool = False) -> sqlite3.Connection:
         """Create a database connection."""
         uri = False

py/services/recipe_scanner.py

@@ -1547,7 +1547,7 @@ class RecipeScanner:
                     self._cache.raw_data = recipes
                     self._update_folder_metadata(self._cache)
                     self._sort_cache_sync()
-                    # Backfill source_path from JSON files if missing (schema migration)
+                    # Backfill source_path from JSON files if missing (one-shot schema migration)
                     if self._backfill_source_path_if_needed(recipes, json_paths):
                         self._cache.image_id_map = self._build_image_id_map()
                         self._persistent_cache.save_cache(
@@ -1574,7 +1574,7 @@ class RecipeScanner:
                     self._cache.raw_data = recipes
                     self._update_folder_metadata(self._cache)
                     self._sort_cache_sync()
-                    # Backfill source_path from JSON files if missing (schema migration)
+                    # Backfill source_path from JSON files if missing (one-shot schema migration)
                     self._backfill_source_path_if_needed(recipes, json_paths)
                     self._cache.image_id_map = self._build_image_id_map()
                     # Persist updated cache
@@ -1711,6 +1711,9 @@ class RecipeScanner:
 
         return recipes, changed, json_paths
 
+    # Metadata key recording that the one-shot source_path backfill has run.
+    _SOURCE_PATH_BACKFILL_MARKER = "source_path_backfilled"
+
     def _backfill_source_path_if_needed(
         self,
         recipes: List[Dict[str, Any]],
@@ -1718,8 +1721,21 @@ class RecipeScanner:
     ) -> bool:
         """Backfill source_path from recipe JSON files if missing from cache.
 
+        This is a one-shot schema migration: once it has run, a completion
+        marker is stored in the persistent cache metadata and later startups
+        skip it entirely. Recipes without a source_path in their JSON file
+        would otherwise be re-read and re-parsed on every startup. New or
+        changed recipe files still get source_path from the normal parse path
+        during reconciliation.
+
         Returns True if any recipes were updated (caller should persist cache).
         """
+        cache = self._persistent_cache
+        if (
+            cache is not None
+            and cache.get_metadata_value(self._SOURCE_PATH_BACKFILL_MARKER) == "1"
+        ):
+            return False
         updated = False
         for recipe in recipes:
             if recipe.get("source_path"):
@@ -1737,6 +1753,8 @@ class RecipeScanner:
                     updated = True
             except Exception:
                 pass
+        if cache is not None:
+            cache.set_metadata_value(self._SOURCE_PATH_BACKFILL_MARKER, "1")
         return updated
 
     def _full_directory_scan_sync(

tests/routes/test_download_queue_handlers.py

@@ -0,0 +1,186 @@
+"""Handler-level tests for download queue terminal/status transitions.
+
+Regression test: a "not found in queue" outcome must be returned as HTTP 200
+with ``success: false``, not 404. The browser extension's apiFetch treats any
+404 as a missing endpoint and retries a legacy URL, producing spurious
+``/api/downloads/queue/complete ... 404`` warnings on every completion.
+"""
+
+import json
+import logging
+from pathlib import Path
+
+import pytest
+from aiohttp import web
+from aiohttp.test_utils import make_mocked_request
+
+from py.routes.handlers.model_handlers import ModelDownloadHandler
+from py.services.download_queue_service import DownloadQueueService
+
+
+def _make_handler() -> ModelDownloadHandler:
+    return ModelDownloadHandler(
+        ws_manager=None,  # pyright: ignore[reportArgumentType] - unused by queue endpoints
+        logger=logging.getLogger("test-download-queue"),
+        download_use_case=None,  # pyright: ignore[reportArgumentType] - unused by queue endpoints
+        download_coordinator=None,  # pyright: ignore[reportArgumentType] - unused by queue endpoints
+    )
+
+
+def _queue_request(path: str, query: dict[str, str]) -> web.Request:
+    query_string = "&".join(f"{key}={value}" for key, value in query.items())
+    return make_mocked_request("GET", f"{path}?{query_string}")
+
+
+@pytest.fixture
+def queue_service(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> DownloadQueueService:
+    """Return a tmp-backed DownloadQueueService and stub the singleton."""
+    service = DownloadQueueService(db_path=str(tmp_path / "queue.sqlite"))
+
+    async def fake_get_instance(_cls: object = None) -> DownloadQueueService:
+        return service
+
+    monkeypatch.setattr(DownloadQueueService, "get_instance", fake_get_instance)
+    return service
+
+
+@pytest.mark.asyncio
+async def test_complete_missing_download_returns_200_not_404(
+    queue_service: DownloadQueueService,
+) -> None:
+    """Completing a download that is not queued is a business outcome."""
+    handler = _make_handler()
+    response = await handler.complete_download_in_queue(
+        _queue_request(
+            "/api/lm/downloads/queue/complete",
+            {"download_id": "dl-nope", "status": "completed"},
+        )
+    )
+    assert response.status == 200
+    text = response.text
+    assert text is not None
+    assert json.loads(text) == {
+        "success": False,
+        "error": "Download not found in queue",
+    }
+    # The failed completion must not have side effects on the queue.
+    assert await queue_service.get_queue() == []
+
+
+@pytest.mark.asyncio
+async def test_complete_queued_download_returns_success(
+    queue_service: DownloadQueueService,
+) -> None:
+    """The happy path still moves the item to history with HTTP 200."""
+    await queue_service.add_to_queue(download_id="dl-1", model_id=1)
+    handler = _make_handler()
+    response = await handler.complete_download_in_queue(
+        _queue_request(
+            "/api/lm/downloads/queue/complete",
+            {"download_id": "dl-1", "status": "completed"},
+        )
+    )
+    assert response.status == 200
+    text = response.text
+    assert text is not None
+    payload = json.loads(text)
+    assert payload["success"] is True
+    # The returned item reflects the pre-transition queue record; the
+    # terminal status lands in history.
+    history = await queue_service.get_history()
+    assert len(history["items"]) == 1
+    assert history["items"][0]["status"] == "completed"
+
+
+@pytest.mark.asyncio
+async def test_status_missing_download_returns_200_not_404(
+    queue_service: DownloadQueueService,
+) -> None:
+    """Status updates for unknown items also return 200 with success: false."""
+    handler = _make_handler()
+    response = await handler.update_download_queue_status(
+        _queue_request(
+            "/api/lm/downloads/queue/status",
+            {"download_id": "dl-nope", "status": "downloading"},
+        )
+    )
+    assert response.status == 200
+    text = response.text
+    assert text is not None
+    assert json.loads(text) == {
+        "success": False,
+        "error": "Download not found in queue",
+    }
+    assert await queue_service.get_queue() == []
+
+
+@pytest.mark.asyncio
+async def test_status_queued_download_returns_success(
+    queue_service: DownloadQueueService,
+) -> None:
+    """The happy path still updates the queue item with HTTP 200."""
+    await queue_service.add_to_queue(download_id="dl-2", model_id=2)
+    handler = _make_handler()
+    response = await handler.update_download_queue_status(
+        _queue_request(
+            "/api/lm/downloads/queue/status",
+            {"download_id": "dl-2", "status": "downloading"},
+        )
+    )
+    assert response.status == 200
+    text = response.text
+    assert text is not None
+    assert json.loads(text) == {"success": True}
+
+
+@pytest.mark.asyncio
+async def test_retry_missing_history_returns_200_not_404(
+    queue_service: DownloadQueueService,
+) -> None:
+    """Retrying a history entry that no longer exists is a business outcome."""
+    handler = _make_handler()
+    response = await handler.retry_download_from_history(
+        _queue_request(
+            "/api/lm/downloads/history/retry",
+            {"download_id": "dl-nope"},
+        )
+    )
+    assert response.status == 200
+    text = response.text
+    assert text is not None
+    assert json.loads(text) == {
+        "success": False,
+        "error": "History item not found or not retryable",
+    }
+    # No side effects: history and queue stay empty.
+    history = await queue_service.get_history()
+    assert history["items"] == []
+    assert await queue_service.get_queue() == []
+
+
+@pytest.mark.asyncio
+async def test_retry_failed_history_returns_success(
+    queue_service: DownloadQueueService,
+) -> None:
+    """The happy path still re-queues a retryable history entry with HTTP 200."""
+    await queue_service.add_to_history(
+        download_id="dl-fail", model_id=1, status="failed"
+    )
+    handler = _make_handler()
+    response = await handler.retry_download_from_history(
+        _queue_request(
+            "/api/lm/downloads/history/retry",
+            {"download_id": "dl-fail"},
+        )
+    )
+    assert response.status == 200
+    text = response.text
+    assert text is not None
+    payload = json.loads(text)
+    assert payload["success"] is True
+    # The retried item is re-queued under a fresh download_id.
+    queue = await queue_service.get_queue()
+    assert len(queue) == 1
+    assert queue[0]["status"] == "queued"

tests/services/test_downloaded_version_history_service.py

@@ -1,5 +1,7 @@
 from pathlib import Path
 
+import threading
+
 import pytest
 
 from py.services.downloaded_version_history_service import (
@@ -70,6 +72,37 @@ async def test_download_history_bulk_lookup(tmp_path: Path) -> None:
     }
 
 
+@pytest.mark.asyncio
+async def test_mark_downloaded_bulk_writes_off_event_loop(
+    tmp_path: Path, monkeypatch
+) -> None:
+    """The executemany upsert + commit must not run on the event loop thread."""
+    db_path = tmp_path / "download-history.sqlite"
+    service = DownloadedVersionHistoryService(
+        str(db_path),
+        settings_manager=DummySettings(),
+    )
+
+    loop_thread = threading.get_ident()
+    write_threads: list[int] = []
+    original_write = service._mark_downloaded_bulk_sync
+
+    def tracking_write(payload):
+        write_threads.append(threading.get_ident())
+        return original_write(payload)
+
+    monkeypatch.setattr(service, "_mark_downloaded_bulk_sync", tracking_write)
+
+    await service.mark_downloaded_bulk(
+        "lora",
+        [{"model_id": 7, "version_id": 701, "file_path": "/m/x.safetensors"}],
+        source="scan",
+    )
+
+    assert write_threads and write_threads[0] != loop_thread
+    assert await service.has_been_downloaded("lora", 701) is True
+
+
 @pytest.mark.asyncio
 async def test_per_file_history_tracking(tmp_path: Path) -> None:
     """Per-file records coexist with the version-level row (#1058)."""

tests/services/test_model_scanner.py

@@ -4,6 +4,7 @@ import asyncio
 import json
 import os
 import sqlite3
+import threading
 import time
 from collections.abc import Iterator
 from pathlib import Path
@@ -391,6 +392,74 @@ async def test_load_persisted_cache_populates_cache(tmp_path: Path, monkeypatch)
     assert ws_stub.payloads[-1]['progress'] == 1
 
 
+@pytest.mark.asyncio
+async def test_load_persisted_cache_rebuilds_off_event_loop(tmp_path: Path, monkeypatch):
+    """The SQLite read and per-model rebuild must not run on the event loop."""
+    monkeypatch.setenv('LORA_MANAGER_DISABLE_PERSISTENT_CACHE', '0')
+    db_path = tmp_path / 'cache.sqlite'
+    store = PersistentModelCache(db_path=str(db_path))
+
+    file_path = tmp_path / 'one.txt'
+    file_path.write_text('one', encoding='utf-8')
+    normalized = _normalize_path(file_path)
+
+    raw_model = {
+        'file_path': normalized,
+        'file_name': 'one',
+        'model_name': 'one',
+        'folder': '',
+        'size': 3,
+        'modified': 123.0,
+        'sha256': 'hash-one',
+        'base_model': 'test',
+        'preview_url': '',
+        'preview_nsfw_level': 0,
+        'from_civitai': True,
+        'favorite': False,
+        'notes': '',
+        'usage_tips': '',
+        'exclude': False,
+        'db_checked': False,
+        'last_checked_at': 0.0,
+        'tags': ['alpha'],
+        'civitai': {'id': 11, 'modelId': 22, 'name': 'ver', 'trainedWords': ['abc']},
+    }
+
+    store.save_cache('dummy', [raw_model], {'hash-one': [normalized]}, [])
+
+    monkeypatch.setattr(model_scanner, 'get_persistent_cache', lambda: store)
+
+    scanner = DummyScanner(tmp_path)
+    ws_stub = RecordingWebSocketManager()
+    monkeypatch.setattr(model_scanner, 'ws_manager', ws_stub)
+
+    loop_thread = threading.get_ident()
+    worker_threads: List[int] = []
+
+    original_load_cache = store.load_cache
+    def tracking_load_cache(model_type):
+        worker_threads.append(threading.get_ident())
+        return original_load_cache(model_type)
+    monkeypatch.setattr(store, 'load_cache', tracking_load_cache)
+
+    original_adjust = scanner.adjust_cached_entry
+    def tracking_adjust(entry):
+        worker_threads.append(threading.get_ident())
+        return original_adjust(entry)
+    monkeypatch.setattr(scanner, 'adjust_cached_entry', tracking_adjust)
+
+    loaded = await scanner._load_persisted_cache('dummy')
+    assert loaded is True
+
+    # Both the SQLite read and the per-entry adjustment ran off the loop
+    assert len(worker_threads) == 2
+    assert all(tid != loop_thread for tid in worker_threads)
+
+    cache = await scanner.get_cached_data()
+    assert len(cache.raw_data) == 1
+    assert cache.raw_data[0]['file_path'] == normalized
+
+
 @pytest.mark.asyncio
 async def test_update_single_model_cache_persists_changes(tmp_path: Path, monkeypatch):
     monkeypatch.setenv('LORA_MANAGER_DISABLE_PERSISTENT_CACHE', '0')

tests/services/test_pending_delete_service.py

@@ -16,6 +16,7 @@ import errno
 import json
 import os
 import shutil
+import threading
 import time
 from collections.abc import Iterator
 from pathlib import Path
@@ -1683,6 +1684,33 @@ async def test_reg_g_reconciliation_finds_external_batches(
     assert service._known_batch_dirs.get("ext-fresh") == str(fresh_dir)
 
 
+# (g2) reconciliation runs the filesystem walk off the event loop so a large
+#      or slow library cannot block startup
+async def test_reg_g2_reconciliation_walk_runs_in_worker_thread(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    root = tmp_path / "loras"
+    root.mkdir()
+    (root / PENDING_DELETE_DIR_NAME).mkdir()
+    await _register_model_root(monkeypatch, lora_roots=[root])
+
+    loop_thread = threading.get_ident()
+    walk_threads: List[int] = []
+    real_walk = os.walk
+
+    def _recording_walk(*args: Any, **kwargs: Any):
+        walk_threads.append(threading.get_ident())
+        return real_walk(*args, **kwargs)
+
+    monkeypatch.setattr(os, "walk", _recording_walk)
+
+    service = await PendingDeleteService.get_instance()
+    await service._reconcile_scan_roots()
+
+    assert walk_threads, "reconciliation never walked the model roots"
+    assert all(thread_id != loop_thread for thread_id in walk_threads)
+
+
 # (h) _find_batch_dir with cleared registry locates + registers (restart sim)
 async def test_reg_h_find_batch_dir_restart_simulation(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch

tests/services/test_recipe_scanner_source_path_backfill.py

@@ -0,0 +1,104 @@
+"""Tests for the one-shot source_path backfill in RecipeScanner."""
+
+import json
+from pathlib import Path
+
+import pytest
+
+from py.services.persistent_recipe_cache import PersistentRecipeCache
+from py.services.recipe_scanner import RecipeScanner
+
+
+@pytest.fixture
+def scanner_with_cache(tmp_path: Path):
+    """RecipeScanner instance backed by a real persistent cache in tmp_path."""
+    cache = PersistentRecipeCache(db_path=str(tmp_path / "recipe_cache.sqlite"))
+    scanner = RecipeScanner.__new__(RecipeScanner)
+    scanner._persistent_cache = cache
+    return scanner, cache
+
+
+def _write_recipe_json(path: Path, recipe_id: str, source_path: str | None = None) -> None:
+    data = {"id": recipe_id}
+    if source_path is not None:
+        data["source_path"] = source_path
+    path.write_text(json.dumps(data), encoding="utf-8")
+
+
+def test_backfill_processes_recipes_when_marker_absent(scanner_with_cache, tmp_path: Path):
+    """Without the completion marker, missing source_path values are backfilled."""
+    scanner, cache = scanner_with_cache
+
+    json_with_source = tmp_path / "r1.recipe.json"
+    _write_recipe_json(json_with_source, "r1", source_path="https://civitai.com/images/1")
+    json_without_source = tmp_path / "r2.recipe.json"
+    _write_recipe_json(json_without_source, "r2")
+
+    recipes = [
+        {"id": "r1", "source_path": ""},
+        {"id": "r2", "source_path": ""},
+        {"id": "r3", "source_path": "https://civitai.com/images/3"},
+    ]
+    json_paths = {"r1": str(json_with_source), "r2": str(json_without_source)}
+
+    updated = scanner._backfill_source_path_if_needed(recipes, json_paths)
+
+    assert updated is True
+    assert recipes[0]["source_path"] == "https://civitai.com/images/1"
+    # JSON legitimately has no source_path: stays empty, no error
+    assert recipes[1]["source_path"] == ""
+    assert recipes[2]["source_path"] == "https://civitai.com/images/3"
+    # The run records the completion marker
+    assert (
+        cache.get_metadata_value(RecipeScanner._SOURCE_PATH_BACKFILL_MARKER) == "1"
+    )
+
+
+def test_backfill_is_skipped_once_marker_is_set(scanner_with_cache, tmp_path: Path, monkeypatch):
+    """The second initialization must not re-read recipe JSON files."""
+    scanner, cache = scanner_with_cache
+
+    json_path = tmp_path / "r1.recipe.json"
+    _write_recipe_json(json_path, "r1", source_path="https://civitai.com/images/1")
+
+    recipes = [{"id": "r1", "source_path": ""}]
+    json_paths = {"r1": str(json_path)}
+
+    # First run: backfills and records the marker
+    assert scanner._backfill_source_path_if_needed(recipes, json_paths) is True
+    assert recipes[0]["source_path"] == "https://civitai.com/images/1"
+
+    # Second run: simulate a fresh startup where the cache still lacks
+    # source_path. Under the old behavior the file would be re-read and
+    # re-parsed; now the marker must suppress any file access.
+    recipes = [{"id": "r1", "source_path": ""}]
+
+    import os
+
+    real_exists = os.path.exists
+
+    def _failing_exists(path):
+        if str(path).endswith(".recipe.json"):
+            raise AssertionError("backfill touched the filesystem despite marker")
+        return real_exists(path)
+
+    monkeypatch.setattr(os.path, "exists", _failing_exists)
+
+    updated = scanner._backfill_source_path_if_needed(recipes, json_paths)
+
+    assert updated is False
+    assert recipes[0]["source_path"] == ""
+
+
+def test_backfill_marker_does_not_suppress_reconcile_parsed_source_path(scanner_with_cache):
+    """The marker only gates the backfill; parsed recipes keep their source_path."""
+    scanner, cache = scanner_with_cache
+    cache.set_metadata_value(RecipeScanner._SOURCE_PATH_BACKFILL_MARKER, "1")
+
+    # A recipe that arrived from the normal parse path with a source_path is
+    # left untouched by the (skipped) backfill.
+    recipes = [{"id": "r1", "source_path": "https://civitai.com/images/9"}]
+    updated = scanner._backfill_source_path_if_needed(recipes, {})
+
+    assert updated is False
+    assert recipes[0]["source_path"] == "https://civitai.com/images/9"

tests/test_persistent_recipe_cache.py

@@ -552,6 +552,27 @@ class TestPersistentRecipeCache:
         assert loaded is not None
         assert loaded.image_id_map == {"222": "new-only"}
 
+    def test_metadata_value_roundtrip(self, temp_db_path):
+        """set_metadata_value/get_metadata_value store and replace values."""
+        cache = PersistentRecipeCache(db_path=temp_db_path)
+
+        assert cache.get_metadata_value("source_path_backfilled") is None
+
+        cache.set_metadata_value("source_path_backfilled", "1")
+        assert cache.get_metadata_value("source_path_backfilled") == "1"
+
+        cache.set_metadata_value("source_path_backfilled", "2")
+        assert cache.get_metadata_value("source_path_backfilled") == "2"
+
+    def test_metadata_value_survives_save_cache(self, temp_db_path, sample_recipes):
+        """A full save_cache must not drop unrelated cache_metadata entries."""
+        cache = PersistentRecipeCache(db_path=temp_db_path)
+
+        cache.set_metadata_value("source_path_backfilled", "1")
+        cache.save_cache(sample_recipes)
+
+        assert cache.get_metadata_value("source_path_backfilled") == "1"
+
 
 class TestHasWorkflowColumn:
     """has_workflow column persistence (plan 3.1)."""

tests/test_recipe_fts_index_validation.py

@@ -181,3 +181,88 @@ class TestFTSIndexValidation:
         # Search should still work
         results = fts.search("anime")
         assert "recipe-001" in results
+
+    @staticmethod
+    def _forbid_content_scan(monkeypatch):
+        """Fail the test if validation falls back to scanning the FTS content table."""
+
+        def _raise(*args, **kwargs):
+            raise AssertionError("validate_index scanned the FTS content table")
+
+        monkeypatch.setattr(RecipeFTSIndex, "get_indexed_count", _raise)
+        monkeypatch.setattr(RecipeFTSIndex, "get_indexed_recipe_ids", _raise)
+
+    def test_validate_index_with_metadata_does_not_scan(
+        self, temp_db_path, sample_recipes, monkeypatch
+    ):
+        """Validation must rely on stored metadata, not content-table scans."""
+        fts = RecipeFTSIndex(db_path=temp_db_path)
+        fts.build_index(sample_recipes)
+
+        self._forbid_content_scan(monkeypatch)
+
+        result = fts.validate_index(3, {"recipe-001", "recipe-002", "recipe-003"})
+        assert result is True
+
+        # Mismatches are also detected from metadata alone
+        result = fts.validate_index(4, {"recipe-001", "recipe-002", "recipe-003"})
+        assert result is False
+        result = fts.validate_index(3, {"recipe-001", "recipe-002", "recipe-999"})
+        assert result is False
+
+    def test_validate_index_legacy_fallback_writes_metadata(
+        self, temp_db_path, sample_recipes, monkeypatch
+    ):
+        """Indexes built by older versions validate via a one-time scan, then metadata."""
+        import sqlite3
+
+        fts = RecipeFTSIndex(db_path=temp_db_path)
+        fts.build_index(sample_recipes)
+
+        # Simulate a legacy index: drop the fingerprint metadata
+        conn = sqlite3.connect(temp_db_path)
+        try:
+            conn.execute(
+                "DELETE FROM fts_metadata WHERE key = ?",
+                (RecipeFTSIndex._FINGERPRINT_METADATA_KEY,)
+            )
+            conn.commit()
+        finally:
+            conn.close()
+
+        # Fallback validation scans once and succeeds
+        result = fts.validate_index(3, {"recipe-001", "recipe-002", "recipe-003"})
+        assert result is True
+
+        # Metadata was written, so the next validation needs no scan
+        self._forbid_content_scan(monkeypatch)
+        result = fts.validate_index(3, {"recipe-001", "recipe-002", "recipe-003"})
+        assert result is True
+
+    def test_validate_index_metadata_tracks_incremental_mutations(
+        self, temp_db_path, sample_recipes, monkeypatch
+    ):
+        """add_recipe/remove_recipe keep the validation metadata in sync."""
+        fts = RecipeFTSIndex(db_path=temp_db_path)
+        fts.build_index(sample_recipes[:2])
+
+        fts.add_recipe(sample_recipes[2])
+        fts.remove_recipe("recipe-001")
+
+        self._forbid_content_scan(monkeypatch)
+
+        result = fts.validate_index(2, {"recipe-002", "recipe-003"})
+        assert result is True
+
+    def test_validate_index_after_clear_uses_metadata(
+        self, temp_db_path, sample_recipes, monkeypatch
+    ):
+        """clear() resets the validation metadata to the empty index state."""
+        fts = RecipeFTSIndex(db_path=temp_db_path)
+        fts.build_index(sample_recipes)
+        fts.clear()
+
+        self._forbid_content_scan(monkeypatch)
+
+        assert fts.validate_index(0, set()) is True
+        assert fts.validate_index(3, {"recipe-001", "recipe-002", "recipe-003"}) is False