feat(discovery): AI tagging, semantic search, similarity, duplicate finder, folder management

Adds a full discovery and organisation layer to the gallery. ## AI Tagger - WD tagger (ONNX via ort) with DirectML/CPU acceleration - Per-image and per-folder tag queuing; configurable batch size and confidence threshold; model downloaded on first use via ureq/zip - Tags stored with source ('ai' | 'user'); user tags are never overwritten by AI; AI tags protected from accidental promotion - Tagger pause/resume per folder; in-flight batches discarded cleanly on pause or cancellation without leaving jobs stuck in processing ## Semantic & Tag Search - CLIP text-query embedding via candle (HuggingFace hub model) - Progressive candidate doubling with filter post-processing so folder/rating/media-kind filters do not silently under-return results - Tag search with DB-backed pagination (offset + COUNT total) - Filename search unchanged; prefix syntax: s:, t:, f: in search bar ## Similarity Search - HNSW in-memory index (hnsw_rs) with monotonic embedding_revision counter for cache invalidation; build_index retries if a concurrent write advances the revision during construction - Folder-scoped similar search uses brute-force cosine scan (no k limit) - Region-based similarity: crop an arbitrary rectangle in the lightbox, embed it with CLIP, find the nearest images globally or within folder - Pagination for both similar and region results; scope toggle (all media / current folder) re-runs the query without reopening image ## Duplicate Finder - Three-phase detection: stat (size) → sample hash (4×16 KB windows) → full-file hash (xxh3, large files only) to eliminate false positives - Filesystem-first deletion: only removes DB rows for files successfully deleted from disk; failed files remain visible for retry - Persisted scan cache (SQLite) with invalidation on reindex and deletion; both global and per-folder scopes invalidated after any deletion ## Tag Cloud & Explore - Visual cluster explore: k-means over CLIP embeddings, configurable k, representative thumbnail per cluster; cache keyed on xxh3 of embedding set - Tag explore: ranked tag list with image counts and representative thumbnail per tag; invalidated when AI tagging completes or folder removed - Tag autocomplete for search bar ## Folder Management - Inline rename (display name only, not OS rename) - Relocate: file-picker to update folder path; all image paths rewritten using SUBSTR prefix replacement to avoid corrupting paths that contain the folder name as a substring - Missing-folder recovery banner: Locate or Remove, never silent deletion - Right-click context menu on sidebar items: Reindex, Rename, Locate, Remove; hover buttons preserved alongside context menu ## Background Tasks - Per-folder progress panel with embedding, tagging, caption, and thumbnail stage breakdown - Retry button per task for failed embeddings and tagging jobs - Completed-task notifications (system tray via tauri-plugin-notification) - Scan errors shown inline; WalkDir permission errors protect existing records from deletion rather than cascading data loss ## Backend correctness - sqlite-vec DML performed outside transactions throughout (virtual-table limitation); embedding revision incremented on delete as well as insert - Tagging job discard checks status = 'processing' so paused jobs reset to 'pending' are also skipped, not just cancelled ones - Stale async responses in Lightbox guarded with cancellation flags and currentImageIdRef for both tag-load and tag-add callbacks - Duplicate cache cleared on reindex; folder-removal clears vector rows outside transaction before deleting the folder row
2026-06-07 22:43:16 +00:00
parent 8905baf4a5
commit 0ca4d142d8
32 changed files with 9539 additions and 760 deletions
@@ -1,7 +1,9 @@
+use crate::captioner::{self, FlorenceCaptioner};
 use crate::db::{self, DbPool, EmbeddingJob, FolderJobProgress, ImageRecord, IndexedMediaEntry};
 use crate::embedder::{embedding_source_path, ClipImageEmbedder};
 use crate::media::{probe_video_metadata, MediaTools};
 use crate::storage::{detect_storage_profile, RuntimeAdaptiveProfile, StorageProfile};
+use crate::tagger::{self, WdTagger};
 use crate::thumbnail;
 use crate::vector;
 use anyhow::Result;
@@ -9,7 +11,6 @@ use rayon::prelude::*;
 use serde::Serialize;
 use std::collections::{HashMap, HashSet};
 use std::path::{Path, PathBuf};
