feat(versions): link at ingest, hide in listings, version endpoints

XMP (PNG iTXt DerivedFrom, native chunk reader) then file-name convention
links at ingest; unlinked name candidates are retried each tick. Versions
are hidden from /photos (SQL for the recursive listing, hidden-row set for
the walk and tag branches), /memories, reels, /photos/exif, the GPS summary,
unified search and the media gauges while their original is reachable in the
view; include_versions=true lists them. /photos gains the parallel arrays
photo_display_paths / photo_version_counts, memories / search hits /
/photos/exif rows gain display_path / version_count, all skipped when empty.
GET /image/versions and PUT /image/versions/display (library optional).
Tags and favorites on a version path bind to the original; faces, CLIP and
duplicate grouping skip version hashes; insight generation redirects to the
original. Orphan GC and the back-ref refresh cover the new tables.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/ai/handlers.rs

@@ -430,7 +430,17 @@ pub async fn generate_insight_handler(
         &mut request.persona_id,
     );
 
+    // Insights describe the original: asking for one on a photo version
+    // (#191) runs against the original it belongs to.
     let normalized_path = normalize_path(&request.file_path);
+    let normalized_path = app_state
+        .version_dao
+        .lock()
+        .expect("Unable to lock VersionDao")
+        .original_path_of(&normalized_path)
+        .ok()
+        .flatten()
+        .unwrap_or(normalized_path);
     let library = app_state.primary_library();
     let gen_type = InsightGenerationType::Standard;
 
@@ -748,7 +758,17 @@ pub async fn generate_agentic_insight_handler(
         &mut request.persona_id,
     );
 
+    // Insights describe the original: asking for one on a photo version
+    // (#191) runs against the original it belongs to.
     let normalized_path = normalize_path(&request.file_path);
+    let normalized_path = app_state
+        .version_dao
+        .lock()
+        .expect("Unable to lock VersionDao")
+        .original_path_of(&normalized_path)
+        .ok()
+        .flatten()
+        .unwrap_or(normalized_path);
     let library = app_state.primary_library();
     let gen_type = InsightGenerationType::Agentic;
 

src/ai/insight_generator.rs

