diff --git a/SKILLS.md b/SKILLS.md index 2ffa3ed..bc35e55 100644 --- a/SKILLS.md +++ b/SKILLS.md @@ -21,8 +21,10 @@ searchable content or hide slow first inputs behind warm percentiles. ## Runtime and setup Use tools/embeddinggemma2/model-manifest.json and its pinned runtime builder. -Build/stage CPU workers for VM tests. Production acceleration requires separate -hardware evidence and reliable CPU fallback. Verify size/SHA-256, cache reuse, +Normal distribution builds and staging use `auto`, with GPU support and CPU +fallback in the same runtime. Force `SEARCHI_FORCE_CPU=1` for every Windows VM +inference test; explicit CPU builds are internal test tools only. Production +acceleration requires separate hardware evidence. Verify size/SHA-256, cache reuse, interrupted downloads, errors, and retries. Keep inference local. Never substitute MiniLM, simulated vectors, or hosted inference. Ship the worker with the app. diff --git a/crates/searchi-service/src/lib.rs b/crates/searchi-service/src/lib.rs index 6c01200..ca7df0c 100644 --- a/crates/searchi-service/src/lib.rs +++ b/crates/searchi-service/src/lib.rs @@ -289,22 +289,36 @@ impl HybridSearch { /// storage errors leave the previous generation active. A crashed build is /// reclaimed on the next explicit rebuild under an OS advisory lock. pub fn index_embeddings(&mut self) -> Result { - let previous_fingerprint = self.generation_fingerprint.clone(); - self.generation_fingerprint = format!( + let fingerprint = format!( "{}:text-only", generation_fingerprint(self.embedder.fingerprint()) ); + self.rebuild_generation(fingerprint, Self::build) + } + + /// 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, + ) -> Result { let _lock = self.store.build_lock()?; - let generation = self.store.begin(&self.generation_fingerprint)?; - let built = self.build(generation); + 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) => { - self.store.activate(generation)?; - Ok(report) - } + Ok(report) => Ok(report), Err(error) => { - self.store.abandon(generation)?; 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) } } diff --git a/crates/searchi-service/src/media.rs b/crates/searchi-service/src/media.rs index 3f2539d..1783b99 100644 --- a/crates/searchi-service/src/media.rs +++ b/crates/searchi-service/src/media.rs @@ -17,8 +17,7 @@ impl HybridSearch { tools: &FfmpegTools, limits: &MediaLimits, ) -> Result { - let previous_fingerprint = self.generation_fingerprint.clone(); - self.generation_fingerprint = generation_fingerprint(self.embedder.fingerprint()); + let mut fingerprint = generation_fingerprint(self.embedder.fingerprint()); let defaults = MediaLimits::default(); if limits.max_frames != defaults.max_frames || limits.frame_interval_ms != defaults.frame_interval_ms @@ -29,7 +28,7 @@ impl HybridSearch { || limits.max_duration_ms != defaults.max_duration_ms || limits.max_file_bytes != defaults.max_file_bytes { - self.generation_fingerprint.push_str(&format!( + fingerprint.push_str(&format!( ":custom-{}-{}-{}-{}-{}-{}-{}-{}", limits.max_frames, limits.frame_interval_ms, @@ -41,14 +40,12 @@ impl HybridSearch { limits.max_file_bytes )); } - let _lock = self.store.build_lock()?; - let generation = self.store.begin(&self.generation_fingerprint)?; - let built = (|| { - let mut report = self.build(generation)?; + self.rebuild_generation(fingerprint, |service, generation| { + let mut report = service.build(generation)?; let mut after = 0; let mut media_count = 0usize; loop { - let files = self.store.media_batch(after)?; + let files = service.store.media_batch(after)?; if files.is_empty() { break; } @@ -64,7 +61,7 @@ impl HybridSearch { "Media generation exceeds 10000-file resource budget" ); let before_chunks = report.chunks; - match self.build_media_file( + match service.build_media_file( generation, id, path, @@ -78,7 +75,7 @@ impl HybridSearch { if error.downcast_ref::().is_some() { return Err(error); } - self.store.media_failure( + service.store.media_failure( generation, &path.to_string_lossy(), &format!("{error:#}"), @@ -96,18 +93,7 @@ impl HybridSearch { } } Ok(report) - })(); - match built { - Ok(report) => { - self.store.activate(generation)?; - Ok(report) - } - Err(error) => { - self.store.abandon(generation)?; - self.generation_fingerprint = previous_fingerprint; - Err(error) - } - } + }) } #[allow(clippy::too_many_arguments)] diff --git a/crates/searchi-service/tests/google_cpu.rs b/crates/searchi-service/tests/google_cpu.rs index c5f9ae1..6e5b99d 100644 --- a/crates/searchi-service/tests/google_cpu.rs +++ b/crates/searchi-service/tests/google_cpu.rs @@ -78,3 +78,125 @@ fn real_google_search_persists_and_suppresses_changed_deleted_sources() -> Resul .all(|hit| hit.filename != "garden.txt")); Ok(()) } + +#[test] +#[ignore = "Requires explicitly provisioned Google model and CPU worker"] +fn real_google_failed_policy_rebuild_preserves_live_reader_and_retry() -> Result<()> { + use fs2::FileExt; + use rusqlite::Connection; + use searchi_service::{FfmpegTools, MediaLimits}; + use std::fs::OpenOptions; + + assert_eq!(std::env::var("SEARCHI_FORCE_CPU")?, "1"); + let model = std::env::var_os("SEARCHI_GEMMA_MODEL_DIR").expect("Set SEARCHI_GEMMA_MODEL_DIR"); + let worker = std::env::var_os("SEARCHI_GEMMA_WORKER").expect("Set SEARCHI_GEMMA_WORKER"); + let temp = tempfile::tempdir()?; + let root = temp.path().join("documents"); + fs::create_dir(&root)?; + let db = temp.path().join("index.db"); + fs::write( + root.join("agreement.txt"), + "The supplier partnership may be ended with sixty days written notice.", + )?; + let mut lexical = SearchIndex::open(&db)?; + lexical.index_folder(&root)?; + let mut service = HybridSearch::open(&db, model, worker, 1)?; + service.index_embeddings()?; + let connection = Connection::open(&db)?; + let active = |connection: &Connection| -> Result<(i64, String)> { + Ok(connection.query_row( + "SELECT g.id,g.fingerprint FROM semantic_generations g JOIN semantic_meta m ON m.active_generation=g.id", + [], + |row| Ok((row.get(0)?, row.get(1)?)), + )?) + }; + let assert_reader = |service: &mut HybridSearch| -> Result<()> { + let hits = service.search_semantic( + "How can we stop working with the vendor?", + 10, + &SearchOptions::default(), + )?; + assert_eq!(hits.len(), 1); + assert_eq!(hits[0].filename, "agreement.txt"); + assert!(hits[0].snippet.contains("sixty days")); + Ok(()) + }; + assert_reader(&mut service)?; + let original = active(&connection)?; + // No media files are present: exercise the production media-policy rebuild + // with genuine document inference, without invoking any decoder fixture. + let tools = FfmpegTools { + ffmpeg: std::env::current_exe()?, + ffprobe: std::env::current_exe()?, + }; + let limits = MediaLimits::default(); + let lock_path = format!("{}.semantic.lock", db.canonicalize()?.to_string_lossy()); + let lock = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(lock_path)?; + lock.try_lock_exclusive()?; + let failure = service + .index_embeddings_with_media(&tools, &limits) + .unwrap_err(); + assert!(format!("{failure:#}").contains("Another semantic rebuild")); + FileExt::unlock(&lock)?; + assert_eq!(active(&connection)?, original); + assert_reader(&mut service)?; + + connection.execute_batch("CREATE TRIGGER reject_generation_begin BEFORE INSERT ON semantic_generations BEGIN SELECT RAISE(ABORT,'rejected generation begin'); END;")?; + assert!(service + .index_embeddings_with_media(&tools, &limits) + .is_err()); + assert_eq!(active(&connection)?, original); + assert_reader(&mut service)?; + connection.execute_batch("DROP TRIGGER reject_generation_begin; + CREATE TRIGGER reject_generation_activation BEFORE UPDATE OF active_generation ON semantic_meta BEGIN SELECT RAISE(ABORT,'rejected generation activation'); END;")?; + let failure = service + .index_embeddings_with_media(&tools, &limits) + .unwrap_err(); + assert!(format!("{failure:#}").contains("rejected generation activation")); + assert_eq!(active(&connection)?, original); + assert_reader(&mut service)?; + let generations: i64 = + connection.query_row("SELECT count(*) FROM semantic_generations", [], |row| { + row.get(0) + })?; + assert_eq!(generations, 1, "Failed activation must discard staging"); + + 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;")?; + let failure = service + .index_embeddings_with_media(&tools, &limits) + .unwrap_err(); + let message = format!("{failure:#}"); + assert!(message.contains("rejected generation activation")); + assert!(message.contains("rejected generation cleanup")); + assert_eq!(active(&connection)?, original); + assert_reader(&mut service)?; + connection.execute_batch( + "DROP TRIGGER reject_generation_activation; DROP TRIGGER reject_generation_cleanup;", + )?; + let retry = service.index_embeddings_with_media(&tools, &limits)?; + assert_eq!(retry.passages, 1); + assert_reader(&mut service)?; + let media_active = active(&connection)?; + assert_ne!(media_active.1, original.1); + + // Exercise the reverse media-to-text policy change through the same helper. + lock.try_lock_exclusive()?; + assert!(service.index_embeddings().is_err()); + FileExt::unlock(&lock)?; + assert_eq!(active(&connection)?, media_active); + assert_reader(&mut service)?; + connection.execute_batch("CREATE TRIGGER reject_generation_activation BEFORE UPDATE OF active_generation ON semantic_meta BEGIN SELECT RAISE(ABORT,'rejected generation activation'); END;")?; + assert!(service.index_embeddings().is_err()); + assert_eq!(active(&connection)?, media_active); + assert_reader(&mut service)?; + connection.execute_batch("DROP TRIGGER reject_generation_activation;")?; + service.index_embeddings()?; + assert_eq!(active(&connection)?.1, original.1); + assert_reader(&mut service)?; + Ok(()) +} diff --git a/docs/SEMANTIC.md b/docs/SEMANTIC.md index 20e3d4b..4d67130 100644 --- a/docs/SEMANTIC.md +++ b/docs/SEMANTIC.md @@ -118,7 +118,10 @@ dispatches supported CPU SIMD kernels with portable fallback. The Cargo config contains the Windows NK_DYNAMIC override required for static NumKong linkage. Staging vectors and a real HNSW graph are persisted with an ANN blob SHA256. -Activation is atomic. Failed rebuilds preserve the active generation; the next +Activation is atomic. Failed rebuilds preserve the active generation and the +same search session's reader policy, including failures acquiring the build +lock, starting a generation, publishing it or discarding failed staging. The +original failure remains visible if cleanup also fails. The next rebuild reclaims abandoned staging under a cross-process lock. Search uses a consistent read snapshot. diff --git a/docs/VALIDATION.md b/docs/VALIDATION.md index f58d129..6ae4186 100644 --- a/docs/VALIDATION.md +++ b/docs/VALIDATION.md @@ -7,6 +7,16 @@ and `SEARCHI_FORCE_CPU=1`. ## Current integration checkpoint (2026-10-12) +The subsequent rebuild-policy fix passes the Windows workspace suite (93 +passed, 12 explicitly ignored) and strict service Clippy. A separately executed +genuine Google CPU regression verifies that the same reader retains its +supplier-contract result after lock, begin, activation and cleanup failures, +in both text-only and media-policy directions, and that retry succeeds. +SQLite triggers inject storage failures; actual Google document vectors supply +the searchable generation. No synthetic relevance vectors or media decoding +are used in that test. The additional ignored test accounts for the changed +default-suite ignore count. + The reliability checkpoint passes 93 workspace tests on Windows and 93 on native Linux (Ubuntu 24.04 under WSL); each explicitly ignores 11 optional integration tests. Linux uses its native Rust toolchain and an isolated ext4