Merge pull request 'Photo versions follow-ups (#194)' (#198) from fix/194-version-followups into master

Reviewed-on: #198

CLAUDE.md

@@ -379,8 +379,10 @@ for the `link_versions` binary).
   (1) a PNG's XMP `xmpMM:DerivedFrom/stRef:documentID = blake3:<hex>` (iTXt chunk
   `XML:com.adobe.xmp`, read by hand, no decode); (2) the file-name convention next
   to an indexed original in the same library (same dir, or the parent of
-  `aiedits/`). Name links are retried every tick (`sweep_name_links`), so an
-  original that wasn't hashed yet, or arrives later, still links. Only *new* rows
+  `aiedits/`). Name links are retried every tick (`sweep_name_links`, paged by a
+  per-library resume cursor that wraps, so candidates that never resolve can't starve
+  the ones behind them), so an original that wasn't hashed yet, or arrives later,
+  still links. Only *new* rows
   are examined, so files ingested before this existed need the `link_versions`
   binary once.
 - **Date/GPS/camera copy**: a version row takes its original's date triple, GPS and
@@ -397,16 +399,23 @@ for the `link_versions` binary).
   normal photo, so nothing disappears silently. Applied in `/photos` (SQL for the
   recursive listing, `hidden_version_rows` for the folder walk and tag branch),
   `/memories` and reels, `/photos/exif`, `/photos/gps-summary`, unified search, the
-  camera counts and the media gauges; `include_versions=true` lists them.
+  camera counts and the media gauges; `include_versions=true` lists them. The non-recursive `/photos` `dirs` also drops a
+  folder whose indexed contents are all hidden versions (typically `aiedits/`;
+  `dirs_holding_only_hidden_versions`, one indexed range scan per dir).
 - **Display file** is fetched through the original's library, so the display
   choice only counts when the chosen version has a row there (else the original
   shows; `PUT /image/versions/display` answers 400). Recording a link also drops
   the version's `face_detections`, CLIP columns and `clip_failures`.
 - **Tags and favorites** written or read on a version path bind to the original
   (`canonical_rel_path` in the tag/favorite DAOs, `get_rel_paths_sharing_content`
-  for reads). Insight generation on a version path runs against the original. A
-  *manual date* set on a version path is overwritten by the copy above: send the
-  original's path.
+  for reads). Annotations are keyed by rel_path alone, so a path that also names a
+  non-version file in any library is never redirected. Linking merges favorites and
+  tags already recorded on the version's own path into the original (union, duplicates
+  dropped; `link_version`). Insight generation on a version path runs against the
+  original. A *manual date* set or cleared on a version path (`POST /image/exif/date`,
+  `/clear`) is applied to the original's row instead, as the copy above would overwrite
+  it. Precomputed reels are not invalidated by a display change: reels render the
+  original's path, never the display file.
 - **Excluded from derived work** (`NOT_A_VERSION_CLAUSE`): face backlog and ingest
   candidates, face stats (numerator and denominator), CLIP backlog and index, and
   exact/perceptual duplicate grouping. Thumbnails and previews *are* generated.
@@ -421,7 +430,8 @@ for the `link_versions` binary).
   TSV. `exact` rows (XMP, file name) are what ingest would have linked and apply
   with `--apply`; `review` rows (ComfyUI prompt `LoadImage` input, Upscayl-style
   names, nearest pHash) apply only from a reviewed TSV (`--reviewed`), each checked
-  against the current hash; `unresolved` rows are listed, never guessed. `--apply`
+  against the current version hash and the TSV's `original_hash` (stale rows are refused
+  and listed); `unresolved` rows are listed, never guessed. `--apply`
   takes a `VACUUM INTO` snapshot first.
 
 ### Enhance on demand (#192)

src/bin/link_versions.rs

