Reuse completed Google embeddings and resume semantic staging

SKILLS.md

@@ -73,6 +73,17 @@ SEARCHI_GEMMA_MODEL_DIR and SEARCHI_GEMMA_WORKER, then run:
 cargo test -p searchi-semantic --test gemma_cpu -- --ignored --nocapture
 ```
 
+Incremental reuse/recovery additionally requires verified CPU FFmpeg helpers in
+SEARCHI_TEST_FFMPEG and SEARCHI_TEST_FFPROBE. Run the actual Google integration:
+
+```sh
+cargo test -p searchi-service --test incremental_google_cpu --test google_cpu --locked -- --ignored --nocapture --test-threads=1
+```
+
+Completed-group reuse must preserve actual vector bytes and remap current source
+citations. Do not substitute simulated vectors or treat zero encoder calls as a
+foreground latency benchmark. See docs/INCREMENTAL-SEMANTIC.md for remaining gates.
+
 Record actual outputs, hardware, corpus size, background preparation, first-input
 latency and warm latency in the relevant validation report. Unexecuted platform
 checks stay open. Compilation alone is not UI interaction verification.

crates/searchi-service/src/lib.rs

@@ -1,6 +1,7 @@
 //! Shared offline hybrid search using real Google embeddings and persisted,
 //! filtered HNSW retrieval. Production-scale validation remains a release gate.
 mod media;
+mod pipeline;
 mod routing;
 pub use routing::SemanticOrigin;
 mod store;
@@ -131,6 +132,14 @@ pub struct EmbeddingReport {
     pub chunks: usize,
     pub generation: i64,
     pub model_fingerprint: String,
+    /// Completed groups copied or resumed without document/media inference.
+    pub reused_sources: usize,
+    pub reused_chunks: usize,
+    pub encoded_sources: usize,
+    pub document_inference_calls: usize,
+    pub image_inference_calls: usize,
+    pub audio_inference_calls: usize,
+    pub resumed: bool,
 }
 
 #[derive(Debug, Serialize)]
@@ -206,7 +215,7 @@ impl HybridSearch {
             });
         }
         ensure!(
-            version == 6,
+            (6..=7).contains(&version),
             "Unsupported semantic schema version {version}"
         );
         let (generation,fingerprint,revision,current):(Option<i64>,Option<String>,Option<i64>,i64)=connection.query_row(
@@ -285,88 +294,16 @@ impl HybridSearch {
         Ok(())
     }
 
-    /// Build a replacement generation in bounded CPU batches. Inference or
+    /// Build or resume a replacement generation in bounded batches. Completed
-    /// storage errors leave the previous generation active. A crashed build is
+    /// groups can be reused; failures retain staging work and the ready reader.
-    /// reclaimed on the next explicit rebuild under an OS advisory lock.
     pub fn index_embeddings(&mut self) -> Result<EmbeddingReport> {
         let fingerprint = format!(
             "{}:text-only",
             generation_fingerprint(self.embedder.fingerprint())
         );
-        self.rebuild_generation(fingerprint, Self::build)
+        self.rebuild_generation(fingerprint, |service, generation, work| {
-    }
+            service.build(generation, work)
-
+        })
-    /// Keep the reader's policy aligned with the active generation on every
-    /// failure boundary, including lock admission, activation and cleanup.
-    fn rebuild_generation(
-        &mut self,
-        fingerprint: String,
-        build: impl FnOnce(&mut Self, i64) -> Result<EmbeddingReport>,
-    ) -> Result<EmbeddingReport> {
-        let _lock = self.store.build_lock()?;
-        let generation = self.store.begin(&fingerprint)?;
-        let previous_fingerprint = std::mem::replace(&mut self.generation_fingerprint, fingerprint);
-        let built = build(self, generation).and_then(|report| {
-            self.store.activate(generation)?;
-            Ok(report)
-        });
-        match built {
-            Ok(report) => Ok(report),
-            Err(error) => {
-                self.generation_fingerprint = previous_fingerprint;
-                if let Err(cleanup) = self.store.abandon(generation) {
-                    return Err(error.context(format!(
-                        "Failed to discard staging generation {generation}: {cleanup:#}"
-                    )));
-                }
-                Err(error)
-            }
-        }
-    }
-
-    fn build(&mut self, generation: i64) -> Result<EmbeddingReport> {
-        let mut report = EmbeddingReport {
-            media_files: 0,
-            media_failures: 0,
-            failures: Vec::new(),
-            passages: 0,
-            chunks: 0,
-            generation,
-            model_fingerprint: self.generation_fingerprint.clone(),
-        };
-        let mut after_id = 0;
-        loop {
-            let passages = self.lexical.passage_batch(after_id, 64)?;
-            if passages.is_empty() {
-                break;
-            }
-            for passage in passages {
-                after_id = passage.id;
-                if passage.text.trim().is_empty() {
-                    continue;
-                }
-                let title = bounded_title(&passage.filename);
-                let windows = passage_windows(&mut *self.embedder, &passage.text, title)?;
-                ensure!(
-                    report.chunks + windows.len() <= MAX_SCAN_CHUNKS,
-                    "Corpus exceeds ANN generation resource budget of {MAX_SCAN_CHUNKS} chunks"
-                );
-                for (batch_index, batch) in windows.chunks(MAX_BATCH).enumerate() {
-                    let texts: Vec<_> = batch.iter().map(|window| window.text.as_str()).collect();
-                    let vectors = self.embedder.embed_documents(&texts, title)?;
-                    self.store.insert(
-                        generation,
-                        &passage,
-                        batch_index * MAX_BATCH,
-                        batch,
-                        &vectors,
-                    )?;
-                }
-                report.chunks += windows.len();
-                report.passages += 1;
-            }
-        }
-        Ok(report)
     }
 
     /// Filter each retrieval lane before ranking, preserve filename priority,

crates/searchi-service/src/media.rs

@@ -1,5 +1,7 @@
+use super::pipeline::BuildWork;
 use super::*;
 use searchi_media::*;
+use store::SourceTask;
 
 #[derive(Debug)]
 struct InferenceFailure(anyhow::Error);
@@ -40,8 +42,8 @@ impl HybridSearch {
                 limits.max_file_bytes
             ));
         }
-        self.rebuild_generation(fingerprint, |service, generation| {
+        self.rebuild_generation(fingerprint, |service, generation, work| {
-            let mut report = service.build(generation)?;
+            service.build(generation, work)?;
             let mut after = 0;
             let mut media_count = 0usize;
             loop {
@@ -60,42 +62,142 @@ impl HybridSearch {
                         media_count <= 10000,
                         "Media generation exceeds 10000-file resource budget"
                     );
-                    let before_chunks = report.chunks;
+                    let before_chunks = work.report.chunks;
-                    match service.build_media_file(
+                    let identity = MediaIdentity::read(
-                        generation,
-                        id,
                         path,
-                        kind,
+                        if kind == MediaKind::Image {
-                        tools,
+                            64 * 1024 * 1024
-                        limits,
+                        } else {
-                        &mut report,
+                            limits.max_file_bytes
-                    ) {
+                        },
-                        Ok(()) => report.media_files += 1,
+                    );
+                    let task = SourceTask {
+                        key: super::pipeline::task_key(&["media", &path.to_string_lossy()]),
+                        path: path.to_string_lossy().into_owned(),
+                        source_type: "media".into(),
+                        source_hash: identity
+                            .as_ref()
+                            .map(|i| i.sha256.clone())
+                            .unwrap_or_default(),
+                        title_hash: String::new(),
+                        policy: service.generation_fingerprint.clone(),
+                        passage_id: -id,
+                    };
+                    work.track(&task)?;
+                    service.store.stage_task(generation, &task)?;
+                    let result = identity.and_then(|identity| {
+                        service.build_media_source(
+                            generation, id, path, kind, tools, limits, &identity, &task, work,
+                        )?;
+                        work.media.push(identity);
+                        Ok(())
+                    });
+                    match result {
+                        Ok(()) => work.report.media_files += 1,
                         Err(error) => {
+                            if let Err(cleanup) =
+                                service
+                                    .store
+                                    .fail_source(generation, &task, &format!("{error:#}"))
+                            {
+                                return Err(error.context(format!(
+                                    "Failed to discard incomplete media source: {cleanup:#}"
+                                )));
+                            }
                             if error.downcast_ref::<InferenceFailure>().is_some() {
                                 return Err(error);
                             }
                             service.store.media_failure(
                                 generation,
-                                &path.to_string_lossy(),
+                                &task.path,
                                 &format!("{error:#}"),
                             )?;
-                            report.chunks = before_chunks;
+                            work.report.chunks = before_chunks;
-                            if report.failures.len() < 1000 {
+                            if work.report.failures.len() < 1000 {
-                                report.failures.push(MediaFailure {
+                                work.report.failures.push(MediaFailure {
-                                    path: path.to_string_lossy().into_owned(),
+                                    path: task.path,
                                     message: format!("{error:#}").chars().take(2048).collect(),
                                 });
                             }
-                            report.media_failures += 1;
+                            work.report.media_failures += 1;
                         }
                     }
                 }
             }
-            Ok(report)
+            Ok(())
         })
     }
 
+    #[allow(clippy::too_many_arguments)]
+    fn build_media_source(
+        &mut self,
+        generation: i64,
+        id: i64,
+        path: &Path,
+        kind: MediaKind,
+        tools: &FfmpegTools,
+        limits: &MediaLimits,
+        identity: &MediaIdentity,
+        task: &SourceTask,
+        work: &mut BuildWork,
+    ) -> Result<()> {
+        if let Some(count) = self
+            .store
+            .reuse_media_source(generation, task, id, identity)?
+        {
+            ensure!(
+                work.report.chunks + count <= MAX_SCAN_CHUNKS,
+                "Media exceeds generation chunk budget"
+            );
+            if kind != MediaKind::Image {
+                // Metadata remains part of music search after path remapping.
+                let metadata = probe(tools, path, limits)?;
+                self.refresh_media_metadata(path, identity, &metadata)?;
+            }
+            identity.ensure_unchanged()?;
+            work.report.chunks += count;
+            work.report.reused_chunks += count;
+            work.report.reused_sources += 1;
+            return Ok(());
+        }
+        let before = work.report.chunks;
+        self.build_media_file(
+            generation,
+            id,
+            path,
+            kind,
+            tools,
+            limits,
+            identity,
+            &mut work.report,
+        )?;
+        self.store
+            .complete_source(generation, task, work.report.chunks - before)?;
+        work.report.encoded_sources += 1;
+        Ok(())
+    }
+
+    fn refresh_media_metadata(
+        &mut self,
+        path: &Path,
+        identity: &MediaIdentity,
+        metadata: &MediaProbe,
+    ) -> Result<()> {
+        let tags = metadata
+            .tags
+            .iter()
+            .map(|(k, v)| format!("{k}: {v}"))
+            .collect::<Vec<_>>()
+            .join("\n");
+        self.lexical.set_media_metadata(
+            &path.to_string_lossy(),
+            &tags,
+            metadata.duration_ms,
+            identity.bytes,
+            identity.modified_ns,
+        )
+    }
+
     #[allow(clippy::too_many_arguments)]
     fn build_media_file(
         &mut self,
@@ -105,16 +207,9 @@ impl HybridSearch {
         kind: MediaKind,
         tools: &FfmpegTools,
         limits: &MediaLimits,
+        identity: &MediaIdentity,
         report: &mut EmbeddingReport,
     ) -> Result<()> {
-        let identity = MediaIdentity::read(
-            path,
-            if kind == MediaKind::Image {
-                64 * 1024 * 1024
-            } else {
-                limits.max_file_bytes
-            },
-        )?;
         let mut ordinal = 0;
         if kind == MediaKind::Image {
             validate_image(path)?;
@@ -122,29 +217,18 @@ impl HybridSearch {
                 report.chunks < MAX_SCAN_CHUNKS,
                 "Media exceeds generation chunk budget"
             );
+            report.image_inference_calls += 1;
             let vectors = self
                 .embedder
                 .embed_images(&[path], 280)
                 .map_err(|e| anyhow::Error::new(InferenceFailure(e)))?;
             ensure!(vectors.len() == 1, "Image embedding count mismatch");
             self.store
-                .insert_media(generation, id, ordinal, &identity, "image", 0, &vectors[0])?;
+                .insert_media(generation, id, ordinal, identity, "image", 0, &vectors[0])?;
             report.chunks += 1;
         } else {
             let metadata = probe(tools, path, limits)?;
-            let tags = metadata
+            self.refresh_media_metadata(path, identity, &metadata)?;
-                .tags
-                .iter()
-                .map(|(k, v)| format!("{k}: {v}"))
-                .collect::<Vec<_>>()
-                .join("\n");
-            self.lexical.set_media_metadata(
-                &path.to_string_lossy(),
-                &tags,
-                metadata.duration_ms,
-                identity.bytes,
-                identity.modified_ns,
-            )?;
             if metadata.has_video && kind == MediaKind::Video {
                 let mut seen = std::collections::HashSet::new();
                 let mut scenes = 0;
@@ -168,6 +252,7 @@ impl HybridSearch {
                             report.chunks < MAX_SCAN_CHUNKS,
                             "Media exceeds generation chunk budget"
                         );
+                        report.image_inference_calls += 1;
                         let vectors = self
                             .embedder
                             .embed_images(&[frame.path.as_path()], 70)
@@ -177,7 +262,7 @@ impl HybridSearch {
                             generation,
                             id,
                             ordinal,
-                            &identity,
+                            identity,
                             "video",
                             frame.timestamp_ms,
                             &vectors[0],
@@ -194,6 +279,7 @@ impl HybridSearch {
                         "Audio exceeds generation chunk budget"
                     );
                     let pcm = decode_audio(tools, path, window, limits)?;
+                    report.audio_inference_calls += 1;
                     let vector = self
                         .embedder
                         .embed_audio(&pcm)
@@ -202,7 +288,7 @@ impl HybridSearch {
                         generation,
                         id,
                         ordinal,
-                        &identity,
+                        identity,
                         "audio",
                         window.start_ms,
                         &vector,

crates/searchi-service/src/pipeline.rs

@@ -0,0 +1,272 @@
+//! Durable bounded semantic source tasks. Only completed source groups may
+//! supply vectors; tokenizer/inference work stays outside SQLite transactions.
+use super::*;
+use searchi_core::IndexedPassage;
+use store::{content_hash, SourceTask};
+
+pub(super) struct BuildWork {
+    pub report: EmbeddingReport,
+    pub keys: Vec<String>,
+    pub media: Vec<searchi_media::MediaIdentity>,
+    documents: HashMap<String, searchi_media::MediaIdentity>,
+}
+
+impl BuildWork {
+    pub fn track(&mut self, task: &SourceTask) -> Result<()> {
+        ensure!(
+            self.keys.len() < MAX_SCAN_CHUNKS,
+            "Semantic source task budget exceeded"
+        );
+        self.keys.push(task.key.clone());
+        Ok(())
+    }
+}
+
+pub(super) fn task_key(parts: &[&str]) -> String {
+    let mut input = String::new();
+    for part in parts {
+        input.push_str(&format!("{}:", part.len()));
+        input.push_str(part);
+    }
+    content_hash(&input)
+}
+
+pub(super) fn text_task(passage: &IndexedPassage, policy: &str) -> Result<SourceTask> {
+    let key = task_key(&[
+        "text",
+        &passage.path,
+        &passage.source.kind,
+        &passage.source.number.to_string(),
+    ]);
+    Ok(SourceTask {
+        key,
+        path: passage.path.clone(),
+        source_type: "text".into(),
+        source_hash: content_hash(&passage.text),
+        title_hash: content_hash(bounded_title(&passage.filename)),
+        policy: policy.into(),
+        passage_id: passage.id,
+    })
+}
+
+impl HybridSearch {
+    pub(super) fn rebuild_generation(
+        &mut self,
+        fingerprint: String,
+        build: impl FnOnce(&mut Self, i64, &mut BuildWork) -> Result<()>,
+    ) -> Result<EmbeddingReport> {
+        let lock = self.store.build_lock()?;
+        let (generation, resumed) = self.store.start_or_resume(&fingerprint, &lock)?;
+        // This revision covers the entire discovery/build/finalization pass.
+        // A concurrent lexical update aborts publication; next invocation
+        // reconciles and resumes all still-compatible completed groups.
+        let revision = self.store.revision()?;
+        let previous_fingerprint = std::mem::replace(&mut self.generation_fingerprint, fingerprint);
+        let mut work = BuildWork {
+            report: EmbeddingReport {
+                media_files: 0,
+                media_failures: 0,
+                failures: Vec::new(),
+                passages: 0,
+                chunks: 0,
+                generation,
+                model_fingerprint: self.generation_fingerprint.clone(),
+                reused_sources: 0,
+                reused_chunks: 0,
+                encoded_sources: 0,
+                document_inference_calls: 0,
+                image_inference_calls: 0,
+                audio_inference_calls: 0,
+                resumed,
+            },
+            keys: Vec::new(),
+            media: Vec::new(),
+            documents: HashMap::new(),
+        };
+        let result = build(self, generation, &mut work).and_then(|()| {
+            for identity in work.media.iter().chain(work.documents.values()) {
+                // Full bytes, not just stat metadata: in-place changes can
+                // preserve size and timestamps, including on mounted volumes.
+                let current = searchi_media::MediaIdentity::read(&identity.path, identity.bytes)?;
+                ensure!(
+                    current.sha256 == identity.sha256
+                        && current.modified_ns == identity.modified_ns,
+                    "Source changed during semantic indexing; retry required"
+                );
+            }
+            self.store
+                .reconcile_tasks(generation, &work.keys, revision)?;
+            self.store.activate_checked(generation, revision)?;
+            Ok(work.report)
+        });
+        if result.is_err() {
+            self.generation_fingerprint = previous_fingerprint;
+        }
+        result
+    }
+
+    pub(super) fn build(&mut self, generation: i64, work: &mut BuildWork) -> Result<()> {
+        let mut after_id = 0;
+        loop {
+            let passages = self.lexical.passage_batch(after_id, 64)?;
+            if passages.is_empty() {
+                break;
+            }
+            for passage in passages {
+                after_id = passage.id;
+                if passage.text.trim().is_empty() {
+                    continue;
+                }
+                if !work.documents.contains_key(&passage.path) {
+                    ensure!(
+                        work.documents.len() < MAX_SCAN_CHUNKS,
+                        "Document source budget exceeded"
+                    );
+                    let identity = searchi_media::MediaIdentity::read(
+                        Path::new(&passage.path),
+                        32 * 1024 * 1024,
+                    )?;
+                    work.documents.insert(passage.path.clone(), identity);
+                }
+                let task = text_task(&passage, &self.generation_fingerprint)?;
+                work.track(&task)?;
+                self.store.stage_task(generation, &task)?;
+                let result = self.build_text_source(generation, &task, &passage, work);
+                if let Err(error) = result {
+                    if let Err(cleanup) =
+                        self.store
+                            .fail_source(generation, &task, &format!("{error:#}"))
+                    {
+                        return Err(error.context(format!(
+                            "Failed to discard incomplete text source: {cleanup:#}"
+                        )));
+                    }
+                    return Err(error);
+                }
+            }
+        }
+        Ok(())
+    }
+
+    fn build_text_source(
+        &mut self,
+        generation: i64,
+        task: &SourceTask,
+        passage: &IndexedPassage,
+        work: &mut BuildWork,
+    ) -> Result<()> {
+        if let Some(count) = self.store.reuse_text_source(generation, task, passage)? {
+            ensure!(
+                work.report.chunks + count <= MAX_SCAN_CHUNKS,
+                "Corpus exceeds ANN chunk budget"
+            );
+            work.report.chunks += count;
+            work.report.passages += 1;
+            work.report.reused_chunks += count;
+            work.report.reused_sources += 1;
+            return Ok(());
+        }
+        let title = bounded_title(&passage.filename);
+        let windows = passage_windows(&mut *self.embedder, &passage.text, title)?;
+        ensure!(
+            work.report.chunks + windows.len() <= MAX_SCAN_CHUNKS,
+            "Corpus exceeds ANN generation resource budget of {MAX_SCAN_CHUNKS} chunks"
+        );
+        for (batch_index, batch) in windows.chunks(MAX_BATCH).enumerate() {
+            let texts: Vec<_> = batch.iter().map(|window| window.text.as_str()).collect();
+            work.report.document_inference_calls += 1;
+            let vectors = self.embedder.embed_documents(&texts, title)?;
+            self.store.insert(
+                generation,
+                passage,
+                batch_index * MAX_BATCH,
+                batch,
+                &vectors,
+            )?;
+        }
+        self.store
+            .complete_source(generation, task, windows.len())?;
+        work.report.chunks += windows.len();
+        work.report.passages += 1;
+        work.report.encoded_sources += 1;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Fault-only provider: never returns a vector or claims semantic quality.
+    struct RejectInference;
+    impl EmbeddingProvider for RejectInference {
+        fn fingerprint(&self) -> &str {
+            "fault-injection-only"
+        }
+        fn dimensions(&self) -> usize {
+            GEMMA_DIMENSIONS
+        }
+        fn split_windows(&mut self, text: &str, _: &str) -> Result<Vec<TextWindow>> {
+            Ok(vec![TextWindow {
+                text: text.into(),
+                start_byte: 0,
+                end_byte: text.len(),
+            }])
+        }
+        fn embed_documents(&mut self, _: &[&str], _: &str) -> Result<Vec<Vec<f32>>> {
+            anyhow::bail!("injected primary inference failure")
+        }
+        fn embed_query(&mut self, _: &str) -> Result<Vec<f32>> {
+            anyhow::bail!("fault provider never supplies query vectors")
+        }
+    }
+
+    #[test]
+    fn cleanup_failure_keeps_primary_error_reader_policy_and_staging() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let root = temp.path().join("corpus");
+        std::fs::create_dir(&root)?;
+        std::fs::write(
+            root.join("notes.txt"),
+            "A real source whose inference is deliberately rejected.",
+        )?;
+        let db = temp.path().join("index.db");
+        SearchIndex::open(&db)?.index_folder(&root)?;
+        let mut service = HybridSearch::with_provider(&db, Box::new(RejectInference))?;
+        let prior_policy = service.generation_fingerprint.clone();
+        let connection = rusqlite::Connection::open(&db)?;
+        connection.execute_batch("CREATE TRIGGER reject_task_cleanup BEFORE UPDATE OF state ON semantic_tasks WHEN new.state='failed' BEGIN SELECT RAISE(ABORT,'injected task cleanup failure'); END;")?;
+        let error = service.index_embeddings().unwrap_err();
+        let message = format!("{error:#}");
+        assert!(
+            message.contains("injected primary inference failure"),
+            "{message}"
+        );
+        assert!(
+            message.contains("injected task cleanup failure"),
+            "{message}"
+        );
+        assert_eq!(service.generation_fingerprint, prior_policy);
+        let staging: usize =
+            connection.query_row("SELECT count(*) FROM semantic_generations", [], |r| {
+                r.get(0)
+            })?;
+        assert_eq!(staging, 1);
+        let state: String =
+            connection.query_row("SELECT state FROM semantic_tasks", [], |r| r.get(0))?;
+        assert_eq!(
+            state, "running",
+            "Failed cleanup transaction must roll back"
+        );
+        connection.execute_batch("DROP TRIGGER reject_task_cleanup;")?;
+        let error = service.index_embeddings().unwrap_err();
+        assert!(format!("{error:#}").contains("injected primary inference failure"));
+        assert_eq!(service.generation_fingerprint, prior_policy);
+        let staging: usize =
+            connection.query_row("SELECT count(*) FROM semantic_generations", [], |r| {
+                r.get(0)
+            })?;
+        assert_eq!(staging, 1, "Resume must retain the same staging generation");
+        Ok(())
+    }
+}

crates/searchi-service/src/store.rs

@@ -1,3 +1,5 @@
+mod incremental;
+pub(crate) use incremental::SourceTask;
 use std::{
     collections::{HashMap, HashSet},
     fs::{File, OpenOptions},
@@ -47,7 +49,7 @@ pub(crate) struct Store {
     eligibility_cache: Option<EligibilityCache>,
 }
 
-pub(crate) struct BuildLock(File);
+pub(crate) struct BuildLock(File, PathBuf);
 impl Drop for BuildLock {
     fn drop(&mut self) {
         let _ = FileExt::unlock(&self.0);
@@ -95,7 +97,7 @@ impl Store {
             |row| row.get(0),
         )?;
         ensure!(
-            (1..=6).contains(&version),
+            (1..=7).contains(&version),
             "Unsupported semantic schema version {version}"
         );
         initialization.execute_batch("
@@ -159,6 +161,7 @@ impl Store {
             }
         }
         initialization.execute_batch("CREATE TABLE IF NOT EXISTS semantic_media_errors(generation INTEGER NOT NULL REFERENCES semantic_generations(id) ON DELETE CASCADE,path TEXT NOT NULL,message TEXT NOT NULL,PRIMARY KEY(generation,path))")?;
+        incremental::migrate(&initialization)?;
         initialization.commit()?;
         let canonical = db.canonicalize()?;
         let lock_path = PathBuf::from(format!("{}.semantic.lock", canonical.to_string_lossy()));
@@ -180,9 +183,10 @@ impl Store {
             .open(&self.lock_path)?;
         file.try_lock_exclusive()
             .context("Another semantic rebuild is running")?;
-        Ok(BuildLock(file))
+        Ok(BuildLock(file, self.lock_path.clone()))
     }
 
+    #[cfg(test)]
     pub(crate) fn begin(&mut self, fingerprint: &str) -> Result<i64> {
         let transaction = self.connection.transaction()?;
         // Only called while holding the cross-process lock: abandoned staging
@@ -299,7 +303,12 @@ impl Store {
         Ok(())
     }
 
+    #[cfg(test)]
     pub(crate) fn activate(&mut self, generation: i64) -> Result<()> {
+        self.activate_revision(generation, None)
+    }
+
+    fn activate_revision(&mut self, generation: i64, revision: Option<i64>) -> Result<()> {
         // A failed graph build or serialization leaves the old generation active.
         let ann = self.build_ann(generation)?;
         let length = ann.serialized_length();
@@ -310,7 +319,21 @@ impl Store {
         let mut serialized = vec![0_u8; length];
         ann.save_to_buffer(&mut serialized)?;
         let checksum = format!("{:x}", Sha256::digest(&serialized));
-        let transaction = self.connection.transaction()?;
+        let transaction = self
+            .connection
+            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
+        if let Some(expected) = revision {
+            let current: i64 = transaction.query_row(
+                "SELECT corpus_revision FROM semantic_meta WHERE singleton=1",
+                [],
+                |r| r.get(0),
+            )?;
+            ensure!(
+                current == expected,
+                "Indexed corpus changed during semantic rebuild; retry required"
+            );
+            incremental::validate_completion(&transaction, generation)?;
+        }
         ensure!(
             transaction.query_row(
                 "SELECT EXISTS(SELECT 1 FROM semantic_generations WHERE id=?1)",
@@ -389,6 +412,7 @@ impl Store {
         Ok(ann)
     }
 
+    #[cfg(test)]
     pub(crate) fn abandon(&mut self, generation: i64) -> Result<()> {
         self.connection.execute("DELETE FROM semantic_generations WHERE id=?1 AND id!=COALESCE((SELECT active_generation FROM semantic_meta WHERE singleton=1),-1)", [generation])?;
         Ok(())

crates/searchi-service/src/store/incremental.rs

@@ -0,0 +1,671 @@
+//! Completed-source manifests and resumable staging. Legacy generations stay
+//! searchable but cannot supply vectors without explicit completion evidence.
+use super::*;
+
+#[derive(Debug, Clone)]
+pub(crate) struct SourceTask {
+    pub key: String,
+    pub path: String,
+    /// `text` for one extracted passage, `media` for the entire media file.
+    pub source_type: String,
+    pub source_hash: String,
+    pub title_hash: String,
+    pub policy: String,
+    pub passage_id: i64,
+}
+
+pub(super) fn migrate(connection: &Connection) -> Result<()> {
+    for (table, name, declaration) in [
+        (
+            "semantic_generations",
+            "incremental",
+            "INTEGER NOT NULL DEFAULT 0",
+        ),
+        ("semantic_chunks", "task_key", "TEXT"),
+    ] {
+        let exists: bool = connection.query_row(
+            "SELECT EXISTS(SELECT 1 FROM pragma_table_info(?1) WHERE name=?2)",
+            params![table, name],
+            |r| r.get(0),
+        )?;
+        if !exists {
+            connection.execute_batch(&format!(
+                "ALTER TABLE {table} ADD COLUMN {name} {declaration}"
+            ))?;
+        }
+    }
+    connection.execute_batch("CREATE TABLE IF NOT EXISTS semantic_tasks(
+        generation INTEGER NOT NULL REFERENCES semantic_generations(id) ON DELETE CASCADE,
+        task_key TEXT NOT NULL,path TEXT NOT NULL,source_type TEXT NOT NULL,
+        source_hash TEXT NOT NULL,title_hash TEXT NOT NULL,policy TEXT NOT NULL,
+        passage_id INTEGER NOT NULL,state TEXT NOT NULL CHECK(state IN ('pending','running','done','failed')),
+        attempts INTEGER NOT NULL DEFAULT 0,expected_chunks INTEGER,message TEXT,
+        PRIMARY KEY(generation,task_key));
+        CREATE INDEX IF NOT EXISTS semantic_task_reuse ON semantic_tasks(source_hash,title_hash,policy,source_type,state);
+        CREATE INDEX IF NOT EXISTS semantic_chunk_task ON semantic_chunks(generation,task_key);
+        UPDATE semantic_meta SET schema_version=7 WHERE singleton=1;")?;
+    Ok(())
+}
+
+pub(super) fn validate_completion(connection: &Connection, generation: i64) -> Result<()> {
+    let incomplete: bool = connection.query_row("SELECT EXISTS(SELECT 1 FROM semantic_tasks t WHERE generation=?1 AND
+        (state NOT IN ('done','failed') OR (state='done' AND (expected_chunks IS NULL OR expected_chunks!=(SELECT count(*) FROM semantic_chunks c WHERE c.generation=t.generation AND c.task_key=t.task_key))) OR
+        (state='failed' AND EXISTS(SELECT 1 FROM semantic_chunks c WHERE c.generation=t.generation AND c.task_key=t.task_key))))",[generation], |r|r.get(0))?;
+    ensure!(
+        !incomplete,
+        "Semantic generation contains incomplete source tasks"
+    );
+    let unowned: bool = connection.query_row("SELECT EXISTS(SELECT 1 FROM semantic_chunks c LEFT JOIN semantic_tasks t ON t.generation=c.generation AND t.task_key=c.task_key WHERE c.generation=?1 AND (t.state IS NULL OR t.state!='done'))",[generation],|r|r.get(0))?;
+    ensure!(
+        !unowned,
+        "Semantic generation contains chunks without completed sources"
+    );
+    Ok(())
+}
+
+impl Store {
+    pub(crate) fn revision(&self) -> Result<i64> {
+        Ok(self.connection.query_row(
+            "SELECT corpus_revision FROM semantic_meta WHERE singleton=1",
+            [],
+            |r| r.get(0),
+        )?)
+    }
+
+    /// Caller must retain the OS lock through reconciliation and publication.
+    pub(crate) fn start_or_resume(
+        &mut self,
+        fingerprint: &str,
+        lock: &BuildLock,
+    ) -> Result<(i64, bool)> {
+        ensure!(
+            lock.1 == self.lock_path,
+            "Build lock belongs to a different index"
+        );
+        ensure!(
+            !fingerprint.is_empty() && fingerprint.len() <= 4096,
+            "Invalid generation fingerprint"
+        );
+        let tx = self
+            .connection
+            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
+        let generation: Option<i64>=tx.query_row("SELECT id FROM semantic_generations WHERE id!=COALESCE((SELECT active_generation FROM semantic_meta WHERE singleton=1),-1) AND fingerprint=?1 AND dimensions=?2 AND incremental=1 ORDER BY id DESC LIMIT 1",params![fingerprint,self.dimensions as i64],|r|r.get(0)).optional()?;
+        let resumed = generation.is_some();
+        let id = if let Some(id) = generation {
+            id
+        } else {
+            tx.execute("INSERT INTO semantic_generations(fingerprint,dimensions,corpus_revision,incremental) SELECT ?1,?2,corpus_revision,1 FROM semantic_meta WHERE singleton=1",params![fingerprint,self.dimensions as i64])?;
+            tx.last_insert_rowid()
+        };
+        tx.execute("DELETE FROM semantic_generations WHERE id!=?1 AND id!=COALESCE((SELECT active_generation FROM semantic_meta WHERE singleton=1),-1)",[id])?;
+        // Untagged chunks are writes interrupted before the source transaction.
+        tx.execute("UPDATE semantic_tasks SET state='pending',expected_chunks=NULL WHERE generation=?1 AND state='done' AND
+            (expected_chunks IS NULL OR expected_chunks!=(SELECT count(*) FROM semantic_chunks c WHERE c.generation=semantic_tasks.generation AND c.task_key=semantic_tasks.task_key))",[id])?;
+        tx.execute("DELETE FROM semantic_chunks WHERE generation=?1 AND NOT EXISTS(SELECT 1 FROM semantic_tasks t WHERE t.generation=semantic_chunks.generation AND t.task_key=semantic_chunks.task_key AND t.state='done')",[id])?;
+        tx.execute("UPDATE semantic_tasks SET state='pending',expected_chunks=NULL,message=NULL WHERE generation=?1 AND state!='done'",[id])?;
+        tx.commit()?;
+        Ok((id, resumed))
+    }
+
+    pub(crate) fn stage_task(&mut self, generation: i64, task: &SourceTask) -> Result<bool> {
+        validate_task(task)?;
+        let tx = self.connection.transaction()?;
+        let same:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM semantic_tasks WHERE generation=?1 AND task_key=?2 AND source_hash=?3 AND title_hash=?4 AND policy=?5 AND source_type=?6 AND state='done')",params![generation,task.key,task.source_hash,task.title_hash,task.policy,task.source_type],|r|r.get(0))?;
+        if !same {
+            tx.execute(
+                "DELETE FROM semantic_chunks WHERE generation=?1 AND task_key=?2",
+                params![generation, task.key],
+            )?;
+            tx.execute("INSERT INTO semantic_tasks(generation,task_key,path,source_type,source_hash,title_hash,policy,passage_id,state,attempts)
+                VALUES(?1,?2,?3,?4,?5,?6,?7,?8,'running',1)
+                ON CONFLICT(generation,task_key) DO UPDATE SET path=excluded.path,source_type=excluded.source_type,source_hash=excluded.source_hash,title_hash=excluded.title_hash,policy=excluded.policy,passage_id=excluded.passage_id,state='running',attempts=semantic_tasks.attempts+1,expected_chunks=NULL,message=NULL",params![generation,task.key,task.path,task.source_type,task.source_hash,task.title_hash,task.policy,task.passage_id])?;
+        }
+        tx.commit()?;
+        Ok(same)
+    }
+
+    pub(crate) fn complete_source(
+        &mut self,
+        generation: i64,
+        task: &SourceTask,
+        expected_chunks: usize,
+    ) -> Result<()> {
+        ensure!(
+            !task.source_hash.is_empty(),
+            "Cannot complete a source without verified identity"
+        );
+        ensure!(
+            expected_chunks <= MAX_SCAN_CHUNKS,
+            "Source exceeds semantic chunk budget"
+        );
+        let tx = self
+            .connection
+            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
+        let count:i64=tx.query_row("SELECT count(*) FROM semantic_chunks WHERE generation=?1 AND path=?2 AND ((?3='text' AND source_type='text' AND passage_id=?4) OR (?3='media' AND source_type!='text'))",params![generation,task.path,task.source_type,task.passage_id],|r|r.get(0))?;
+        ensure!(
+            usize::try_from(count)? == expected_chunks,
+            "Incomplete source chunk membership"
+        );
+        let total: usize = tx.query_row(
+            "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
+            [generation],
+            |r| r.get(0),
+        )?;
+        ensure!(
+            total <= MAX_SCAN_CHUNKS,
+            "Generation exceeds semantic chunk budget"
+        );
+        tx.execute("UPDATE semantic_chunks SET task_key=?5 WHERE generation=?1 AND path=?2 AND ((?3='text' AND source_type='text' AND passage_id=?4) OR (?3='media' AND source_type!='text'))",params![generation,task.path,task.source_type,task.passage_id,task.key])?;
+        let changed=tx.execute("UPDATE semantic_tasks SET state='done',expected_chunks=?3,path=?4,passage_id=?5,message=NULL WHERE generation=?1 AND task_key=?2 AND source_hash=?6 AND title_hash=?7 AND policy=?8 AND source_type=?9",params![generation,task.key,count,task.path,task.passage_id,task.source_hash,task.title_hash,task.policy,task.source_type])?;
+        ensure!(changed == 1, "Missing matching source task");
+        if task.source_type == "media" {
+            tx.execute(
+                "DELETE FROM semantic_media_errors WHERE generation=?1 AND path=?2",
+                params![generation, task.path],
+            )?;
+        }
+        tx.commit()?;
+        Ok(())
+    }
+
+    pub(crate) fn fail_source(
+        &mut self,
+        generation: i64,
+        task: &SourceTask,
+        message: &str,
+    ) -> Result<()> {
+        let tx = self.connection.transaction()?;
+        tx.execute("DELETE FROM semantic_chunks WHERE generation=?1 AND (task_key=?2 OR (path=?3 AND ((?4='text' AND source_type='text' AND passage_id=?5) OR (?4='media' AND source_type!='text'))))",params![generation,task.key,task.path,task.source_type,task.passage_id])?;
+        tx.execute("UPDATE semantic_tasks SET state='failed',expected_chunks=NULL,message=?3 WHERE generation=?1 AND task_key=?2",params![generation,task.key,message.chars().take(2048).collect::<String>()])?;
+        tx.commit()?;
+        Ok(())
+    }
+
+    pub(crate) fn reconcile_tasks(
+        &mut self,
+        generation: i64,
+        keys: &[String],
+        revision: i64,
+    ) -> Result<()> {
+        ensure!(
+            keys.len() <= MAX_SCAN_CHUNKS,
+            "Generation exceeds source task budget"
+        );
+        let tx = self
+            .connection
+            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
+        let current: i64 = tx.query_row(
+            "SELECT corpus_revision FROM semantic_meta WHERE singleton=1",
+            [],
+            |r| r.get(0),
+        )?;
+        ensure!(
+            current == revision,
+            "Indexed corpus changed during semantic reconciliation; retry required"
+        );
+        tx.execute_batch("CREATE TEMP TABLE IF NOT EXISTS semantic_current_tasks(task_key TEXT PRIMARY KEY); DELETE FROM semantic_current_tasks;")?;
+        for key in keys {
+            tx.execute(
+                "INSERT OR IGNORE INTO semantic_current_tasks VALUES(?1)",
+                [key],
+            )?;
+        }
+        tx.execute("DELETE FROM semantic_chunks WHERE generation=?1 AND (task_key IS NULL OR task_key NOT IN (SELECT task_key FROM semantic_current_tasks))",[generation])?;
+        tx.execute("DELETE FROM semantic_tasks WHERE generation=?1 AND task_key NOT IN (SELECT task_key FROM semantic_current_tasks)",[generation])?;
+        tx.execute("DELETE FROM semantic_media_errors WHERE generation=?1 AND path NOT IN (SELECT path FROM semantic_tasks WHERE generation=?1)",[generation])?;
+        tx.execute(
+            "UPDATE semantic_generations SET corpus_revision=?2 WHERE id=?1",
+            params![generation, revision],
+        )?;
+        tx.commit()?;
+        Ok(())
+    }
+
+    pub(crate) fn activate_checked(&mut self, generation: i64, revision: i64) -> Result<()> {
+        validate_completion(&self.connection, generation)?;
+        self.activate_revision(generation, Some(revision))
+    }
+}
+
+fn validate_task(task: &SourceTask) -> Result<()> {
+    ensure!(
+        !task.key.is_empty()
+            && task.key.len() <= 256
+            && !task.path.is_empty()
+            && task.path.len() <= 32768,
+        "Invalid source task identity"
+    );
+    ensure!(
+        matches!(task.source_type.as_str(), "text" | "media"),
+        "Invalid source task type"
+    );
+    ensure!(
+        (!task.source_hash.is_empty() || task.source_type == "media")
+            && task.source_hash.len() <= 256
+            && task.title_hash.len() <= 256
+            && !task.policy.is_empty()
+            && task.policy.len() <= 4096,
+        "Invalid source completion policy"
+    );
+    Ok(())
+}
+
+impl Store {
+    fn reusable_task(
+        &self,
+        generation: i64,
+        task: &SourceTask,
+    ) -> Result<Option<(i64, String, usize)>> {
+        Ok(self.connection.query_row("SELECT t.generation,t.task_key,t.expected_chunks FROM semantic_tasks t JOIN semantic_generations g ON g.id=t.generation JOIN semantic_generations destination ON destination.id=?1
+            WHERE t.source_hash=?2 AND t.title_hash=?3 AND t.policy=?4 AND t.source_type=?5 AND t.state='done' AND t.expected_chunks IS NOT NULL
+            AND g.fingerprint=destination.fingerprint AND g.dimensions=destination.dimensions
+            AND t.expected_chunks=(SELECT count(*) FROM semantic_chunks c WHERE c.generation=t.generation AND c.task_key=t.task_key)
+            ORDER BY (t.generation=?1) DESC,t.generation DESC LIMIT 1",params![generation,task.source_hash,task.title_hash,task.policy,task.source_type],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?))).optional()?)
+    }
+
+    /// No tokenizer or neural call: only a completed, policy-compatible group
+    /// whose persisted ranges still equal the current extracted bytes qualifies.
+    pub(crate) fn reuse_text_source(
+        &mut self,
+        generation: i64,
+        task: &SourceTask,
+        passage: &IndexedPassage,
+    ) -> Result<Option<usize>> {
+        ensure!(
+            task.source_type == "text"
+                && task.passage_id == passage.id
+                && task.path == passage.path,
+            "Text task mapping differs from current source"
+        );
+        ensure!(
+            task.source_hash == content_hash(&passage.text),
+            "Text task hash differs from source bytes"
+        );
+        let Some((source, key, count)) = self.reusable_task(generation, task)? else {
+            return Ok(None);
+        };
+        ensure!(
+            count <= MAX_SCAN_CHUNKS,
+            "Reusable source exceeds chunk budget"
+        );
+        {
+            let mut stmt=self.connection.prepare("SELECT start_byte,end_byte,snippet,vector,source_type,ordinal,content_hash FROM semantic_chunks WHERE generation=?1 AND task_key=?2 ORDER BY ordinal")?;
+            let mut rows = stmt.query(params![source, key])?;
+            let mut seen = 0usize;
+            while let Some(row) = rows.next()? {
+                let start: usize = row.get(0)?;
+                let end: usize = row.get(1)?;
+                let snippet: String = row.get(2)?;
+                ensure!(
+                    passage.text.get(start..end) == Some(snippet.as_str()),
+                    "Reusable text range differs from current source"
+                );
+                ensure!(
+                    row.get::<_, String>(4)? == "text"
+                        && row.get::<_, usize>(5)? == seen
+                        && row.get::<_, String>(6)? == task.source_hash,
+                    "Corrupt completed text source membership"
+                );
+                validate_vector(&row.get::<_, Vec<u8>>(3)?, self.dimensions)?;
+                seen += 1;
+            }
+            ensure!(seen == count, "Reusable text completion count differs");
+        }
+        self.copy_completed(
+            generation,
+            task,
+            source,
+            &key,
+            count,
+            Some(&passage.source),
+            None,
+        )?;
+        Ok(Some(count))
+    }
+
+    pub(crate) fn reuse_media_source(
+        &mut self,
+        generation: i64,
+        task: &SourceTask,
+        file_id: i64,
+        identity: &searchi_media::MediaIdentity,
+    ) -> Result<Option<usize>> {
+        ensure!(
+            task.source_type == "media"
+                && task.passage_id == -file_id
+                && task.path == identity.path.to_string_lossy()
+                && task.source_hash == identity.sha256,
+            "Media task mapping differs from current source"
+        );
+        let Some((source, key, count)) = self.reusable_task(generation, task)? else {
+            return Ok(None);
+        };
+        ensure!(
+            count <= MAX_SCAN_CHUNKS,
+            "Reusable source exceeds chunk budget"
+        );
+        {
+            let mut stmt=self.connection.prepare("SELECT vector,source_type,ordinal,content_hash FROM semantic_chunks WHERE generation=?1 AND task_key=?2 ORDER BY ordinal")?;
+            let mut rows = stmt.query(params![source, key])?;
+            let mut seen = 0usize;
+            while let Some(row) = rows.next()? {
+                ensure!(
+                    matches!(
+                        row.get::<_, String>(1)?.as_str(),
+                        "image" | "audio" | "video"
+                    ) && row.get::<_, usize>(2)? == seen
+                        && row.get::<_, String>(3)? == task.source_hash,
+                    "Corrupt completed media source membership"
+                );
+                validate_vector(&row.get::<_, Vec<u8>>(0)?, self.dimensions)?;
+                seen += 1;
+            }
+            ensure!(seen == count, "Reusable media completion count differs");
+        }
+        self.copy_completed(generation, task, source, &key, count, None, Some(identity))?;
+        Ok(Some(count))
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    fn copy_completed(
+        &mut self,
+        generation: i64,
+        task: &SourceTask,
+        source: i64,
+        key: &str,
+        count: usize,
+        location: Option<&SourceLocation>,
+        identity: Option<&searchi_media::MediaIdentity>,
+    ) -> Result<()> {
+        let tx = self
+            .connection
+            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
+        let total:usize=tx.query_row("SELECT count(*) FROM semantic_chunks WHERE generation=?1 AND (task_key IS NULL OR task_key!=?2)",params![generation,task.key],|r|r.get(0))?;
+        ensure!(
+            total.saturating_add(count) <= MAX_SCAN_CHUNKS,
+            "Generation exceeds semantic chunk budget"
+        );
+        // A temporary SQL table allows remapping a completed staging task onto
+        // itself without retaining its old citations or allocating all vectors.
+        tx.execute_batch("CREATE TEMP TABLE IF NOT EXISTS semantic_reuse_chunks AS SELECT * FROM semantic_chunks WHERE 0; DELETE FROM semantic_reuse_chunks;")?;
+        tx.execute("INSERT INTO semantic_reuse_chunks SELECT * FROM semantic_chunks WHERE generation=?1 AND task_key=?2",params![source,key])?;
+        tx.execute(
+            "DELETE FROM semantic_chunks WHERE generation=?1 AND task_key=?2",
+            params![generation, task.key],
+        )?;
+        tx.execute("INSERT INTO semantic_chunks(generation,passage_id,ordinal,path,kind,number,content_hash,start_byte,end_byte,snippet,vector,source_type,file_size,file_mtime,task_key)
+            SELECT ?1,?2,ordinal,?3,COALESCE(?4,kind),COALESCE(?5,number),content_hash,start_byte,end_byte,snippet,vector,source_type,COALESCE(?6,file_size),COALESCE(?7,file_mtime),?8 FROM semantic_reuse_chunks",params![generation,task.passage_id,task.path,location.map(|s|s.kind.as_str()),location.map(|s|s.number),identity.map(|i|i.bytes as i64),identity.map(|i|i.modified_ns),task.key])?;
+        let changed=tx.execute("UPDATE semantic_tasks SET state='done',expected_chunks=?3,path=?4,passage_id=?5,message=NULL WHERE generation=?1 AND task_key=?2 AND source_hash=?6 AND title_hash=?7 AND policy=?8 AND source_type=?9",params![generation,task.key,count as i64,task.path,task.passage_id,task.source_hash,task.title_hash,task.policy,task.source_type])?;
+        ensure!(changed == 1, "Missing source task during reuse");
+        if task.source_type == "media" {
+            tx.execute(
+                "DELETE FROM semantic_media_errors WHERE generation=?1 AND path=?2",
+                params![generation, task.path],
+            )?;
+        }
+        tx.execute_batch("DELETE FROM semantic_reuse_chunks;")?;
+        tx.commit()?;
+        Ok(())
+    }
+}
+
+fn validate_vector(bytes: &[u8], dimensions: usize) -> Result<()> {
+    ensure!(
+        bytes.len() == dimensions * 4,
+        "Corrupt reusable vector length"
+    );
+    let mut norm = 0.0f32;
+    for value in bytes.as_chunks::<4>().0 {
+        let value = f32::from_le_bytes(*value);
+        ensure!(value.is_finite(), "Corrupt reusable vector value");
+        norm += value * value;
+    }
+    ensure!(
+        norm.is_finite() && (norm.sqrt() - 1.0).abs() < 1e-3,
+        "Reusable vector is not normalized"
+    );
+    Ok(())
+}
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use searchi_core::SearchIndex;
+
+    // Storage-only vectors exercise durability and membership. They are not
+    // simulated production inference or evidence of semantic relevance.
+    fn vector() -> Vec<f32> {
+        vec![1.0, 0.0]
+    }
+    fn task(p: &IndexedPassage) -> SourceTask {
+        SourceTask {
+            key: format!("text-{}-{}", p.path, p.source.number),
+            path: p.path.clone(),
+            source_type: "text".into(),
+            source_hash: content_hash(&p.text),
+            title_hash: content_hash(&p.filename),
+            policy: "storage-test-window-policy".into(),
+            passage_id: p.id,
+        }
+    }
+    fn insert_complete(
+        store: &mut Store,
+        g: i64,
+        p: &IndexedPassage,
+        t: &SourceTask,
+    ) -> Result<()> {
+        store.stage_task(g, t)?;
+        store.insert(
+            g,
+            p,
+            0,
+            &[TextWindow {
+                text: p.text.clone(),
+                start_byte: 0,
+                end_byte: p.text.len(),
+            }],
+            &[vector()],
+        )?;
+        store.complete_source(g, t, 1)
+    }
+
+    #[test]
+    fn resume_retains_completed_groups_and_discards_unfinished_vectors() -> Result<()> {
+        let dir = tempfile::tempdir()?;
+        let root = dir.path().join("docs");
+        std::fs::create_dir(&root)?;
+        std::fs::write(root.join("first.txt"), "complete source")?;
+        std::fs::write(root.join("second.txt"), "interrupted source")?;
+        let db = dir.path().join("index.db");
+        let mut lexical = SearchIndex::open(&db)?;
+        lexical.index_folder(&root)?;
+        let passages = lexical.passage_batch(0, 64)?;
+        let first = task(&passages[0]);
+        let second = task(&passages[1]);
+        let generation;
+        {
+            let mut store = Store::open(&db, 2)?;
+            let lock = store.build_lock()?;
+            generation = store.start_or_resume("storage-only-model", &lock)?.0;
+            insert_complete(&mut store, generation, &passages[0], &first)?;
+            store.stage_task(generation, &second)?;
+            store.insert(
+                generation,
+                &passages[1],
+                0,
+                &[TextWindow {
+                    text: passages[1].text.clone(),
+                    start_byte: 0,
+                    end_byte: passages[1].text.len(),
+                }],
+                &[vector()],
+            )?;
+            assert!(store
+                .activate_checked(generation, store.revision()?)
+                .is_err());
+        }
+        let mut store = Store::open(&db, 2)?;
+        let lock = store.build_lock()?;
+        assert_eq!(
+            store.start_or_resume("storage-only-model", &lock)?,
+            (generation, true)
+        );
+        assert_eq!(
+            store.connection.query_row(
+                "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
+                [generation],
+                |r| r.get::<_, usize>(0)
+            )?,
+            1
+        );
+        store.stage_task(generation, &first)?;
+        assert_eq!(
+            store.reuse_text_source(generation, &first, &passages[0])?,
+            Some(1)
+        );
+        assert!(store
+            .reuse_text_source(generation, &second, &passages[1])?
+            .is_none());
+        assert!(store
+            .activate_checked(generation, store.revision()?)
+            .is_err());
+        store.reconcile_tasks(generation, &[first.key], store.revision()?)?;
+        store.activate_checked(generation, store.revision()?)?;
+        Ok(())
+    }
+
+    #[test]
+    fn legacy_vectors_remain_searchable_but_do_not_qualify_for_reuse() -> Result<()> {
+        let dir = tempfile::tempdir()?;
+        let root = dir.path().join("docs");
+        std::fs::create_dir(&root)?;
+        std::fs::write(root.join("notes.txt"), "legacy searchable source")?;
+        let db = dir.path().join("index.db");
+        let mut lexical = SearchIndex::open(&db)?;
+        lexical.index_folder(&root)?;
+        let p = lexical.passage_batch(0, 1)?.remove(0);
+        let mut store = Store::open(&db, 2)?;
+        let legacy = store.begin("legacy-policy")?;
+        store.insert(
+            legacy,
+            &p,
+            0,
+            &[TextWindow {
+                text: p.text.clone(),
+                start_byte: 0,
+                end_byte: p.text.len(),
+            }],
+            &[vector()],
+        )?;
+        store.activate(legacy)?;
+        // Reopen a schema6 DB with legacy active chunks and no manifests.
+        store.connection.execute_batch(
+            "DROP INDEX semantic_chunk_task; ALTER TABLE semantic_chunks DROP COLUMN task_key;
+            DROP TABLE semantic_tasks; ALTER TABLE semantic_generations DROP COLUMN incremental;
+            UPDATE semantic_meta SET schema_version=6;",
+        )?;
+        drop(store);
+        let mut store = Store::open(&db, 2)?;
+        assert_eq!(
+            store
+                .search(&vector(), "legacy-policy", 1, &SearchOptions::default())?
+                .len(),
+            1
+        );
+        let lock = store.build_lock()?;
+        let (g, _) = store.start_or_resume("legacy-policy", &lock)?;
+        let t = task(&p);
+        store.stage_task(g, &t)?;
+        assert_eq!(store.reuse_text_source(g, &t, &p)?, None);
+        assert_eq!(
+            store.active_fingerprint()?.as_deref(),
+            Some("legacy-policy")
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn revision_race_cannot_publish_and_failed_activation_keeps_staging() -> Result<()> {
+        let dir = tempfile::tempdir()?;
+        let root = dir.path().join("docs");
+        std::fs::create_dir(&root)?;
+        std::fs::write(root.join("notes.txt"), "source before change")?;
+        let db = dir.path().join("index.db");
+        let mut lexical = SearchIndex::open(&db)?;
+        lexical.index_folder(&root)?;
+        let p = lexical.passage_batch(0, 1)?.remove(0);
+        let mut store = Store::open(&db, 2)?;
+        let lock = store.build_lock()?;
+        let (g, _) = store.start_or_resume("checked-policy", &lock)?;
+        let t = task(&p);
+        insert_complete(&mut store, g, &p, &t)?;
+        let old_revision = store.revision()?;
+        std::fs::write(root.join("notes.txt"), "source after change longer")?;
+        lexical.index_folder(&root)?;
+        assert!(store.activate_checked(g, old_revision).is_err());
+        assert_eq!(store.active_fingerprint()?, None);
+        assert_eq!(store.start_or_resume("checked-policy", &lock)?, (g, true));
+        assert_eq!(
+            store.connection.query_row(
+                "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
+                [g],
+                |r| r.get::<_, usize>(0)
+            )?,
+            1
+        );
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod completion_failure_tests {
+    use super::*;
+    use searchi_core::SearchIndex;
+
+    #[test]
+    fn failed_completion_rolls_back_chunk_membership_and_malformed_ranges_never_reuse() -> Result<()>
+    {
+        // SQLite durability fixture, not neural inference evidence.
+        let dir = tempfile::tempdir()?;
+        let root = dir.path().join("docs");
+        std::fs::create_dir(&root)?;
+        std::fs::write(root.join("notes.txt"), "UTF-8 source with 中文 preserved")?;
+        let db = dir.path().join("index.db");
+        let mut lexical = SearchIndex::open(&db)?;
+        lexical.index_folder(&root)?;
+        let p = lexical.passage_batch(0, 1)?.remove(0);
+        let t = SourceTask {
+            key: "durability-task".into(),
+            path: p.path.clone(),
+            source_type: "text".into(),
+            source_hash: content_hash(&p.text),
+            title_hash: content_hash(&p.filename),
+            policy: "exact-window-test".into(),
+            passage_id: p.id,
+        };
+        let mut store = Store::open(&db, 2)?;
+        let lock = store.build_lock()?;
+        let (g, _) = store.start_or_resume("storage-geometry", &lock)?;
+        store.stage_task(g, &t)?;
+        store.insert(
+            g,
+            &p,
+            0,
+            &[TextWindow {
+                text: p.text.clone(),
+                start_byte: 0,
+                end_byte: p.text.len(),
+            }],
+            &[vec![1.0, 0.0]],
+        )?;
+        assert!(store.complete_source(g, &t, 2).is_err());
+        let (state,owned):(String,usize)=store.connection.query_row("SELECT state,(SELECT count(*) FROM semantic_chunks WHERE generation=?1 AND task_key IS NOT NULL) FROM semantic_tasks WHERE generation=?1 AND task_key=?2",params![g,t.key],|r|Ok((r.get(0)?,r.get(1)?)))?;
+        assert_eq!(state, "running");
+        assert_eq!(owned, 0);
+        assert!(store.activate_checked(g, store.revision()?).is_err());
+        store.complete_source(g, &t, 1)?;
+        store.connection.execute("UPDATE semantic_chunks SET start_byte=1,snippet='different snippet' WHERE generation=?1",[g])?;
+        assert!(store.reuse_text_source(g, &t, &p).is_err());
+        assert!(store.active_fingerprint()?.is_none());
+        Ok(())
+    }
+}

crates/searchi-service/tests/google_cpu.rs

@@ -164,22 +164,31 @@ fn real_google_failed_policy_rebuild_preserves_live_reader_and_retry() -> Result
         connection.query_row("SELECT count(*) FROM semantic_generations", [], |row| {
             row.get(0)
         })?;
-    assert_eq!(generations, 1, "Failed activation must discard staging");
+    assert_eq!(
-
+        generations, 2,
-    connection.execute_batch("CREATE TRIGGER reject_generation_cleanup BEFORE DELETE ON semantic_generations WHEN old.id != (SELECT active_generation FROM semantic_meta WHERE singleton=1) BEGIN SELECT RAISE(ABORT,'rejected generation cleanup'); END;")?;
+        "Failed activation must retain resumable staging"
+    );
+    // A repeated publication failure must preserve the same completed staging
+    // generation and the live reader; it must not create or abandon another one.
     let failure = service
         .index_embeddings_with_media(&tools, &limits)
         .unwrap_err();
-    let message = format!("{failure:#}");
+    assert!(format!("{failure:#}").contains("rejected generation activation"));
-    assert!(message.contains("rejected generation activation"));
-    assert!(message.contains("rejected generation cleanup"));
     assert_eq!(active(&connection)?, original);
     assert_reader(&mut service)?;
-    connection.execute_batch(
+    assert_eq!(
-        "DROP TRIGGER reject_generation_activation; DROP TRIGGER reject_generation_cleanup;",
+        connection.query_row("SELECT count(*) FROM semantic_generations", [], |row| row
-    )?;
+            .get::<_, i64>(
+            0
+        ))?,
+        2
+    );
+    connection.execute_batch("DROP TRIGGER reject_generation_activation;")?;
     let retry = service.index_embeddings_with_media(&tools, &limits)?;
     assert_eq!(retry.passages, 1);
+    assert!(retry.resumed);
+    assert_eq!(retry.reused_sources, 1);
+    assert_eq!(retry.document_inference_calls, 0);
     assert_reader(&mut service)?;
     let media_active = active(&connection)?;
     assert_ne!(media_active.1, original.1);
@@ -195,7 +204,9 @@ fn real_google_failed_policy_rebuild_preserves_live_reader_and_retry() -> Result
     assert_eq!(active(&connection)?, media_active);
     assert_reader(&mut service)?;
     connection.execute_batch("DROP TRIGGER reject_generation_activation;")?;
-    service.index_embeddings()?;
+    let reverse_retry = service.index_embeddings()?;
+    assert!(reverse_retry.resumed);
+    assert_eq!(reverse_retry.document_inference_calls, 0);
     assert_eq!(active(&connection)?.1, original.1);
     assert_reader(&mut service)?;
     Ok(())

crates/searchi-service/tests/incremental_google_cpu.rs

@@ -0,0 +1,417 @@
+//! Incremental acceptance uses genuine Google vectors, never stand-in embeddings.
+use anyhow::Result;
+use rusqlite::Connection;
+use searchi_core::SearchIndex;
+use searchi_semantic::{GemmaEmbedder, TextWindow};
+use searchi_service::{EmbeddingProvider, HybridSearch, SearchOptions};
+use std::{
+    fs,
+    path::Path,
+    sync::{Arc, Mutex},
+};
+
+#[derive(Default, Debug, Clone, PartialEq, Eq)]
+struct Calls {
+    documents: usize,
+    images: usize,
+    audio: usize,
+}
+struct CountedGoogle {
+    inner: GemmaEmbedder,
+    calls: Arc<Mutex<Calls>>,
+}
+impl EmbeddingProvider for CountedGoogle {
+    fn fingerprint(&self) -> &str {
+        EmbeddingProvider::fingerprint(&self.inner)
+    }
+    fn dimensions(&self) -> usize {
+        EmbeddingProvider::dimensions(&self.inner)
+    }
+    fn split_windows(&mut self, text: &str, title: &str) -> Result<Vec<TextWindow>> {
+        self.inner.split_windows(text, title)
+    }
+    fn embed_documents(&mut self, texts: &[&str], title: &str) -> Result<Vec<Vec<f32>>> {
+        self.calls.lock().unwrap().documents += texts.len();
+        self.inner.embed_documents(texts, title)
+    }
+    fn embed_query(&mut self, text: &str) -> Result<Vec<f32>> {
+        self.inner.embed_query(text)
+    }
+    fn embed_code_query(&mut self, text: &str) -> Result<Vec<f32>> {
+        self.inner.embed_code_query(text)
+    }
+    fn embed_images(&mut self, paths: &[&Path], budget: u32) -> Result<Vec<Vec<f32>>> {
+        self.calls.lock().unwrap().images += paths.len();
+        self.inner.embed_images_with_budget(paths, budget)
+    }
+    fn embed_audio(&mut self, pcm: &[f32]) -> Result<Vec<f32>> {
+        self.calls.lock().unwrap().audio += 1;
+        self.inner.embed_audio_pcm(pcm)
+    }
+    fn worker_info(&self) -> Option<searchi_semantic::WorkerInfo> {
+        Some(self.inner.info.clone())
+    }
+}
+fn google(db: &Path) -> Result<(HybridSearch, Arc<Mutex<Calls>>)> {
+    assert_eq!(std::env::var("SEARCHI_FORCE_CPU")?, "1");
+    let model = std::env::var_os("SEARCHI_GEMMA_MODEL_DIR")
+        .expect("Verified Google model directory required");
+    let worker =
+        std::env::var_os("SEARCHI_GEMMA_WORKER").expect("Normal auto runtime worker required");
+    let inner = GemmaEmbedder::load(model, worker, 1)?;
+    assert_eq!(inner.info.device, "cpu");
+    let calls = Arc::new(Mutex::new(Calls::default()));
+    let service = HybridSearch::with_provider(
+        db,
+        Box::new(CountedGoogle {
+            inner,
+            calls: calls.clone(),
+        }),
+    )?;
+    Ok((service, calls))
+}
+type Chunk = (String, String, i64, i64, i64, String, Vec<u8>);
+fn chunks(db: &Path) -> Result<Vec<Chunk>> {
+    let connection = Connection::open(db)?;
+    let mut query = connection.prepare("SELECT path,kind,number,start_byte,end_byte,snippet,vector FROM semantic_chunks WHERE generation=(SELECT active_generation FROM semantic_meta WHERE singleton=1) ORDER BY path,ordinal")?;
+    let rows = query.query_map([], |r| {
+        Ok((
+            r.get(0)?,
+            r.get(1)?,
+            r.get(2)?,
+            r.get(3)?,
+            r.get(4)?,
+            r.get(5)?,
+            r.get(6)?,
+        ))
+    })?;
+    Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
+}
+
+#[test]
+#[ignore = "Requires verified Google assets and the normal native worker; CPU only"]
+fn genuine_google_reuses_text_code_and_remaps_moves_but_reencodes_titles() -> Result<()> {
+    let temp = tempfile::tempdir()?;
+    let root = temp.path().join("sources");
+    fs::create_dir(&root)?;
+    let document = root.join("agreement.txt");
+    fs::write(
+        &document,
+        "The supplier partnership may be ended with sixty days written notice.",
+    )?;
+    fs::write(
+        root.join("retry.rs"),
+        "pub fn exponential_backoff(attempt: u32) -> u64 { 100 * 2_u64.pow(attempt.min(8)) }\n",
+    )?;
+    let db = temp.path().join("index.db");
+    let mut lexical = SearchIndex::open(&db)?;
+    lexical.index_folder(&root)?;
+    let (mut service, calls) = google(&db)?;
+    service.index_embeddings()?;
+    let first = chunks(&db)?;
+    assert_eq!(first.len(), 2);
+    let initial = calls.lock().unwrap().clone();
+    assert_eq!(initial.documents, 2);
+    service.index_embeddings()?;
+    assert_eq!(
+        *calls.lock().unwrap(),
+        initial,
+        "Unchanged groups must use zero inference"
+    );
+    assert_eq!(
+        chunks(&db)?,
+        first,
+        "Real vector bytes and locations must survive reuse"
+    );
+
+    let moved = root.join("nested");
+    fs::create_dir(&moved)?;
+    fs::rename(&document, moved.join("agreement.txt"))?;
+    lexical.index_folder(&root)?;
+    service.index_embeddings()?;
+    assert_eq!(
+        *calls.lock().unwrap(),
+        initial,
+        "Folder-only moves retain the Google title"
+    );
+    let after_move = chunks(&db)?;
+    let original = first
+        .iter()
+        .find(|c| c.0.ends_with("agreement.txt"))
+        .unwrap();
+    let relocated = after_move
+        .iter()
+        .find(|c| c.0.ends_with("agreement.txt"))
+        .unwrap();
+    assert_eq!(relocated.6, original.6);
+    assert!(Path::new(&relocated.0).starts_with(moved.canonicalize()?));
+    let results = service.search_semantic(
+        "How can we stop working with the vendor?",
+        10,
+        &SearchOptions::default(),
+    )?;
+    assert_eq!(results[0].filename, "agreement.txt");
+    assert!(results[0].source.is_some());
+
+    fs::rename(
+        moved.join("agreement.txt"),
+        moved.join("supplier-contract.txt"),
+    )?;
+    lexical.index_folder(&root)?;
+    service.index_embeddings()?;
+    assert_eq!(
+        calls.lock().unwrap().documents,
+        initial.documents + 1,
+        "Basename changes alter Google's document title prompt"
+    );
+    fs::write(
+        moved.join("supplier-contract.txt"),
+        "The supplier partnership may be ended with ninety days written notice.",
+    )?;
+    lexical.index_folder(&root)?;
+    service.index_embeddings()?;
+    assert_eq!(calls.lock().unwrap().documents, initial.documents + 2);
+    let code_before = first.iter().find(|c| c.0.ends_with("retry.rs")).unwrap();
+    let latest = chunks(&db)?;
+    assert_eq!(
+        latest.iter().find(|c| c.0.ends_with("retry.rs")).unwrap(),
+        code_before
+    );
+    Ok(())
+}
+
+fn ffmpeg(tools: &searchi_service::FfmpegTools, output: &Path, arguments: &[&str]) -> Result<()> {
+    let result = std::process::Command::new(&tools.ffmpeg)
+        .args([
+            "-hide_banner",
+            "-loglevel",
+            "error",
+            "-nostdin",
+            "-y",
+            "-threads",
+            "1",
+        ])
+        .args(arguments)
+        .arg(output)
+        .output()?;
+    anyhow::ensure!(
+        result.status.success(),
+        "CPU fixture encoding failed: {}",
+        String::from_utf8_lossy(&result.stderr)
+    );
+    Ok(())
+}
+
+#[test]
+#[ignore = "Requires verified Google assets, normal worker and CPU FFmpeg helpers"]
+fn genuine_google_mixed_media_reuse_checks_bytes_despite_identical_file_metadata() -> Result<()> {
+    let tools = searchi_service::FfmpegTools {
+        ffmpeg: std::env::var_os("SEARCHI_TEST_FFMPEG")
+            .expect("Set SEARCHI_TEST_FFMPEG")
+            .into(),
+        ffprobe: std::env::var_os("SEARCHI_TEST_FFPROBE")
+            .expect("Set SEARCHI_TEST_FFPROBE")
+            .into(),
+    };
+    let temp = tempfile::tempdir()?;
+    let root = temp.path().join("mixed");
+    fs::create_dir(&root)?;
+    fs::write(
+        root.join("agreement.txt"),
+        "The supplier may terminate the partnership with sixty days written notice.",
+    )?;
+    fs::write(
+        root.join("retry.rs"),
+        "pub fn retry_delay(attempt: u32) -> u64 { 100 * 2_u64.pow(attempt.min(8)) }\n",
+    )?;
+    let image = root.join("color.bmp");
+    ffmpeg(
+        &tools,
+        &image,
+        &[
+            "-f",
+            "lavfi",
+            "-i",
+            "color=c=red:s=64x64",
+            "-frames:v",
+            "1",
+            "-c:v",
+            "bmp",
+        ],
+    )?;
+    ffmpeg(
+        &tools,
+        &root.join("tone.wav"),
+        &[
+            "-f",
+            "lavfi",
+            "-i",
+            "sine=frequency=440:duration=1",
+            "-c:a",
+            "pcm_s16le",
+            "-ar",
+            "16000",
+            "-ac",
+            "1",
+        ],
+    )?;
+    ffmpeg(
+        &tools,
+        &root.join("scene.mp4"),
+        &[
+            "-f",
+            "lavfi",
+            "-i",
+            "color=c=blue:s=64x64:r=2:d=1",
+            "-an",
+            "-c:v",
+            "mpeg4",
+        ],
+    )?;
+    let db = temp.path().join("index.db");
+    let mut lexical = SearchIndex::open(&db)?;
+    lexical.index_folder(&root)?;
+    let (mut service, calls) = google(&db)?;
+    let limits = searchi_service::MediaLimits::default();
+    service.index_embeddings_with_media(&tools, &limits)?;
+    let first = chunks(&db)?;
+    for name in [
+        "agreement.txt",
+        "retry.rs",
+        "color.bmp",
+        "tone.wav",
+        "scene.mp4",
+    ] {
+        assert!(
+            first.iter().any(|chunk| chunk.0.ends_with(name)),
+            "Missing actual source {name}"
+        );
+    }
+    let initial = calls.lock().unwrap().clone();
+    assert_eq!(initial.documents, 2);
+    assert!(initial.images >= 2 && initial.audio >= 1);
+    service.index_embeddings_with_media(&tools, &limits)?;
+    assert_eq!(
+        *calls.lock().unwrap(),
+        initial,
+        "All five complete source groups must avoid repeated inference"
+    );
+    assert_eq!(chunks(&db)?, first);
+
+    let metadata = fs::metadata(&image)?;
+    let before = fs::read(&image)?;
+    ffmpeg(
+        &tools,
+        &image,
+        &[
+            "-f",
+            "lavfi",
+            "-i",
+            "color=c=green:s=64x64",
+            "-frames:v",
+            "1",
+            "-c:v",
+            "bmp",
+        ],
+    )?;
+    assert_eq!(fs::metadata(&image)?.len(), metadata.len());
+    assert_ne!(fs::read(&image)?, before);
+    fs::File::options()
+        .write(true)
+        .open(&image)?
+        .set_times(fs::FileTimes::new().set_modified(metadata.modified()?))?;
+    assert_eq!(fs::metadata(&image)?.modified()?, metadata.modified()?);
+    // Core intentionally sees unchanged size/mtime: semantic reuse must still
+    // hash actual media bytes before accepting a completion manifest.
+    lexical.index_folder(&root)?;
+    service.index_embeddings_with_media(&tools, &limits)?;
+    let changed = calls.lock().unwrap().clone();
+    assert_eq!(changed.documents, initial.documents);
+    assert_eq!(changed.audio, initial.audio);
+    assert_eq!(changed.images, initial.images + 1);
+    let latest = chunks(&db)?;
+    for chunk in first.iter().filter(|chunk| !chunk.0.ends_with("color.bmp")) {
+        assert!(
+            latest.contains(chunk),
+            "Unaffected vector and citation changed"
+        );
+    }
+    assert_ne!(
+        latest
+            .iter()
+            .find(|chunk| chunk.0.ends_with("color.bmp"))
+            .unwrap()
+            .6,
+        first
+            .iter()
+            .find(|chunk| chunk.0.ends_with("color.bmp"))
+            .unwrap()
+            .6
+    );
+    Ok(())
+}
+
+#[test]
+#[ignore = "Requires verified Google assets and normal native worker; CPU only"]
+fn genuine_google_activation_failure_resumes_completed_vectors_after_reopen() -> Result<()> {
+    let temp = tempfile::tempdir()?;
+    let root = temp.path().join("documents");
+    fs::create_dir(&root)?;
+    let file = root.join("agreement.txt");
+    fs::write(
+        &file,
+        "The supplier agreement ends with sixty days written notice.",
+    )?;
+    let db = temp.path().join("index.db");
+    let mut lexical = SearchIndex::open(&db)?;
+    lexical.index_folder(&root)?;
+    let (mut service, calls) = google(&db)?;
+    service.index_embeddings()?;
+    let connection = Connection::open(&db)?;
+    let active = connection.query_row(
+        "SELECT active_generation FROM semantic_meta WHERE singleton=1",
+        [],
+        |r| r.get::<_, i64>(0),
+    )?;
+    fs::write(
+        &file,
+        "The supplier agreement ends with ninety days written notice.",
+    )?;
+    lexical.index_folder(&root)?;
+    connection.execute_batch("CREATE TRIGGER reject_incremental_activation BEFORE UPDATE OF active_generation ON semantic_meta BEGIN SELECT RAISE(ABORT,'injected incremental activation failure'); END;")?;
+    let failure = service.index_embeddings().unwrap_err();
+    assert!(format!("{failure:#}").contains("injected incremental activation failure"));
+    assert_eq!(
+        connection.query_row(
+            "SELECT active_generation FROM semantic_meta WHERE singleton=1",
+            [],
+            |r| r.get::<_, i64>(0)
+        )?,
+        active
+    );
+    assert_eq!(calls.lock().unwrap().documents, 2);
+    assert!(
+        service
+            .search_semantic("supplier agreement", 10, &SearchOptions::default())?
+            .is_empty(),
+        "Changed old source cannot remain readable"
+    );
+    drop(service);
+    connection.execute_batch("DROP TRIGGER reject_incremental_activation;")?;
+    let (mut reopened, resumed_calls) = google(&db)?;
+    reopened.index_embeddings()?;
+    assert_eq!(
+        resumed_calls.lock().unwrap().documents,
+        0,
+        "Durable completed source must survive reopen after failed activation"
+    );
+    let results = reopened.search_semantic(
+        "How much notice ends the supplier agreement?",
+        10,
+        &SearchOptions::default(),
+    )?;
+    assert_eq!(results.len(), 1);
+    assert!(results[0].snippet.contains("ninety days"));
+    assert!(results[0].source.is_some());
+    Ok(())
+}

docs/INCREMENTAL-SEMANTIC-VALIDATION.md

@@ -0,0 +1,63 @@
+# Completed-source reuse validation
+
+2026-10-13. Windows VM, CPU-only inference (`SEARCHI_FORCE_CPU=1`), normal
+hardware-detecting worker, pinned Google EmbeddingGemma 2 backbone/projector,
+one inference thread. FFmpeg/ffprobe are the verified app-local helpers.
+
+## Genuine Google integration
+
+`cargo test -p searchi-service --test incremental_google_cpu --test google_cpu
+--locked -- --ignored --nocapture --test-threads=1` executes actual models.
+A counting adapter forwards every operation to `GemmaEmbedder`; no synthetic
+vectors satisfy the integration assertions.
+
+- Document and code initial encoding: two calls. Unchanged rebuild and folder
+  move: zero extra calls, bit-identical vectors, remapped current paths/locations.
+  A basename rename adds one call because the document title is in Google's
+  prompt. A single content change adds one call; unrelated code stays identical.
+- Mixed document/code/image/audio/video fixture: five source groups. Unchanged
+  rebuild makes zero additional document/image/audio encoder calls and retains
+  exact vector bytes and citations. Replacing the BMP while preserving its
+  size and exact mtime adds one image call; the other four groups stay identical.
+- Rejected activation retains the prior active generation. Reopening resumes
+  completed changed-document vectors with zero document calls; updated text
+  citations are returned after publication. Existing policy-change regression
+  covers lock admission, generation creation, repeated activation failures,
+  same-instance reader policy and resumed retry in both directions.
+
+Synthetic colors/sine waves in media fixtures prove reuse, not media relevance.
+Tests are correctness checks, not latency or throughput benchmarks.
+
+## Storage and failure boundaries
+
+Four unit regressions cover schema-6 legacy search without unsafe reuse,
+completed versus partial restart state, revision-checked publication,
+transactional completion failure and malformed stored text windows. Another
+fault-only pipeline test covers primary inference error preservation when
+partial-source cleanup also fails. Its provider returns no simulated vectors.
+
+## Executed checks
+
+- Windows workspace: 98 passed, 15 explicitly ignored; no failures.
+- Native Ubuntu 24.04 headless workspace: 98 passed, 15 explicitly ignored.
+- Windows native desktop unit suite: 17 passed.
+- Strict Windows/Linux workspace and Windows native-desktop Clippy passed;
+  Windows workspace/desktop formatting passed.
+- Final-source Windows real Google tests above: five passed. The two existing
+  lifecycle/policy tests took 25.31 s and the three incremental tests took
+  138.90 s. Concurrent compilation/resource contention was present; these
+  elapsed suite times are not application search benchmarks.
+- Final-source native Linux actual Google integration: the same five tests
+  passed; existing lifecycle/policy suite 49.77 s, incremental suite 173.59 s.
+  Ubuntu 24.04 WSL x86-64 with two allocated CPUs, one inference thread,
+  isolated verified models/normal auto worker and pinned native FFmpeg helpers.
+  Cold Clippy compilation ran concurrently; these are not latency comparisons.
+
+## Remaining gates
+
+Actual process termination during a multi-batch source, simultaneous real-model
+source mutation, wider policy changes, near-limit pruning, minimum-ABI/older-CPU
+checks and first-query/foreground latency during indexing remain open. Full file
+hash validation currently precedes ANN construction; direct file mutation during
+that construction before watcher reconciliation needs a later validation callback.
+See [INCREMENTAL-SEMANTIC.md](INCREMENTAL-SEMANTIC.md).

docs/INCREMENTAL-SEMANTIC.md

@@ -1,8 +1,13 @@
 # Incremental semantic indexing and recovery
 
-Status: implementation design and acceptance gates; not implemented. Current
+Status: completed-source reuse and compatible staging recovery implemented in
-semantic rebuilds regenerate vectors and the ANN graph. Interrupted staging is
+schema 7. Unchanged vectors are copied with current citations; the ANN graph
-discarded by the next rebuild. The previous ready generation remains searchable.
+is rebuilt. Incomplete source groups are discarded before retry. The previous
+ready generation and its reader policy remain available on failures.
+
+CPU tests use the normal hardware-detecting runtime forced to CPU.
+Recorded integration evidence and remaining gates are below. This is not a
+production readiness or media relevance claim.
 
 ## Reuse requires complete sources
 
@@ -17,7 +22,7 @@ encoder/preprocessing/sampling policy. Size and modification time alone are
 insufficient. Existing media chunks are written progressively; their presence
 does not prove that a file completed. A crash can leave an incomplete group.
 
-Add a versioned completion manifest per source group: generation, task key,
+The versioned completion manifest records per source group: generation, task key,
 source type, encoder/policy key, source hash, title hash where applicable,
 expected chunk count and completion state. Associate chunks with their group.
 Only completed groups can supply reusable vectors. Historical generations
@@ -51,20 +56,25 @@ preserve both the previous active generation and the same reader's policy.
 
 ## Acceptance gates
 
-- [ ] Migrate existing databases while preserving active-generation search.
+- [x] Schema-6 storage migration preserves active-generation search; legacy
-- [ ] Rebuild mixed text/code/image/audio/video fixtures unchanged with zero
+  chunks without completion evidence cannot supply reusable vectors.
+- [x] Rebuild mixed text/code/image/audio/video fixtures unchanged with zero
   document/media inference calls and identical vectors and source locations.
-- [ ] Change one document and one media file; re-encode only affected groups.
+- [x] Change one document and one image; re-encode only affected groups.
-- [ ] Re-encode text after a basename rename; reuse a folder-only move with
+  Image regression preserves file size and exact mtime to exercise SHA identity.
-  current citations. Verify media bytes before reusing a renamed media file.
+- [x] Re-encode text after a basename rename; reuse a folder-only move with
+  current citations.
+- [ ] Verify media bytes and current citations after a media-file rename.
 - [ ] Regenerate affected media when sampling policy changes; retain eligible
   text groups when component encoder-key reuse is implemented.
 - [ ] Terminate after a committed batch and resume without repeating completed
   inference. Never reuse or publish an unfinished source group.
 - [ ] Mutate or delete a queued/running source and prove stale vectors cannot
   activate or become readable.
-- [ ] Inject storage/activation failure, preserve existing results and resume.
+- [x] Inject activation failure, preserve existing results and resume completed
-- [ ] Use genuine Google CPU inference for these integration checks. Storage
+  staging with zero document inference; cleanup-failure unit regression retains
+  the primary error and old reader policy.
+- [x] Use genuine Google CPU inference for reuse integration checks. Storage
   fixtures alone do not establish retrieval quality or reuse equivalence.
 - [ ] Measure preparation, the first foreground query and warm p95 while
   resumed indexing runs; record actual inference/reuse counts and resource use.
@@ -72,3 +82,24 @@ preserve both the previous active generation and the same reader's policy.
 Foreground query priority and bounded persistent workers remain part of the
 first-use requirement. Faster unchanged rebuilds alone do not prove responsive
 search while indexing.
+
+## Current limits
+
+Reuse requires the complete generation policy to match. Component-level reuse
+across media sampling changes remains open. Durable staging resumes at completed
+source boundaries; a real process-kill test during a multi-batch source remains
+open. Storage tests cover discarded partial groups and corpus revision races.
+
+Full source hashes are checked before ANN construction; indexed changes during
+ANN construction are rejected by the final transaction's revision guard. A direct
+filesystem mutation during ANN construction before watcher reconciliation still
+needs a later validation callback and concurrent real-model regression.
+
+Obsolete completed groups in resumed staging are removed at final reconciliation.
+Near the 100,000-chunk ceiling, replacement work can temporarily hit the budget
+before this cleanup and fail safely. Bounded discovery/pruning before inference
+is the next improvement. Hashing, media metadata probes and graph rebuilding still
+have costs: zero encoder calls does not mean zero work or prove fast first input.
+
+Synthetic colored-image and sine-wave fixtures prove actual vector reuse and
+citation equality; they do not establish useful media relevance.

docs/ROADMAP.md

@@ -100,7 +100,7 @@ Status: implementation started; no production release yet. Windows VM tests use
   Warm p95 alone does not satisfy the first-use requirement.
 - [ ] Test migrations, interrupted work, invalid documents, inaccessible/offline roots and cloud placeholders.
 - [ ] Automated native build matrix and documented supported cross-compilation paths.
-- [x] Native Linux headless workspace tests: 93 passed, 11 explicitly ignored.
+- [x] Native Linux headless workspace tests: 98 passed, 15 explicitly ignored.
 - [x] Native Linux Google text/code/media reference parity and service
   lifecycle/rebuild recovery: seven genuine CPU integration tests passed.
 - [x] Unsigned Linux x86-64 headless archive: verify extracted checksums/modes,
@@ -143,7 +143,13 @@ Audit and measured priorities: [EMBEDDINGGEMMA2-AUDIT.md](EMBEDDINGGEMMA2-AUDIT.
 - [x] Windows x64 baseline/optimized CPU runtime selection and OS AVX-state
   checks, verified with real packaged baseline and optimized inference.
 - [ ] Validate CPU selection on older hardware and remaining architectures.
-- [ ] Incrementally reuse unchanged document/media vectors in atomic generations.
+- [x] Completed-source vector reuse in atomic schema-7 generations, verified
+  with genuine Google CPU document/code/image/audio/video fixtures.
+- [x] Resume completed staging sources after failed publication; incomplete
+  groups are purged under the OS build lock before retry.
+- [ ] Process-kill/concurrent-source integration, pre-inference pruning near
+  chunk limits, component-policy reuse and indexing/foreground measurements.
+  See [INCREMENTAL-SEMANTIC.md](INCREMENTAL-SEMANTIC.md).
 - [ ] Benchmark genuine token-budgeted multi-sequence batching and isolation.
 - [ ] Implement and measure shared foreground/indexer resource budgets/priority.
 - [ ] Attribute startup cost; measure smaller query context and selective encoders.