Add the duplicate originals review, from the sidebar and from Settings

"Duplicate originals" appears under the trash in the collections
sidebar while the catalog holds any, and Settings says how many there
are beside the other whole-library passes. Both open one page: every
group with its picture and paths, the copy that stays (tap another path
to change it), a per-group Include box, what the survivor will gain and
any flag, label or face conflict, and why a group was skipped.

The summary is the dry run -- "N groups, M files to trash, K skipped" --
and nothing moves until "Check" has read the copies and "Move M copies
to trash" is pressed. Both run on workers with progress on the page, in
the activity register and, for the move, on the library status line;
Stop ends a job between groups. When it ends the grid, the sidebar and
the trash are refreshed and the survivors' judgements are written to
their sidecars and XMP the way a rating keystroke writes them.

The page is paginated at 30 groups, so a redraw decodes 30 thumbnails
and previews 30 merges whatever the size of the library. Back and
Escape leave it like its own Back button.

ui/dr-ui/src/duplicates.rs

@@ -39,7 +39,9 @@
 //! there, and consolidating again treats it as already moved.
 
 use std::path::PathBuf;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::mpsc::{Receiver, Sender};
+use std::sync::Arc;
 
 use dr_catalog::duplicates::{Copy, Group, Outcome, Probe, PROBE_WINDOW};
 use dr_catalog::{trash, Catalog};
@@ -102,11 +104,6 @@ impl Reviewed {
             check: Check::Unchecked,
         }
     }
-
-    /// Whether the group goes ahead when the button is pressed.
-    pub fn planned(&self) -> bool {
-        self.include && matches!(self.check, Check::Same(_))
-    }
 }
 
 /// The dry-run figures the review shows above the list.
@@ -481,10 +478,11 @@ pub fn spawn_check(
     catalog_path: PathBuf,
     cache_dir: PathBuf,
     groups: Vec<(usize, Group)>,
+    stop: Arc<AtomicBool>,
 ) -> Receiver<DupMessage> {
     let (tx, rx) = std::sync::mpsc::channel();
     std::thread::spawn(move || {
-        let stopped = run_check(&conn, &catalog_path, cache_dir, groups, &tx).err();
+        let stopped = run_check(&conn, &catalog_path, cache_dir, groups, &stop, &tx).err();
         let _ = tx.send(DupMessage::Finished { stopped });
     });
     rx
@@ -495,6 +493,7 @@ fn run_check(
     catalog_path: &std::path::Path,
     cache_dir: PathBuf,
     groups: Vec<(usize, Group)>,
+    stop: &AtomicBool,
     tx: &Sender<DupMessage>,
 ) -> Result<(), String> {
     let catalog = Catalog::open(catalog_path).map_err(|e| e.to_string())?;
@@ -504,6 +503,9 @@ fn run_check(
         let backend = crate::remote::connect(conn).map_err(|e| e.to_string())?;
         let total = groups.len();
         for (done, (index, group)) in groups.into_iter().enumerate() {
+            if stop.load(Ordering::Relaxed) {
+                return Err("Stopped".into());
+            }
             // The server going away ends the job: every group after this
             // one would fail the same way, and none of them is a verdict.
             let (check, probes) =
@@ -736,16 +738,18 @@ pub async fn consolidate_group(
 
 /// Consolidate groups on a worker, one at a time.
 ///
-/// Closing the review stops the job *between* groups, never inside one.
+/// `stop` ends the job *between* groups, never inside one.
 pub fn spawn_consolidate(
     conn: Connection,
     catalog_path: PathBuf,
     cache_dir: PathBuf,
     plans: Vec<Plan>,
+    stop: Arc<AtomicBool>,
 ) -> Receiver<DupMessage> {
     let (tx, rx) = std::sync::mpsc::channel();
     std::thread::spawn(move || {
-        let stopped = run_consolidate(&conn, &catalog_path, cache_dir, plans, &tx).err();
+        let stopped =
+            run_consolidate(&conn, &catalog_path, cache_dir, plans, &stop, &tx).err();
         let _ = tx.send(DupMessage::Finished { stopped });
     });
     rx
@@ -756,6 +760,7 @@ fn run_consolidate(
     catalog_path: &std::path::Path,
     cache_dir: PathBuf,
     plans: Vec<Plan>,
+    stop: &AtomicBool,
     tx: &Sender<DupMessage>,
 ) -> Result<(), String> {
     let catalog = Catalog::open(catalog_path).map_err(|e| e.to_string())?;
@@ -765,6 +770,10 @@ fn run_consolidate(
         let backend = crate::remote::connect(conn).map_err(|e| e.to_string())?;
         let total = plans.len();
         for (done, plan) in plans.iter().enumerate() {
+            // Between groups, never inside one.
+            if stop.load(Ordering::Relaxed) {
+                return Err("Stopped".into());
+            }
             let msg = match consolidate_group(&*backend, &catalog, &cache, plan).await {
                 Ok(outcome) => DupMessage::Consolidated {
                     index: plan.index,
@@ -1002,6 +1011,7 @@ mod tests {
                 self.catalog_path.clone(),
                 self.cache.clone(),
                 groups,
+                Arc::default(),
             );
             for msg in rx {
                 match msg {
@@ -1019,6 +1029,7 @@ mod tests {
                 self.catalog_path.clone(),
                 self.cache.clone(),
                 plans,
+                Arc::default(),
             );
             let (mut done, mut failed) = (0, Vec::new());
             for msg in rx {