Keep the peer's run marker on adopted faces, so they are not re-exported as ours

Adopting an image from a peer's face shard stamped its run marker as now,
and the export reads a catalog marker newer than the shard's as a
re-index. So every adopted image went straight back out under this
device's client id: 14,100 adopted, 15,457 "newly indexed" on the next
pass, twenty-two shards of a peer's faces uploaded a second time.

merge_shard now carries the peer's indexed_at into the local index, and
the import writes that marker into face_index; where an older peer's shard
carries none, the store takes the catalog's, so the two agree either way
and the export finds nothing to send.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

core/dr-catalog/src/face_shard.rs

@@ -474,14 +474,26 @@ impl FaceShardStore {
             rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
         )?;
 
-        let mut q =
-            src.prepare("SELECT file_id, model_id, faces_found, source_edge FROM indexed")?;
-        let images: Vec<(i64, String, i64, i64)> = q
-            .query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)))?
+        // The peer's marker travels with the image: it is what lets
+        // `import_from_shards` record the adoption under the time the peer
+        // indexed it, and so what keeps `export_to_shards` from reading the
+        // adoption as a re-index and sending the peer's faces back out under
+        // this device's name. A shard from before the column has none.
+        let mut q = src.prepare(&format!(
+            "SELECT file_id, model_id, faces_found, source_edge, {} FROM indexed",
+            match has_column(&src, "indexed", "indexed_at") {
+                Ok(true) => "indexed_at",
+                _ => "NULL",
+            }
+        ))?;
+        let images: Vec<(i64, String, i64, i64, Option<i64>)> = q
+            .query_map([], |r| {
+                Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?))
+            })?
             .collect::<Result<_, _>>()?;
 
         let mut adopted = 0;
-        for (file_id, model_id, _found, edge) in images {
+        for (file_id, model_id, _found, edge, indexed_at) in images {
             if self.contains(file_id as u64, &model_id) {
                 continue;
             }
@@ -501,12 +513,22 @@ impl FaceShardStore {
             let faces: Vec<SharedFace> = fq
                 .query_map(rusqlite::params![file_id, &model_id], read_shared_face)?
                 .collect::<Result<_, _>>()?;
-            self.put_image(file_id as u64, &model_id, edge as u32, &faces)?;
+            self.put_image_at(file_id as u64, &model_id, edge as u32, &faces, indexed_at)?;
             adopted += 1;
         }
         Ok(adopted)
     }
 
+    /// Record when the catalog indexed a held image, for an entry that
+    /// arrived without a marker — a peer's shard from before the column.
+    pub fn set_indexed_at(&self, file_id: u64, model_id: &str, at: i64) -> Result<(), CatalogError> {
+        self.index.execute(
+            "UPDATE entries SET indexed_at = ?3 WHERE file_id = ?1 AND model_id = ?2",
+            rusqlite::params![file_id as i64, model_id, at],
+        )?;
+        Ok(())
+    }
+
     /// Read back everything held for one image.
     pub fn get_image(
         &self,
@@ -865,6 +887,31 @@ pub fn import_from_shards(
             edge,
             &local,
         )?;
+        // The peer's marker, not this moment. `record_detections` stamps the
+        // run as now, and `export_to_shards` reads a marker newer than the
+        // shard's as a re-index — so every adopted image went straight back
+        // out as this device's own work: 14,100 adopted, 15,457 "newly
+        // indexed" on the next pass, and twenty-two shards of a peer's faces
+        // uploaded again under a second name. Where the peer's shard carried
+        // no marker, the store takes the catalog's, so the two agree either
+        // way and the export sees nothing to send.
+        match store.indexed_at(file_id as u64, &held) {
+            Some(theirs) => {
+                conn.execute(
+                    "UPDATE face_index SET indexed_at = ?3
+                      WHERE image_id = ?1 AND model_id = ?2",
+                    rusqlite::params![image_id, held, theirs],
+                )?;
+            }
+            None => {
+                let ours: i64 = conn.query_row(
+                    "SELECT indexed_at FROM face_index WHERE image_id = ?1 AND model_id = ?2",
+                    rusqlite::params![image_id, held],
+                    |r| r.get(0),
+                )?;
+                store.set_indexed_at(file_id as u64, &held, ours)?;
+            }
+        }
         adopted += 1;
     }
     Ok(adopted)
@@ -1322,6 +1369,11 @@ mod catalog_round_trip {
         assert!((got[0].landmarks[2].0 - 0.15).abs() < 1e-5);
         let emb = faces::embeddings(&b, "w600k_mbf").unwrap();
         assert!(emb.iter().any(|e| e.embedding[0] == 1));
+
+        // And what B adopted is not B's work: its next export sends nothing.
+        // Adopting used to stamp the run as now, so every adopted image went
+        // back out under B's name as a re-index.
+        assert_eq!(export_to_shards(&b, &mut store_b, "w600k_mbf").unwrap(), 0);
     }
 
     /// The desktop switched to a stronger detector part-way through the