@@ -179,6 +179,9 @@ fn main() -> anyhow::Result<()> {
         Arc::new(Mutex::new(Box::new(SqliteVersionDao::new())));
     let mut guard = dao.lock().unwrap();
     let report = backfill::apply(guard.as_mut(), &proposals, &reviewed, &rows);
+    for why in &report.stale_details {
+        eprintln!("  stale reviewed row refused: {why}");
+    }
     let synced = guard
         .sync_version_metadata()
         .map_err(|e| anyhow::anyhow!("{e:?}"))?;

src/database/mod.rs

@@ -323,10 +323,11 @@ impl FavoriteDao for SqliteFavoriteDao {
 
         let mut connection = self.connection.lock().expect("Unable to get FavoriteDao");
         // Favoriting a photo version favorites its original (#191).
-        let favorite_path = &version_dao::canonical_rel_path(connection.deref_mut(), favorite_path)
-            .ok()
-            .flatten()
-            .unwrap_or_else(|| favorite_path.to_string());
+        let favorite_path =
+            &version_dao::canonical_rel_path(connection.deref_mut(), None, favorite_path)
+                .ok()
+                .flatten()
+                .unwrap_or_else(|| favorite_path.to_string());
 
         if favorites
             .filter(userid.eq(user_id).and(rel_path.eq(&favorite_path)))
@@ -349,10 +350,11 @@ impl FavoriteDao for SqliteFavoriteDao {
         use schema::favorites::dsl::*;
 
         let mut connection = self.connection.lock().unwrap();
-        let favorite_path = version_dao::canonical_rel_path(connection.deref_mut(), &favorite_path)
-            .ok()
-            .flatten()
-            .unwrap_or(favorite_path);
+        let favorite_path =
+            version_dao::canonical_rel_path(connection.deref_mut(), None, &favorite_path)
+                .ok()
+                .flatten()
+                .unwrap_or(favorite_path);
         diesel::delete(favorites)
             .filter(userid.eq(user_id).and(rel_path.eq(favorite_path)))
             .execute(connection.deref_mut())

src/database/version_dao.rs

@@ -188,6 +188,15 @@ pub trait VersionDao: Sync + Send {
     /// the original's rel_path (same library when possible).
     fn original_path_of(&mut self, rel_path: &str) -> Result<Option<String>, DbError>;
 
+    /// `(library_id, rel_path)` of the original of the version at
+    /// `(library_id, rel_path)`, when it is a linked version with a live
+    /// original. For per-library writes that must land on the original.
+    fn original_location_of(
+        &mut self,
+        library_id: i32,
+        rel_path: &str,
+    ) -> Result<Option<(i32, String)>, DbError>;
+
     /// Which of `hashes` are versions of something.
     fn version_hashes_among(&mut self, hashes: &[String]) -> Result<HashSet<String>, DbError>;
 
@@ -228,14 +237,28 @@ pub trait VersionDao: Sync + Send {
         view_library_ids: &[i32],
     ) -> Result<HashSet<(i32, String)>, DbError>;
 
+    /// Of `dirs` (rel paths), those that hold indexed files and where every
+    /// indexed file anywhere underneath is a version hidden in the view.
+    /// Such a folder would open empty. A folder with no indexed rows at all
+    /// is not reported (nothing says it is safe to hide). One indexed range
+    /// scan per dir; no filesystem walk.
+    fn dirs_holding_only_hidden_versions(
+        &mut self,
+        view_library_ids: &[i32],
+        dirs: &[String],
+    ) -> Result<HashSet<String>, DbError>;
+
     /// Rows in a library whose file name follows the version convention
     /// (`<name>.<kind>[.<n>].<ext>`), have a hash and aren't linked yet.
-    /// Returns `(rel_path, content_hash)`, oldest first.
+    /// Returns `(id, rel_path, content_hash)` ordered by row id, starting
+    /// after `after_id` (the caller's resume cursor, so rows that never
+    /// resolve can't starve the ones behind them).
     fn unlinked_name_candidates(
         &mut self,
         library_id: i32,
+        after_id: i32,
         limit: i64,
-    ) -> Result<Vec<(String, String)>, DbError>;
+    ) -> Result<Vec<(i32, String, String)>, DbError>;
 
     /// Copy the original's date, GPS and camera fields onto every version
     /// row whose copy is stale. Returns the number of rows updated.
@@ -279,26 +302,93 @@ fn placeholders(n: usize) -> String {
 /// If `rel_path` is a linked version whose original has a live row, the
 /// original's rel_path (same library when possible). Tag and favorite
 /// writes on a version path bind to the original through this.
+///
+/// `library_id` scopes the version lookup to one library's row; `None` is
+/// for the library-agnostic tag/favorite DAOs, which only know a rel_path.
+/// Annotations are keyed by rel_path alone, so in either case a path that
+/// *also* names a file in some library that is not a linked version stays
+/// itself: redirecting it would hijack that photo's own tags and favorites.
 pub fn canonical_rel_path(
     conn: &mut SqliteConnection,
+    library_id: Option<i32>,
     rel_path: &str,
 ) -> QueryResult<Option<String>> {
+    Ok(canonical_location(conn, library_id, rel_path)?.map(|(_, p)| p))
+}
+
+/// [`canonical_rel_path`] with the original's library too, for callers that
+/// write per-library rows (the manual date lives on the `image_exif` row).
+pub fn canonical_location(
+    conn: &mut SqliteConnection,
+    library_id: Option<i32>,
+    rel_path: &str,
+) -> QueryResult<Option<(i32, String)>> {
     #[derive(QueryableByName)]
     struct Row {
+        #[diesel(sql_type = Integer)]
+        library_id: i32,
         #[diesel(sql_type = Text)]
         rel_path: String,
     }
     let rows = sql_query(
-        "SELECT o.rel_path AS rel_path FROM image_exif v \
+        "SELECT o.library_id AS library_id, o.rel_path AS rel_path FROM image_exif v \
          JOIN photo_versions pv ON pv.version_hash = v.content_hash \
          JOIN image_exif o ON o.content_hash = pv.original_hash \
-         WHERE v.rel_path = ? \
+         WHERE v.rel_path = ?1 AND (?2 IS NULL OR v.library_id = ?2) \
+           AND NOT EXISTS (SELECT 1 FROM image_exif x WHERE x.rel_path = v.rel_path \
+                             AND NOT EXISTS (SELECT 1 FROM photo_versions p \
+                                              WHERE p.version_hash = x.content_hash)) \
          ORDER BY (o.library_id = v.library_id) DESC, o.id \
          LIMIT 1",
     )
     .bind::<Text, _>(rel_path)
+    .bind::<Nullable<Integer>, _>(library_id)
     .load::<Row>(conn)?;
-    Ok(rows.into_iter().next().map(|r| r.rel_path))
+    Ok(rows.into_iter().next().map(|r| (r.library_id, r.rel_path)))
+}
+
+/// Fold favorites and tags recorded on a version's own path into its
+/// original's, now that reads and writes on the version path canonicalise
+/// to the original and could never reach them again. Set-valued data is
+/// unioned (duplicates dropped), then the version path's rows are removed.
+/// A no-op unless the version has a live original (`canonical_rel_path`).
+fn merge_annotations_into_original(
+    conn: &mut SqliteConnection,
+    library_id: i32,
+    version_rel_path: &str,
+) -> QueryResult<()> {
+    let Some(original) = canonical_rel_path(conn, Some(library_id), version_rel_path)? else {
+        return Ok(());
+    };
+    if original == version_rel_path {
+        return Ok(());
+    }
+    sql_query(
+        "INSERT INTO favorites (userid, rel_path) \
+         SELECT f.userid, ?2 FROM favorites f WHERE f.rel_path = ?1 \
+           AND NOT EXISTS (SELECT 1 FROM favorites g WHERE g.userid = f.userid AND g.rel_path = ?2)",
+    )
+    .bind::<Text, _>(version_rel_path)
+    .bind::<Text, _>(&original)
+    .execute(conn)?;
+    sql_query("DELETE FROM favorites WHERE rel_path = ?")
+        .bind::<Text, _>(version_rel_path)
+        .execute(conn)?;
+    sql_query(
+        "INSERT INTO tagged_photo (rel_path, tag_id, created_time, content_hash) \
+         SELECT ?2, t.tag_id, t.created_time, \
+                (SELECT content_hash FROM image_exif WHERE rel_path = ?2 \
+                  AND content_hash IS NOT NULL LIMIT 1) \
+         FROM tagged_photo t WHERE t.rel_path = ?1 \
+           AND NOT EXISTS (SELECT 1 FROM tagged_photo u WHERE u.tag_id = t.tag_id AND u.rel_path = ?2)",
+    )
+    .bind::<Text, _>(version_rel_path)
+    .bind::<Text, _>(&original)
+    .execute(conn)?;
+    sql_query("DELETE FROM tagged_photo WHERE rel_path = ?")
+        .bind::<Text, _>(version_rel_path)
+        .execute(conn)?;
+    Ok(())
 }
 
 /// Root original of `hash`: follows `photo_versions` until a hash that is
@@ -421,6 +511,8 @@ impl VersionDao for SqliteVersionDao {
                     .bind::<Text, _>(reference)
                     .execute(conn)?;
                 }
+
+                merge_annotations_into_original(conn, link.library_id, &link.rel_path)?;
                 Ok(LinkOutcome::Linked)
             })
         })