@@ -2544,7 +2544,7 @@ Return ONLY the summary, nothing else."#,
         );
         let rows = {
             let mut dao = self.exif_dao.lock().expect("Unable to lock ExifDao");
-            match dao.query_by_exif(None, None, None, None, None, range) {
+            match dao.query_by_exif(None, None, None, None, None, range, false) {
                 Ok(r) => r,
                 Err(e) => return format!("Error querying photos: {:?}", e),
             }

src/backfill.rs

@@ -18,7 +18,7 @@ use prometheus::IntGauge;
 use crate::camera_tz;
 use crate::content_hash;
 use crate::database::models::PrecomputedReel;
-use crate::database::{ExifDao, PrecomputedReelDao, UtcBackfillCandidate};
+use crate::database::{ExifDao, PrecomputedReelDao, UtcBackfillCandidate, VersionDao};
 use crate::date_resolver;
 use crate::face_watch;
 use crate::faces;
@@ -827,12 +827,14 @@ pub fn backfill_missing_content_hashes(
 /// content_hash recorded in image_exif but no row in face_detections
 /// yet. Re-querying image_exif here picks up rows the EXIF write loop
 /// just inserted alongside any pre-existing rows the watcher walked
-/// over — covers both new uploads and the initial backlog scan.
+/// over — covers both new uploads and the initial backlog scan. Photo
+/// versions (#191) are never candidates: faces come from the original only.
 pub fn build_face_candidates(
     library: &libraries::Library,
     files: &[(PathBuf, String)],
     exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
     face_dao: &Arc<Mutex<Box<dyn faces::FaceDao>>>,
+    version_dao: &Arc<Mutex<Box<dyn VersionDao>>>,
 ) -> Vec<face_watch::FaceCandidate> {
     // Restrict to image files; videos aren't face-scanned in v1 (kamadak
     // doesn't even register them in image_exif).
@@ -859,12 +861,22 @@ pub fn build_face_candidates(
         }
     }
 
+    let hashes: Vec<String> = hash_by_path.values().cloned().collect();
+    let versions = version_dao
+        .lock()
+        .expect("version dao")
+        .version_hashes_among(&hashes)
+        .unwrap_or_default();
+
     let mut candidates = Vec::new();
     let mut dao = face_dao.lock().expect("face dao");
     for rel_path in image_paths {
         let Some(hash) = hash_by_path.get(&rel_path) else {
             continue;
         };
+        if versions.contains(hash) {
+            continue;
+        }
         match dao.already_scanned(hash) {
             Ok(true) => continue,
             Ok(false) => candidates.push(face_watch::FaceCandidate {
@@ -891,9 +903,11 @@ mod tests {
 
     use crate::database::models::{InsertImageExif, InsertLibrary};
     use crate::database::test::in_memory_db_connection;
-    use crate::database::{ExifDao, SqliteExifDao, schema};
+    use crate::database::version_dao::NewVersionLink;
+    use crate::database::{ExifDao, SqliteExifDao, SqliteVersionDao, schema};
     use crate::faces::{FaceDao, SqliteFaceDao};
     use crate::libraries::Library;
+    use diesel::sqlite::SqliteConnection;
 
     /// Everything `setup` hands back to a test: tempdir, library, shared
     /// connection, and the two DAOs. Aliased to keep clippy's
@@ -1148,7 +1162,9 @@ mod tests {
             (tmp.path().join("unhashed.jpg"), "unhashed.jpg".to_string()),
             (tmp.path().join("clip.mp4"), "clip.mp4".to_string()),
         ];
-        let candidates = build_face_candidates(&library, &files, &exif_dao, &face_dao);
+        let version_dao = version_dao_for(&_conn);
+        let candidates =
+            build_face_candidates(&library, &files, &exif_dao, &face_dao, &version_dao);
 
         assert_eq!(
             candidates.len(),
@@ -1159,6 +1175,58 @@ mod tests {
         assert_eq!(candidates[0].content_hash, "hash-fresh");
     }
 
+    fn version_dao_for(conn: &Arc<Mutex<SqliteConnection>>) -> Arc<Mutex<Box<dyn VersionDao>>> {
+        Arc::new(Mutex::new(Box::new(SqliteVersionDao::from_shared(
+            Arc::clone(conn),
+        ))))
+    }
+
+    #[test]
+    fn build_face_candidates_skips_photo_versions() {
+        let (tmp, library, conn, exif_dao, face_dao) = setup();
+        insert_exif(&exif_dao, 1, "orig.jpg", Some("hash-orig"));
+        insert_exif(
+            &exif_dao,
+            1,
+            "aiedits/orig.jpg.restored.png",
+            Some("hash-ver"),
+        );
+        let version_dao = version_dao_for(&conn);
+        version_dao
+            .lock()
+            .unwrap()
+            .link_version(&NewVersionLink {
+                version_hash: "hash-ver".into(),
+                original_hash: "hash-orig".into(),
+                kind: "restored".into(),
+                source: "xmp".into(),
+                library_id: 1,
+                rel_path: "aiedits/orig.jpg.restored.png".into(),
+                references: vec![],
+            })
+            .unwrap();
+        let files: Vec<(PathBuf, String)> = vec![
+            (tmp.path().join("orig.jpg"), "orig.jpg".to_string()),
+            (
+                tmp.path().join("aiedits/orig.jpg.restored.png"),
+                "aiedits/orig.jpg.restored.png".to_string(),
+            ),
+        ];
+        let candidates =
+            build_face_candidates(&library, &files, &exif_dao, &face_dao, &version_dao);
+        assert_eq!(candidates.len(), 1);
+        assert_eq!(candidates[0].rel_path, "orig.jpg");
+
+        // The SQL backlog drain agrees.
+        let unscanned = face_dao
+            .lock()
+            .unwrap()
+            .list_unscanned_candidates(1, 10)
+            .unwrap();
+        assert_eq!(unscanned.len(), 1);
+        assert_eq!(unscanned[0].0, "orig.jpg");
+    }
+
     // --- True-instant backfill drain (ADR-0001, ticket #118) -------------
 
     use crate::database::SqlitePrecomputedReelDao;

src/clip_search.rs

@@ -82,6 +82,10 @@ pub struct SearchHit {
     /// Cosine similarity in [-1, 1]. In practice OpenAI CLIP returns
     /// 0.10–0.40 for the typical photo library.
     pub score: f32,
+    /// `display_path` / `version_count` (#191); absent for a photo without
+    /// versions. `rel_path` is always the original's.
+    #[serde(flatten)]
+    pub versions: crate::versions::VersionFields,
 }
 
 #[derive(Debug, Serialize)]
@@ -230,11 +234,13 @@ pub async fn score_photos(
 /// to a row are skipped. Shared by both endpoints.
 pub fn resolve_hits(
     exif_dao: &Mutex<Box<dyn ExifDao>>,
+    version_dao: &Mutex<Box<dyn crate::database::VersionDao>>,
     scored: &[(f32, String)],
 ) -> Vec<SearchHit> {
     if scored.is_empty() {
         return Vec::new();
     }
+    let summaries = crate::versions::load_summaries(version_dao);
     let hashes: Vec<String> = scored.iter().map(|(_, h)| h.clone()).collect();
     let mut dao = exif_dao.lock().expect("exif dao");
     let path_map = dao.get_rel_paths_for_hashes(&hashes).unwrap_or_else(|e| {
@@ -258,11 +264,15 @@ pub fn resolve_hits(
             .get(hash)
             .and_then(|paths| paths.first().cloned())
             .unwrap_or(row.file_path);
+        let versions = crate::versions::VersionFields::from_summary(
+            summaries.get(&(row.library_id, rel_path.clone())),
+        );
         results.push(SearchHit {
             library_id: row.library_id,
             rel_path,
             content_hash: hash.clone(),
             score: *score,
+            versions,
         });
     }
     results
@@ -343,7 +353,7 @@ pub async fn search_photos(
         let end = (offset + limit).min(total_matching);
         scored.hits[offset..end].to_vec()
     };
-    let results = resolve_hits(&exif_dao, &page);
+    let results = resolve_hits(&exif_dao, &state.version_dao, &page);
 
     Ok(HttpResponse::Ok().json(SearchResponse {
         query: q_text,

src/data/mod.rs

@@ -108,6 +108,19 @@ pub struct PhotosResponse {
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub photo_libraries: Vec<i32>,
 
+    /// Photo versions (#191), parallel to `photos`. `photos[i]` is always
+    /// the *original's* path (it anchors tags, sort order and dates).
+    /// `photo_display_paths[i]` is the file to show instead when the global
+    /// display choice isn't the original, else `null`; skipped when no photo
+    /// on the page has one.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub photo_display_paths: Vec<Option<String>>,
+
+    /// How many versions stand behind `photos[i]`; skipped when the page has
+    /// none, so responses without versions are unchanged.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub photo_version_counts: Vec<u32>,
+
     // Pagination metadata (only present when limit is set)
     #[serde(skip_serializing_if = "Option::is_none")]
     pub total_count: Option<i64>,
@@ -180,6 +193,13 @@ pub struct FilesRequest {
     /// inside a group.
     #[serde(default)]
     pub include_duplicates: Option<bool>,
+
+    /// When true, list photo versions (#191) as photos of their own instead
+    /// of hiding them behind their original. Default false. Counterpart of
+    /// `include_duplicates`; a version whose original is not reachable in
+    /// the request's view is listed either way.
+    #[serde(default)]
+    pub include_versions: Option<bool>,
 }
 
 #[derive(Copy, Clone, Deserialize, PartialEq, Debug)]
@@ -495,6 +515,11 @@ pub struct ExifSummary {
     /// Capture wall clock to display (naive ISO-8601; see
     /// `MetadataResponse::wall_clock`).
     pub wall_clock: Option<String>,
+    /// `display_path` / `version_count` (#191); absent for a photo without
+    /// versions. `file_path` is always the original's (versions are hidden
+    /// from this listing while their original is reachable).
+    #[serde(flatten)]
+    pub versions: crate::versions::VersionFields,
 }
 
 impl ExifSummary {
@@ -525,6 +550,7 @@ impl ExifSummary {
             date_utc: r.date_utc,
             camera_offset_minutes: r.camera_offset_minutes,
             date_taken_tz_source: r.date_taken_tz_source,
+            versions: Default::default(),
         }
     }
 }

src/database/mod.rs

@@ -8,6 +8,7 @@ use tracing_opentelemetry::OpenTelemetrySpanExt;
 use crate::database::models::{
     Favorite, ImageExif, InsertFavorite, InsertImageExif, InsertUser, User,
 };
+use crate::database::version_dao::{NOT_A_VERSION_CLAUSE, visible_version_clause};
 use crate::otel::trace_db_call;
 
 /// Wire shape for a single member of a duplicate group, returned by
@@ -311,6 +312,11 @@ impl FavoriteDao for SqliteFavoriteDao {
         use schema::favorites::dsl::*;
 
         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());
 
         if favorites
             .filter(userid.eq(user_id).and(rel_path.eq(&favorite_path)))
@@ -332,9 +338,14 @@ impl FavoriteDao for SqliteFavoriteDao {
     fn remove_favorite(&mut self, user_id: i32, favorite_path: String) {
         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);
         diesel::delete(favorites)
             .filter(userid.eq(user_id).and(rel_path.eq(favorite_path)))
-            .execute(self.connection.lock().unwrap().deref_mut())
+            .execute(connection.deref_mut())
             .unwrap();
     }
 
@@ -403,6 +414,10 @@ pub trait ExifDao: Sync + Send {
     /// or the permanent floor) through `date_view::row_true_instant`'s
     /// per-source fallback. Every row returned satisfies
     /// [`InstantRange::contains`](crate::date_view::InstantRange::contains).
+    ///
+    /// `include_versions=false` drops version rows whose original is
+    /// reachable in the view (`library_id`, or every library when `None`);
+    /// see [`visible_version_clause`].
     fn query_by_exif(
         &mut self,
         library_id: Option<i32>,
@@ -411,6 +426,7 @@ pub trait ExifDao: Sync + Send {
         lens_model: Option<&str>,
         gps_bounds: Option<(f64, f64, f64, f64)>, // (min_lat, max_lat, min_lon, max_lon)
         date_range: Option<crate::date_view::InstantRange>,
+        include_versions: bool,
     ) -> Result<Vec<ImageExif>, DbError>;
 
     /// Get distinct camera makes with counts
@@ -655,6 +671,10 @@ pub trait ExifDao: Sync + Send {
     ///
     /// Returns full rows so the caller can read the date triple
     /// (`date_utc` / `camera_offset_minutes` / `date_taken_tz_source`).
+    ///
+    /// `view_library_ids` is the whole view the caller is assembling (it
+    /// queries one library at a time): a photo version is left out while its
+    /// original has a row in any of them.
     fn get_memories_in_window(
         &mut self,
         library_id: i32,
@@ -662,6 +682,7 @@ pub trait ExifDao: Sync + Send {
         conv_hi: i64,
         unconv_lo: i64,
         unconv_hi: i64,
+        view_library_ids: &[i32],
     ) -> Result<Vec<ImageExif>, DbError>;
 
     /// Return image rows that have a `content_hash` but no `phash_64`,
@@ -811,11 +832,16 @@ pub trait ExifDao: Sync + Send {
     /// `duplicate_of_hash IS NOT NULL` so the default photo listing hides
     /// demoted siblings; the Apollo duplicates modal passes `true` to
     /// see both survivors and demoted members inside a group.
+    ///
+    /// `include_versions=false` likewise hides photo versions while their
+    /// original is reachable in the same view (an unscoped call counts every
+    /// library; a scoped one only `library_ids`).
     fn list_rel_paths_for_libraries(
         &mut self,
         library_ids: &[i32],
         path_prefix: Option<&str>,
         include_duplicates: bool,
+        include_versions: bool,
     ) -> Result<Vec<(i32, String)>, DbError>;
 
     /// Delete a single image_exif row scoped to `(library_id, rel_path)`.
@@ -899,6 +925,12 @@ fn classify_image_exif_error(e: &anyhow::Error) -> DbErrorKind {
     }
 }
 
+#[derive(diesel::QueryableByName)]
+struct OriginalHashRow {
+    #[diesel(sql_type = diesel::sql_types::Text)]
+    original_hash: String,
+}
+
 pub struct SqliteExifDao {
     connection: Arc<Mutex<SqliteConnection>>,
 }
@@ -1103,6 +1135,7 @@ impl ExifDao for SqliteExifDao {
         lens_model_filter: Option<&str>,
         gps_bounds: Option<(f64, f64, f64, f64)>,
         date_range: Option<crate::date_view::InstantRange>,
+        include_versions: bool,
     ) -> Result<Vec<ImageExif>, DbError> {
         trace_db_call("query", "query_by_exif", |_span| {
             use schema::image_exif::dsl::*;
@@ -1110,6 +1143,13 @@ impl ExifDao for SqliteExifDao {
             let mut connection = self.connection.lock().expect("Unable to get ExifDao");
             let mut query = image_exif.into_boxed();
 
+            if !include_versions {
+                let view: Vec<i32> = library_id_filter.into_iter().collect();
+                query = query.filter(diesel::dsl::sql::<diesel::sql_types::Bool>(
+                    &visible_version_clause(&view),
+                ));
+            }
+
             // Library scope (most-selective filter — apply first so the
             // `(library_id, ...)` indexes are eligible).
             if let Some(lib_id) = library_id_filter {
@@ -1177,6 +1217,9 @@ impl ExifDao for SqliteExifDao {
 
             image_exif
                 .filter(camera_make.is_not_null())
+                .filter(diesel::dsl::sql::<diesel::sql_types::Bool>(
+                    &visible_version_clause(&[]),
+                ))
                 .group_by(camera_make)
                 .select((camera_make, count(id)))
                 .order(count(id).desc())
@@ -1202,6 +1245,9 @@ impl ExifDao for SqliteExifDao {
             image_exif
                 .filter(camera_make.is_not_null())
                 .filter(camera_model.is_not_null())
+                .filter(diesel::dsl::sql::<diesel::sql_types::Bool>(
+                    &visible_version_clause(&[]),
+                ))
                 .group_by((camera_make, camera_model))
                 .select((camera_make, camera_model, count(id)))
                 .order(count(id).desc())
@@ -1262,6 +1308,9 @@ impl ExifDao for SqliteExifDao {
             // Query all photos with non-null GPS coordinates
             let mut query = image_exif
                 .filter(gps_latitude.is_not_null().and(gps_longitude.is_not_null()))
+                .filter(diesel::dsl::sql::<diesel::sql_types::Bool>(
+                    &visible_version_clause(&[]),
+                ))
                 .into_boxed();
 
             // Apply path filtering
@@ -1572,6 +1621,10 @@ impl ExifDao for SqliteExifDao {
                 .filter(library_id.eq(library_id_val))
                 .filter(clip_embedding.is_null())
                 .filter(content_hash.is_not_null())
+                // Versions are never embedded: invented detail would pollute search.
+                .filter(diesel::dsl::sql::<diesel::sql_types::Bool>(
+                    NOT_A_VERSION_CLAUSE,
+                ))
                 .filter(diesel::dsl::not(diesel::dsl::exists(failed)))
                 .select((rel_path, content_hash.assume_not_null()))
                 .order(id.asc())
@@ -1685,6 +1738,9 @@ impl ExifDao for SqliteExifDao {
             let mut query = image_exif
                 .filter(content_hash.is_not_null())
                 .filter(clip_embedding.is_not_null())
+                .filter(diesel::dsl::sql::<diesel::sql_types::Bool>(
+                    NOT_A_VERSION_CLAUSE,
+                ))
                 .into_boxed();
             if !library_ids_val.is_empty() {
                 query = query.filter(library_id.eq_any(library_ids_val));
@@ -1879,6 +1935,7 @@ impl ExifDao for SqliteExifDao {
         conv_hi: i64,
         unconv_lo: i64,
         unconv_hi: i64,
+        view_library_ids: &[i32],
     ) -> Result<Vec<ImageExif>, DbError> {
         trace_db_call("query", "get_memories_in_window", |_span| {
             use crate::database::schema::image_exif::dsl as d;
@@ -1887,6 +1944,9 @@ impl ExifDao for SqliteExifDao {
 
             d::image_exif
                 .filter(d::library_id.eq(library_id))
+                .filter(diesel::dsl::sql::<diesel::sql_types::Bool>(
+                    &visible_version_clause(view_library_ids),
+                ))
                 .filter(d::date_taken.is_not_null())
                 .filter(d::date_taken.ge(unconv_lo))
                 .filter(d::date_taken.le(unconv_hi))
@@ -1938,13 +1998,27 @@ impl ExifDao for SqliteExifDao {
                 .map_err(|e| anyhow::anyhow!("Query error: {}", e))?
                 .flatten();
 
+            // A version shares its annotations (tags, favorites) with its
+            // original, so its hash set includes the original's.
             let paths = match hash {
-                Some(h) => image_exif
+                Some(h) => {
-                    .filter(content_hash.eq(h))
+                    let original: Option<String> = diesel::sql_query(
-                    .select(rel_path)
+                        "SELECT original_hash FROM photo_versions WHERE version_hash = ?",
-                    .distinct()
+                    )
-                    .load::<String>(connection.deref_mut())
+                    .bind::<diesel::sql_types::Text, _>(&h)
-                    .map_err(|e| anyhow::anyhow!("Query error: {}", e))?,
+                    .get_result::<OriginalHashRow>(connection.deref_mut())
+                    .optional()
+                    .map_err(|e| anyhow::anyhow!("Query error: {}", e))?
+                    .map(|r| r.original_hash);
+                    let mut hashes = vec![h];
+                    hashes.extend(original);
+                    image_exif
+                        .filter(content_hash.eq_any(hashes))
+                        .select(rel_path)
+                        .distinct()
+                        .load::<String>(connection.deref_mut())
+                        .map_err(|e| anyhow::anyhow!("Query error: {}", e))?
+                }
                 None => vec![rel_path_val.to_string()],
             };
 
@@ -2043,6 +2117,7 @@ impl ExifDao for SqliteExifDao {
         library_ids: &[i32],
         path_prefix: Option<&str>,
         include_duplicates: bool,
+        include_versions: bool,
     ) -> Result<Vec<(i32, String)>, DbError> {
         trace_db_call("query", "list_rel_paths_for_libraries", |_span| {
             use schema::image_exif::dsl::*;
@@ -2098,6 +2173,12 @@ impl ExifDao for SqliteExifDao {
                 }
             }
 
+            if !include_versions {
+                query = query.filter(diesel::dsl::sql::<diesel::sql_types::Bool>(
+                    &visible_version_clause(library_ids),
+                ));
+            }
+
             query
                 .load::<(i32, String)>(connection.deref_mut())
                 .map_err(|e| anyhow::anyhow!("Query error: {}", e))
@@ -2263,6 +2344,9 @@ impl ExifDao for SqliteExifDao {
             use schema::image_exif::dsl::*;
             let mut q = image_exif
                 .filter(content_hash.eq_any(&dup_hashes))
+                .filter(diesel::dsl::sql::<diesel::sql_types::Bool>(
+                    NOT_A_VERSION_CLAUSE,
+                ))
                 .select((
                     library_id,
                     rel_path,
@@ -2343,6 +2427,9 @@ impl ExifDao for SqliteExifDao {
             let mut q = image_exif
                 .filter(content_hash.is_not_null())
                 .filter(phash_64.is_not_null())
+                .filter(diesel::dsl::sql::<diesel::sql_types::Bool>(
+                    NOT_A_VERSION_CLAUSE,
+                ))
                 .select((
                     library_id,
                     rel_path,
@@ -2421,6 +2508,9 @@ impl ExifDao for SqliteExifDao {
 
             let mut q = image_exif
                 .filter(content_hash.is_not_null())
+                .filter(diesel::dsl::sql::<diesel::sql_types::Bool>(
+                    NOT_A_VERSION_CLAUSE,
+                ))
                 .select((library_id, rel_path))
                 .into_boxed();
             if let Some(lib) = library_id_filter {
@@ -2749,13 +2839,13 @@ mod exif_dao_tests {
 
         // Union: both rows.
         let all = dao
-            .query_by_exif(None, None, None, None, None, None)
+            .query_by_exif(None, None, None, None, None, None, false)
             .unwrap();
         assert_eq!(all.len(), 2);
 
         // Scoped to lib 2: only archive row.
         let lib2 = dao
-            .query_by_exif(Some(2), None, None, None, None, None)
+            .query_by_exif(Some(2), None, None, None, None, None, false)
             .unwrap();
         assert_eq!(lib2.len(), 1);
         assert_eq!(lib2[0].file_path, "archive/a.jpg");
@@ -2908,6 +2998,7 @@ mod exif_dao_tests {
                 None,
                 None,
                 jul5_ny(Some(JUL5_NY_START), Some(JUL5_NY_END)),
+                false,
             )
             .unwrap();
         assert_eq!(paths(rows), vec!["redated.jpg"]);
@@ -2949,6 +3040,7 @@ mod exif_dao_tests {
                 None,
                 None,
                 jul5_ny(Some(JUL5_NY_START), Some(JUL5_NY_END)),
+                false,
             )
             .unwrap();
         assert_eq!(paths(rows), vec!["jul5_early.jpg"]);
@@ -2979,6 +3071,7 @@ mod exif_dao_tests {
                 None,
                 None,
                 jul5_ny(Some(JUL5_NY_START), Some(JUL5_NY_END)),
+                false,
             )
             .unwrap();
         assert_eq!(paths(in_day), vec!["tokyo.jpg"]);
@@ -2992,6 +3085,7 @@ mod exif_dao_tests {
                 None,
                 None,
                 jul5_ny(Some(1_720_195_200), Some(JUL5_NY_END)),
+                false,
             )
             .unwrap();
         assert!(afternoon.is_empty());
@@ -3021,6 +3115,7 @@ mod exif_dao_tests {
                 None,
                 None,
                 jul5_ny(Some(JUL5_NY_START), None),
+                false,
             )
             .unwrap();
         assert_eq!(paths(from_only), vec!["a.jpg"]);
@@ -3032,12 +3127,13 @@ mod exif_dao_tests {
                 None,
                 None,
                 jul5_ny(None, Some(JUL5_NY_START)),
+                false,
             )
             .unwrap();
         assert!(to_only.is_empty());
         assert!(jul5_ny(None, None).is_none());
         let all = dao
-            .query_by_exif(None, None, None, None, None, None)
+            .query_by_exif(None, None, None, None, None, None, false)
             .unwrap();
         assert_eq!(paths(all), vec!["a.jpg", "undated.jpg"]);
     }
@@ -3221,7 +3317,9 @@ mod exif_dao_tests {
         insert_row_with_source(&mut dao, 1, "nodate.jpg", None, None);
 
         let (clo, chi, ulo, uhi) = wide_windows(now_ts, 20);
-        let rows = dao.get_memories_in_window(1, clo, chi, ulo, uhi).unwrap();
+        let rows = dao
+            .get_memories_in_window(1, clo, chi, ulo, uhi, &[])
+            .unwrap();
         let paths: std::collections::HashSet<String> =
             rows.into_iter().map(|r| r.file_path).collect();
         assert!(
@@ -3254,8 +3352,12 @@ mod exif_dao_tests {
             Some("exif"),
         );
         let (clo, chi, ulo, uhi) = wide_windows(now_ts, 20);
-        let lib1 = dao.get_memories_in_window(1, clo, chi, ulo, uhi).unwrap();
+        let lib1 = dao
-        let lib2 = dao.get_memories_in_window(2, clo, chi, ulo, uhi).unwrap();
+            .get_memories_in_window(1, clo, chi, ulo, uhi, &[])
+            .unwrap();
+        let lib2 = dao
+            .get_memories_in_window(2, clo, chi, ulo, uhi, &[])
+            .unwrap();
         assert_eq!(lib1.len(), 1);
         assert_eq!(lib1[0].file_path, "main/x.jpg");
         assert_eq!(lib2.len(), 1);
@@ -3320,7 +3422,9 @@ mod exif_dao_tests {
         let chi = now_ts - year;
         let ulo = now_ts - year - 86_400;
         let uhi = now_ts;
-        let rows = dao.get_memories_in_window(1, clo, chi, ulo, uhi).unwrap();
+        let rows = dao
+            .get_memories_in_window(1, clo, chi, ulo, uhi, &[])
+            .unwrap();
         let paths: std::collections::HashSet<String> =
             rows.into_iter().map(|r| r.file_path).collect();
         assert!(

src/database/version_dao.rs

@@ -170,6 +170,10 @@ pub trait VersionDao: Sync + Send {
     /// The link row for a version hash, if any.
     fn get_link(&mut self, version_hash: &str) -> Result<Option<PhotoVersionRow>, DbError>;
 
+    /// If `rel_path` is a linked version whose original is still indexed,
+    /// the original's rel_path (same library when possible).
+    fn original_path_of(&mut self, rel_path: &str) -> Result<Option<String>, DbError>;
+
     /// Which of `hashes` are versions of something.
     fn version_hashes_among(&mut self, hashes: &[String]) -> Result<HashSet<String>, DbError>;
 
@@ -191,8 +195,11 @@ pub trait VersionDao: Sync + Send {
     fn get_display(&mut self, original_hash: &str) -> Result<Option<String>, DbError>;
 
     /// Set (`Some`) or clear (`None`) the global display choice.
-    fn set_display(&mut self, original_hash: &str, display_hash: Option<&str>)
+    fn set_display(
-    -> Result<(), DbError>;
+        &mut self,
+        original_hash: &str,
+        display_hash: Option<&str>,
+    ) -> Result<(), DbError>;
 
     /// Per original `(library_id, rel_path)` that has at least one live
     /// version: the version count and display file. Empty when no photo has
@@ -406,6 +413,15 @@ impl VersionDao for SqliteVersionDao {
         .map_err(|e| DbError::log(DbErrorKind::QueryError, e))
     }
 
+    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)
+                .map_err(|e| anyhow::anyhow!("Query error: {}", e))
+        })
+        .map_err(|e| DbError::log(DbErrorKind::QueryError, e))
+    }
+
     fn version_hashes_among(&mut self, hashes: &[String]) -> Result<HashSet<String>, DbError> {
         trace_db_call("query", "version_hashes_among", |_span| {
             #[derive(QueryableByName)]
@@ -912,10 +928,7 @@ mod tests {
         );
         // Original deleted: the version shows up everywhere.
         exec(&c, "DELETE FROM image_exif WHERE rel_path = 'a/orig.jpg'");
-        assert_eq!(
+        assert_eq!(visible(&[]), vec!["a/orig.jpg.restored.png", "b/other.jpg"]);
-            visible(&[]),
-            vec!["a/orig.jpg.restored.png", "b/other.jpg"]
-        );
     }
 
     #[test]

src/duplicates.rs

@@ -65,6 +65,13 @@ lazy_static! {
     static ref FOLDER_PAIR_CACHE: Mutex<Option<FolderPairCacheEntry>> = Mutex::new(None);
 }
 
+/// Drop both caches. Called when a version link appears: the version
+/// leaves the duplicate groups, so cached clusters that include it are stale.
+pub fn invalidate_caches() {
+    invalidate_perceptual_cache();
+    invalidate_folder_pair_cache();
+}
+
 /// Drop the perceptual-cluster cache. Called from `resolve`/`unresolve`
 /// so the next modal open reflects the soft-mark change immediately.
 fn invalidate_perceptual_cache() {

src/faces.rs

@@ -569,12 +569,16 @@ impl FaceDao for SqliteFaceDao {
             // be filtered client-side and re-pulled every tick forever
             // because no marker row is written for excluded paths.
             let ext_predicate = image_path_predicate("rel_path");
+            // Photo versions (#191) are never scanned: invented detail
+            // must not seed person clusters.
+            let not_version = crate::database::version_dao::not_a_version_clause_for("e");
             let sql = format!(
                 "SELECT rel_path, content_hash \
                  FROM image_exif e \
                  WHERE library_id = ? \
                    AND content_hash IS NOT NULL \
                    AND {ext_predicate} \
+                   AND {not_version} \
                    AND NOT EXISTS ( \
                      SELECT 1 FROM face_detections f \
                      WHERE f.content_hash = e.content_hash \
@@ -928,7 +932,11 @@ impl FaceDao for SqliteFaceDao {
             // hash can have many rows (multiple detected faces) but we
             // want it counted once.
             let scanned: i64 = {
-                let mut q = face_detections::table.into_boxed();
+                let mut q = face_detections::table
+                    .filter(diesel::dsl::sql::<diesel::sql_types::Bool>(
+                        "content_hash NOT IN (SELECT version_hash FROM photo_versions)",
+                    ))
+                    .into_boxed();
                 if let Some(lib) = library_id {
                     q = q.filter(face_detections::library_id.eq(lib));
                 }
@@ -944,6 +952,9 @@ impl FaceDao for SqliteFaceDao {
             let with_faces: i64 = {
                 let mut q = face_detections::table
                     .filter(face_detections::status.eq("detected"))
+                    .filter(diesel::dsl::sql::<diesel::sql_types::Bool>(
+                        "content_hash NOT IN (SELECT version_hash FROM photo_versions)",
+                    ))
                     .into_boxed();
                 if let Some(lib) = library_id {
                     q = q.filter(face_detections::library_id.eq(lib));
@@ -960,6 +971,9 @@ impl FaceDao for SqliteFaceDao {
             let no_faces: i64 = {
                 let mut q = face_detections::table
                     .filter(face_detections::status.eq("no_faces"))
+                    .filter(diesel::dsl::sql::<diesel::sql_types::Bool>(
+                        "content_hash NOT IN (SELECT version_hash FROM photo_versions)",
+                    ))
                     .into_boxed();
                 if let Some(lib) = library_id {
                     q = q.filter(face_detections::library_id.eq(lib));
@@ -976,6 +990,9 @@ impl FaceDao for SqliteFaceDao {
             let failed: i64 = {
                 let mut q = face_detections::table
                     .filter(face_detections::status.eq("failed"))
+                    .filter(diesel::dsl::sql::<diesel::sql_types::Bool>(
+                        "content_hash NOT IN (SELECT version_hash FROM photo_versions)",
+                    ))
                     .into_boxed();
                 if let Some(lib) = library_id {
                     q = q.filter(face_detections::library_id.eq(lib));
@@ -1006,10 +1023,14 @@ impl FaceDao for SqliteFaceDao {
             // below 100% for the duration of that backfill.
             let total_photos: i64 = {
                 let ext_predicate = image_path_predicate("rel_path");
+                // Versions are never face-scanned, so they stay out of the
+                // denominator (and the numerators above).
+                let not_version = crate::database::version_dao::NOT_A_VERSION_CLAUSE;
                 let row: CountRow = if let Some(lib) = library_id {
                     let sql = format!(
                         "SELECT COUNT(DISTINCT content_hash) AS count FROM image_exif \
-                         WHERE library_id = ? AND content_hash IS NOT NULL AND {ext_predicate}"
+                         WHERE library_id = ? AND content_hash IS NOT NULL AND {ext_predicate} \
+                           AND {not_version}"
                     );
                     diesel::sql_query(sql)
                         .bind::<diesel::sql_types::Integer, _>(lib)
@@ -1018,7 +1039,8 @@ impl FaceDao for SqliteFaceDao {
                 } else {
                     let sql = format!(
                         "SELECT COUNT(DISTINCT content_hash) AS count FROM image_exif \
-                         WHERE content_hash IS NOT NULL AND {ext_predicate}"
+                         WHERE content_hash IS NOT NULL AND {ext_predicate} \
+                           AND {not_version}"
                     );
                     diesel::sql_query(sql)
                         .get_result(conn.deref_mut())

src/files.rs

@@ -342,6 +342,9 @@ pub async fn list_photos<TagD: TagDao, FS: FileSystemAccess>(
                 req.lens_model.as_deref(),
                 gps_bounds,
                 date_range,
+                // Hiding happens in the listing below; this is only the
+                // set the listing is intersected with.
+                true,
             )
             .unwrap_or_else(|e| {
                 warn!("EXIF query failed: {:?}", e);
@@ -391,6 +394,7 @@ pub async fn list_photos<TagD: TagDao, FS: FileSystemAccess>(
     };
 
     let search_recursively = req.recursive.unwrap_or(false);
+    let include_versions = req.include_versions.unwrap_or(false);
     if let Some(tag_ids) = &req.tag_ids
         && search_recursively
     {
@@ -498,6 +502,21 @@ pub async fn list_photos<TagD: TagDao, FS: FileSystemAccess>(
                 })
                 .collect::<Vec<(FileWithTagCount, i32)>>()
         })
+        .map(|paired| {
+            // Versions stay hidden behind their original (tag hits can be
+            // either file, since tags bind to the original's path).
+            if include_versions {
+                return paired;
+            }
+            let hidden = hidden_version_rows(&app_state, library.map(|l| l.id));
+            if hidden.is_empty() {
+                return paired;
+            }
+            paired
+                .into_iter()
+                .filter(|(f, lib_id)| !hidden.contains(&(*lib_id, f.file_name.clone())))
+                .collect()
+        })
         .map(|paired| {
             // Handle sorting - use helper function that supports EXIF date sorting and pagination
             let sort_type = req.sort.unwrap_or(NameAsc);
@@ -545,10 +564,14 @@ pub async fn list_photos<TagD: TagDao, FS: FileSystemAccess>(
             span.set_attribute("total_count", total_count.to_string());
             span.set_status(Status::Ok);
 
+            let (photo_display_paths, photo_version_counts) =
+                version_arrays(&app_state, &tagged_files, &photo_libraries);
             HttpResponse::Ok().json(PhotosResponse {
                 photos: tagged_files,
                 dirs: vec![],
                 photo_libraries,
+                photo_display_paths,
+                photo_version_counts,
                 total_count: pagination_metadata.0,
                 has_more: pagination_metadata.1,
                 next_offset: pagination_metadata.2,
@@ -583,7 +606,7 @@ pub async fn list_photos<TagD: TagDao, FS: FileSystemAccess>(
         let include_duplicates = req.include_duplicates.unwrap_or(false);
         let rows = {
             let mut dao = exif_dao.lock().expect("Unable to get ExifDao");
-            dao.list_rel_paths_for_libraries(&lib_ids, prefix, include_duplicates)
+            dao.list_rel_paths_for_libraries(&lib_ids, prefix, include_duplicates, include_versions)
                 .unwrap_or_else(|e| {
                     warn!("list_rel_paths_for_libraries failed: {:?}", e);
                     Vec::new()
@@ -681,6 +704,21 @@ pub async fn list_photos<TagD: TagDao, FS: FileSystemAccess>(
         }
     }
 
+    // The filesystem walk knows nothing about versions: drop the ones hidden
+    // behind their original in this view (the recursive branch did it in SQL).
+    if !search_recursively && !include_versions {
+        let hidden = hidden_version_rows(&app_state, library.map(|l| l.id));
+        if !hidden.is_empty() {
+            let (names, libs): (Vec<String>, Vec<i32>) = file_names
+                .into_iter()
+                .zip(file_libraries)
+                .filter(|(name, lib_id)| !hidden.contains(&(*lib_id, name.clone())))
+                .unzip();
+            file_names = names;
+            file_libraries = libs;
+        }
+    }
+
     if !any_library_resolved {
         error!("Bad photos request: {}", req.path);
         span.set_status(Status::error("Invalid path"));
@@ -871,16 +909,68 @@ pub async fn list_photos<TagD: TagDao, FS: FileSystemAccess>(
     span.set_attribute("total_count", total_count.to_string());
     span.set_status(Status::Ok);
 
+    let (photo_display_paths, photo_version_counts) =
+        version_arrays(&app_state, &response_files, &response_libraries);
     HttpResponse::Ok().json(PhotosResponse {
         photos: response_files,
         dirs,
         photo_libraries: response_libraries,
+        photo_display_paths,
+        photo_version_counts,
         total_count: pagination_metadata.0,
         has_more: pagination_metadata.1,
         next_offset: pagination_metadata.2,
     })
 }
 
+/// `(library_id, rel_path)` of the version rows hidden in the request's view
+/// (`library` = the resolved `?library=`; `None` = every library). A lookup
+/// failure hides nothing: better a duplicate tile than a missing photo.
+fn hidden_version_rows(
+    app_state: &AppState,
+    library: Option<i32>,
+) -> std::collections::HashSet<(i32, String)> {
+    let view: Vec<i32> = library.into_iter().collect();
+    app_state
+        .version_dao
+        .lock()
+        .expect("Unable to lock VersionDao")
+        .hidden_version_rows(&view)
+        .unwrap_or_else(|e| {
+            warn!("hidden_version_rows failed: {:?}", e);
+            Default::default()
+        })
+}
+
+/// The `/photos` parallel arrays for a page: `photo_display_paths` (the file
+/// to show when the global choice isn't the original) and
+/// `photo_version_counts`. Each is empty, so skipped on the wire, when no
+/// photo on the page has anything to report.
+fn version_arrays(
+    app_state: &AppState,
+    photos: &[String],
+    libraries: &[i32],
+) -> (Vec<Option<String>>, Vec<u32>) {
+    let summaries = crate::versions::load_summaries(&app_state.version_dao);
+    if summaries.is_empty() {
+        return (Vec::new(), Vec::new());
+    }
+    let mut display: Vec<Option<String>> = Vec::with_capacity(photos.len());
+    let mut counts: Vec<u32> = Vec::with_capacity(photos.len());
+    for (path, lib) in photos.iter().zip(libraries) {
+        let s = summaries.get(&(*lib, path.clone()));
+        display.push(s.and_then(|s| s.display.as_ref().map(|(_, p)| p.clone())));
+        counts.push(s.map_or(0, |s| s.version_count));
+    }
+    if display.iter().all(Option::is_none) {
+        display.clear();
+    }
+    if counts.iter().all(|c| *c == 0) {
+        counts.clear();
+    }
+    (display, counts)
+}
+
 fn sort(
     files: Vec<FileWithTagCount>,
     file_libraries: Vec<i32>,
@@ -1253,13 +1343,20 @@ pub async fn list_exif_summary(
         None,
         None,
         crate::date_view::InstantRange::new(req.date_from, req.date_to, display_zone, server_zone),
+        false,
     ) {
         Ok(rows) => {
+            let summaries = crate::versions::load_summaries(&app_state.version_dao);
             let photos: Vec<ExifSummary> = rows
                 .into_iter()
                 .map(|r| {
                     let name = library_names.get(&r.library_id).cloned();
-                    ExifSummary::from_row(r, name, display_zone, server_zone)
+                    let versions = crate::versions::VersionFields::from_summary(
+                        summaries.get(&(r.library_id, r.file_path.clone())),
+                    );
+                    let mut summary = ExifSummary::from_row(r, name, display_zone, server_zone);
+                    summary.versions = versions;
+                    summary
                 })
                 .collect();
 
@@ -1622,6 +1719,7 @@ mod tests {
             _: Option<&str>,
             _: Option<(f64, f64, f64, f64)>,
             _: Option<crate::date_view::InstantRange>,
+            _include_versions: bool,
         ) -> Result<Vec<crate::database::models::ImageExif>, DbError> {
             Ok(Vec::new())
         }
@@ -1748,6 +1846,7 @@ mod tests {
             _conv_hi: i64,
             _unconv_lo: i64,
             _unconv_hi: i64,
+            _view_library_ids: &[i32],
         ) -> Result<Vec<crate::database::models::ImageExif>, DbError> {
             Ok(Vec::new())
         }
@@ -1794,6 +1893,7 @@ mod tests {
             _library_ids: &[i32],
             _path_prefix: Option<&str>,
             _include_duplicates: bool,
+            _include_versions: bool,
         ) -> Result<Vec<(i32, String)>, DbError> {
             Ok(vec![])
         }

src/handlers/favorites.rs

@@ -44,6 +44,8 @@ pub async fn favorites(
                 photos: favorites,
                 dirs: Vec::new(),
                 photo_libraries: Vec::new(),
+                photo_display_paths: Vec::new(),
+                photo_version_counts: Vec::new(),
                 total_count: None,
                 has_more: None,
                 next_offset: None,

src/lib.rs

@@ -46,6 +46,7 @@ pub mod testhelpers;
 pub mod thumbnails;
 pub mod unified_search;
 pub mod utils;
+pub mod versions;
 pub mod video;
 pub mod watcher;
 

src/library_maintenance.rs

@@ -282,6 +282,37 @@ pub fn refresh_back_refs(conn: &mut SqliteConnection) -> usize {
     });
     total += updated;
 
+    // photo_versions — (library_id, rel_path) of the version file; same
+    // repoint as photo_insights.
+    let updated = sql_query(
+        "UPDATE photo_versions \
+         SET library_id = ( \
+                 SELECT ie.library_id FROM image_exif ie \
+                 WHERE ie.content_hash = photo_versions.version_hash \
+                 ORDER BY ie.id LIMIT 1 \
+             ), \
+             rel_path = ( \
+                 SELECT ie.rel_path FROM image_exif ie \
+                 WHERE ie.content_hash = photo_versions.version_hash \
+                 ORDER BY ie.id LIMIT 1 \
+             ) \
+         WHERE EXISTS ( \
+                 SELECT 1 FROM image_exif ie \
+                 WHERE ie.content_hash = photo_versions.version_hash \
+             ) \
+           AND NOT EXISTS ( \
+                 SELECT 1 FROM image_exif ie \
+                 WHERE ie.library_id = photo_versions.library_id \
+                   AND ie.rel_path  = photo_versions.rel_path \
+             )",
+    )
+    .execute(conn)
+    .unwrap_or_else(|e| {
+        warn!("back-ref refresh: photo_versions update failed: {:?}", e);
+        0
+    });
+    total += updated;
+
     if total > 0 {
         info!("back-ref refresh: updated {} hash-keyed row(s)", total);
     }
@@ -299,6 +330,8 @@ pub struct GcStats {
     pub deleted_face_detections: usize,
     pub deleted_tagged_photo: usize,
     pub deleted_photo_insights: usize,
+    /// Version links (and display choices) whose version file is gone.
+    pub deleted_photo_versions: usize,
     /// Hashes dropped from the pending set because they re-appeared
     /// in image_exif (e.g. user remounted a backup that was briefly
     /// missing).
@@ -311,12 +344,16 @@ impl GcStats {
             || self.deleted_face_detections > 0
             || self.deleted_tagged_photo > 0
             || self.deleted_photo_insights > 0
+            || self.deleted_photo_versions > 0
             || self.revived > 0
     }
 
     #[allow(dead_code)]
     pub fn total_deleted(&self) -> usize {
-        self.deleted_face_detections + self.deleted_tagged_photo + self.deleted_photo_insights
+        self.deleted_face_detections
+            + self.deleted_tagged_photo
+            + self.deleted_photo_insights
+            + self.deleted_photo_versions
     }
 }
 
@@ -433,10 +470,11 @@ pub fn run_orphan_gc(
 
     if !to_delete.is_empty() {
         match delete_hash_keyed_rows(conn, &to_delete) {
-            Ok((faces, tags, insights)) => {
+            Ok((faces, tags, insights, versions)) => {
                 stats.deleted_face_detections = faces;
                 stats.deleted_tagged_photo = tags;
                 stats.deleted_photo_insights = insights;
+                stats.deleted_photo_versions = versions;
                 // Drop deleted hashes from pending so we don't try to
                 // re-delete them next tick (they'll have already been
                 // removed from the orphan set).
@@ -452,12 +490,13 @@ pub fn run_orphan_gc(
 
     if stats.changed() {
         info!(
-            "orphan-gc: {} new orphan hash(es) marked, {} revived; deleted {} face_detections / {} tagged_photo / {} photo_insights row(s) (pending: {})",
+            "orphan-gc: {} new orphan hash(es) marked, {} revived; deleted {} face_detections / {} tagged_photo / {} photo_insights / {} photo_versions row(s) (pending: {})",
             stats.newly_marked,
             stats.revived,
             stats.deleted_face_detections,
             stats.deleted_tagged_photo,
             stats.deleted_photo_insights,
+            stats.deleted_photo_versions,
             state.pending.len(),
         );
     } else {
@@ -503,6 +542,12 @@ fn collect_orphan_hashes(conn: &mut SqliteConnection) -> QueryResult<HashSet<Str
              SELECT content_hash FROM tagged_photo    WHERE content_hash IS NOT NULL \
              UNION ALL \
              SELECT content_hash FROM photo_insights  WHERE content_hash IS NOT NULL \
+             UNION ALL \
+             SELECT version_hash FROM photo_versions \
+             UNION ALL \
+             SELECT display_hash FROM photo_version_display \
+             UNION ALL \
+             SELECT original_hash FROM photo_version_display \
          ) AS derived \
          WHERE content_hash NOT IN ( \
              SELECT content_hash FROM image_exif WHERE content_hash IS NOT NULL \
@@ -514,13 +559,19 @@ fn collect_orphan_hashes(conn: &mut SqliteConnection) -> QueryResult<HashSet<Str
 }
 
 /// Delete every hash-keyed row whose `content_hash` is in `hashes`.
-/// Returns `(faces, tagged_photo, photo_insights)`.
+/// Returns `(faces, tagged_photo, photo_insights, photo_versions)`.
+///
+/// For photo versions (#191): a link whose *version* file is gone, its
+/// reference rows, and any display choice naming the gone file (reads fall
+/// back to the original until then) or the gone original. A link whose
+/// *original* is gone is kept: the version shows as a standalone photo and
+/// the link revives if the original returns.
 fn delete_hash_keyed_rows(
     conn: &mut SqliteConnection,
     hashes: &[String],
-) -> QueryResult<(usize, usize, usize)> {
+) -> QueryResult<(usize, usize, usize, usize)> {
     if hashes.is_empty() {
-        return Ok((0, 0, 0));
+        return Ok((0, 0, 0, 0));
     }
 
     use crate::database::schema::{face_detections, photo_insights, tagged_photo};
@@ -535,7 +586,27 @@ fn delete_hash_keyed_rows(
         diesel::delete(photo_insights::table.filter(photo_insights::content_hash.eq_any(hashes)))
             .execute(conn)?;
 
-    Ok((faces, tags, insights))
+    use crate::database::schema::{
+        photo_version_display, photo_version_references, photo_versions,
+    };
+    let versions =
+        diesel::delete(photo_versions::table.filter(photo_versions::version_hash.eq_any(hashes)))
+            .execute(conn)?;
+    diesel::delete(
+        photo_version_references::table
+            .filter(photo_version_references::version_hash.eq_any(hashes)),
+    )
+    .execute(conn)?;
+    let displays = diesel::delete(
+        photo_version_display::table.filter(
+            photo_version_display::display_hash
+                .eq_any(hashes)
+                .or(photo_version_display::original_hash.eq_any(hashes)),
+        ),
+    )
+    .execute(conn)?;
+
+    Ok((faces, tags, insights, versions + displays))
 }
 
 #[cfg(test)]

src/main.rs

@@ -326,6 +326,8 @@ fn main() -> std::io::Result<()> {
                 .service(handlers::favorites::favorites)
                 .service(handlers::favorites::put_add_favorite)
                 .service(handlers::favorites::delete_favorite)
+                .service(versions::get_versions)
+                .service(versions::put_display)
                 .service(handlers::image::get_file_metadata)
                 .service(handlers::image::set_image_gps)
                 .service(handlers::image::set_image_date)

src/memories.rs

@@ -172,6 +172,10 @@ pub struct MemoryItem {
     /// the camera's clock, which can differ from the owner's day that
     /// `created` groups under. Render verbatim (ADR-0001).
     pub wall_clock: Option<String>,
+    /// `display_path` / `version_count` (#191); absent for a photo without
+    /// versions. The item's `path` is always the original's.
+    #[serde(flatten)]
+    pub versions: crate::versions::VersionFields,
 }
 
 #[derive(Debug, Serialize)]
@@ -652,6 +656,11 @@ pub fn gather_memory_items(
     // memory in the client's tz, used as the primary sort key.
     let mut memories_with_dates: Vec<(MemoryItem, NaiveDate)> = Vec::new();
 
+    // A photo version is left out of the memories while its original is
+    // reachable in this view (the selected library, else every library).
+    let view_library_ids: Vec<i32> = library.map(|l| vec![l.id]).unwrap_or_default();
+    let version_summaries = crate::versions::load_summaries(&app_state.version_dao);
+
     for lib in &libraries_to_scan {
         let base = Path::new(&lib.root_path);
         let effective = lib.effective_excluded_dirs(&app_state.excluded_dirs);
@@ -659,7 +668,14 @@ pub fn gather_memory_items(
 
         let rows = match exif_dao.lock() {
             Ok(mut dao) => {
-                match dao.get_memories_in_window(lib.id, conv_lo, conv_hi, unconv_lo, unconv_hi) {
+                match dao.get_memories_in_window(
+                    lib.id,
+                    conv_lo,
+                    conv_hi,
+                    unconv_lo,
+                    unconv_hi,
+                    &view_library_ids,
+                ) {
                     Ok(rows) => rows,
                     Err(e) => {
                         warn!(
@@ -697,6 +713,9 @@ pub fn gather_memory_items(
 
             let created = crate::date_view::row_true_instant(&row, display_zone, server_zone);
             let wall_clock = crate::date_view::row_wall_clock(&row, display_zone, server_zone);
+            let versions = crate::versions::VersionFields::from_summary(
+                version_summaries.get(&(lib.id, row.file_path.clone())),
+            );
             memories_with_dates.push((
                 MemoryItem {
                     path: row.file_path,
@@ -707,6 +726,7 @@ pub fn gather_memory_items(
                     camera_offset_minutes: row.camera_offset_minutes,
                     date_taken_tz_source: row.date_taken_tz_source,
                     wall_clock,
+                    versions,
                 },
                 display_day,
             ));
@@ -1228,6 +1248,7 @@ mod tests {
             camera_offset_minutes: None,
             date_taken_tz_source: None,
             wall_clock: None,
+            versions: Default::default(),
         }
     }
 
@@ -1746,6 +1767,7 @@ mod tests {
                 camera_offset_minutes: None,
                 date_taken_tz_source: None,
                 wall_clock: None,
+                versions: Default::default(),
             };
             let v: serde_json::Value =
                 serde_json::to_value(&MemoriesResponse { items: vec![item] }).unwrap();

src/reels/selector.rs

@@ -433,6 +433,7 @@ mod tests {
                 camera_offset_minutes: None,
                 date_taken_tz_source: None,
                 wall_clock: None,
+                versions: Default::default(),
             },
             memories::MemoryItem {
                 path: "b".into(),
@@ -443,6 +444,7 @@ mod tests {
                 camera_offset_minutes: None,
                 date_taken_tz_source: None,
                 wall_clock: None,
+                versions: Default::default(),
             },
             memories::MemoryItem {
                 path: "c".into(),
@@ -453,6 +455,7 @@ mod tests {
                 camera_offset_minutes: None,
                 date_taken_tz_source: None,
                 wall_clock: None,
+                versions: Default::default(),
             },
         ];
         assert_eq!(
@@ -472,6 +475,7 @@ mod tests {
             camera_offset_minutes: None,
             date_taken_tz_source: None,
             wall_clock: None,
+            versions: Default::default(),
         }
     }
     fn item_at(ts: i64, name: &str) -> memories::MemoryItem {

src/state.rs

@@ -13,7 +13,7 @@ use crate::database::{
     LocationHistoryDao, PrecomputedReelDao, SearchHistoryDao, SqliteCalendarEventDao,
     SqliteDailySummaryDao, SqliteExifDao, SqliteInsightDao, SqliteInsightGenerationJobDao,
     SqliteKnowledgeDao, SqliteLocationHistoryDao, SqlitePrecomputedReelDao, SqliteSearchHistoryDao,
-    SqliteUserAiPrefsDao, UserAiPrefsDao, connect,
+    SqliteUserAiPrefsDao, SqliteVersionDao, UserAiPrefsDao, VersionDao, connect,
 };
 use crate::database::{PreviewDao, SqlitePreviewDao};
 use crate::faces;
@@ -98,6 +98,10 @@ pub struct AppState {
     /// client; read by the nightly pre-generation scheduler.
     #[allow(dead_code)]
     pub user_ai_prefs_dao: Arc<Mutex<Box<dyn UserAiPrefsDao>>>,
+    /// Photo-version links and the global display choice (#191). Read by
+    /// the listings that annotate a photo with its versions and by the
+    /// `/image/versions` endpoints.
+    pub version_dao: Arc<Mutex<Box<dyn VersionDao>>>,
 }
 
 impl AppState {
@@ -142,6 +146,7 @@ impl AppState {
         insight_job_handles: Arc<Mutex<HashMap<i32, tokio::task::AbortHandle>>>,
         precomputed_reel_dao: Arc<Mutex<Box<dyn PrecomputedReelDao>>>,
         user_ai_prefs_dao: Arc<Mutex<Box<dyn UserAiPrefsDao>>>,
+        version_dao: Arc<Mutex<Box<dyn VersionDao>>>,
     ) -> Self {
         assert!(
             !libraries_vec.is_empty(),
@@ -202,6 +207,7 @@ impl AppState {
             insight_job_handles,
             precomputed_reel_dao,
             user_ai_prefs_dao,
+            version_dao,
         }
     }
 
@@ -290,6 +296,9 @@ impl Default for AppState {
         let user_ai_prefs_dao: Arc<Mutex<Box<dyn UserAiPrefsDao>>> =
             Arc::new(Mutex::new(Box::new(SqliteUserAiPrefsDao::new())));
 
+        let version_dao: Arc<Mutex<Box<dyn VersionDao>>> =
+            Arc::new(Mutex::new(Box::new(SqliteVersionDao::new())));
+
         // Load base path and ensure the primary library row reflects it.
         let base_path = env::var("BASE_PATH").expect("BASE_PATH was not set in the env");
         let mut seed_conn = connect();
@@ -374,6 +383,7 @@ impl Default for AppState {
             insight_job_handles,
             precomputed_reel_dao,
             user_ai_prefs_dao,
+            version_dao,
         )
     }
 }
@@ -552,6 +562,7 @@ impl AppState {
             Arc::new(Mutex::new(HashMap::new())), // placeholder for test
             Arc::new(Mutex::new(Box::new(SqlitePrecomputedReelDao::new()))), // placeholder for test
             Arc::new(Mutex::new(Box::new(SqliteUserAiPrefsDao::new()))), // placeholder for test
+            Arc::new(Mutex::new(Box::new(SqliteVersionDao::new()))), // placeholder for test
         )
     }
 }

src/tags.rs

@@ -604,6 +604,16 @@ pub trait TagDao: Send + Sync {
     fn delete_tag(&mut self, id: i32) -> anyhow::Result<bool>;
 }
 
+/// The path a tag read/write on `path` applies to: the original's when
+/// `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)
+        .ok()
+        .flatten()
+        .unwrap_or_else(|| path.to_string())
+}
+
 pub struct SqliteTagDao {
     connection: Arc<Mutex<SqliteConnection>>,
 }
@@ -670,6 +680,8 @@ impl TagDao for SqliteTagDao {
             span.set_attribute("path", path.to_string());
 
             debug!("Getting Tags for path: {:?}", path);
+            // A photo version carries its original's tags.
+            let path = &canonical_path(conn.deref_mut(), path);
             tags::table
                 .left_join(tagged_photo::table)
                 .filter(tagged_photo::rel_path.eq(&path))
@@ -861,6 +873,7 @@ impl TagDao for SqliteTagDao {
             span.set_attribute("tag_name", tag_name.to_string());
             span.set_attribute("path", path.to_string());
 
+            let path = &canonical_path(conn.deref_mut(), path);
             tags::table
                 .filter(tags::name.eq(tag_name))
                 .get_result::<Tag>(conn.deref_mut())
@@ -893,6 +906,10 @@ impl TagDao for SqliteTagDao {
             span.set_attribute("path", path.to_string());
             span.set_attribute("tag_id", tag_id.to_string());
 
+            // Tagging a photo version tags its original (photo versions,
+            // #191): the tag binds to the original's bytes and path.
+            let path = &canonical_path(conn.deref_mut(), path);
+
             // Eagerly populate content_hash so this tag follows the bytes,
             // not the path (see CLAUDE.md "Multi-library data model").
             // None is fine — the reconciliation pass will backfill once

src/thumbnails.rs

@@ -343,9 +343,36 @@ pub fn create_thumbnails(libs: &[libraries::Library], excluded_dirs: &[String])
 }
 
 pub fn update_media_counts(media_dir: &Path, excluded_dirs: &[String]) {
+    update_media_counts_excluding(media_dir, excluded_dirs, &std::collections::HashSet::new());
+}
+
+/// As [`update_media_counts`], leaving out the files in `hidden` (rel paths
+/// under `media_dir`): photo versions (#191) are the same photo as their
+/// original, so the gauges count it once.
+pub fn update_media_counts_excluding(
+    media_dir: &Path,
+    excluded_dirs: &[String],
+    hidden: &std::collections::HashSet<String>,
+) {
+    let (image_count, video_count) = count_media(media_dir, excluded_dirs, hidden);
+    IMAGE_GAUGE.set(image_count);
+    VIDEO_GAUGE.set(video_count);
+}
+
+fn count_media(
+    media_dir: &Path,
+    excluded_dirs: &[String],
+    hidden: &std::collections::HashSet<String>,
+) -> (i64, i64) {
     let mut image_count = 0;
     let mut video_count = 0;
     for entry in crate::file_scan::walk_library_files(media_dir, excluded_dirs) {
+        if !hidden.is_empty()
+            && let Ok(rel) = entry.path().strip_prefix(media_dir)
+            && hidden.contains(&rel.to_string_lossy().replace('\\', "/"))
+        {
+            continue;
+        }
         if is_image(&entry) {
             image_count += 1;
         } else if is_video(&entry) {
@@ -353,8 +380,7 @@ pub fn update_media_counts(media_dir: &Path, excluded_dirs: &[String]) {
         }
     }
 
-    IMAGE_GAUGE.set(image_count);
+    (image_count, video_count)
-    VIDEO_GAUGE.set(video_count);
 }
 
 pub fn is_image(entry: &DirEntry) -> bool {
@@ -395,6 +421,21 @@ mod tests {
         assert_eq!(s, PathBuf::from("/thumbs/notes.unsupported"));
     }
 
+    #[test]
+    fn update_media_counts_leaves_out_hidden_versions() {
+        let tmp = TempDir::new().expect("tempdir");
+        fs::create_dir(tmp.path().join("aiedits")).unwrap();
+        fs::write(tmp.path().join("a.jpg"), b"").unwrap();
+        fs::write(tmp.path().join("aiedits/a.jpg.restored.png"), b"").unwrap();
+        let hidden = std::collections::HashSet::from(["aiedits/a.jpg.restored.png".to_string()]);
+        assert_eq!(count_media(tmp.path(), &[], &hidden), (1, 0));
+        assert_eq!(
+            count_media(tmp.path(), &[], &Default::default()),
+            (2, 0),
+            "nothing hidden: both files count"
+        );
+    }
+
     /// Smoke-test update_media_counts: build a tempdir with two images
     /// and one video, run the walker, and assert the gauges line up.
     /// Exercises the is_image / is_video classifier on real DirEntry

src/unified_search.rs

@@ -459,12 +459,21 @@ pub async fn unified_search<TagD: TagDao>(
                     display_zone,
                     server_zone,
                 ),
+                // Versions are dropped below, against the request's own
+                // library scope (this query is unscoped).
+                true,
             )
             .unwrap_or_else(|e| {
                 log::warn!("unified_search: query_by_exif failed: {e:?}");
                 Vec::new()
             })
         };
+        let hidden_versions = state
+            .version_dao
+            .lock()
+            .expect("version dao")
+            .hidden_version_rows(&library_ids)
+            .unwrap_or_default();
 
         candidate = rows
             .into_iter()
@@ -473,6 +482,10 @@ pub async fn unified_search<TagD: TagDao>(
                 if !library_ids.is_empty() && !library_ids.contains(&row.library_id) {
                     return false;
                 }
+                // A version stays behind its original.
+                if hidden_versions.contains(&(row.library_id, row.file_path.clone())) {
+                    return false;
+                }
                 // Precise GPS distance (the EXIF query only did a coarse box).
                 if let Some((lat, lon, radius_km)) = gps {
                     match (row.gps_latitude, row.gps_longitude) {
@@ -569,7 +582,7 @@ pub async fn unified_search<TagD: TagDao>(
             );
             let total_matching = hits.len();
             let page = paginate(&hits, offset, limit);
-            let results = resolve_hits(&exif_dao, &page);
+            let results = resolve_hits(&exif_dao, &state.version_dao, &page);
             HttpResponse::Ok().json(UnifiedResponse {
                 query: nl,
                 interpreted: interpreted(&echo, resolved_place),
@@ -593,6 +606,7 @@ pub async fn unified_search<TagD: TagDao>(
             let results: Vec<SearchHit> = if offset >= total_matching {
                 Vec::new()
             } else {
+                let summaries = crate::versions::load_summaries(&state.version_dao);
                 candidate[offset..end]
                     .iter()
                     .map(|r| SearchHit {
@@ -600,6 +614,9 @@ pub async fn unified_search<TagD: TagDao>(
                         rel_path: r.file_path.clone(),
                         content_hash: r.content_hash.clone().unwrap_or_default(),
                         score: 0.0,
+                        versions: crate::versions::VersionFields::from_summary(
+                            summaries.get(&(r.library_id, r.file_path.clone())),
+                        ),
                     })
                     .collect()
             };

src/versions/handlers.rs

@@ -0,0 +1,252 @@
+//! `GET /image/versions` and `PUT /image/versions/display`.
+//!
+//! Both take the path of the original *or* of any version, and answer about
+//! the whole set. The display choice is global (not per user).
+
+use std::sync::Mutex;
+
+use actix_web::{HttpResponse, Responder, get, put, web};
+use log::warn;
+use serde::{Deserialize, Serialize};
+
+use crate::data::Claims;
+use crate::database::ExifDao;
+use crate::database::version_dao::VersionDao;
+use crate::libraries;
+use crate::state::AppState;
+use crate::utils::normalize_path;
+
+#[derive(Debug, Deserialize)]
+pub struct VersionsQuery {
+    pub path: String,
+    pub library: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct SetDisplayBody {
+    pub path: String,
+    /// Content hash of the version to display; `null` / absent = the original.
+    #[serde(default)]
+    pub display_hash: Option<String>,
+    pub library: Option<String>,
+}
+
+#[derive(Debug, Serialize, PartialEq)]
+pub struct FileRef {
+    pub path: String,
+    pub library_id: i32,
+    pub hash: String,
+}
+
+#[derive(Debug, Serialize, PartialEq)]
+pub struct ReferenceEntry {
+    pub hash: String,
+    /// Where the reference photo lives, if it has been ingested.
+    pub path: Option<String>,
+    pub library_id: Option<i32>,
+}
+
+#[derive(Debug, Serialize, PartialEq)]
+pub struct VersionEntry {
+    pub path: String,
+    pub library_id: i32,
+    pub hash: String,
+    pub kind: String,
+    pub source: String,
+    pub created_at: i64,
+    pub references: Vec<ReferenceEntry>,
+    pub width: Option<i32>,
+    pub height: Option<i32>,
+}
+
+#[derive(Debug, Serialize, PartialEq)]
+pub struct VersionsResponse {
+    /// `null` when the original's file is gone (its versions then show up as
+    /// standalone photos).
+    pub original: Option<FileRef>,
+    pub versions: Vec<VersionEntry>,
+    /// The globally displayed version; `null` = the original.
+    pub display_hash: Option<String>,
+}
+
+/// The photo a request path belongs to.
+struct Resolved {
+    original_hash: String,
+    /// Library of the row the path named; preferred when picking files.
+    library_id: i32,
+}
+
+fn resolve(
+    app_state: &AppState,
+    exif_dao: &Mutex<Box<dyn ExifDao>>,
+    path: &str,
+    library: Option<&str>,
+) -> Result<Resolved, HttpResponse> {
+    let library_id = match libraries::resolve_library_param_state(app_state, library) {
+        Ok(lib) => lib.map(|l| l.id),
+        Err(msg) => return Err(HttpResponse::BadRequest().body(msg)),
+    };
+    let path = normalize_path(path);
+    let mut rows = exif_dao
+        .lock()
+        .expect("Unable to lock ExifDao")
+        .get_exif_batch(library_id, std::slice::from_ref(&path))
+        .map_err(|_| HttpResponse::InternalServerError().finish())?;
+    // Without `library`, a path can name rows in several libraries. Take the
+    // earliest hashed one; it then resolves to its original below, so a
+    // version in one library and its original in another agree.
+    rows.sort_by_key(|r| r.id);
+    let Some((row, hash)) = rows
+        .iter()
+        .find_map(|r| r.content_hash.clone().map(|h| (r, h)))
+    else {
+        return Err(HttpResponse::NotFound().body("no indexed file at that path"));
+    };
+    let link = app_state
+        .version_dao
+        .lock()
+        .expect("Unable to lock VersionDao")
+        .get_link(&hash)
+        .map_err(|_| HttpResponse::InternalServerError().finish())?;
+    Ok(Resolved {
+        original_hash: link.map(|l| l.original_hash).unwrap_or(hash),
+        library_id: row.library_id,
+    })
+}
+
+/// Everything about the set around `original_hash`. Reads fall back to the
+/// original when the stored display choice has no live row.
+pub fn build_response(
+    dao: &mut dyn VersionDao,
+    original_hash: &str,
+    prefer_library: i32,
+) -> Result<VersionsResponse, crate::database::DbError> {
+    let original = dao
+        .files_for_hashes(&[original_hash.to_string()], Some(prefer_library))?
+        .remove(original_hash)
+        .map(|(library_id, path)| FileRef {
+            path,
+            library_id,
+            hash: original_hash.to_string(),
+        });
+
+    let files = dao.list_versions(original_hash)?;
+    let mut versions = Vec::with_capacity(files.len());
+    for f in files {
+        let reference_hashes = dao.list_references(&f.version_hash)?;
+        let located = dao.files_for_hashes(&reference_hashes, Some(f.library_id))?;
+        let references = reference_hashes
+            .into_iter()
+            .map(|hash| {
+                let found = located.get(&hash);
+                ReferenceEntry {
+                    path: found.map(|(_, p)| p.clone()),
+                    library_id: found.map(|(l, _)| *l),
+                    hash,
+                }
+            })
+            .collect();
+        versions.push(VersionEntry {
+            path: f.rel_path,
+            library_id: f.library_id,
+            hash: f.version_hash,
+            kind: f.kind,
+            source: f.source,
+            created_at: f.created_at,
+            references,
+            width: f.width,
+            height: f.height,
+        });
+    }
+
+    let display_hash = dao
+        .get_display(original_hash)?
+        .filter(|d| versions.iter().any(|v| &v.hash == d));
+    Ok(VersionsResponse {
+        original,
+        versions,
+        display_hash,
+    })
+}
+
+#[get("/image/versions")]
+#[tracing::instrument(name = "get_versions", skip_all)]
+pub async fn get_versions(
+    _: Claims,
+    q: web::Query<VersionsQuery>,
+    app_state: web::Data<AppState>,
+    exif_dao: web::Data<Mutex<Box<dyn ExifDao>>>,
+) -> impl Responder {
+    let resolved = match resolve(&app_state, &exif_dao, &q.path, q.library.as_deref()) {
+        Ok(r) => r,
+        Err(resp) => return resp,
+    };
+    let mut dao = app_state
+        .version_dao
+        .lock()
+        .expect("Unable to lock VersionDao");
+    match build_response(dao.as_mut(), &resolved.original_hash, resolved.library_id) {
+        Ok(body) => HttpResponse::Ok().json(body),
+        Err(e) => {
+            warn!("get_versions failed: {:?}", e);
+            HttpResponse::InternalServerError().finish()
+        }
+    }
+}
+
+#[put("/image/versions/display")]
+#[tracing::instrument(name = "put_version_display", skip_all)]
+pub async fn put_display(
+    _: Claims,
+    // The client also sends `library` as a query parameter, like its other
+    // path-keyed calls; the body field wins when both are given.
+    query: web::Query<LibraryQuery>,
+    body: web::Json<SetDisplayBody>,
+    app_state: web::Data<AppState>,
+    exif_dao: web::Data<Mutex<Box<dyn ExifDao>>>,
+) -> impl Responder {
+    let library = body.library.as_deref().or(query.library.as_deref());
+    let resolved = match resolve(&app_state, &exif_dao, &body.path, library) {
+        Ok(r) => r,
+        Err(resp) => return resp,
+    };
+    let mut dao = app_state
+        .version_dao
+        .lock()
+        .expect("Unable to lock VersionDao");
+
+    // Choosing the original is the same as clearing the choice.
+    let wanted = body
+        .display_hash
+        .as_deref()
+        .filter(|h| *h != resolved.original_hash);
+    if let Some(hash) = wanted {
+        let current =
+            match build_response(dao.as_mut(), &resolved.original_hash, resolved.library_id) {
+                Ok(r) => r,
+                Err(e) => {
+                    warn!("put_display lookup failed: {:?}", e);
+                    return HttpResponse::InternalServerError().finish();
+                }
+            };
+        if !current.versions.iter().any(|v| v.hash == hash) {
+            return HttpResponse::BadRequest().body("display_hash is not a version of this photo");
+        }
+    }
+    if let Err(e) = dao.set_display(&resolved.original_hash, wanted) {
+        warn!("put_display failed: {:?}", e);
+        return HttpResponse::InternalServerError().finish();
+    }
+    match build_response(dao.as_mut(), &resolved.original_hash, resolved.library_id) {
+        Ok(body) => HttpResponse::Ok().json(body),
+        Err(e) => {
+            warn!("put_display response failed: {:?}", e);
+            HttpResponse::InternalServerError().finish()
+        }
+    }
+}
+
+#[derive(Debug, Deserialize)]
+pub struct LibraryQuery {
+    pub library: Option<String>,
+}

src/versions/mod.rs

@@ -0,0 +1,384 @@
+//! Photo versions (#191): AI restorations / upscales / edits that sit near
+//! their original and are shown as one photo.
+//!
+//! ImageApi only *detects* versions; an external process (the ComfyUI node
+//! in `comfyui-imageapi-versions`) writes them into the library, so the
+//! read-only rule for library roots holds. This module owns
+//!
+//! - the link rules at ingest ([`link_new_files`], [`sweep_name_links`]):
+//!   the PNG's XMP `DerivedFrom` hash first, then the
+//!   `<name>.<kind>[.<n>].<ext>` file-name convention;
+//! - the HTTP surface (`GET /image/versions`, `PUT /image/versions/display`).
+//!
+//! The data model lives in [`crate::database::version_dao`]. The link is the
+//! original's `content_hash`, never a path: ComfyUI writes through a UNC
+//! share and ImageApi reads `/media/...`, and files get moved.
+
+pub mod handlers;
+pub mod xmp;
+
+use std::path::{Path, PathBuf};
+use std::sync::{Arc, LazyLock, Mutex};
+
+use log::{debug, info, warn};
+use regex::Regex;
+
+use crate::database::version_dao::{
+    KINDS, LinkOutcome, NewVersionLink, SOURCE_NAME, SOURCE_XMP, VersionDao, is_known_kind,
+};
+use crate::database::{ExifDao, version_dao};
+use crate::libraries::Library;
+
+pub use handlers::{get_versions, put_display};
+
+/// Directory the producer saves versions into, next to the original.
+pub const VERSIONS_SUBFOLDER: &str = "aiedits";
+
+/// Cap on name-convention candidates examined per library per tick.
+const NAME_SWEEP_LIMIT: i64 = 500;
+
+static VERSION_NAME: LazyLock<Regex> = LazyLock::new(|| {
+    let kinds = KINDS.join("|");
+    Regex::new(&format!(
+        r"(?i)^(?P<name>.+)\.(?P<kind>{kinds})(?:\.(?P<n>\d+))?\.(?P<ext>[a-z0-9]+)$"
+    ))
+    .unwrap()
+});
+
+/// A file name that follows the producer's `<original name>.<kind>[.<n>].<ext>`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct VersionName {
+    /// The original's full file name, extension included.
+    pub original_name: String,
+    pub kind: &'static str,
+}
+
+/// Parse a bare file name (no directory). The original's extension stays in
+/// `original_name`, so `IMG_1.NEF` and `IMG_1.JPG` are told apart.
+pub fn parse_version_name(file_name: &str) -> Option<VersionName> {
+    let caps = VERSION_NAME.captures(file_name)?;
+    let kind_text = caps["kind"].to_ascii_lowercase();
+    let kind = KINDS.iter().find(|k| **k == kind_text)?;
+    Some(VersionName {
+        original_name: caps["name"].to_string(),
+        kind,
+    })
+}
+
+/// Rel paths where the original of a name-convention file may sit, best
+/// first: the same directory, then the parent of an `aiedits/` directory.
+pub fn original_candidates(rel_path: &str) -> Vec<String> {
+    let (dir, file_name) = match rel_path.rsplit_once('/') {
+        Some((d, f)) => (d, f),
+        None => ("", rel_path),
+    };
+    let Some(parsed) = parse_version_name(file_name) else {
+        return Vec::new();
+    };
+    let join = |d: &str| {
+        if d.is_empty() {
+            parsed.original_name.clone()
+        } else {
+            format!("{d}/{}", parsed.original_name)
+        }
+    };
+    let mut out = vec![join(dir)];
+    let (parent, leaf) = match dir.rsplit_once('/') {
+        Some((p, l)) => (p, l),
+        None => ("", dir),
+    };
+    if leaf.eq_ignore_ascii_case(VERSIONS_SUBFOLDER) {
+        out.push(join(parent));
+    }
+    out
+}
+
+/// Resolve the name-convention link of `rel_path` (in `library`) against the
+/// `image_exif` rows of that library. `None` when the name doesn't follow the
+/// convention or no candidate original has a hash yet (the sweep retries).
+pub fn name_link(
+    library: &Library,
+    rel_path: &str,
+    version_hash: &str,
+    exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
+) -> Option<NewVersionLink> {
+    let file_name = rel_path.rsplit('/').next().unwrap_or(rel_path);
+    let kind = parse_version_name(file_name)?.kind;
+    let candidates = original_candidates(rel_path);
+    let rows = exif_dao
+        .lock()
+        .expect("Unable to lock ExifDao")
+        .get_exif_batch(Some(library.id), &candidates)
+        .ok()?;
+    // First candidate (in preference order) that has a hash of its own.
+    let original_hash = candidates.iter().find_map(|c| {
+        rows.iter()
+            .find(|r| &r.file_path == c)
+            .and_then(|r| r.content_hash.clone())
+            .filter(|h| h != version_hash)
+    })?;
+    Some(NewVersionLink {
+        version_hash: version_hash.to_string(),
+        original_hash,
+        kind: kind.to_string(),
+        source: SOURCE_NAME.to_string(),
+        library_id: library.id,
+        rel_path: rel_path.to_string(),
+        references: Vec::new(),
+    })
+}
+
+/// Resolve the XMP link of a PNG on disk. `None` when the file isn't a PNG,
+/// can't be read, or carries no `DerivedFrom` blake3 id.
+pub fn xmp_link(
+    library: &Library,
+    rel_path: &str,
+    abs_path: &Path,
+    version_hash: &str,
+) -> Option<NewVersionLink> {
+    let is_png = abs_path
+        .extension()
+        .is_some_and(|e| e.eq_ignore_ascii_case("png"));
+    if !is_png {
+        return None;
+    }
+    let parsed = match xmp::read_version_xmp(abs_path) {
+        Ok(p) => p?,
+        Err(e) => {
+            debug!(
+                "versions: could not read XMP of {}: {}",
+                abs_path.display(),
+                e
+            );
+            return None;
+        }
+    };
+    let file_name = rel_path.rsplit('/').next().unwrap_or(rel_path);
+    let kind = parsed
+        .kind
+        .filter(|k| is_known_kind(k))
+        .or_else(|| parse_version_name(file_name).map(|n| n.kind.to_string()))
+        .unwrap_or_else(|| "edit".to_string());
+    Some(NewVersionLink {
+        version_hash: version_hash.to_string(),
+        original_hash: parsed.original_hash,
+        kind,
+        source: SOURCE_XMP.to_string(),
+        library_id: library.id,
+        rel_path: rel_path.to_string(),
+        references: parsed.references,
+    })
+}
+
+/// Record a link, logging the outcome. Returns true if a new link was made.
+pub fn record_link(version_dao: &Arc<Mutex<Box<dyn VersionDao>>>, link: &NewVersionLink) -> bool {
+    let outcome = version_dao
+        .lock()
+        .expect("Unable to lock VersionDao")
+        .link_version(link);
+    match outcome {
+        Ok(LinkOutcome::Linked) => {
+            info!(
+                "versions: linked {} -> original {}.. ({}, {})",
+                link.rel_path,
+                &link.original_hash[..12.min(link.original_hash.len())],
+                link.kind,
+                link.source
+            );
+            true
+        }
+        Ok(LinkOutcome::AlreadyLinked) => false,
+        Ok(LinkOutcome::Rejected(why)) => {
+            warn!("versions: refused link for {}: {}", link.rel_path, why);
+            false
+        }
+        Err(e) => {
+            warn!("versions: link failed for {}: {:?}", link.rel_path, e);
+            false
+        }
+    }
+}
+
+/// Link freshly ingested files by their embedded XMP. `files` are the
+/// `(absolute, rel_path)` pairs the watcher just registered; rows without a
+/// hash yet are skipped (an XMP-less file is the name sweep's job anyway).
+/// Returns the number of new links.
+pub fn link_new_files(
+    library: &Library,
+    files: &[(PathBuf, String)],
+    exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
+    version_dao: &Arc<Mutex<Box<dyn VersionDao>>>,
+) -> usize {
+    let pngs: Vec<&(PathBuf, String)> = files
+        .iter()
+        .filter(|(p, _)| p.extension().is_some_and(|e| e.eq_ignore_ascii_case("png")))
+        .collect();
+    if pngs.is_empty() {
+        return 0;
+    }
+    let rels: Vec<String> = pngs.iter().map(|(_, r)| r.clone()).collect();
+    let rows = exif_dao
+        .lock()
+        .expect("Unable to lock ExifDao")
+        .get_exif_batch(Some(library.id), &rels)
+        .unwrap_or_default();
+
+    let mut linked = 0;
+    for (abs, rel) in pngs {
+        let Some(hash) = rows
+            .iter()
+            .find(|r| &r.file_path == rel)
+            .and_then(|r| r.content_hash.clone())
+        else {
+            continue;
+        };
+        if let Some(link) = xmp_link(library, rel, abs, &hash)
+            && record_link(version_dao, &link)
+        {
+            linked += 1;
+        }
+    }
+    linked
+}
+
+/// Link name-convention files whose original is (now) known. Idempotent and
+/// DB-only, so it also serves as the retry for "original not hashed yet" and
+/// "original ingested after the version". Returns the number of new links.
+pub fn sweep_name_links(
+    library: &Library,
+    exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
+    version_dao: &Arc<Mutex<Box<dyn VersionDao>>>,
+) -> usize {
+    let candidates = match version_dao
+        .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;
+        }
+    };
+    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)
+        {
+            linked += 1;
+        }
+    }
+    linked
+}
+
+/// Sweep names and invalidate what a new link makes stale. The watcher calls
+/// this once per library per tick, after ingest.
+pub fn link_pending(
+    library: &Library,
+    exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
+    version_dao: &Arc<Mutex<Box<dyn VersionDao>>>,
+) {
+    if sweep_name_links(library, exif_dao, version_dao) > 0 {
+        crate::duplicates::invalidate_caches();
+    }
+}
+
+/// The version facts a per-photo record (a memory, a search hit) carries.
+/// Flattened into the record; absent for a photo without versions, so
+/// responses stay byte-identical for libraries that have none.
+///
+/// `display_path` is `null` when the original is displayed, a rel_path when a
+/// version is. `/photos` carries the same facts as the parallel arrays
+/// `photo_display_paths` / `photo_version_counts` instead.
+#[derive(Debug, Clone, Default, serde::Serialize)]
+pub struct VersionFields {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub display_path: Option<Option<String>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub version_count: Option<u32>,
+}
+
+impl VersionFields {
+    pub fn from_summary(summary: Option<&version_dao::PhotoVersionSummary>) -> Self {
+        match summary {
+            Some(s) if s.version_count > 0 => VersionFields {
+                display_path: Some(s.display.as_ref().map(|(_, p)| p.clone())),
+                version_count: Some(s.version_count),
+            },
+            _ => VersionFields::default(),
+        }
+    }
+}
+
+/// Load the per-photo version summaries once for a response. Empty (and
+/// cheap) when nothing has versions; a DB failure degrades to "no versions"
+/// rather than failing the listing.
+pub fn load_summaries(
+    version_dao: &Mutex<Box<dyn VersionDao>>,
+) -> std::collections::HashMap<(i32, String), version_dao::PhotoVersionSummary> {
+    version_dao
+        .lock()
+        .expect("Unable to lock VersionDao")
+        .photo_summaries()
+        .unwrap_or_else(|e| {
+            warn!("versions: could not load summaries: {:?}", e);
+            Default::default()
+        })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parses_the_producer_naming_convention() {
+        let n = parse_version_name("scan_0042.jpg.restored.png").unwrap();
+        assert_eq!(n.original_name, "scan_0042.jpg");
+        assert_eq!(n.kind, "restored");
+        let n = parse_version_name("scan_0042.jpg.restored.2.png").unwrap();
+        assert_eq!(n.original_name, "scan_0042.jpg");
+        let n = parse_version_name("IMG_1.NEF.upscale.png").unwrap();
+        assert_eq!(n.original_name, "IMG_1.NEF");
+        assert_eq!(n.kind, "upscale");
+        assert_eq!(
+            parse_version_name("a.b.c.EDIT.12.PNG")
+                .unwrap()
+                .original_name,
+            "a.b.c"
+        );
+    }
+
+    #[test]
+    fn rejects_names_outside_the_convention() {
+        assert!(parse_version_name("holiday.png").is_none());
+        assert!(parse_version_name("restored.png").is_none());
+        assert!(parse_version_name("x.jpg.blurred.png").is_none());
+        assert!(parse_version_name("x.jpg.restored").is_none());
+        assert!(parse_version_name("x.restored.png.bak").is_none());
+        assert!(parse_version_name(".restored.png").is_none());
+    }
+
+    #[test]
+    fn candidates_prefer_the_same_dir_then_the_parent_of_aiedits() {
+        assert_eq!(
+            original_candidates("2004/aiedits/scan.jpg.restored.png"),
+            vec!["2004/aiedits/scan.jpg", "2004/scan.jpg"]
+        );
+        assert_eq!(
+            original_candidates("2004/scan.jpg.restored.png"),
+            vec!["2004/scan.jpg"]
+        );
+        assert_eq!(
+            original_candidates("aiedits/scan.jpg.edit.3.png"),
+            vec!["aiedits/scan.jpg", "scan.jpg"]
+        );
+        assert_eq!(
+            original_candidates("scan.jpg.upscale.png"),
+            vec!["scan.jpg"]
+        );
+        assert!(original_candidates("2004/plain.png").is_empty());
+    }
+}

src/versions/xmp.rs

@@ -0,0 +1,312 @@
+//! Reading the version link out of a PNG: the `XML:com.adobe.xmp` iTXt
+//! chunk the ComfyUI producer writes (`comfyui-imageapi-versions`), and the
+//! ComfyUI `prompt` tEXt chunk the backfill uses for files without XMP.
+//!
+//! The chunk stream is walked by hand: no pixel decode, and the `png`
+//! crate's text API would still inflate every IDAT's neighbours for no gain.
+//! IDAT payloads are seeked over, so a multi-megabyte restoration costs a
+//! handful of small reads on a network share.
+
+use std::fs::File;
+use std::io::{self, BufReader, Read, Seek, SeekFrom};
+use std::path::Path;
+use std::sync::LazyLock;
+
+use regex::Regex;
+
+/// iTXt keyword exiftool, Lightroom and the producer put XMP under.
+pub const XMP_KEYWORD: &str = "XML:com.adobe.xmp";
+
+/// Text chunks bigger than this are skipped rather than read: ComfyUI's
+/// `prompt` / `workflow` JSON is tens of KiB, an XMP packet around 1 KiB.
+const MAX_TEXT_CHUNK: u32 = 4 * 1024 * 1024;
+
+const PNG_SIGNATURE: [u8; 8] = [0x89, b'P', b'N', b'G', 0x0D, 0x0A, 0x1A, 0x0A];
+
+/// What a version PNG's XMP says about its provenance.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct VersionXmp {
+    /// `xmpMM:DerivedFrom/stRef:documentID`: the original's `content_hash`.
+    pub original_hash: String,
+    /// `xmpMM:Ingredients[]/stRef:documentID`: reference photos.
+    pub references: Vec<String>,
+    /// `imageapi:versionKind`, as written (not yet validated).
+    pub kind: Option<String>,
+}
+
+/// Value of the first text chunk (`tEXt` or uncompressed `iTXt`) whose
+/// keyword is in `keywords`, for each keyword found. Compressed chunks are
+/// skipped (the producer never compresses). `Ok(vec![])` for a file that
+/// isn't a PNG.
+pub fn read_png_text(path: &Path, keywords: &[&str]) -> io::Result<Vec<(String, String)>> {
+    let mut reader = BufReader::new(File::open(path)?);
+    let mut sig = [0u8; 8];
+    if reader.read_exact(&mut sig).is_err() || sig != PNG_SIGNATURE {
+        return Ok(Vec::new());
+    }
+    let mut found: Vec<(String, String)> = Vec::new();
+    loop {
+        let mut head = [0u8; 8];
+        if reader.read_exact(&mut head).is_err() {
+            break; // truncated file: use what we have
+        }
+        let len = u32::from_be_bytes([head[0], head[1], head[2], head[3]]);
+        let kind = &head[4..8];
+        let is_text = kind == b"tEXt" || kind == b"iTXt";
+        if is_text && len <= MAX_TEXT_CHUNK {
+            let mut data = vec![0u8; len as usize];
+            if reader.read_exact(&mut data).is_err() {
+                break;
+            }
+            let parsed = if kind == b"tEXt" {
+                parse_text(&data)
+            } else {
+                parse_itxt(&data)
+            };
+            if let Some((key, value)) = parsed
+                && keywords.contains(&key.as_str())
+                && !found.iter().any(|(k, _)| *k == key)
+            {
+                found.push((key, value));
+                if found.len() == keywords.len() {
+                    break;
+                }
+            }
+            reader.seek_relative(4)?; // CRC
+        } else {
+            reader.seek(SeekFrom::Current(i64::from(len) + 4))?;
+        }
+        if kind == b"IEND" {
+            break;
+        }
+    }
+    Ok(found)
+}
+
+/// `keyword\0text` (Latin-1).
+fn parse_text(data: &[u8]) -> Option<(String, String)> {
+    let nul = data.iter().position(|b| *b == 0)?;
+    let key = String::from_utf8_lossy(&data[..nul]).into_owned();
+    let value: String = data[nul + 1..].iter().map(|b| *b as char).collect();
+    Some((key, value))
+}
+
+/// `keyword\0 compressed(1) method(1) lang\0 translated\0 text` (UTF-8);
+/// `None` for compressed chunks.
+fn parse_itxt(data: &[u8]) -> Option<(String, String)> {
+    let nul = data.iter().position(|b| *b == 0)?;
+    let key = String::from_utf8_lossy(&data[..nul]).into_owned();
+    let rest = data.get(nul + 1..)?;
+    if *rest.first()? != 0 {
+        return None;
+    }
+    let rest = rest.get(2..)?; // flag + method
+    let lang_end = rest.iter().position(|b| *b == 0)?;
+    let rest = &rest[lang_end + 1..];
+    let translated_end = rest.iter().position(|b| *b == 0)?;
+    let text = &rest[translated_end + 1..];
+    Some((key, String::from_utf8_lossy(text).into_owned()))
+}
+
+/// The XMP packet of a PNG, if it has one.
+pub fn read_png_xmp(path: &Path) -> io::Result<Option<String>> {
+    Ok(read_png_text(path, &[XMP_KEYWORD])?
+        .into_iter()
+        .next()
+        .map(|(_, v)| v))
+}
+
+static DERIVED_FROM: LazyLock<Regex> = LazyLock::new(|| {
+    // Element form (`<xmpMM:DerivedFrom ...>...</xmpMM:DerivedFrom>`) or the
+    // attribute form exiftool may write (`<xmpMM:DerivedFrom stRef:documentID=".."/>`).
+    Regex::new(r"(?s)<xmpMM:DerivedFrom\b([^>]*?)(?:/>|>(.*?)</xmpMM:DerivedFrom>)").unwrap()
+});
+static INGREDIENTS: LazyLock<Regex> = LazyLock::new(|| {
+    Regex::new(r"(?s)<xmpMM:Ingredients\b[^>]*>(.*?)</xmpMM:Ingredients>").unwrap()
+});
+static DOCUMENT_ID: LazyLock<Regex> = LazyLock::new(|| {
+    Regex::new(r#"documentID(?:\s*=\s*["']|\s*>\s*)blake3:([0-9a-fA-F]{64})"#).unwrap()
+});
+static VERSION_KIND: LazyLock<Regex> = LazyLock::new(|| {
+    Regex::new(r#"imageapi:versionKind(?:\s*=\s*["']|\s*>\s*)([A-Za-z]+)"#).unwrap()
+});
+
+/// Parse the link fields out of an XMP packet. `None` when it has no
+/// `DerivedFrom` blake3 id (an ordinary photo's XMP, or someone else's).
+pub fn parse_version_xmp(xmp: &str) -> Option<VersionXmp> {
+    let derived = DERIVED_FROM.captures(xmp)?;
+    let region = format!(
+        "{} {}",
+        derived.get(1).map_or("", |m| m.as_str()),
+        derived.get(2).map_or("", |m| m.as_str())
+    );
+    let original_hash = DOCUMENT_ID.captures(&region)?[1].to_ascii_lowercase();
+
+    let mut references: Vec<String> = Vec::new();
+    if let Some(ing) = INGREDIENTS.captures(xmp) {
+        for cap in DOCUMENT_ID.captures_iter(&ing[1]) {
+            let h = cap[1].to_ascii_lowercase();
+            if !references.contains(&h) {
+                references.push(h);
+            }
+        }
+    }
+    let kind = VERSION_KIND
+        .captures(xmp)
+        .map(|c| c[1].to_ascii_lowercase());
+    Some(VersionXmp {
+        original_hash,
+        references,
+        kind,
+    })
+}
+
+/// Parse the version link out of a PNG file; `None` if it has none.
+pub fn read_version_xmp(path: &Path) -> io::Result<Option<VersionXmp>> {
+    Ok(read_png_xmp(path)?.and_then(|x| parse_version_xmp(&x)))
+}
+
+#[cfg(test)]
+pub(crate) mod test_png {
+    use std::io::Write;
+
+    fn chunk(out: &mut Vec<u8>, kind: &[u8; 4], data: &[u8]) {
+        out.extend_from_slice(&(data.len() as u32).to_be_bytes());
+        out.extend_from_slice(kind);
+        out.extend_from_slice(data);
+        out.extend_from_slice(&[0, 0, 0, 0]); // CRC: the reader never checks it
+    }
+
+    /// A PNG-shaped byte stream: signature, IHDR, optional tEXt `prompt`,
+    /// optional XMP iTXt, an IDAT, IEND. Not decodable; enough for the
+    /// chunk walker (and unique per `salt`, so each file hashes differently).
+    pub fn png_bytes(xmp: Option<&str>, prompt: Option<&str>, salt: u8) -> Vec<u8> {
+        let mut out = super::PNG_SIGNATURE.to_vec();
+        chunk(&mut out, b"IHDR", &[0, 0, 0, 1, 0, 0, 0, 1, 8, 2, 0, 0, 0]);
+        if let Some(p) = prompt {
+            let mut d = b"prompt\0".to_vec();
+            d.extend_from_slice(p.as_bytes());
+            chunk(&mut out, b"tEXt", &d);
+        }
+        if let Some(x) = xmp {
+            let mut d = b"XML:com.adobe.xmp\0\0\0\0\0".to_vec();
+            d.extend_from_slice(x.as_bytes());
+            chunk(&mut out, b"iTXt", &d);
+        }
+        chunk(&mut out, b"IDAT", &[salt; 32]);
+        chunk(&mut out, b"IEND", &[]);
+        out
+    }
+
+    pub fn write_png(path: &std::path::Path, xmp: Option<&str>, prompt: Option<&str>, salt: u8) {
+        if let Some(dir) = path.parent() {
+            std::fs::create_dir_all(dir).unwrap();
+        }
+        std::fs::File::create(path)
+            .unwrap()
+            .write_all(&png_bytes(xmp, prompt, salt))
+            .unwrap();
+    }
+
+    /// The packet the producer writes, built the way `versioning.build_xmp` does.
+    pub fn producer_xmp(original: &str, kind: &str, references: &[&str]) -> String {
+        let ingredients = if references.is_empty() {
+            String::new()
+        } else {
+            let items: String = references
+                .iter()
+                .map(|h| {
+                    format!(
+                        "<rdf:li rdf:parseType=\"Resource\"><stRef:documentID>blake3:{h}</stRef:documentID>\
+                         <stRef:filePath>ref.jpg</stRef:filePath></rdf:li>"
+                    )
+                })
+                .collect();
+            format!("<xmpMM:Ingredients><rdf:Bag>{items}</rdf:Bag></xmpMM:Ingredients>")
+        };
+        format!(
+            "<?xpacket begin=\"\u{feff}\" id=\"W5M0MpCehiHzreSzNTczkc9d\"?><x:xmpmeta xmlns:x=\"adobe:ns:meta/\">\
+             <rdf:RDF xmlns:rdf=\"http://www.w3.org/1999/02/22-rdf-syntax-ns#\"><rdf:Description rdf:about=\"\" \
+             xmlns:xmpMM=\"http://ns.adobe.com/xap/1.0/mm/\" xmlns:stRef=\"http://ns.adobe.com/xap/1.0/sType/ResourceRef#\" \
+             xmlns:imageapi=\"https://git.crkc.xyz/Apps/ImageApi/ns/versions/1.0/\">\
+             <xmpMM:DerivedFrom rdf:parseType=\"Resource\"><stRef:documentID>blake3:{original}</stRef:documentID>\
+             <stRef:filePath>orig.jpg</stRef:filePath></xmpMM:DerivedFrom>{ingredients}\
+             <imageapi:versionKind>{kind}</imageapi:versionKind></rdf:Description></rdf:RDF></x:xmpmeta><?xpacket end=\"w\"?>"
+        )
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::test_png::*;
+    use super::*;
+
+    const ORIG: &str = "d5fa14088629f680f4f8e5815df473cc4684a97fc937da588baf72d2693dea3a";
+    const REF: &str = "0000000000000000000000000000000000000000000000000000000000000001";
+
+    /// The packet embedded in a real archive restoration (copied verbatim).
+    const REAL_XMP: &str = r#"<?xpacket begin="" id="W5M0MpCehiHzreSzNTczkc9d"?><x:xmpmeta xmlns:x="adobe:ns:meta/"><rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"><rdf:Description rdf:about="" xmlns:xmpMM="http://ns.adobe.com/xap/1.0/mm/" xmlns:stRef="http://ns.adobe.com/xap/1.0/sType/ResourceRef#" xmlns:imageapi="https://git.crkc.xyz/Apps/ImageApi/ns/versions/1.0/"><xmpMM:DerivedFrom rdf:parseType="Resource"><stRef:documentID>blake3:d5fa14088629f680f4f8e5815df473cc4684a97fc937da588baf72d2693dea3a</stRef:documentID><stRef:filePath>019_19.JPG</stRef:filePath></xmpMM:DerivedFrom><imageapi:versionKind>restored</imageapi:versionKind></rdf:Description></rdf:RDF></x:xmpmeta><?xpacket end="w"?>"#;
+
+    #[test]
+    fn parses_the_real_producer_packet() {
+        let x = parse_version_xmp(REAL_XMP).unwrap();
+        assert_eq!(x.original_hash, ORIG);
+        assert_eq!(x.kind.as_deref(), Some("restored"));
+        assert!(x.references.is_empty());
+    }
+
+    #[test]
+    fn parses_references_and_keeps_them_apart_from_the_original() {
+        let x = parse_version_xmp(&producer_xmp(ORIG, "edit", &[REF])).unwrap();
+        assert_eq!(x.original_hash, ORIG);
+        assert_eq!(x.references, vec![REF]);
+        assert_eq!(x.kind.as_deref(), Some("edit"));
+    }
+
+    #[test]
+    fn accepts_the_attribute_form_other_writers_use() {
+        let xmp = format!(
+            r#"<x:xmpmeta><rdf:RDF><rdf:Description><xmpMM:DerivedFrom stRef:documentID="blake3:{ORIG}" stRef:filePath="a.jpg"/></rdf:Description></rdf:RDF></x:xmpmeta>"#
+        );
+        assert_eq!(parse_version_xmp(&xmp).unwrap().original_hash, ORIG);
+    }
+
+    #[test]
+    fn ignores_xmp_without_a_blake3_derived_from() {
+        assert!(parse_version_xmp("<x:xmpmeta><dc:title>hi</dc:title></x:xmpmeta>").is_none());
+        let other = "<xmpMM:DerivedFrom><stRef:documentID>xmp.did:1234</stRef:documentID></xmpMM:DerivedFrom>";
+        assert!(parse_version_xmp(other).is_none());
+        // An ingredient alone is not a link.
+        let only_ing = format!(
+            "<xmpMM:Ingredients><rdf:Bag><rdf:li><stRef:documentID>blake3:{REF}</stRef:documentID></rdf:li></rdf:Bag></xmpMM:Ingredients>"
+        );
+        assert!(parse_version_xmp(&only_ing).is_none());
+    }
+
+    #[test]
+    fn reads_xmp_and_prompt_chunks_from_a_png_file() {
+        let dir = tempfile::tempdir().unwrap();
+        let p = dir.path().join("v.png");
+        write_png(
+            &p,
+            Some(&producer_xmp(ORIG, "restored", &[])),
+            Some(r#"{"1": {"class_type": "LoadImage"}}"#),
+            1,
+        );
+        let got = read_version_xmp(&p).unwrap().unwrap();
+        assert_eq!(got.original_hash, ORIG);
+        let text = read_png_text(&p, &["prompt"]).unwrap();
+        assert_eq!(text[0].1, r#"{"1": {"class_type": "LoadImage"}}"#);
+    }
+
+    #[test]
+    fn plain_png_and_non_png_have_no_link() {
+        let dir = tempfile::tempdir().unwrap();
+        let p = dir.path().join("plain.png");
+        write_png(&p, None, None, 2);
+        assert!(read_version_xmp(&p).unwrap().is_none());
+        let j = dir.path().join("x.jpg");
+        std::fs::write(&j, b"\xff\xd8\xff\xe0 not a png").unwrap();
+        assert!(read_version_xmp(&j).unwrap().is_none());
+    }
+}

src/watcher.rs

@@ -28,7 +28,7 @@ use crate::content_hash;
 use crate::database::models::InsertImageExif;
 use crate::database::{
     ExifDao, PrecomputedReelDao, PreviewDao, SqliteExifDao, SqlitePrecomputedReelDao,
-    SqlitePreviewDao, SqliteUserAiPrefsDao, UserAiPrefsDao,
+    SqlitePreviewDao, SqliteUserAiPrefsDao, SqliteVersionDao, UserAiPrefsDao, VersionDao,
 };
 use crate::date_resolver;
 use crate::exif;
@@ -42,6 +42,7 @@ use crate::perceptual_hash;
 use crate::tags;
 use crate::tags::SqliteTagDao;
 use crate::thumbnails;
+use crate::versions;
 use crate::video;
 use crate::video::actors::{
     GeneratePreviewClipMessage, QueueVideosMessage, VideoPlaylistManager, VideoToQueue,
@@ -329,6 +330,9 @@ pub fn watch_files(
         let preview_dao = Arc::new(Mutex::new(
             Box::new(SqlitePreviewDao::new()) as Box<dyn PreviewDao>
         ));
+        let version_dao = Arc::new(Mutex::new(
+            Box::new(SqliteVersionDao::new()) as Box<dyn VersionDao>
+        ));
         let face_dao = Arc::new(Mutex::new(
             Box::new(faces::SqliteFaceDao::new()) as Box<dyn faces::FaceDao>
         ));
@@ -561,6 +565,7 @@ pub fn watch_files(
                         Arc::clone(&preview_dao),
                         Arc::clone(&face_dao),
                         Arc::clone(&watcher_tag_dao),
+                        Arc::clone(&version_dao),
                         face_client.clone(),
                         &effective_excludes,
                         None,
@@ -582,6 +587,7 @@ pub fn watch_files(
                         Arc::clone(&preview_dao),
                         Arc::clone(&face_dao),
                         Arc::clone(&watcher_tag_dao),
+                        Arc::clone(&version_dao),
                         face_client.clone(),
                         &effective_excludes,
                         Some(check_since),
@@ -590,8 +596,26 @@ pub fn watch_files(
                     );
                 }
 
-                // Update media counts per library (metric aggregates across all)
+                // Name-convention versions whose original is (now) known,
-                thumbnails::update_media_counts(Path::new(&lib.root_path), &effective_excludes);
+                // incl. the retry for originals that weren't hashed yet.
+                versions::link_pending(lib, &exif_dao, &version_dao);
+
+                // Update media counts per library (metric aggregates across all).
+                // A version is the same photo as its original, so it isn't counted.
+                let hidden_versions: HashSet<String> = version_dao
+                    .lock()
+                    .expect("version_dao poisoned")
+                    .hidden_version_rows(&[])
+                    .unwrap_or_default()
+                    .into_iter()
+                    .filter(|(lib_id, _)| *lib_id == lib.id)
+                    .map(|(_, rel)| rel)
+                    .collect();
+                thumbnails::update_media_counts_excluding(
+                    Path::new(&lib.root_path),
+                    &effective_excludes,
+                    &hidden_versions,
+                );
 
                 // Missing-file detection: prune image_exif rows whose
                 // source file is no longer on disk. Per-library, so we
@@ -634,6 +658,21 @@ pub fn watch_files(
                 }
                 let _ = crate::database::reconcile::run(&mut conn);
 
+                // Photo versions: copy the original's date / GPS / camera
+                // onto version rows whose copy is missing or stale (a
+                // manual date override, a re-resolve). DB-only, idempotent.
+                match version_dao
+                    .lock()
+                    .expect("version_dao poisoned")
+                    .sync_version_metadata()
+                {
+                    Ok(n) if n > 0 => {
+                        info!("versions: refreshed date/GPS on {} version row(s)", n);
+                    }
+                    Ok(_) => {}
+                    Err(e) => warn!("versions: metadata sync failed: {:?}", e),
+                }
+
                 // Back-ref refresh: hash-keyed rows whose
                 // (library_id, rel_path) tuple no longer matches any
                 // image_exif row but whose hash still does. After a
@@ -707,6 +746,7 @@ pub fn process_new_files(
     preview_dao: Arc<Mutex<Box<dyn PreviewDao>>>,
     face_dao: Arc<Mutex<Box<dyn faces::FaceDao>>>,
     tag_dao: Arc<Mutex<Box<dyn tags::TagDao>>>,
+    version_dao: Arc<Mutex<Box<dyn VersionDao>>>,
     face_client: crate::ai::face_client::FaceClient,
     excluded_dirs: &[String],
     modified_since: Option<SystemTime>,
@@ -755,6 +795,10 @@ pub fn process_new_files(
 
     let mut new_files_found = false;
     let mut files_needing_row = Vec::new();
+    // Files whose image_exif row was written this tick (version linking
+    // looks only at these; re-reading every PNG's XMP on each full scan
+    // would be wasted I/O on a network share).
+    let mut newly_registered: Vec<(PathBuf, String)> = Vec::new();
 
     // Register every image/video file in image_exif. Rows without EXIF
     // still carry library_id, rel_path, content_hash, and size_bytes so
@@ -796,6 +840,7 @@ pub fn process_new_files(
         );
 
         for (file_path, relative_path) in files_needing_row {
+            let registered_file = (file_path.clone(), relative_path.clone());
             let timestamp = Utc::now().timestamp();
 
             // Hash + size from filesystem metadata — always attempted so
@@ -899,10 +944,25 @@ pub fn process_new_files(
                 );
             } else {
                 debug!("Registered {} in image_exif", relative_path);
+                newly_registered.push(registered_file);
             }
         }
     }
 
+    // ── Photo versions (#191) ──────────────────────────────────────────
+    // Link before any derived work below, so a version is never face-scanned
+    // or CLIP-encoded in the tick it appears. XMP first (PNG from the
+    // ComfyUI node; survives renames and moves), then the file-name
+    // convention. Both run after every new row exists, so a version walked
+    // before its original in the same batch still finds it.
+    if !newly_registered.is_empty() {
+        let linked = versions::link_new_files(library, &newly_registered, &exif_dao, &version_dao)
+            + versions::sweep_name_links(library, &exif_dao, &version_dao);
+        if linked > 0 {
+            crate::duplicates::invalidate_caches();
+        }
+    }
+
     // ── Face detection pass ────────────────────────────────────────────
     // Run after EXIF writes so newly-registered files have their
     // content_hash populated. Skipped wholesale when face_client is
@@ -920,7 +980,8 @@ pub fn process_new_files(
         // ensures small/medium deploys self-heal without operator
         // action.
         backfill::backfill_missing_content_hashes(&files, library, &exif_dao);
-        let candidates = backfill::build_face_candidates(library, &files, &exif_dao, &face_dao);
+        let candidates =
+            backfill::build_face_candidates(library, &files, &exif_dao, &face_dao, &version_dao);
         debug!(
             "face_watch: scan tick — {} image file(s) walked, {} candidate(s) (library '{}', modified_since={})",
             files