-use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Mutex, OnceLock};
 use std::time::{Duration, Instant};
 use tauri::{AppHandle, Emitter};
@@ -22,29 +23,97 @@ const IMAGE_EXTENSIONS: &[&str] = &[
 const VIDEO_EXTENSIONS: &[&str] = &["mp4", "mov", "m4v", "webm"];

 const JOB_PROGRESS_EMIT_INTERVAL: Duration = Duration::from_millis(750);
+const CAPTION_BATCH_SIZE: usize = 1;

 static LAST_JOB_PROGRESS_EMIT: OnceLock<Mutex<HashMap<i64, Instant>>> = OnceLock::new();
 static ACTIVE_INDEXING_FOLDERS: OnceLock<Mutex<HashSet<i64>>> = OnceLock::new();

-static THUMBNAIL_WORKER_PAUSED: AtomicBool = AtomicBool::new(false);
-static METADATA_WORKER_PAUSED: AtomicBool = AtomicBool::new(false);
-static EMBEDDING_WORKER_PAUSED: AtomicBool = AtomicBool::new(false);
+static PAUSED_WORKER_FOLDERS: OnceLock<Mutex<PausedWorkerFolders>> = OnceLock::new();

-pub fn set_worker_paused(worker: &str, paused: bool) {
-    match worker {
-        "thumbnail" => THUMBNAIL_WORKER_PAUSED.store(paused, Ordering::Relaxed),
-        "metadata" => METADATA_WORKER_PAUSED.store(paused, Ordering::Relaxed),
-        "embedding" => EMBEDDING_WORKER_PAUSED.store(paused, Ordering::Relaxed),
-        _ => {}
+#[derive(Default)]
+struct PausedWorkerFolders {
+    thumbnail: HashSet<i64>,
+    metadata: HashSet<i64>,
+    embedding: HashSet<i64>,
+    caption: HashSet<i64>,
+    tagging: HashSet<i64>,
+}
+
+#[derive(Clone, Copy)]
+pub struct FolderWorkerPausedState {
+    pub thumbnail: bool,
+    pub metadata: bool,
+    pub embedding: bool,
+    pub caption: bool,
+    pub tagging: bool,
+}
+
+pub fn set_worker_paused(worker: &str, folder_id: i64, paused: bool) {
+    if let Ok(mut paused_folders) = PAUSED_WORKER_FOLDERS
+        .get_or_init(|| Mutex::new(PausedWorkerFolders::default()))
+        .lock()
+    {
+        let folder_set = match worker {
+            "thumbnail" => Some(&mut paused_folders.thumbnail),
+            "metadata" => Some(&mut paused_folders.metadata),
+            "embedding" => Some(&mut paused_folders.embedding),
+            "caption" => Some(&mut paused_folders.caption),
+            "tagging" => Some(&mut paused_folders.tagging),
+            _ => None,
+        };
+
+        if let Some(folder_set) = folder_set {
+            if paused {
+                folder_set.insert(folder_id);
+            } else {
+                folder_set.remove(&folder_id);
+            }
+        }
    }
 }

-pub fn get_worker_paused_states() -> [bool; 3] {
-    [
-        THUMBNAIL_WORKER_PAUSED.load(Ordering::Relaxed),
-        METADATA_WORKER_PAUSED.load(Ordering::Relaxed),
-        EMBEDDING_WORKER_PAUSED.load(Ordering::Relaxed),
-    ]
+pub fn get_worker_paused_states(folder_ids: &[i64]) -> HashMap<i64, FolderWorkerPausedState> {
+    let Ok(paused_folders) = PAUSED_WORKER_FOLDERS
+        .get_or_init(|| Mutex::new(PausedWorkerFolders::default()))
+        .lock()
+    else {
+        return HashMap::new();
+    };
+
+    folder_ids
+        .iter()
+        .copied()
+        .map(|folder_id| {
+            (
+                folder_id,
+                FolderWorkerPausedState {
+                    thumbnail: paused_folders.thumbnail.contains(&folder_id),
+                    metadata: paused_folders.metadata.contains(&folder_id),
+                    embedding: paused_folders.embedding.contains(&folder_id),
+                    caption: paused_folders.caption.contains(&folder_id),
+                    tagging: paused_folders.tagging.contains(&folder_id),
+                },
+            )
+        })
+        .collect()
+}
+
+fn paused_folder_ids(worker: &str) -> HashSet<i64> {
+    let Ok(paused_folders) = PAUSED_WORKER_FOLDERS
+        .get_or_init(|| Mutex::new(PausedWorkerFolders::default()))
+        .lock()
+    else {
+        return HashSet::new();
+    };
+
+    match worker {
+        "thumbnail" => paused_folders.thumbnail.clone(),
+        "metadata" => paused_folders.metadata.clone(),
+        "embedding" => paused_folders.embedding.clone(),
+        "caption" => paused_folders.caption.clone(),
+        "tagging" => paused_folders.tagging.clone(),
+        _ => HashSet::new(),
+    }
 }
 static FOLDER_STORAGE_PROFILES: OnceLock<Mutex<HashMap<i64, RuntimeAdaptiveProfile>>> =
    OnceLock::new();
@@ -78,11 +147,50 @@ pub struct MediaJobProgressEvent {

 pub fn index_folder(app: AppHandle, pool: DbPool, folder_id: i64, folder_path: PathBuf) {
    std::thread::spawn(move || {
+        set_folder_indexing_state(folder_id, true);
+
+        // If the folder path no longer exists on disk, record the error and
+        // emit done. Images are intentionally kept in the DB so the user can
+        // choose to relocate the folder or remove it explicitly — they should
+        // not be silently destroyed.
+        if !folder_path.is_dir() {
+            let error_msg = format!("Folder not found: {}", folder_path.display());
+            eprintln!("Indexing error for folder {}: {}", folder_id, error_msg);
+            if let Ok(conn) = pool.get() {
+                let _ = db::set_folder_scan_error(&conn, folder_id, &error_msg);
+            }
+            emit_progress(
+                &app,
+                &IndexProgress {
+                    folder_id,
+                    total: 0,
+                    indexed: 0,
+                    current_file: String::new(),
+                    done: true,
+                },
+            );
+            set_folder_indexing_state(folder_id, false);
+            return;
+        }
+
        let storage_profile = detect_storage_profile(&folder_path);
        set_folder_storage_profile(folder_id, RuntimeAdaptiveProfile::new(storage_profile));
-        set_folder_indexing_state(folder_id, true);
-        if let Err(error) = do_index(app, pool, folder_id, folder_path) {
-            eprintln!("Indexing error: {}", error);
+        if let Err(error) = do_index(app.clone(), &pool, folder_id, folder_path) {
+            eprintln!("Indexing error for folder {}: {}", folder_id, error);
+            if let Ok(conn) = pool.get() {
+                let _ = db::set_folder_scan_error(&conn, folder_id, &error.to_string());
+            }
+            // Always emit done so the frontend reloads and recovers from partial state.
+            emit_progress(
+                &app,
+                &IndexProgress {
+                    folder_id,
+                    total: 0,
+                    indexed: 0,
+                    current_file: String::new(),
+                    done: true,
+                },
+            );
        }
        set_folder_indexing_state(folder_id, false);
    });
@@ -95,10 +203,6 @@ pub fn start_thumbnail_worker(
    cache_dir: PathBuf,
 ) {
    std::thread::spawn(move || loop {
-        if THUMBNAIL_WORKER_PAUSED.load(Ordering::Relaxed) {
-            std::thread::sleep(std::time::Duration::from_millis(500));
-            continue;
-        }
        if let Err(error) = process_thumbnail_batch(&app, &pool, &media_tools, &cache_dir) {
            eprintln!("Thumbnail worker error: {}", error);
        }
@@ -108,10 +212,6 @@ pub fn start_thumbnail_worker(

 pub fn start_metadata_worker(app: AppHandle, pool: DbPool, media_tools: MediaTools) {
    std::thread::spawn(move || loop {
-        if METADATA_WORKER_PAUSED.load(Ordering::Relaxed) {
-            std::thread::sleep(std::time::Duration::from_millis(500));
-            continue;
-        }
        if let Err(error) = process_metadata_batch(&app, &pool, &media_tools) {
            eprintln!("Metadata worker error: {}", error);
        }
@@ -124,10 +224,6 @@ pub fn start_embedding_worker(app: AppHandle, pool: DbPool) {
        let mut embedder: Option<ClipImageEmbedder> = None;
        println!("Embedding worker started.");
        loop {
-            if EMBEDDING_WORKER_PAUSED.load(Ordering::Relaxed) {
-                std::thread::sleep(std::time::Duration::from_millis(500));
-                continue;
-            }
            if let Err(error) = process_embedding_batch(&app, &pool, &mut embedder) {
                eprintln!("Embedding worker error: {}", error);
            }
@@ -136,7 +232,49 @@ pub fn start_embedding_worker(app: AppHandle, pool: DbPool) {
    });
 }

-fn do_index(app: AppHandle, pool: DbPool, folder_id: i64, folder_path: PathBuf) -> Result<()> {
+pub fn start_caption_worker(app: AppHandle, pool: DbPool, app_data_dir: PathBuf) {
+    std::thread::spawn(move || {
+        let mut captioner: Option<FlorenceCaptioner> = None;
+        println!("Caption worker started.");
+        loop {
+            // If the acceleration setting changed, drop the cached session so
+            // the next batch picks it up with the new execution provider.
+            if captioner::CAPTION_SESSION_DIRTY.swap(false, std::sync::atomic::Ordering::Relaxed) {
+                println!("Caption worker: acceleration setting changed — resetting session.");
+                captioner = None;
+            }
+            if let Err(error) = process_caption_batch(&app, &pool, &app_data_dir, &mut captioner) {
+                eprintln!("Caption worker error: {}", error);
+                captioner = None;
+            }
+            std::thread::sleep(std::time::Duration::from_millis(750));
+        }
+    });
+}
+
+pub fn start_tagging_worker(app: AppHandle, pool: DbPool, app_data_dir: PathBuf) {
+    std::thread::spawn(move || {
+        let mut tagger_instance: Option<WdTagger> = None;
+        println!("Tagging worker started.");
+        loop {
+            // If the acceleration setting changed, drop the cached session so
+            // the next batch picks it up with the new execution provider.
+            if tagger::TAGGER_SESSION_DIRTY.swap(false, std::sync::atomic::Ordering::Relaxed) {
+                println!("Tagging worker: acceleration setting changed — resetting session.");
+                tagger_instance = None;
+            }
+            if let Err(error) =
+                process_tagging_batch(&app, &pool, &app_data_dir, &mut tagger_instance)
+            {
+                eprintln!("Tagging worker error: {}", error);
+                tagger_instance = None;
+            }
+            std::thread::sleep(std::time::Duration::from_millis(750));
+        }
+    });
+}
+
+fn do_index(app: AppHandle, pool: &DbPool, folder_id: i64, folder_path: PathBuf) -> Result<()> {
    let existing_entries = {
        let conn = pool.get()?;
        db::get_folder_media_index(&conn, folder_id)?
@@ -146,12 +284,32 @@ fn do_index(app: AppHandle, pool: DbPool, folder_id: i64, folder_path: PathBuf)
        .map(|entry| (entry.path.clone(), entry))
        .collect::<HashMap<_, _>>();

+    // Collect traversal errors separately. An unreadable subdirectory means we
+    // cannot distinguish absent files from inaccessible ones; tracking errors
+    // lets us skip the deletion step for paths under affected subtrees.
+    let mut unreadable_prefixes: Vec<String> = Vec::new();
    let media_paths: Vec<PathBuf> = WalkDir::new(&folder_path)
        .follow_links(true)
        .into_iter()
-        .filter_map(|entry| entry.ok())
-        .filter(|entry| entry.file_type().is_file() && is_supported_media(entry.path()))
-        .map(|entry| entry.path().to_path_buf())
+        .filter_map(|entry| match entry {
+            Ok(e) if e.file_type().is_file() && is_supported_media(e.path()) => {
+                Some(e.path().to_path_buf())
+            }
+            Ok(_) => None,
+            Err(err) => {
+                // Record the inaccessible path so we can protect its descendants
+                // from the missing-file deletion pass.
+                if let Some(path) = err.path() {
+                    unreadable_prefixes.push(path.to_string_lossy().into_owned());
+                }
+                eprintln!(
+                    "WalkDir error while scanning folder {}: {}",
+                    folder_path.display(),
+                    err
+                );
+                None
+            }
+        })
        .collect();

    let total = media_paths.len();
@@ -197,7 +355,7 @@ fn do_index(app: AppHandle, pool: DbPool, folder_id: i64, folder_path: PathBuf)
                    images: committed,
                },
            );
-            emit_folder_job_progress(&app, &pool, &[folder_id]);
+            emit_folder_job_progress(&app, &pool, &[folder_id], false);
        }

        processed += path_chunk.len();
@@ -224,6 +382,17 @@ fn do_index(app: AppHandle, pool: DbPool, folder_id: i64, folder_path: PathBuf)
    let missing_ids = existing_by_path
        .values()
        .filter(|entry| !seen_paths.contains(&entry.path))
+        .filter(|entry| {
+            // If this path lives under a subtree that WalkDir couldn't read,
+            // we don't know whether the file is gone or just temporarily
+            // inaccessible — keep the record to avoid false deletions.
+            // Use Path::starts_with (component-aware) so a sibling directory
+            // whose name shares a prefix is not incorrectly protected.
+            let entry_path = std::path::Path::new(&entry.path);
+            unreadable_prefixes
+                .iter()
+                .all(|prefix| !entry_path.starts_with(prefix))
+        })
        .map(|entry| entry.id)
        .collect::<Vec<_>>();

@@ -232,7 +401,14 @@ fn do_index(app: AppHandle, pool: DbPool, folder_id: i64, folder_path: PathBuf)
        if !missing_ids.is_empty() {
            db::delete_images_by_ids(&conn, &missing_ids)?;
        }
+        let _ = db::backfill_embedding_jobs(&conn)?;
        db::update_folder_count(&conn, folder_id)?;
+        let _ = db::clear_folder_scan_error(&conn, folder_id);
+        // Invalidate duplicate scan cache — any reindex can change file contents
+        // or the set of files, making cached duplicate groups stale.
+        let folder_scope = format!("folder:{}", folder_id);
+        let _ = db::clear_duplicate_scan_cache(&conn, &folder_scope);
+        let _ = db::clear_duplicate_scan_cache(&conn, "all");
    }

    emit_progress(
@@ -245,7 +421,7 @@ fn do_index(app: AppHandle, pool: DbPool, folder_id: i64, folder_path: PathBuf)
            done: true,
        },
    );
-    emit_folder_job_progress(&app, &pool, &[folder_id]);
+    emit_folder_job_progress(&app, &pool, &[folder_id], true);
    Ok(())
 }

@@ -302,6 +478,14 @@ fn build_record(
        embedding_model: Some(vector::CLIP_MODEL_NAME.to_string()),
        embedding_updated_at: None,
        embedding_error: None,
+        generated_caption: None,
+        caption_model: None,
+        caption_updated_at: None,
+        caption_error: None,
+        ai_rating: None,
+        ai_tagger_model: None,
+        ai_tagged_at: None,
+        ai_tagger_error: None,
    })
 }

@@ -335,11 +519,13 @@ fn process_thumbnail_batch(
        with_db_write_lock(|| {
            let mut conn = pool.get()?;
            let active_folders = active_indexing_folders();
+            let paused_folders = paused_folder_ids("thumbnail");
            let worker_batch_size = max_worker_batch_size(&active_folders);
            let worker_fetch_size = max_worker_fetch_size(&active_folders);
            db::claim_thumbnail_jobs(
                &mut conn,
                &active_folders,
+                &paused_folders,
                worker_fetch_size,
                worker_batch_size,
            )
@@ -428,7 +614,7 @@ fn process_thumbnail_batch(
                images: updated_images,
            },
        );
-        emit_folder_job_progress(app, pool, &folder_ids.into_iter().collect::<Vec<_>>());
+        emit_folder_job_progress(app, pool, &folder_ids.into_iter().collect::<Vec<_>>(), true);
    }

    Ok(())
@@ -439,11 +625,13 @@ fn process_metadata_batch(app: &AppHandle, pool: &DbPool, media_tools: &MediaToo
        with_db_write_lock(|| {
            let mut conn = pool.get()?;
            let active_folders = active_indexing_folders();
+            let paused_folders = paused_folder_ids("metadata");
            let worker_batch_size = max_worker_batch_size(&active_folders);
            let worker_fetch_size = max_worker_fetch_size(&active_folders);
            db::claim_metadata_jobs(
                &mut conn,
                &active_folders,
+                &paused_folders,
                worker_fetch_size,
                worker_batch_size,
            )
@@ -506,7 +694,7 @@ fn process_metadata_batch(app: &AppHandle, pool: &DbPool, media_tools: &MediaToo
                images: updated_images,
            },
        );
-        emit_folder_job_progress(app, pool, &folder_ids.into_iter().collect::<Vec<_>>());
+        emit_folder_job_progress(app, pool, &folder_ids.into_iter().collect::<Vec<_>>(), true);
    }

    Ok(())
@@ -518,14 +706,11 @@ fn process_embedding_batch(
    embedder: &mut Option<ClipImageEmbedder>,
 ) -> Result<()> {
    let batch_started_at = Instant::now();
-    if embedder.is_none() {
-        *embedder = Some(ClipImageEmbedder::new()?);
-    }
-
    let claim_started_at = Instant::now();
+    let paused_folders = paused_folder_ids("embedding");
    let jobs = with_db_write_lock(|| {
        let mut conn = pool.get()?;
-        db::claim_embedding_jobs(&mut conn, EMBEDDING_BATCH_SIZE)
+        db::claim_embedding_jobs(&mut conn, &paused_folders, EMBEDDING_BATCH_SIZE)
    })?;
    let claim_elapsed = claim_started_at.elapsed();

@@ -533,9 +718,18 @@ fn process_embedding_batch(
        return Ok(());
    }

+    if embedder.is_none() {
+        *embedder = Some(ClipImageEmbedder::new()?);
+    }
+
    println!("Embedding batch claimed: {} items", jobs.len());
    let folder_ids = jobs.iter().map(|job| job.folder_id).collect::<HashSet<_>>();
-    emit_folder_job_progress(app, pool, &folder_ids.iter().copied().collect::<Vec<_>>());
+    emit_folder_job_progress(
+        app,
+        pool,
+        &folder_ids.iter().copied().collect::<Vec<_>>(),
+        false,
+    );
    let embedder = embedder.as_ref().expect("embedder should be initialized");

    let infer_started_at = Instant::now();
@@ -639,7 +833,7 @@ fn process_embedding_batch(
                images: updated_images,
            },
        );
-        emit_folder_job_progress(app, pool, &folder_ids.into_iter().collect::<Vec<_>>());
+        emit_folder_job_progress(app, pool, &folder_ids.into_iter().collect::<Vec<_>>(), true);
    }

    let write_elapsed = write_started_at.elapsed();
@@ -652,6 +846,237 @@ fn process_embedding_batch(
    Ok(())
 }

+fn process_caption_batch(
+    app: &AppHandle,
+    pool: &DbPool,
+    app_data_dir: &Path,
+    captioner: &mut Option<FlorenceCaptioner>,
+) -> Result<()> {
+    if !captioner::caption_model_status(app_data_dir).ready {
+        return Ok(());
+    }
+
+    let paused_folders = paused_folder_ids("caption");
+    let jobs = with_db_write_lock(|| {
+        let mut conn = pool.get()?;
+        db::claim_caption_jobs(&mut conn, &paused_folders, CAPTION_BATCH_SIZE)
+    })?;
+
+    if jobs.is_empty() {
+        return Ok(());
+    }
+
+    if captioner.is_none() {
+        match FlorenceCaptioner::new(app_data_dir) {
+            Ok(model) => *captioner = Some(model),
+            Err(error) => {
+                with_db_write_lock(|| {
+                    let conn = pool.get()?;
+                    db::requeue_caption_jobs(
+                        &conn,
+                        &jobs.iter().map(|job| job.image_id).collect::<Vec<_>>(),
+                    )
+                })?;
+                return Err(error);
+            }
+        }
+    }
+
+    let folder_ids = jobs.iter().map(|job| job.folder_id).collect::<HashSet<_>>();
+    emit_folder_job_progress(
+        app,
+        pool,
+        &folder_ids.iter().copied().collect::<Vec<_>>(),
+        false,
+    );
+
+    let captioner = captioner
+        .as_mut()
+        .expect("captioner should be initialized before caption batch processing");
+    let caption_results = jobs
+        .iter()
+        .map(|job| (job.clone(), captioner.generate(Path::new(&job.path))))
+        .collect::<Vec<_>>();
+
+    let updated_images = with_db_write_lock(|| {
+        let mut conn = pool.get()?;
+        let tx = conn.transaction()?;
+        let mut updated_images = Vec::with_capacity(caption_results.len());
+
+        for (job, caption_result) in &caption_results {
+            if db::is_caption_job_cancelled(&tx, job.image_id)? {
+                tx.execute(
+                    "DELETE FROM caption_jobs WHERE image_id = ?1",
+                    [job.image_id],
+                )?;
+                continue;
+            }
+            match caption_result {
+                Ok(caption) => {
+                    updated_images.push(db::update_generated_caption(
+                        &tx,
+                        job.image_id,
+                        caption,
+                        captioner::FLORENCE_CAPTION_MODEL_NAME,
+                    )?);
+                }
+                Err(error) => {
+                    db::mark_caption_failed(&tx, job.image_id, &error.to_string())?;
+                    updated_images.push(db::get_image_by_id(&tx, job.image_id)?);
+                }
+            }
+        }
+
+        tx.commit()?;
+        Ok(updated_images)
+    })?;
+
+    if !updated_images.is_empty() {
+        let folder_ids = updated_images
+            .iter()
+            .map(|image| image.folder_id)
+            .collect::<HashSet<_>>();
+        emit_media_updates(
+            app,
+            &MediaUpdateBatch {
+                images: updated_images,
+            },
+        );
+        emit_folder_job_progress(app, pool, &folder_ids.into_iter().collect::<Vec<_>>(), true);
+    }
+
+    Ok(())
+}
+
+fn process_tagging_batch(
+    app: &AppHandle,
+    pool: &DbPool,
+    app_data_dir: &Path,
+    tagger_instance: &mut Option<WdTagger>,
+) -> Result<()> {
+    if !tagger::tagger_model_status(app_data_dir).ready {
+        return Ok(());
+    }
+
+    let paused_folders = paused_folder_ids("tagging");
+    let batch_size = crate::tagger::tagger_batch_size(app_data_dir);
+    let jobs = with_db_write_lock(|| {
+        let mut conn = pool.get()?;
+        db::claim_tagging_jobs(&mut conn, &paused_folders, batch_size)
+    })?;
+
+    if jobs.is_empty() {
+        return Ok(());
+    }
+
+    if tagger_instance.is_none() {
+        match WdTagger::new(app_data_dir) {
+            Ok(model) => *tagger_instance = Some(model),
+            Err(error) => {
+                with_db_write_lock(|| {
+                    let conn = pool.get()?;
+                    db::requeue_tagging_jobs(
+                        &conn,
+                        &jobs.iter().map(|job| job.image_id).collect::<Vec<_>>(),
+                    )
+                })?;
+                return Err(error);
+            }
+        }
+    }
+
+    let folder_ids = jobs.iter().map(|job| job.folder_id).collect::<HashSet<_>>();
+    emit_folder_job_progress(
+        app,
+        pool,
+        &folder_ids.iter().copied().collect::<Vec<_>>(),
+        false,
+    );
+
+    let tagger_ref = tagger_instance
+        .as_mut()
+        .expect("tagger should be initialized before tagging batch processing");
+
+    let tag_results = jobs
+        .iter()
+        .map(|job| {
+            (
+                job.clone(),
+                tagger_ref.run(Path::new(&job.path), tagger::DEFAULT_MAX_TAGS),
+            )
+        })
+        .collect::<Vec<_>>();
+
+    let updated_images = with_db_write_lock(|| {
+        let mut conn = pool.get()?;
+        let tx = conn.transaction()?;
+        let mut updated_images = Vec::with_capacity(tag_results.len());
+
+        for (job, tag_result) in &tag_results {
+            // If the job is no longer in 'processing' state it was either
+            // cancelled or reset to 'pending' (pause) while inference was running.
+            // Discard the result in both cases — for cancelled rows also delete
+            // the row; for paused rows leave it so the job is retried later.
+            if !db::is_tagging_job_processing(&tx, job.image_id)? {
+                tx.execute(
+                    "DELETE FROM tagging_jobs WHERE image_id = ?1 AND status = 'cancelled'",
+                    [job.image_id],
+                )?;
+                continue;
+            }
+            match tag_result {
+                Ok(output) => {
+                    let tag_pairs: Vec<(String, f64)> = output
+                        .tags
+                        .iter()
+                        .map(|t| (t.tag.clone(), t.confidence as f64))
+                        .collect();
+                    db::update_ai_tags(
+                        &tx,
+                        job.image_id,
+                        &tag_pairs,
+                        &output.rating,
+                        tagger::WD_TAGGER_MODEL_NAME,
+                    )?;
+                }
+                Err(error) => {
+                    db::mark_tagging_failed(&tx, job.image_id, &error.to_string())?;
+                }
+            }
+            updated_images.push(db::get_image_by_id(&tx, job.image_id)?);
+        }
+
+        tx.commit()?;
+        Ok(updated_images)
+    })
+    .or_else(|db_err| {
+        // The DB write failed.  Try to requeue the claimed jobs so they aren't
+        // left stuck in 'processing' until the next app restart.
+        let image_ids: Vec<i64> = jobs.iter().map(|job| job.image_id).collect();
+        let _ = with_db_write_lock(|| {
+            let conn = pool.get()?;
+            db::requeue_tagging_jobs(&conn, &image_ids)
+        });
+        Err(db_err)
+    })?;
+
+    if !updated_images.is_empty() {
+        let folder_ids = updated_images
+            .iter()
+            .map(|image| image.folder_id)
+            .collect::<HashSet<_>>();
+        emit_media_updates(
+            app,
+            &MediaUpdateBatch {
+                images: updated_images,
+            },
+        );
+        emit_folder_job_progress(app, pool, &folder_ids.into_iter().collect::<Vec<_>>(), true);
+    }
+
+    Ok(())
+}
+
 fn active_indexing_folders() -> HashSet<i64> {
    ACTIVE_INDEXING_FOLDERS
        .get_or_init(|| Mutex::new(HashSet::new()))
@@ -737,7 +1162,7 @@ fn emit_media_updates(app: &AppHandle, batch: &MediaUpdateBatch) {
    let _ = app.emit("media-updated", batch);
 }

-fn emit_folder_job_progress(app: &AppHandle, pool: &DbPool, folder_ids: &[i64]) {
+pub fn emit_folder_job_progress(app: &AppHandle, pool: &DbPool, folder_ids: &[i64], force: bool) {
    let mut unique_folder_ids = folder_ids.iter().copied().collect::<Vec<_>>();
    unique_folder_ids.sort_unstable();
    unique_folder_ids.dedup();
@@ -749,10 +1174,11 @@ fn emit_folder_job_progress(app: &AppHandle, pool: &DbPool, folder_ids: &[i64])
        Err(_) => return,
    };
    unique_folder_ids.retain(|folder_id| {
-        let should_emit = tracker
-            .get(folder_id)
-            .map(|last_emit| now.duration_since(*last_emit) >= JOB_PROGRESS_EMIT_INTERVAL)
-            .unwrap_or(true);
+        let should_emit = force
+            || tracker
+                .get(folder_id)
+                .map(|last_emit| now.duration_since(*last_emit) >= JOB_PROGRESS_EMIT_INTERVAL)
+                .unwrap_or(true);
        if should_emit {
            tracker.insert(*folder_id, now);
        }