@@ -445,7 +537,20 @@ impl VersionDao for SqliteVersionDao {
     fn original_path_of(&mut self, rel_path: &str) -> Result<Option<String>, DbError> {
         trace_db_call("query", "original_path_of", |_span| {
             let mut conn = self.connection.lock().expect("Unable to get VersionDao");
-            canonical_rel_path(&mut conn, rel_path)
+            canonical_rel_path(&mut conn, None, rel_path)
+                .map_err(|e| anyhow::anyhow!("Query error: {}", e))
+        })
+        .map_err(|e| DbError::log(DbErrorKind::QueryError, e))
+    }
+
+    fn original_location_of(
+        &mut self,
+        library_id: i32,
+        rel_path: &str,
+    ) -> Result<Option<(i32, String)>, DbError> {
+        trace_db_call("query", "original_location_of", |_span| {
+            let mut conn = self.connection.lock().expect("Unable to get VersionDao");
+            canonical_location(&mut conn, Some(library_id), rel_path)
                 .map_err(|e| anyhow::anyhow!("Query error: {}", e))
         })
         .map_err(|e| DbError::log(DbErrorKind::QueryError, e))
@@ -720,14 +825,82 @@ impl VersionDao for SqliteVersionDao {
         .map_err(|e| DbError::log(DbErrorKind::QueryError, e))
     }
 
+    fn dirs_holding_only_hidden_versions(
+        &mut self,
+        view_library_ids: &[i32],
+        dirs: &[String],
+    ) -> Result<HashSet<String>, DbError> {
+        if dirs.is_empty() {
+            return Ok(HashSet::new());
+        }
+        trace_db_call("query", "dirs_holding_only_hidden_versions", |_span| {
+            #[derive(QueryableByName)]
+            struct Counts {
+                #[diesel(sql_type = BigInt)]
+                total: i64,
+                #[diesel(sql_type = BigInt)]
+                visible: i64,
+            }
+            #[derive(QueryableByName)]
+            struct Id {
+                #[diesel(sql_type = Integer)]
+                id: i32,
+            }
+            let mut conn = self.connection.lock().expect("Unable to get VersionDao");
+            // Scan the view's libraries (all of them when unscoped) so the
+            // (library_id, rel_path) index serves the prefix range.
+            let scan: Vec<i32> = if view_library_ids.is_empty() {
+                sql_query("SELECT id FROM libraries")
+                    .load::<Id>(&mut *conn)?
+                    .into_iter()
+                    .map(|r| r.id)
+                    .collect()
+            } else {
+                view_library_ids.to_vec()
+            };
+            let list = scan
+                .iter()
+                .map(i32::to_string)
+                .collect::<Vec<_>>()
+                .join(",");
+            let sql = format!(
+                "SELECT COUNT(*) AS total, \
+                        COALESCE(SUM(CASE WHEN {} THEN 1 ELSE 0 END), 0) AS visible \
+                 FROM image_exif \
+                 WHERE library_id IN ({list}) AND rel_path >= ? AND rel_path < ?",
+                visible_version_clause(view_library_ids)
+            );
+            let mut out = HashSet::new();
+            for dir in dirs {
+                let dir = dir.trim_end_matches('/');
+                if dir.is_empty() {
+                    continue;
+                }
+                // '0' is the character after '/', closing the prefix range.
+                let c = sql_query(&sql)
+                    .bind::<Text, _>(format!("{dir}/"))
+                    .bind::<Text, _>(format!("{dir}0"))
+                    .get_result::<Counts>(&mut *conn)?;
+                if c.total > 0 && c.visible == 0 {
+                    out.insert(dir.to_string());
+                }
+            }
+            Ok(out)
+        })
+        .map_err(|e| DbError::log(DbErrorKind::QueryError, e))
+    }
+
     fn unlinked_name_candidates(
         &mut self,
         library_id: i32,
+        after_id: i32,
         limit: i64,
-    ) -> Result<Vec<(String, String)>, DbError> {
+    ) -> Result<Vec<(i32, String, String)>, DbError> {
         trace_db_call("query", "unlinked_name_candidates", |_span| {
             #[derive(QueryableByName)]
             struct Row {
+                #[diesel(sql_type = Integer)]
+                id: i32,
                 #[diesel(sql_type = Text)]
                 rel_path: String,
                 #[diesel(sql_type = Text)]
@@ -739,8 +912,8 @@ impl VersionDao for SqliteVersionDao {
                 .collect::<Vec<_>>()
                 .join(" OR ");
             let sql = format!(
-                "SELECT rel_path, content_hash FROM image_exif \
-                 WHERE library_id = ? AND content_hash IS NOT NULL AND ({kinds}) \
+                "SELECT id, rel_path, content_hash FROM image_exif \
+                 WHERE library_id = ? AND id > ? AND content_hash IS NOT NULL AND ({kinds}) \
                    AND NOT EXISTS (SELECT 1 FROM photo_versions pv \
                                     WHERE pv.version_hash = image_exif.content_hash) \
                  ORDER BY id LIMIT ?"
@@ -748,11 +921,12 @@ impl VersionDao for SqliteVersionDao {
             let mut conn = self.connection.lock().expect("Unable to get VersionDao");
             sql_query(sql)
                 .bind::<Integer, _>(library_id)
+                .bind::<Integer, _>(after_id)
                 .bind::<BigInt, _>(limit)
                 .load::<Row>(&mut *conn)
                 .map(|rows| {
                     rows.into_iter()
-                        .map(|r| (r.rel_path, r.content_hash))
+                        .map(|r| (r.id, r.rel_path, r.content_hash))
                         .collect()
                 })
                 .map_err(|e| anyhow::anyhow!("Query error: {}", e))
@@ -1007,20 +1181,120 @@ mod tests {
             .unwrap();
         let mut conn = c.lock().unwrap();
         assert_eq!(
-            canonical_rel_path(&mut conn, "aiedits/orig.jpg.restored.png").unwrap(),
+            canonical_rel_path(&mut conn, None, "aiedits/orig.jpg.restored.png").unwrap(),
             Some("orig.jpg".to_string())
         );
-        assert_eq!(canonical_rel_path(&mut conn, "orig.jpg").unwrap(), None);
+        assert_eq!(
+            canonical_rel_path(&mut conn, None, "orig.jpg").unwrap(),
+            None
+        );
         drop(conn);
         // Original gone: nothing to bind to, the version is a normal photo.
         exec(&c, "DELETE FROM image_exif WHERE content_hash = 'o'");
         let mut conn = c.lock().unwrap();
         assert_eq!(
-            canonical_rel_path(&mut conn, "aiedits/orig.jpg.restored.png").unwrap(),
+            canonical_rel_path(&mut conn, None, "aiedits/orig.jpg.restored.png").unwrap(),
             None
         );
     }
 
+    #[test]
+    fn canonical_path_is_scoped_to_the_versions_library() {
+        let (mut dao, c) = dao();
+        insert_exif(&c, 1, "album_a/orig.jpg", "o");
+        insert_exif(&c, 1, "album_a/aiedits/orig.jpg.edit.png", "v");
+        dao.link_version(&link("v", "o", "album_a/aiedits/orig.jpg.edit.png"))
+            .unwrap();
+        // Library 2 has an unrelated, unlinked photo at the same path.
+        insert_exif(&c, 2, "album_a/aiedits/orig.jpg.edit.png", "other");
+        let mut conn = c.lock().unwrap();
+        let p = "album_a/aiedits/orig.jpg.edit.png";
+        assert_eq!(
+            canonical_rel_path(&mut conn, Some(2), p).unwrap(),
+            None,
+            "library 2's row is not a version"
+        );
+        // Annotations are path-keyed, so a path that also names a
+        // standalone photo is never redirected away from it.
+        assert_eq!(canonical_rel_path(&mut conn, None, p).unwrap(), None);
+        assert_eq!(
+            canonical_rel_path(&mut conn, Some(1), p).unwrap(),
+            None,
+            "shared path with a standalone photo stays put"
+        );
+        drop(conn);
+        exec(&c, "DELETE FROM image_exif WHERE content_hash = 'other'");
+        let mut conn = c.lock().unwrap();
+        assert_eq!(
+            canonical_rel_path(&mut conn, Some(1), p).unwrap(),
+            Some("album_a/orig.jpg".to_string())
+        );
+        assert_eq!(canonical_rel_path(&mut conn, Some(2), p).unwrap(), None);
+    }
+
+    #[test]
+    fn linking_merges_annotations_from_the_version_path_into_the_original() {
+        let (mut dao, c) = dao();
+        insert_exif(&c, 1, "album_a/orig.jpg", "o");
+        insert_exif(&c, 1, "album_a/aiedits/orig.jpg.edit.png", "v");
+        let v = "album_a/aiedits/orig.jpg.edit.png";
+        let o = "album_a/orig.jpg";
+        for t in ["t1", "t2"] {
+            exec(
+                &c,
+                &format!("INSERT INTO tags (name, created_time) VALUES ('{t}', 0)"),
+            );
+        }
+        // Tag 1 on both paths (duplicate), tag 2 only on the version.
+        exec(
+            &c,
+            &format!(
+                "INSERT INTO tagged_photo (rel_path, tag_id, created_time) \
+             SELECT '{o}', id, 1 FROM tags WHERE name = 't1'"
+            ),
+        );
+        exec(
+            &c,
+            &format!(
+                "INSERT INTO tagged_photo (rel_path, tag_id, created_time) \
+             SELECT '{v}', id, 2 FROM tags"
+            ),
+        );
+        // User 1 favorited both; user 2 only the version.
+        exec(
+            &c,
+            &format!(
+                "INSERT INTO favorites (userid, rel_path) VALUES (1, '{o}'), (1, '{v}'), (2, '{v}')"
+            ),
+        );
+
+        dao.link_version(&link("v", "o", v)).unwrap();
+
+        let rows = |sql: &str| -> Vec<(String, i32)> {
+            #[derive(QueryableByName)]
+            struct R {
+                #[diesel(sql_type = Text)]
+                p: String,
+                #[diesel(sql_type = Integer)]
+                k: i32,
+            }
+            sql_query(sql)
+                .load::<R>(&mut *c.lock().unwrap())
+                .unwrap()
+                .into_iter()
+                .map(|r| (r.p, r.k))
+                .collect()
+        };
+        assert_eq!(
+            rows("SELECT rel_path AS p, userid AS k FROM favorites ORDER BY userid"),
+            vec![(o.to_string(), 1), (o.to_string(), 2)]
+        );
+        assert_eq!(
+            rows("SELECT rel_path AS p, tag_id AS k FROM tagged_photo ORDER BY tag_id"),
+            vec![(o.to_string(), 1), (o.to_string(), 2)]
+        );
+    }
+
     #[test]
     fn sync_copies_date_gps_and_camera_from_a_converted_original() {
         let (mut dao, c) = dao();
@@ -1089,8 +1363,18 @@ mod tests {
         insert_exif(&c, 2, "a/y.jpg.restored.png", "h4");
         dao.link_version(&link("h2", "ox", "a/x.jpg.edit.2.png"))
             .unwrap_or(LinkOutcome::Linked);
-        let got = dao.unlinked_name_candidates(1, 100).unwrap();
-        assert_eq!(got, vec![("a/x.jpg.restored.png".to_string(), "h1".into())]);
+        let got = dao.unlinked_name_candidates(1, 0, 100).unwrap();
+        assert_eq!(got.len(), 1);
+        assert_eq!(
+            (got[0].1.as_str(), got[0].2.as_str()),
+            ("a/x.jpg.restored.png", "h1")
+        );
+        // The cursor skips rows at or before it.
+        assert!(
+            dao.unlinked_name_candidates(1, got[0].0, 100)
+                .unwrap()
+                .is_empty()
+        );
     }
 
     #[test]

src/files.rs

@@ -725,6 +725,23 @@ pub async fn list_photos<TagD: TagDao, FS: FileSystemAccess>(
         return HttpResponse::BadRequest().finish();
     }
 
+    // A folder whose indexed contents are all hidden versions (typically
+    // `aiedits/`) would open empty: leave it out unless versions are asked
+    // for. Recursive listings are flat and carry no dirs.
+    if !search_recursively && !include_versions && !dirs_set.is_empty() {
+        let candidates: Vec<String> = dirs_set.iter().cloned().collect();
+        let view: Vec<i32> = library.into_iter().map(|l| l.id).collect();
+        match app_state
+            .version_dao
+            .lock()
+            .expect("Unable to lock VersionDao")
+            .dirs_holding_only_hidden_versions(&view, &candidates)
+        {
+            Ok(empty) => dirs_set.retain(|d| !empty.contains(d.trim_end_matches('/'))),
+            Err(e) => warn!("dirs_holding_only_hidden_versions failed: {:?}", e),
+        }
+    }
+
     let dirs: Vec<String> = dirs_set.into_iter().collect();
 
     info!(

src/handlers/image.rs

@@ -926,6 +926,29 @@ fn earliest_fs_time(response: &MetadataResponse) -> Option<i64> {
     }
 }
 
+/// A manual date on a photo version (#191) is overwritten by the next
+/// metadata sync (`sync_version_metadata` re-copies the original's date), so
+/// the edit is redirected to the original, as tag writes are. Returns the
+/// `(library, path)` the edit applies to: the original's when `path` is a
+/// linked version with a live original, else the inputs unchanged.
+fn date_edit_target<'a>(
+    app_state: &'a AppState,
+    library: &'a libraries::Library,
+    path: &str,
+) -> (&'a libraries::Library, String) {
+    let original = app_state
+        .version_dao
+        .lock()
+        .expect("Unable to lock VersionDao")
+        .original_location_of(library.id, path)
+        .ok()
+        .flatten();
+    match original {
+        Some((lib_id, rel)) => (app_state.library_by_id(lib_id).unwrap_or(library), rel),
+        None => (library, path.to_string()),
+    }
+}
+
 #[post("/image/exif/date")]
 #[tracing::instrument(name = "set_image_date", skip_all)]
 pub async fn set_image_date(
@@ -949,6 +972,7 @@ pub async fn set_image_date(
     // Path normalization matches set_image_gps so a Windows-import client
     // doesn't end up with a backslash variant that misses the row.
     let normalized_path = body.path.replace('\\', "/");
+    let (library, normalized_path) = date_edit_target(&app_state, library, &normalized_path);
 
     // The user's display zone — the wall the UI sent is rendered in this
     // zone (ADR-0001, ticket #115). NULL / unparseable prefs fall back to
@@ -1012,6 +1036,7 @@ pub async fn clear_image_date(
     };
 
     let normalized_path = body.path.replace('\\', "/");
+    let (library, normalized_path) = date_edit_target(&app_state, library, &normalized_path);
 
     let updated = {
         let mut dao = exif_dao.lock().expect("Unable to lock ExifDao");

src/tags.rs

@@ -608,7 +608,7 @@ pub trait TagDao: Send + Sync {
 /// `path` is a linked photo version whose original is still indexed, else
 /// `path` itself.
 fn canonical_path(conn: &mut SqliteConnection, path: &str) -> String {
-    crate::database::version_dao::canonical_rel_path(conn, path)
+    crate::database::version_dao::canonical_rel_path(conn, None, path)
         .ok()
         .flatten()
         .unwrap_or_else(|| path.to_string())

src/versions/backfill.rs

@@ -110,6 +110,8 @@ pub struct ReviewedLink {
     pub method: String,
     pub library_id: i32,
     pub version_path: String,
+    /// The original's path as the dry run proposed it (may be empty).
+    pub original_path: String,
     pub kind: String,
     pub version_hash: String,
     pub original_hash: String,
@@ -140,6 +142,7 @@ pub fn parse_reviewed(tsv: &str) -> (Vec<ReviewedLink>, usize) {
             method: f[1].to_string(),
             library_id: f[2].parse().unwrap(),
             version_path: f[3].to_string(),
+            original_path: f[4].to_string(),
             kind: f[6].to_string(),
             version_hash: f[7].to_string(),
             original_hash: f[8].to_string(),
@@ -634,6 +637,8 @@ pub struct ApplyReport {
     pub refused: usize,
     /// Reviewed rows whose version row no longer matches the file in the TSV.
     pub stale: usize,
+    /// One line per stale reviewed row, saying what no longer matches.
+    pub stale_details: Vec<String>,
 }
 
 /// Write the links. `exact` proposals apply as they are; `reviewed` come
@@ -651,6 +656,8 @@ pub fn apply(
         .map(|r| ((r.library_id, r.rel_path.as_str()), r.content_hash.as_str()))
         .collect();
 
+    let live_hashes: HashSet<&str> = rows.iter().map(|r| r.content_hash.as_str()).collect();
+
     let mut links: Vec<NewVersionLink> = Vec::new();
     for p in exact.iter().filter(|p| p.status == Status::Exact) {
         links.push(NewVersionLink {
@@ -669,8 +676,32 @@ pub fn apply(
         });
     }
     for r in reviewed {
-        if current.get(&(r.library_id, r.version_path.as_str())) != Some(&r.version_hash.as_str()) {
+        let stale = if current.get(&(r.library_id, r.version_path.as_str()))
+            != Some(&r.version_hash.as_str())
+        {
+            Some("version file now holds different bytes (or is gone)")
+        } else if !live_hashes.contains(r.original_hash.as_str()) {
+            Some("original_hash is not in the index any more")
+        } else if !r.original_path.is_empty()
+            && current
+                .get(&(r.library_id, r.original_path.as_str()))
+                .is_some_and(|h| *h != r.original_hash.as_str())
+        {
+            Some("original path now holds different bytes than original_hash")
+        } else {
+            None
+        };
+        if let Some(why) = stale {
             report.stale += 1;
+            report.stale_details.push(format!(
+                "{} -> {}: {why}",
+                r.version_path,
+                if r.original_path.is_empty() {
+                    &r.original_hash
+                } else {
+                    &r.original_path
+                }
+            ));
             continue;
         }
         links.push(NewVersionLink {
@@ -977,6 +1008,19 @@ mod tests {
         stale_rows[2].content_hash = "changed".into();
         let report = apply(&mut dao, &[], &reviewed, &stale_rows);
         assert_eq!(report.stale, 1);
+
+        // So is one whose original changed since the dry run: its hash is
+        // gone from the index, or its path holds other bytes.
+        let mut gone = rows.clone();
+        gone[0].content_hash = "ha-edited".into();
+        let report = apply(&mut dao, &[], &reviewed, &gone);
+        assert_eq!(report.stale, 1);
+        assert_eq!(report.linked + report.already_linked, 0);
+        assert!(report.stale_details[0].contains("d/aiedits/o_00001_.png"));
+        let mut swapped = rows.clone();
+        swapped[0].content_hash = "hb".into(); // d/a.jpg now holds hb's bytes
+        let report = apply(&mut dao, &[], &reviewed, &swapped);
+        assert_eq!(report.stale, 1);
     }
     #[test]
     fn apply_drops_derived_data_of_an_already_ingested_version() {
@@ -995,16 +1039,17 @@ mod tests {
             "INSERT INTO clip_failures (library_id, rel_path, content_hash, error, failed_at) VALUES (1, 'v.png', 'v', 'boom', 0)",
         );
         let mut dao = SqliteVersionDao::from_shared(conn.clone());
-        let rows = vec![row("v.png", "v", None)];
         let reviewed = vec![ReviewedLink {
             method: "phash".into(),
             library_id: 1,
             version_path: "v.png".into(),
+            original_path: "o.jpg".into(),
             kind: "edit".into(),
             version_hash: "v".into(),
             original_hash: "o".into(),
             references: vec![],
         }];
+        let rows = vec![row("v.png", "v", None), row("o.jpg", "o", None)];
         assert_eq!(apply(&mut dao, &[], &reviewed, &rows).linked, 1);
 
         #[derive(diesel::QueryableByName)]

src/versions/integration_tests.rs

@@ -253,6 +253,46 @@ async fn name_link_pairs_raw_and_jpeg_with_the_right_original() {
     );
 }
 
+#[actix_rt::test]
+async fn unresolvable_name_candidates_do_not_starve_later_ones() {
+    let fx = fx();
+    // Many matching rows with no original anywhere, then one linkable pair.
+    for i in 0..120 {
+        fx.put(
+            &format!("album_a/aiedits/orphan{i}.jpg.edit.png"),
+            format!("orphan {i}").as_bytes(),
+        );
+    }
+    let original = fx.put("album_b/real.jpg", b"real original");
+    let version = fx.put("album_b/aiedits/real.jpg.restored.png", b"real version");
+
+    let exif = fx.exif_arc();
+    let mut cursor = 0;
+    let mut total = 0;
+    // One 50-row page per call: the old "ORDER BY id LIMIT n" would return
+    // the same 50 orphans forever.
+    for _ in 0..5 {
+        total += sweep_name_links_from(
+            &fx.library,
+            &exif,
+            &fx.state.version_dao,
+            &mut cursor,
+            50,
+            1,
+        );
+    }
+    assert_eq!(total, 1);
+    let link = fx
+        .state
+        .version_dao
+        .lock()
+        .unwrap()
+        .get_link(&version)
+        .unwrap()
+        .unwrap();
+    assert_eq!(link.original_hash, original);
+}
+
 #[actix_rt::test]
 async fn name_link_waits_for_the_original_to_be_hashed() {
     let fx = fx();
@@ -393,6 +433,103 @@ async fn photos_hides_versions_and_adds_parallel_arrays() {
     );
 }
 
+#[actix_rt::test]
+async fn folders_holding_only_hidden_versions_are_left_out_of_dirs() {
+    let (fx, _o, _v1, _v2) = seeded();
+    // A second folder whose aiedits holds a version that is NOT hidden
+    // (its original isn't indexed), and a folder nothing is indexed under.
+    fx.put_png("2004/album_a/aiedits/lost.jpg.edit.png", None, 20);
+    std::fs::create_dir_all(fx.root.join("2004/not_indexed")).unwrap();
+    let app = photos_app!(fx);
+
+    let dirs = |body: &serde_json::Value| -> Vec<String> {
+        let mut d: Vec<String> = body["dirs"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .map(|v| v.as_str().unwrap().to_string())
+            .collect();
+        d.sort();
+        d
+    };
+    let (_, body) = get_json!(app, "/photos?path=2004");
+    assert_eq!(dirs(&body), vec!["2004/album_a", "2004/not_indexed"]);
+
+    let (_, body) = get_json!(app, "/photos?path=2004&include_versions=true");
+    assert_eq!(
+        dirs(&body),
+        vec!["2004/aiedits", "2004/album_a", "2004/not_indexed"]
+    );
+}
+
+#[actix_rt::test]
+async fn a_manual_date_on_a_version_path_lands_on_the_original() {
+    let (fx, original, v1, _v2) = seeded();
+    fx.exec(&format!(
+        "UPDATE image_exif SET date_taken = 1500000000, date_utc = 1500000000, \
+         date_taken_source = 'exif' WHERE content_hash = '{original}'"
+    ));
+    let app = awtest::init_service(
+        App::new()
+            .service(crate::handlers::image::set_image_date)
+            .service(crate::handlers::image::clear_image_date)
+            .app_data(fx.state.clone())
+            .app_data(fx.exif.clone()),
+    )
+    .await;
+    let post = |uri: &str, body: serde_json::Value| {
+        awtest::TestRequest::post()
+            .uri(uri)
+            .insert_header(("Authorization", format!("Bearer {}", Fx::token())))
+            .set_json(body)
+            .to_request()
+    };
+    let resp = awtest::call_service(
+        &app,
+        post(
+            "/image/exif/date",
+            serde_json::json!({"path": "2004/aiedits/scan.jpg.restored.png", "date_taken": 1_400_000_000}),
+        ),
+    )
+    .await;
+    assert_eq!(resp.status().as_u16(), 200);
+
+    assert_eq!(
+        fx.count(&format!(
+            "SELECT COUNT(*) AS n FROM image_exif WHERE content_hash = '{original}' \
+             AND date_taken_source = 'manual'"
+        )),
+        1,
+        "the original carries the manual date"
+    );
+    assert_eq!(
+        fx.count(&format!(
+            "SELECT COUNT(*) AS n FROM image_exif WHERE content_hash = '{v1}' \
+             AND date_taken_source = 'manual'"
+        )),
+        0,
+        "the version row is left to the metadata sync"
+    );
+
+    // The same redirect applies to clearing it.
+    let resp = awtest::call_service(
+        &app,
+        post(
+            "/image/exif/date/clear",
+            serde_json::json!({"path": "2004/aiedits/scan.jpg.restored.png"}),
+        ),
+    )
+    .await;
+    assert!(resp.status().is_success());
+    assert_eq!(
+        fx.count(&format!(
+            "SELECT COUNT(*) AS n FROM image_exif WHERE content_hash = '{original}' \
+             AND date_taken_source = 'manual'"
+        )),
+        0
+    );
+}
+
 #[actix_rt::test]
 async fn photos_without_versions_is_unchanged() {
     let fx = fx();

src/versions/mod.rs

@@ -266,26 +266,83 @@ pub fn sweep_name_links(
     exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
     version_dao: &Arc<Mutex<Box<dyn VersionDao>>>,
 ) -> usize {
-    let candidates = match version_dao
+    let mut cursor = name_sweep_cursors()
         .lock()
-        .expect("Unable to lock VersionDao")
-        .unlinked_name_candidates(library.id, NAME_SWEEP_LIMIT)
-    {
-        Ok(c) => c,
-        Err(e) => {
-            warn!(
-                "versions: name sweep failed for '{}': {:?}",
-                library.name, e
-            );
-            return 0;
-        }
-    };
+        .expect("name sweep cursors")
+        .get(&library.id)
+        .copied()
+        .unwrap_or(0);
+    let linked = sweep_name_links_from(
+        library,
+        exif_dao,
+        version_dao,
+        &mut cursor,
+        NAME_SWEEP_LIMIT,
+        NAME_SWEEP_PAGES,
+    );
+    name_sweep_cursors()
+        .lock()
+        .expect("name sweep cursors")
+        .insert(library.id, cursor);
+    linked
+}
+
+/// Pages examined per sweep call.
+const NAME_SWEEP_PAGES: usize = 4;
+
+/// Per-library resume point (last candidate row id visited) for the name
+/// sweep. Candidates that never resolve (no original in the library) stay
+/// matching forever; paging past them keeps them from starving the rows
+/// behind them. In memory only: a restart just rescans from the start.
+fn name_sweep_cursors() -> &'static Mutex<std::collections::HashMap<i32, i32>> {
+    static CURSORS: std::sync::OnceLock<Mutex<std::collections::HashMap<i32, i32>>> =
+        std::sync::OnceLock::new();
+    CURSORS.get_or_init(Default::default)
+}
+
+/// One sweep: up to `max_pages` pages of `page` candidates from `cursor`,
+/// advancing it and wrapping to 0 past the last candidate (then continuing
+/// from the start, so a call never ends having examined nothing).
+fn sweep_name_links_from(
+    library: &Library,
+    exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
+    version_dao: &Arc<Mutex<Box<dyn VersionDao>>>,
+    cursor: &mut i32,
+    page: i64,
+    max_pages: usize,
+) -> usize {
     let mut linked = 0;
-    for (rel, hash) in candidates {
-        if let Some(link) = name_link(library, &rel, &hash, exif_dao)
-            && record_link(version_dao, &link)
+    let mut wrapped = *cursor == 0;
+    for _ in 0..max_pages {
+        let candidates = match version_dao
+            .lock()
+            .expect("Unable to lock VersionDao")
+            .unlinked_name_candidates(library.id, *cursor, page)
         {
-            linked += 1;
+            Ok(c) => c,
+            Err(e) => {
+                warn!(
+                    "versions: name sweep failed for '{}': {:?}",
+                    library.name, e
+                );
+                return linked;
+            }
+        };
+        let short = (candidates.len() as i64) < page;
+        for (id, rel, hash) in candidates {
+            *cursor = id;
+            if let Some(link) = name_link(library, &rel, &hash, exif_dao)
+                && record_link(version_dao, &link)
+            {
+                linked += 1;
+            }
+        }
+        if short {
+            *cursor = 0;
+            if wrapped {
+                break;
+            }
+            wrapped = true;
         }
     }
     linked