From 9aae3f8216d515a9aa6abf8a2e490019130e6502 Mon Sep 17 00:00:00 2001 From: karl Date: Tue, 13 Oct 2026 09:27:40 +0200 Subject: [PATCH] Prepare text search during media setup and retain query sessions through promotion --- README.md | 9 +- apps/desktop/src-tauri/src/model_download.rs | 11 +- .../src-tauri/src/semantic_commands.rs | 253 +++++++++++++++-- apps/desktop/src/main.ts | 52 +++- apps/desktop/src/search-state.ts | 10 +- apps/desktop/src/semantic-setup-state.ts | 43 +++ .../tests/semantic-setup-state.test.mjs | 53 ++++ crates/searchi-service/src/coverage_tests.rs | 260 ++++++++++++++++++ crates/searchi-service/src/lib.rs | 111 +++++++- crates/searchi-service/src/pipeline.rs | 63 ++++- crates/searchi-service/src/store.rs | 23 +- .../searchi-service/src/store/incremental.rs | 137 ++++++++- .../tests/early_text_google_cpu.rs | 197 +++++++++++++ docs/EARLY-TEXT-SETUP.md | 70 +++++ docs/ROADMAP.md | 8 +- 15 files changed, 1235 insertions(+), 65 deletions(-) create mode 100644 apps/desktop/src/semantic-setup-state.ts create mode 100644 apps/desktop/tests/semantic-setup-state.test.mjs create mode 100644 crates/searchi-service/src/coverage_tests.rs create mode 100644 crates/searchi-service/tests/early_text_google_cpu.rs create mode 100644 docs/EARLY-TEXT-SETUP.md diff --git a/README.md b/README.md index d9c1f7e..375f11d 100644 --- a/README.md +++ b/README.md @@ -131,10 +131,11 @@ Once searchi is running: **Index folder**. 2. Start with a filename or words you remember. These searches become available as files are indexed. -3. Let the local search model finish preparing in the background. First - setup downloads approximately **865 MB** and needs an internet connection. - The app shows its progress; semantic search becomes available when setup - and indexing finish. +3. Let the local search model prepare in the background. First setup downloads + approximately **865 MB** and needs an internet connection. Document and code + search by meaning can begin once the smaller text model and text index are + ready, while the larger media model continues downloading. The app shows + progress and keeps text search available if media setup needs a retry. 4. Describe what you're looking for, select a result and inspect the preview. Press **Enter** to open the selected file in its usual application. diff --git a/apps/desktop/src-tauri/src/model_download.rs b/apps/desktop/src-tauri/src/model_download.rs index fe2a687..7100cf9 100644 --- a/apps/desktop/src-tauri/src/model_download.rs +++ b/apps/desktop/src-tauri/src/model_download.rs @@ -2,6 +2,13 @@ pub use searchi_assets::TOTAL_BYTES; use std::{path::Path, sync::atomic::AtomicU64}; -pub fn provision(directory: &Path, progress: &AtomicU64) -> Result<(), String> { - searchi_assets::provision(directory, progress).map_err(|error| format!("{error:#}")) +/// Notify only after a pinned asset is byte-verified; download continues while +/// desktop orchestration prepares a separately owned query session. +pub fn provision_with_verified_assets( + directory: &Path, + progress: &AtomicU64, + callback: impl FnMut(searchi_assets::VerifiedAsset), +) -> Result<(), String> { + searchi_assets::provision_with_verified_assets(directory, progress, callback) + .map_err(|error| format!("{error:#}")) } diff --git a/apps/desktop/src-tauri/src/semantic_commands.rs b/apps/desktop/src-tauri/src/semantic_commands.rs index 37e4ec2..46e861b 100644 --- a/apps/desktop/src-tauri/src/semantic_commands.rs +++ b/apps/desktop/src-tauri/src/semantic_commands.rs @@ -2,7 +2,9 @@ use crate::{measured_outcome, model_download, AppState, Filters, MeasuredFailure}; use searchi_core::metrics::{MetricsMode, SearchResponse, TraceCollector}; use searchi_core::SearchResult; -use searchi_service::{EmbeddingReport, HybridSearch, InferenceBudget, InferenceRole}; +use searchi_service::{ + EmbeddingReport, HybridSearch, InferenceBudget, InferenceRole, SemanticCoverage, +}; use serde::Serialize; use std::{ path::PathBuf, @@ -17,6 +19,9 @@ pub struct SemanticState { pub model_directory: PathBuf, pub worker: PathBuf, verified: AtomicBool, + text_verified: AtomicBool, + media_verified: AtomicBool, + media_error: Mutex>, downloading: AtomicBool, embedding: AtomicBool, querying: AtomicBool, @@ -47,6 +52,9 @@ impl SemanticState { model_directory, worker, verified: AtomicBool::new(false), + text_verified: AtomicBool::new(false), + media_verified: AtomicBool::new(false), + media_error: Mutex::new(None), downloading: AtomicBool::new(false), embedding: AtomicBool::new(false), querying: AtomicBool::new(false), @@ -148,15 +156,95 @@ impl SemanticState { let semantic = self.clone(); tauri::async_runtime::spawn_blocking(move || { let _guard = WorkGuard(&semantic.downloading); - semantic.record_error(None); - let result = - model_download::provision(&semantic.model_directory, &semantic.downloaded_bytes); - semantic.verified.store(result.is_ok(), Ordering::Release); - let result = result.and_then(|_| semantic.prepare_query_session(database, threads)); - semantic.record_error(result.err()); + let _ = semantic.run_setup(database, threads); }); } + fn run_setup( + self: &std::sync::Arc, + database: PathBuf, + threads: usize, + ) -> Result<(), String> { + self.record_error(None); + self.verified.store(false, Ordering::Release); + self.text_verified.store(false, Ordering::Release); + self.media_verified.store(false, Ordering::Release); + if let Ok(mut error) = self.media_error.lock() { + *error = None; + } + let mut warm = None; + let provision = model_download::provision_with_verified_assets( + &self.model_directory, + &self.downloaded_bytes, + |asset| match asset { + searchi_assets::VerifiedAsset::TextBackbone => { + self.text_verified.store(true, Ordering::Release); + if !self.warmed.load(Ordering::Acquire) { + // Reserve warmup before publishing readiness to indexing admission. + self.warming.store(true, Ordering::Release); + let semantic = self.clone(); + let database = database.clone(); + warm = Some( + std::thread::Builder::new() + .name("searchi-query-prepare".into()) + .spawn(move || { + let result = semantic.prepare_query_session(database, threads); + // Report immediately: the independent projector download can + // continue for minutes after a failed query warmup. + if let Err(error) = &result { + semantic.record_error(Some(error.clone())); + } + result + }) + .map_err(|error| { + let error = + format!("Could not start query preparation: {error}"); + self.record_error(Some(error.clone())); + self.warming.store(false, Ordering::Release); + error + }), + ); + } + } + searchi_assets::VerifiedAsset::MediaProjector => { + self.media_verified.store(true, Ordering::Release); + } + }, + ); + self.verified.store(provision.is_ok(), Ordering::Release); + let prepared = warm + .map(|handle| { + handle.and_then(|handle| { + handle + .join() + .unwrap_or_else(|_| Err("Query preparation failed unexpectedly.".into())) + }) + }) + .unwrap_or(Ok(())); + self.warming.store(false, Ordering::Release); + self.finish_setup(provision, prepared) + } + + fn finish_setup( + &self, + provision: Result<(), String>, + prepared: Result<(), String>, + ) -> Result<(), String> { + if let Err(error) = &prepared { + self.record_error(Some(error.clone())); + } + if let Err(error) = &provision { + if self.text_verified.load(Ordering::Acquire) { + if let Ok(mut media_error) = self.media_error.lock() { + *media_error = Some(error.clone()); + } + } else { + self.record_error(Some(error.clone())); + } + } + prepared.and(provision) + } + /// Linearize resource reset, persisted settings and setup admission. The /// lifecycle lock never spans worker startup, inference or a service wait. pub fn change_resources( @@ -185,6 +273,23 @@ impl SemanticState { Ok(()) } + fn desired_coverage(&self) -> SemanticCoverage { + if self.verified.load(Ordering::Acquire) && self.media_tools.is_some() { + SemanticCoverage::FullMedia + } else { + SemanticCoverage::TextOnly + } + } + + fn can_index(&self, text_indexing_allowed: bool) -> bool { + self.text_verified.load(Ordering::Acquire) + && self.worker.is_file() + && self.warmed.load(Ordering::Acquire) + && !self.warming.load(Ordering::Acquire) + && !self.embedding.load(Ordering::Acquire) + && (self.desired_coverage() == SemanticCoverage::FullMedia || text_indexing_allowed) + } + fn record_error(&self, value: Option) { if let Ok(mut error) = self.error.lock() { *error = value; @@ -196,6 +301,12 @@ impl SemanticState { pub struct SemanticStatus { model: &'static str, ready: bool, + text_model_ready: bool, + media_model_ready: bool, + media_model_error: Option, + indexing_ready: bool, + coverage: Option, + requested_coverage: SemanticCoverage, model_directory: String, worker_ready: bool, downloading: bool, @@ -226,11 +337,23 @@ impl Drop for WorkGuard<'_> { pub fn semantic_status(state: State<'_, AppState>) -> Result { let semantic = &state.semantic; let worker_ready = semantic.worker.is_file(); - let index = - HybridSearch::embedding_status(&state.database).map_err(|error| error.to_string())?; + let desired = semantic.desired_coverage(); + let index = HybridSearch::embedding_status_for_coverage(&state.database, desired) + .map_err(|error| error.to_string())?; + let indexing_ready = semantic.can_index(index.text_indexing_allowed); Ok(SemanticStatus { model: "Google EmbeddingGemma 2", ready: worker_ready && semantic.warmed.load(Ordering::Acquire), + text_model_ready: semantic.text_verified.load(Ordering::Acquire), + media_model_ready: semantic.media_verified.load(Ordering::Acquire), + media_model_error: semantic + .media_error + .lock() + .ok() + .and_then(|error| error.clone()), + indexing_ready, + coverage: index.coverage, + requested_coverage: desired, model_directory: semantic.model_directory.to_string_lossy().into_owned(), worker_ready, downloading: semantic.downloading.load(Ordering::Acquire), @@ -289,13 +412,7 @@ pub async fn setup_model(state: State<'_, AppState>) -> Result<(), String> { drop(admission); tauri::async_runtime::spawn_blocking(move || { let _guard = WorkGuard(&semantic.downloading); - semantic.record_error(None); - let result = - model_download::provision(&semantic.model_directory, &semantic.downloaded_bytes); - semantic.verified.store(result.is_ok(), Ordering::Release); - let result = result.and_then(|_| semantic.prepare_query_session(database, threads)); - semantic.record_error(result.as_ref().err().cloned()); - result + semantic.run_setup(database, threads) }) .await .map_err(|error| error.to_string())? @@ -308,11 +425,13 @@ pub async fn index_embeddings(state: State<'_, AppState>) -> Result) -> Result service.index_text_embeddings_preserving_media() + .and_then(|report| report.ok_or_else(|| anyhow::anyhow!("Full media index retained while media setup finishes."))), + SemanticCoverage::FullMedia => { + let tools = semantic.media_tools.as_ref().ok_or_else(|| anyhow::anyhow!("The media decoder is missing from this installation. Repair or reinstall searchi."))?; + service.index_embeddings_with_media(tools, &semantic.media_limits) + } + } })() .map_err(|error| error.to_string()); // The publisher shares its prepared immutable ANN before activation. @@ -517,6 +640,86 @@ pub async fn hybrid_search_measured( mod tests { use super::*; + #[test] + fn text_admission_requires_verified_warm_session_and_preserves_full_index() { + let runtime = tempfile::NamedTempFile::new().unwrap(); + let state = SemanticState::new( + PathBuf::new(), + runtime.path().into(), + Err("no decoder".into()), + ); + state.text_verified.store(true, Ordering::Release); + assert!(!state.can_index(true)); + state.warmed.store(true, Ordering::Release); + state.downloading.store(true, Ordering::Release); + assert!( + state.can_index(true), + "projector download must not block verified text" + ); + assert!( + !state.can_index(false), + "existing full generation must survive repair" + ); + state.warming.store(true, Ordering::Release); + assert!(!state.can_index(true)); + state.warming.store(false, Ordering::Release); + state.embedding.store(true, Ordering::Release); + assert!(!state.can_index(true)); + state.embedding.store(false, Ordering::Release); + state.text_verified.store(false, Ordering::Release); + assert!( + !state.can_index(true), + "cached warm model cannot authorize unverified indexing" + ); + } + + #[test] + fn full_coverage_requires_pair_verification_and_decoder() { + let state = SemanticState::new(PathBuf::new(), PathBuf::new(), Err("no decoder".into())); + state.verified.store(true, Ordering::Release); + state.media_verified.store(true, Ordering::Release); + assert_eq!(state.desired_coverage(), SemanticCoverage::TextOnly); + } + + #[test] + fn projector_error_preserves_warm_text_and_is_separate_from_query_error() { + let state = SemanticState::new(PathBuf::new(), PathBuf::new(), Err("no decoder".into())); + state.text_verified.store(true, Ordering::Release); + state.warmed.store(true, Ordering::Release); + assert!(state + .finish_setup(Err("projector download failed".into()), Ok(())) + .is_err()); + assert!(state.warmed.load(Ordering::Acquire)); + assert!(state.error.lock().unwrap().is_none()); + assert_eq!( + state.media_error.lock().unwrap().as_deref(), + Some("projector download failed") + ); + assert!(!state.verified.load(Ordering::Acquire)); + } + + #[test] + fn unverified_backbone_and_warmup_errors_remain_query_errors() { + let state = SemanticState::new(PathBuf::new(), PathBuf::new(), Err("no decoder".into())); + state + .finish_setup(Err("backbone invalid".into()), Ok(())) + .unwrap_err(); + assert_eq!( + state.error.lock().unwrap().as_deref(), + Some("backbone invalid") + ); + assert!(state.media_error.lock().unwrap().is_none()); + state.text_verified.store(true, Ordering::Release); + state + .finish_setup(Ok(()), Err("worker startup failed".into())) + .unwrap_err(); + assert_eq!( + state.error.lock().unwrap().as_deref(), + Some("worker startup failed") + ); + assert!(!state.warmed.load(Ordering::Acquire)); + } + #[test] fn failed_resource_save_preserves_ready_session_and_admission() { let state = std::sync::Arc::new(SemanticState::new( diff --git a/apps/desktop/src/main.ts b/apps/desktop/src/main.ts index 3ac371c..c2f4874 100644 --- a/apps/desktop/src/main.ts +++ b/apps/desktop/src/main.ts @@ -16,6 +16,13 @@ import { SEMANTIC_SETTLE_MS, } from "./search-state"; import type { SemanticPhase, SearchFilters } from "./search-state"; +import { + semanticIndexReady, + semanticIndexKey, + semanticRetryVisible, + semanticMediaReady, +} from "./semantic-setup-state"; +import type { SemanticCoverage } from "./semantic-setup-state"; import { fileIcon, icon } from "./icons"; import { SearchMeasurement, @@ -102,6 +109,12 @@ interface SemanticStatus { stale: boolean; pending: number; warming: boolean; + text_model_ready?: boolean; + media_model_ready?: boolean; + media_model_error?: string | null; + indexing_ready?: boolean; + requested_coverage?: SemanticCoverage; + coverage?: SemanticCoverage | null; media_tools_ready?: boolean; media_setup_error?: string | null; media_failures?: { path: string; message: string }[]; @@ -160,6 +173,7 @@ let latestSearch: SearchRequest | null = null; let shownLane: "lexical" | "hybrid" = "lexical"; let folderRevision = 0; let localSemanticError: string | null = null; +let localModelSetupError: string | null = null; let querySemanticFailed = false; let previewPath: string | null = null; const previewRevision = new PreviewRevision(); @@ -297,6 +311,7 @@ function queueSemanticRefresh(delay = 1000) { async function buildSemanticIndex() { if (semanticJobActive || indexingActive) return; + if (semanticStatus && !semanticIndexReady(semanticStatus)) return; semanticJobActive = true; localSemanticError = null; element("semantic-status").textContent = @@ -330,7 +345,9 @@ async function refreshSemanticStatus() { const status = await invoke("semantic_status"); semanticStatus = status; refreshEmptySearchState(); - const error = localSemanticError || status.error; + const setupError = + status.ready && status.media_model_error ? null : localModelSetupError; + const error = localSemanticError || status.error || setupError; if (error && error !== expandedError) { element("semantic-panel").open = true; expandedError = error; @@ -363,13 +380,15 @@ async function refreshSemanticStatus() { message = "Preparing Google semantic search on your device…"; else if (status.downloading) message = - status.downloaded_bytes >= status.total_bytes - ? "Verifying the downloaded Google model…" - : `Downloading Google model · ${Math.round(status.downloaded_bytes / 1_000_000)} of ${Math.round(status.total_bytes / 1_000_000)} MB`; + status.ready && status.text_model_ready + ? `${semanticAvailable(status) ? "Text search ready" : "Text model ready"} · Preparing media model · ${Math.round(status.downloaded_bytes / 1_000_000)} of ${Math.round(status.total_bytes / 1_000_000)} MB` + : status.downloaded_bytes >= status.total_bytes + ? "Verifying the downloaded Google model…" + : `Downloading Google model · ${Math.round(status.downloaded_bytes / 1_000_000)} of ${Math.round(status.total_bytes / 1_000_000)} MB`; else if (status.embedding || semanticJobActive) message = "Building your semantic index… Keyword search remains available."; - else if (status.ready && status.chunks > 0) + else if (semanticAvailable(status) && status.chunks > 0) message = `Semantic search ready · ${status.chunks.toLocaleString()} content segments${status.stale ? " · Updates pending" : ""}${status.media_failures?.length ? ` · ${status.media_failures.length} files could not be indexed` : ""}`; else if (status.ready) message = "Google model ready. Index a folder to search by meaning."; @@ -385,20 +404,29 @@ async function refreshSemanticStatus() { ? `Media decoder setup needs attention: ${status.media_setup_error}. Text search remains available.` : status.worker_info && !status.worker_info.media_supported ? "The installed worker needs a media-capable runtime update. Text search remains available." - : status.media_tools_ready && status.worker_info?.media_supported + : semanticMediaReady(status) && + status.media_tools_ready && + status.worker_info?.media_supported ? "Image, audio, and video encoders are available locally." : "Preparing local image, audio, and video support."; element("semantic-detail").textContent += ` ${mediaDetail}`; + if (status.media_model_error) + element("semantic-detail").textContent += + ` Media model setup needs attention: ${status.media_model_error}. Text search remains available.`; if (status.media_failures?.length) { element("semantic-detail").textContent += ` ${status.media_failures.length} files could not be indexed. The first issue: ${status.media_failures[0].path}: ${status.media_failures[0].message}. Filename search remains available; rebuild after fixing the file.`; } const retry = element("retry-model"); - retry.hidden = !error || status.ready; + retry.hidden = !semanticRetryVisible({ ...status, error }); retry.disabled = working || isCaptureSurface(); const rebuild = element("rebuild-semantic"); rebuild.hidden = !status.ready || (!status.chunks && !status.pending); - rebuild.disabled = working || indexingActive || isCaptureSurface(); + rebuild.disabled = + !semanticIndexReady(status) || + semanticJobActive || + indexingActive || + isCaptureSurface(); element("cpu-threads").disabled = working || indexingActive || isCaptureSurface(); if ( @@ -406,10 +434,11 @@ async function refreshSemanticStatus() { status.ready && status.stale && status.pending > 0 && - !working && + semanticIndexReady(status) && + !semanticJobActive && !indexingActive ) { - const key = `${status.generation}:${status.pending}:${folderRevision}`; + const key = semanticIndexKey(status, folderRevision); if (automaticIndex.claim(key)) { void buildSemanticIndex(); return; @@ -1276,12 +1305,13 @@ element("retry-model").addEventListener("click", async () => { const button = element("retry-model"); button.disabled = true; localSemanticError = null; + localModelSetupError = null; element("semantic-status").textContent = "Retrying Google model setup…"; queueSemanticRefresh(); try { await invoke("setup_model"); } catch (error) { - localSemanticError = String(error); + localModelSetupError = String(error); } finally { button.disabled = false; queueSemanticRefresh(0); diff --git a/apps/desktop/src/search-state.ts b/apps/desktop/src/search-state.ts index 047261c..1042308 100644 --- a/apps/desktop/src/search-state.ts +++ b/apps/desktop/src/search-state.ts @@ -17,9 +17,15 @@ export const SEMANTIC_SETTLE_MS = 120; /** A rebuild retains the previous committed generation. Backend readiness is * withdrawn only while a replacement query graph is being prepared. */ export function semanticAvailable( - status: { ready: boolean; generation: number | null } | null, + status: { + ready: boolean; + generation: number | null; + coverage?: string | null; + } | null, ) { - return Boolean(status?.ready && status.generation !== null); + return Boolean( + status?.ready && status.generation !== null && status.coverage !== null, + ); } /** One active request per lane; intermediate keystrokes collapse to the latest. diff --git a/apps/desktop/src/semantic-setup-state.ts b/apps/desktop/src/semantic-setup-state.ts new file mode 100644 index 0000000..1252d1d --- /dev/null +++ b/apps/desktop/src/semantic-setup-state.ts @@ -0,0 +1,43 @@ +/** Setup and search readiness are independent: media downloads need not block text. */ +export type SemanticCoverage = "text_only" | "full_media"; + +export interface SemanticSetupStatus { + ready: boolean; + generation: number | null; + pending: number; + downloading: boolean; + warming: boolean; + embedding: boolean; + error: string | null; + indexing_ready?: boolean; + text_model_ready?: boolean; + media_model_ready?: boolean; + media_model_error?: string | null; + requested_coverage?: SemanticCoverage; +} + +export function semanticIndexReady(status: SemanticSetupStatus) { + return Boolean( + status.ready && + !status.warming && + !status.embedding && + (status.indexing_ready ?? !status.downloading), + ); +} + +/** Promotion must receive its own automatic attempt even with unchanged counts. */ +export function semanticIndexKey( + status: SemanticSetupStatus, + folderRevision: number, +) { + return `${status.generation}:${status.pending}:${folderRevision}:${status.requested_coverage ?? "full_media"}`; +} + +export function semanticRetryVisible(status: SemanticSetupStatus) { + return Boolean(status.media_model_error || (status.error && !status.ready)); +} + +export function semanticMediaReady(status: SemanticSetupStatus) { + // Legacy capture fixtures and older full-pair status did not expose this field. + return status.media_model_ready ?? (status.ready && !status.downloading); +} diff --git a/apps/desktop/tests/semantic-setup-state.test.mjs b/apps/desktop/tests/semantic-setup-state.test.mjs new file mode 100644 index 0000000..4e8a825 --- /dev/null +++ b/apps/desktop/tests/semantic-setup-state.test.mjs @@ -0,0 +1,53 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { semanticIndexReady, semanticIndexKey, semanticRetryVisible, semanticMediaReady } from '../src/semantic-setup-state.ts'; +import { AutomaticIndexGate, semanticAvailable } from '../src/search-state.ts'; + +const text = () => ({ ready: true, generation: 1, pending: 2, downloading: true, + warming: false, embedding: false, error: null, indexing_ready: true, + text_model_ready: true, media_model_ready: false, requested_coverage: 'text_only' }); + +test('warm verified text can index during media download but not concurrent warmup or indexing', () => { + assert.equal(semanticIndexReady(text()), true); + for (const patch of [{ready:false}, {warming:true}, {embedding:true}, {indexing_ready:false}]) + assert.equal(semanticIndexReady({...text(), ...patch}), false); +}); + +test('media repair cannot authorize a text downgrade of an existing full index', () => { + const status = {...text(), requested_coverage:'full_media', indexing_ready:false}; + assert.equal(semanticAvailable(status), true); + assert.equal(semanticIndexReady(status), false); +}); + +test('an incompatible or unprepared generation never claims semantic availability', () => { + assert.equal(semanticAvailable({...text(), coverage:null}), false); + assert.equal(semanticAvailable({...text(), coverage:'text_only'}), true); + assert.equal(semanticAvailable({...text(), coverage:'full_media'}), true); +}); + +test('coverage promotion gets one new automatic attempt without a repeated polling loop', () => { + const gate = new AutomaticIndexGate(); + const early = semanticIndexKey(text(), 4); + const promoted = semanticIndexKey({...text(), requested_coverage:'full_media'}, 4); + assert.equal(gate.claim(early), true); + assert.equal(gate.claim(early), false); + assert.equal(gate.claim(promoted), true); + assert.equal(gate.claim(promoted), false); +}); + +test('failed media setup leaves text search available and retry visible', () => { + const failed = {...text(), downloading:false, media_model_error:'Download failed'}; + assert.equal(semanticAvailable(failed), true); + assert.equal(semanticRetryVisible(failed), true); + assert.equal(semanticMediaReady(failed), false); + assert.equal(semanticRetryVisible({...text(), error:'Query failed'}), false); + assert.equal(semanticRetryVisible({...text(), ready:false, error:'Setup failed'}), true); +}); + +test('legacy full-pair status remains blocked during download and ready after setup', () => { + const legacy = {ready:true,generation:1,pending:2,downloading:true,warming:false,embedding:false,error:null}; + assert.equal(semanticIndexReady(legacy), false); + assert.equal(semanticMediaReady(legacy), false); + assert.equal(semanticIndexReady({...legacy,downloading:false}), true); + assert.equal(semanticMediaReady({...legacy,downloading:false}), true); +}); diff --git a/crates/searchi-service/src/coverage_tests.rs b/crates/searchi-service/src/coverage_tests.rs new file mode 100644 index 0000000..b796a74 --- /dev/null +++ b/crates/searchi-service/src/coverage_tests.rs @@ -0,0 +1,260 @@ +//! Coverage/publication-control tests use empty ANN graphs, never model vectors. +//! Actual semantic retrieval is separately covered by early_text_google_cpu. +use super::*; +use std::fs; + +struct RejectAllInference; +impl EmbeddingProvider for RejectAllInference { + fn fingerprint(&self) -> &str { + GEMMA_FINGERPRINT + } + fn dimensions(&self) -> usize { + GEMMA_DIMENSIONS + } + fn split_windows(&mut self, _: &str, _: &str) -> Result> { + anyhow::bail!("Unexpected tokenization") + } + fn embed_documents(&mut self, _: &[&str], _: &str) -> Result>> { + anyhow::bail!("Unexpected document inference") + } + fn embed_query(&mut self, _: &str) -> Result> { + anyhow::bail!("No fabricated query vectors") + } +} + +fn fixture() -> Result<(tempfile::TempDir, std::path::PathBuf, SearchIndex, Store)> { + let temp = tempfile::tempdir()?; + let root = temp.path().join("corpus"); + fs::create_dir(&root)?; + fs::write( + root.join("supplier.txt"), + "Supplier cancellation requires written notice.", + )?; + // Filename/coverage control only, not an image inference fixture. + fs::write(root.join("photo.bmp"), b"not an image")?; + let db = temp.path().join("index.db"); + let mut lexical = SearchIndex::open(&db)?; + lexical.index_folder(&root)?; + let store = Store::open(&db, GEMMA_DIMENSIONS)?; + Ok((temp, db, lexical, store)) +} + +fn empty_generation(store: &mut Store, policy: &str) -> Result { + let _lock = store.build_lock()?; + let generation = store.begin(policy)?; + store.activate(generation)?; + Ok(generation) +} + +#[test] +fn coverage_status_distinguishes_media_pending_and_survives_mutation() -> Result<()> { + let (temp, db, mut lexical, mut store) = fixture()?; + let before = HybridSearch::embedding_status_for_coverage(&db, SemanticCoverage::TextOnly)?; + assert_eq!(before.pending, 1); + assert_eq!(before.coverage, None); + assert_eq!(HybridSearch::embedding_status(&db)?.pending, 2); + let policy = format!("{}:text-only", generation_fingerprint(GEMMA_FINGERPRINT)); + empty_generation(&mut store, &policy)?; + let text = HybridSearch::embedding_status_for_coverage(&db, SemanticCoverage::TextOnly)?; + assert_eq!(text.coverage, Some(SemanticCoverage::TextOnly)); + assert!(!text.stale); + assert_eq!( + text.pending, 0, + "Media must not trigger endless text-only rebuilds" + ); + assert!(text.text_indexing_allowed); + let full = HybridSearch::embedding_status(&db)?; + assert!(full.stale); + assert_eq!(full.pending, 2); + fs::write( + temp.path().join("corpus/supplier.txt"), + "A revised supplier cancellation policy.", + )?; + lexical.index_folder(temp.path().join("corpus"))?; + let stale = HybridSearch::embedding_status_for_coverage(&db, SemanticCoverage::TextOnly)?; + assert!(stale.stale); + assert_eq!(stale.pending, 1); + fs::remove_file(temp.path().join("corpus/supplier.txt"))?; + lexical.index_folder(temp.path().join("corpus"))?; + assert_eq!( + HybridSearch::embedding_status_for_coverage(&db, SemanticCoverage::TextOnly)?.pending, + 0 + ); + assert_eq!(HybridSearch::embedding_status(&db)?.pending, 1); + Ok(()) +} + +#[test] +fn guarded_text_setup_preserves_full_publication_even_after_stale_ui_snapshot() -> Result<()> { + let (_temp, db, _lexical, mut publisher) = fixture()?; + let mut early = HybridSearch::with_provider(&db, Box::new(RejectAllInference))?; + assert!( + HybridSearch::embedding_status_for_coverage(&db, SemanticCoverage::TextOnly)? + .text_indexing_allowed + ); + let lock = publisher.build_lock()?; + // A competing full publisher owns the same lock. Early setup must not + // perform source discovery, inference or stage deletion while it is held. + assert!(early + .index_text_embeddings_preserving_media() + .unwrap_err() + .to_string() + .contains("Another semantic rebuild")); + let generation = publisher.begin(&generation_fingerprint(GEMMA_FINGERPRINT))?; + publisher.activate(generation)?; + drop(lock); + assert!(early.index_text_embeddings_preserving_media()?.is_none()); + let status = HybridSearch::embedding_status_for_coverage(&db, SemanticCoverage::TextOnly)?; + assert_eq!(status.generation, Some(generation)); + assert_eq!(status.coverage, Some(SemanticCoverage::FullMedia)); + assert!(!status.text_indexing_allowed); + assert!(!status.stale); + Ok(()) +} + +#[test] +fn custom_full_policy_remains_protected_while_default_status_is_stale() -> Result<()> { + let (_temp, db, _lexical, mut store) = fixture()?; + let policy = format!( + "{}:custom-240-2000-120-10000-2000-1800-14400000-17179869184", + generation_fingerprint(GEMMA_FINGERPRINT) + ); + let generation = empty_generation(&mut store, &policy)?; + let mut early = HybridSearch::with_provider(&db, Box::new(RejectAllInference))?; + assert!(early.index_text_embeddings_preserving_media()?.is_none()); + let text = HybridSearch::embedding_status_for_coverage(&db, SemanticCoverage::TextOnly)?; + assert_eq!(text.coverage, Some(SemanticCoverage::FullMedia)); + assert_eq!(text.generation, Some(generation)); + assert!(!text.stale); + assert!(!text.text_indexing_allowed); + assert!( + HybridSearch::embedding_status(&db)?.stale, + "Default full status keeps its exact-policy contract" + ); + Ok(()) +} + +#[test] +fn active_policy_transition_rejects_other_models_windows_and_dimensions() -> Result<()> { + let (_temp, db, _lexical, mut publisher) = fixture()?; + let graphs = PreparedGraphs::default(); + publisher.share_prepared_graphs(&graphs)?; + let expected = generation_fingerprint(GEMMA_FINGERPRINT); + let generation = empty_generation(&mut publisher, &format!("{expected}:text-only"))?; + let mut reader = Store::open(&db, GEMMA_DIMENSIONS)?; + reader.share_prepared_graphs(&graphs)?; + // Empty graph controls policy acceptance and prepared-graph handoff without + // inventing vectors or claiming search-quality evidence. + let mut trace = TraceCollector::new(MetricsMode::Debug); + reader.search_origin_traced( + &vec![0.0; GEMMA_DIMENSIONS], + &expected, + 5, + &SearchOptions::default(), + None, + &mut trace, + None, + )?; + let spans = trace.finish(0).trace.unwrap(); + assert!(spans + .iter() + .any(|s| s.name == "semantic.ann_load" && s.counters.get("cache_hit") == Some(&1))); + let connection = rusqlite::Connection::open(&db)?; + for policy in [ + "different-model", + &expected.replace("byte64k-overlap512", "byte64k-overlap1024"), + ] { + connection.execute( + "UPDATE semantic_generations SET fingerprint=?1 WHERE id=?2", + rusqlite::params![policy, generation], + )?; + let mut trace = TraceCollector::new(MetricsMode::Debug); + reader.search_origin_traced( + &vec![0.0; GEMMA_DIMENSIONS], + &expected, + 5, + &SearchOptions::default(), + None, + &mut trace, + None, + )?; + assert!(!trace + .finish(0) + .trace + .unwrap() + .iter() + .any(|s| s.name == "semantic.ann_load")); + assert_eq!( + HybridSearch::embedding_status_for_coverage(&db, SemanticCoverage::TextOnly)?.coverage, + None + ); + } + connection.execute( + "UPDATE semantic_generations SET fingerprint=?1,dimensions=512 WHERE id=?2", + rusqlite::params![expected, generation], + )?; + let mut trace = TraceCollector::new(MetricsMode::Debug); + reader.search_origin_traced( + &vec![0.0; GEMMA_DIMENSIONS], + &expected, + 5, + &SearchOptions::default(), + None, + &mut trace, + None, + )?; + assert!(!trace + .finish(0) + .trace + .unwrap() + .iter() + .any(|s| s.name == "semantic.ann_load")); + let incompatible = + HybridSearch::embedding_status_for_coverage(&db, SemanticCoverage::TextOnly)?; + assert_eq!(incompatible.coverage, None); + assert!( + incompatible.text_indexing_allowed, + "Wrong dimensions must not claim compatible full coverage" + ); + connection.execute( + "UPDATE semantic_generations SET dimensions=768,ann_blob=NULL WHERE id=?1", + [generation], + )?; + let unprepared = HybridSearch::embedding_status_for_coverage(&db, SemanticCoverage::TextOnly)?; + assert_eq!(unprepared.coverage, None); + assert!( + !unprepared.text_indexing_allowed, + "Missing ANN must not permit dropping otherwise compatible full media sources" + ); + Ok(()) +} + +#[test] +fn text_task_component_policy_preserves_model_and_window_identity() { + let base = generation_fingerprint(GEMMA_FINGERPRINT); + assert_eq!(policy_base(&format!("{base}:text-only")), base); + let custom = format!("{base}:custom-240-2000-120-10000-2000-1800-14400000-17179869184"); + assert_eq!(policy_base(&custom), base); + assert!(compatible_query_policy( + &format!("{base}:text-only"), + &custom + )); + assert!(compatible_query_policy(&custom, &base)); + for suffix in [":text-only-v2", ":custom-unknown", ":custom-1-2-3-4"] { + let unknown = format!("{base}{suffix}"); + assert_eq!(policy_base(&unknown), unknown); + assert!(!compatible_query_policy(&base, &unknown)); + } + assert!(!compatible_query_policy( + &base, + &base.replace("byte64k-overlap512", "byte64k-overlap1024") + )); + assert_eq!( + serde_json::to_string(&SemanticCoverage::TextOnly).unwrap(), + "\"text_only\"" + ); + assert_eq!( + serde_json::to_string(&SemanticCoverage::FullMedia).unwrap(), + "\"full_media\"" + ); +} diff --git a/crates/searchi-service/src/lib.rs b/crates/searchi-service/src/lib.rs index b5a7e35..9d2f471 100644 --- a/crates/searchi-service/src/lib.rs +++ b/crates/searchi-service/src/lib.rs @@ -53,6 +53,42 @@ fn reusable_policy(expected: &str, active: &str) -> bool { matches!(values,Ok(v) if v.len()==8 && v[0]>=2 && (5000..=30000).contains(&v[3]) && v[4]0 && v[6]<=u32::MAX as u64) } +/// Resolve compatible publication coverage without broadening model or window +/// policy compatibility. A retained reader can start under either coverage. +fn policy_base(policy: &str) -> &str { + if let Some(base) = policy.strip_suffix(":text-only") { + return base; + } + if let Some((base, _)) = policy.rsplit_once(":custom-") { + if reusable_policy(base, policy) { + return base; + } + } + policy +} + +fn compatible_query_policy(expected: &str, active: &str) -> bool { + reusable_policy(policy_base(expected), active) +} + +fn policy_coverage(expected: &str, active: &str) -> Option { + compatible_query_policy(expected, active).then(|| { + if active.ends_with(":text-only") { + SemanticCoverage::TextOnly + } else { + SemanticCoverage::FullMedia + } + }) +} + +/// Requested indexing coverage is separate from warmed model readiness. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum SemanticCoverage { + TextOnly, + FullMedia, +} + /// Model-specific query/document prompts and actual tokenizer windows belong /// to the provider. Storage and ranking never assume a particular dimension. pub trait EmbeddingProvider: Send { @@ -181,6 +217,11 @@ pub struct EmbeddingReport { #[derive(Debug, Serialize)] pub struct EmbeddingStatus { + /// Coverage of a compatible, prepared active generation, if any. + pub coverage: Option, + pub requested_coverage: SemanticCoverage, + /// Early setup must preserve an existing compatible full generation. + pub text_indexing_allowed: bool, pub failures: Vec, pub generation: Option, pub chunks: usize, @@ -236,18 +277,31 @@ impl HybridSearch { /// Cheap local status check. No model, runtime worker, or network is loaded. /// Corpus revision triggers detect file inserts, updates, and deletions. pub fn embedding_status(db: impl AsRef) -> Result { + Self::embedding_status_for_coverage(db, SemanticCoverage::FullMedia) + } + + pub fn embedding_status_for_coverage( + db: impl AsRef, + requested_coverage: SemanticCoverage, + ) -> Result { let connection = rusqlite::Connection::open_with_flags(db, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY)?; connection.busy_timeout(std::time::Duration::from_secs(5))?; - let exists: bool = connection.query_row( + let snapshot = connection.unchecked_transaction()?; + let exists: bool = snapshot.query_row( "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE name='semantic_meta')", [], |row| row.get(0), )?; - let pending: usize = - connection.query_row("SELECT (SELECT count(*) FROM passages)+(SELECT count(*) FROM files WHERE extension IN ('png','jpg','jpeg','webp','bmp','mp4','mkv','mov','avi','webm','m4v','mpeg','mpg','mp3','flac','wav','m4a','ogg','opus','aac','aiff','aif'))", [], |row| row.get(0))?; + let pending: usize = match requested_coverage { + SemanticCoverage::TextOnly => snapshot.query_row("SELECT count(*) FROM passages", [], |row| row.get(0))?, + SemanticCoverage::FullMedia => snapshot.query_row("SELECT (SELECT count(*) FROM passages)+(SELECT count(*) FROM files WHERE extension IN ('png','jpg','jpeg','webp','bmp','mp4','mkv','mov','avi','webm','m4v','mpeg','mpg','mp3','flac','wav','m4a','ogg','opus','aac','aiff','aif'))", [], |row| row.get(0))?, + }; if !exists { return Ok(EmbeddingStatus { + coverage: None, + requested_coverage, + text_indexing_allowed: true, failures: Vec::new(), generation: None, chunks: 0, @@ -256,13 +310,16 @@ impl HybridSearch { pending, }); } - let version: i64 = connection.query_row( + let version: i64 = snapshot.query_row( "SELECT schema_version FROM semantic_meta WHERE singleton=1", [], |row| row.get(0), )?; if version < 6 { return Ok(EmbeddingStatus { + coverage: None, + requested_coverage, + text_indexing_allowed: true, failures: Vec::new(), generation: None, chunks: 0, @@ -275,15 +332,15 @@ impl HybridSearch { (6..=8).contains(&version), "Unsupported semantic schema version {version}" ); - let (generation,fingerprint,revision,current):(Option,Option,Option,i64)=connection.query_row( - "SELECT g.id,g.fingerprint,CASE WHEN g.ann_blob IS NOT NULL AND g.dimensions=768 THEN g.corpus_revision END,m.corpus_revision FROM semantic_meta m LEFT JOIN semantic_generations g ON g.id=m.active_generation WHERE m.singleton=1",[], - |row| Ok((row.get(0)?,row.get(1)?,row.get(2)?,row.get(3)?)))?; - let chunks: usize = connection.query_row( + let (generation,fingerprint,revision,current,dimensions):(Option,Option,Option,i64,Option)=snapshot.query_row( + "SELECT g.id,g.fingerprint,CASE WHEN g.ann_blob IS NOT NULL AND g.dimensions=768 THEN g.corpus_revision END,m.corpus_revision,g.dimensions FROM semantic_meta m LEFT JOIN semantic_generations g ON g.id=m.active_generation WHERE m.singleton=1",[], + |row| Ok((row.get(0)?,row.get(1)?,row.get(2)?,row.get(3)?,row.get(4)?)))?; + let chunks: usize = snapshot.query_row( "SELECT count(*) FROM semantic_chunks WHERE generation=?1", [generation], |row| row.get(0), )?; - let mut failures_query=connection.prepare("SELECT path,message FROM semantic_media_errors WHERE generation=?1 ORDER BY path LIMIT 1000")?; + let mut failures_query=snapshot.prepare("SELECT path,message FROM semantic_media_errors WHERE generation=?1 ORDER BY path LIMIT 1000")?; let failures = failures_query .query_map([generation], |r| { Ok(MediaFailure { @@ -292,9 +349,25 @@ impl HybridSearch { }) })? .collect::>>()?; - let stale = revision != Some(current) - || fingerprint.as_deref() != Some(generation_fingerprint(GEMMA_FINGERPRINT).as_str()); + let expected = generation_fingerprint(GEMMA_FINGERPRINT); + let coverage = if dimensions == Some(GEMMA_DIMENSIONS) { + fingerprint + .as_deref() + .and_then(|active| policy_coverage(&expected, active)) + } else { + None + }; + let prepared_coverage = if revision.is_some() { coverage } else { None }; + let policy_matches = match requested_coverage { + SemanticCoverage::TextOnly => prepared_coverage.is_some(), + // Keep the default full-media status's exact configured policy rule. + SemanticCoverage::FullMedia => fingerprint.as_deref() == Some(expected.as_str()), + }; + let stale = revision != Some(current) || !policy_matches; Ok(EmbeddingStatus { + coverage: prepared_coverage, + requested_coverage, + text_indexing_allowed: coverage != Some(SemanticCoverage::FullMedia), failures, generation, chunks, @@ -374,6 +447,19 @@ impl HybridSearch { }) } + /// Early text setup never replaces a compatible full generation. The + /// decision is made under the publication build lock, not from UI status. + /// `None` means full-media indexing must resume after its assets are ready. + pub fn index_text_embeddings_preserving_media(&mut self) -> Result> { + let fingerprint = format!( + "{}:text-only", + generation_fingerprint(self.embedder.fingerprint()) + ); + self.rebuild_generation_preserving_media(fingerprint, |service, generation, work| { + service.build(generation, work) + }) + } + /// Filter each retrieval lane before ranking, preserve filename priority, /// then fuse keyword and semantic ranks with reciprocal-rank fusion (k=60). pub fn search( @@ -827,3 +913,6 @@ mod tests { Ok(()) } } + +#[cfg(test)] +mod coverage_tests; diff --git a/crates/searchi-service/src/pipeline.rs b/crates/searchi-service/src/pipeline.rs index 508b5fc..5ebcc38 100644 --- a/crates/searchi-service/src/pipeline.rs +++ b/crates/searchi-service/src/pipeline.rs @@ -44,7 +44,7 @@ pub(super) fn text_task(passage: &IndexedPassage, policy: &str) -> Result, build: impl FnOnce(&mut Self, i64, &mut BuildWork) -> Result<()>, ) -> Result { + self.rebuild_generation_guarded(fingerprint, media_limits, false, build)? + .ok_or_else(|| anyhow::anyhow!("Explicit semantic rebuild was unexpectedly deferred")) + } + + pub(super) fn rebuild_generation_preserving_media( + &mut self, + fingerprint: String, + build: impl FnOnce(&mut Self, i64, &mut BuildWork) -> Result<()>, + ) -> Result> { + self.rebuild_generation_guarded(fingerprint, None, true, build) + } + + fn rebuild_generation_guarded( + &mut self, + fingerprint: String, + media_limits: Option<&searchi_media::MediaLimits>, + preserve_full_media: bool, + build: impl FnOnce(&mut Self, i64, &mut BuildWork) -> Result<()>, + ) -> Result> { let lock = self.store.build_lock()?; + if preserve_full_media && self.store.has_compatible_full_generation(&fingerprint)? { + return Ok(None); + } 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 @@ -109,7 +131,7 @@ impl HybridSearch { } Ok(()) })?; - Ok(work.report) + Ok(Some(work.report)) }); if result.is_err() { self.generation_fingerprint = previous_fingerprint; @@ -356,4 +378,41 @@ mod tests { assert_eq!(staging, 1, "Resume must retain the same staging generation"); Ok(()) } + #[test] + fn discovery_and_encoding_use_the_same_normalized_text_component_identity() -> Result<()> { + let temp = tempfile::tempdir()?; + let root = temp.path().join("docs"); + std::fs::create_dir(&root)?; + std::fs::write( + root.join("notes.txt"), + "Exact source bytes before a deliberately failed encoder.", + )?; + let db = temp.path().join("index.db"); + SearchIndex::open(&db)?.index_folder(&root)?; + let mut service = HybridSearch::with_provider(&db, Box::new(RejectInference))?; + let base = generation_fingerprint(service.embedder.fingerprint()); + service.generation_fingerprint = format!("{base}:text-only"); + let discovered = service.discover_tasks(None)?.remove(0); + assert_eq!(discovered.policy, base); + let error = service + .index_text_embeddings_preserving_media() + .unwrap_err(); + assert!(format!("{error:#}").contains("injected primary inference failure")); + let connection = rusqlite::Connection::open(&db)?; + let encoded: (String, String, String, String) = connection.query_row( + "SELECT task_key,source_hash,title_hash,policy FROM semantic_tasks", + [], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), + )?; + assert_eq!( + encoded, + ( + discovered.key, + discovered.source_hash, + discovered.title_hash, + discovered.policy + ) + ); + Ok(()) + } } diff --git a/crates/searchi-service/src/store.rs b/crates/searchi-service/src/store.rs index c60356a..8ebeb82 100644 --- a/crates/searchi-service/src/store.rs +++ b/crates/searchi-service/src/store.rs @@ -82,6 +82,15 @@ impl Store { Ok(()) } + pub(crate) fn has_compatible_full_generation(&self, expected: &str) -> Result { + let active: Option = self.connection.query_row( + "SELECT g.fingerprint FROM semantic_meta m JOIN semantic_generations g ON g.id=m.active_generation WHERE m.singleton=1 AND g.dimensions=?1", + [self.dimensions as i64], |row| row.get(0)).optional()?; + Ok(active.as_deref().is_some_and(|active| { + super::policy_coverage(expected, active) == Some(super::SemanticCoverage::FullMedia) + })) + } + pub(crate) fn active_fingerprint(&self) -> Result> { Ok(self.connection.query_row("SELECT g.fingerprint FROM semantic_meta m JOIN semantic_generations g ON g.id=m.active_generation WHERE m.singleton=1",[],|r|r.get(0)).optional()?) } @@ -525,11 +534,16 @@ impl Store { // A consistent read snapshot covers eligibility and materialization. let generation_span = trace.begin("semantic.generation_snapshot", parent); let snapshot = self.connection.unchecked_transaction()?; - let active:Option=snapshot.query_row("SELECT g.id FROM semantic_generations g JOIN semantic_meta m ON m.active_generation=g.id WHERE g.fingerprint=?1 AND g.dimensions=?2", - params![fingerprint,self.dimensions as i64],|row|row.get(0)).optional()?; - let Some(active) = active else { + let active:Option<(i64,String)>=snapshot.query_row("SELECT g.id,g.fingerprint FROM semantic_generations g JOIN semantic_meta m ON m.active_generation=g.id WHERE g.dimensions=?1", + [self.dimensions as i64],|row|Ok((row.get(0)?,row.get(1)?))).optional()?; + let Some((active, active_fingerprint)) = active else { + trace.end(generation_span, Default::default()); return Ok(Vec::new()); }; + if !super::compatible_query_policy(fingerprint, &active_fingerprint) { + trace.end(generation_span, Default::default()); + return Ok(Vec::new()); + } trace.end(generation_span, Default::default()); if !self .ann_cache @@ -542,7 +556,8 @@ impl Store { [active], |row| row.get(0), )?; - self.prepared_graphs.get(active, fingerprint, &checksum) + self.prepared_graphs + .get(active, &active_fingerprint, &checksum) })?; if let Some(graph) = prepared { self.ann_cache = Some((active, graph)); diff --git a/crates/searchi-service/src/store/incremental.rs b/crates/searchi-service/src/store/incremental.rs index 9f4df1b..36cc17e 100644 --- a/crates/searchi-service/src/store/incremental.rs +++ b/crates/searchi-service/src/store/incremental.rs @@ -398,11 +398,48 @@ impl Store { generation: i64, task: &SourceTask, ) -> Result> { + // Cross-coverage reuse is admitted only for the active generation, + // with at most two metadata identities (active/current staging). Exact + // generation-policy reuse retains the existing all-generation lookup. + // Validate before LIMIT, not by scanning/rejecting matching task rows. + let mut metadata=self.connection.prepare("SELECT g.id,g.fingerprint,g.dimensions,d.fingerprint,d.dimensions FROM semantic_generations d JOIN semantic_generations g ON (g.id=d.id OR g.id=(SELECT active_generation FROM semantic_meta WHERE singleton=1)) WHERE d.id=?1")?; + let generations = metadata.query_map([generation], |row| { + Ok(( + row.get::<_, i64>(0)?, + row.get::<_, String>(1)?, + row.get::<_, usize>(2)?, + row.get::<_, String>(3)?, + row.get::<_, usize>(4)?, + )) + })?; + let mut admitted = Vec::with_capacity(2); + for candidate in generations { + let (id, source_policy, source_dimensions, destination_policy, destination_dimensions) = + candidate?; + let compatible = source_dimensions == destination_dimensions + && (source_policy == destination_policy + || (task.source_type == "text" + && super::super::policy_base(&destination_policy) == task.policy + && super::super::reusable_policy(&task.policy, &destination_policy) + && super::super::reusable_policy(&task.policy, &source_policy))); + if compatible { + admitted.push(id); + } + } + ensure!( + admitted.len() <= 2, + "Donor generation admission exceeds resource budget" + ); + if admitted.is_empty() { + return Ok(None); + } + let first = admitted[0]; + let second = admitted.get(1).copied().unwrap_or(first); 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 (g.fingerprint=destination.fingerprint OR (?5='text' AND t.generation IN (?6,?7))) 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()?) + ORDER BY (t.generation=?1) DESC,t.generation DESC LIMIT 1",params![generation,task.source_hash,task.title_hash,task.policy,task.source_type,first,second],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?))).optional()?) } /// No tokenizer or neural call: only a completed, policy-compatible group @@ -910,6 +947,102 @@ mod tests { ); Ok(()) } + #[test] + fn text_donor_coverage_admission_preserves_strict_identity_and_exact_legacy_lookup( + ) -> Result<()> { + // Storage geometry only; actual Google promotion is tested separately. + let dir = tempfile::tempdir()?; + let root = dir.path().join("docs"); + std::fs::create_dir(&root)?; + std::fs::write( + root.join("notes.txt"), + "A real source with stable exact bytes.", + )?; + let db = dir.path().join("index.db"); + let mut lexical = SearchIndex::open(&db)?; + lexical.index_folder(&root)?; + let passage = lexical.passage_batch(0, 1)?.remove(0); + let mut source_task = task(&passage); + let base = "storage-model:window-v1"; + source_task.policy = base.into(); + let mut store = Store::open(&db, 2)?; + let lock = store.build_lock()?; + let source = store.begin(&format!("{base}:text-only"))?; + insert_complete(&mut store, source, &passage, &source_task)?; + store.activate(source)?; + let destination = store.start_or_resume(base, &lock)?.0; + store.stage_task(destination, &source_task)?; + assert_eq!( + store.reusable_task(destination, &source_task)?.unwrap().0, + source + ); + for wrong in [ + "different-model:window-v1:text-only", + "storage-model:window-v2:text-only", + "storage-model:window-v1:custom-unknown", + ] { + store.connection.execute( + "UPDATE semantic_generations SET fingerprint=?1 WHERE id=?2", + params![wrong, source], + )?; + assert!( + store.reusable_task(destination, &source_task)?.is_none(), + "Forged identical task policy must not admit generation {wrong}" + ); + } + let custom = format!("{base}:custom-240-2000-120-10000-2000-1800-14400000-17179869184"); + store.connection.execute( + "UPDATE semantic_generations SET fingerprint=?1 WHERE id=?2", + params![custom, source], + )?; + assert_eq!( + store.reusable_task(destination, &source_task)?.unwrap().0, + source + ); + store.connection.execute( + "UPDATE semantic_generations SET dimensions=3 WHERE id=?1", + [source], + )?; + assert!(store.reusable_task(destination, &source_task)?.is_none()); + store.connection.execute( + "UPDATE semantic_generations SET dimensions=2,fingerprint=?1 WHERE id=?2", + params![format!("{base}:text-only"), source], + )?; + let mut media_task = source_task.clone(); + media_task.key = "media-control".into(); + media_task.source_type = "media".into(); + media_task.passage_id = -1; + store.stage_task(source, &media_task)?; + store.complete_source(source, &media_task, 0)?; + assert!( + store.reusable_task(destination, &media_task)?.is_none(), + "Media donors still require exact generation policy" + ); + // Exact-policy donors outside the active/current pair remain eligible. + store.connection.execute("INSERT INTO semantic_generations(fingerprint,dimensions,corpus_revision) SELECT ?1,2,corpus_revision FROM semantic_meta WHERE singleton=1",[base])?; + let older = store.connection.last_insert_rowid(); + insert_complete(&mut store, older, &passage, &source_task)?; + assert_eq!( + store.reusable_task(destination, &source_task)?.unwrap().0, + older + ); + assert_eq!( + store.reuse_text_source(destination, &source_task, &passage)?, + Some(1) + ); + let before: Vec = store.connection.query_row( + "SELECT vector FROM semantic_chunks WHERE generation=?1", + [older], + |row| row.get(0), + )?; + let after: Vec = store.connection.query_row( + "SELECT vector FROM semantic_chunks WHERE generation=?1", + [destination], + |row| row.get(0), + )?; + assert_eq!(before, after); + Ok(()) + } } #[cfg(test)] diff --git a/crates/searchi-service/tests/early_text_google_cpu.rs b/crates/searchi-service/tests/early_text_google_cpu.rs new file mode 100644 index 0000000..a769653 --- /dev/null +++ b/crates/searchi-service/tests/early_text_google_cpu.rs @@ -0,0 +1,197 @@ +//! Genuine Google CPU proof: retained query session sees early text and full promotion. +use anyhow::Result; +use searchi_core::SearchIndex; +use searchi_semantic::{Device, GemmaEmbedder, TextWindow, WorkerInfo}; +use searchi_service::{ + EmbeddingProvider, FfmpegTools, HybridSearch, MediaLimits, PreparedGraphs, SearchOptions, + SemanticCoverage, +}; +use std::{ + collections::BTreeMap, + fs, + path::{Path, PathBuf}, + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }, +}; + +struct CountedQuery { + google: GemmaEmbedder, + calls: Arc, +} +impl EmbeddingProvider for CountedQuery { + fn fingerprint(&self) -> &str { + EmbeddingProvider::fingerprint(&self.google) + } + fn dimensions(&self) -> usize { + EmbeddingProvider::dimensions(&self.google) + } + fn split_windows(&mut self, text: &str, title: &str) -> Result> { + self.google.split_windows(text, title) + } + fn embed_documents(&mut self, texts: &[&str], title: &str) -> Result>> { + self.google.embed_documents(texts, title) + } + fn embed_query(&mut self, query: &str) -> Result> { + self.calls.fetch_add(1, Ordering::Relaxed); + self.google.embed_query(query) + } + fn embed_query_debug(&mut self, query: &str) -> Result> { + self.calls.fetch_add(1, Ordering::Relaxed); + self.google.embed_query_debug(query) + } + fn query_diagnostics(&self) -> BTreeMap { + self.google.query_diagnostics() + } + fn worker_info(&self) -> Option { + Some(self.google.info.clone()) + } +} + +fn text_vectors(db: &Path) -> Result)>> { + let connection = rusqlite::Connection::open(db)?; + let mut statement=connection.prepare("SELECT path,vector FROM semantic_chunks WHERE generation=(SELECT active_generation FROM semantic_meta WHERE singleton=1) AND source_type='text' ORDER BY path,ordinal")?; + let rows = statement + .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))? + .collect::>()?; + Ok(rows) +} + +fn assert_retained_trace(response: &searchi_service::SearchResponse) { + let trace = response.metrics.trace.as_ref().expect("Debug trace"); + assert!(trace + .iter() + .any(|span| span.counters.get("query_cache_hit") == Some(&1))); + assert!( + trace + .iter() + .any(|span| span.name == "semantic.ann_load" + && span.counters.get("cache_hit") == Some(&1)) + ); + assert!(!trace.iter().any(|span| matches!( + span.name.as_str(), + "semantic.ann_blob_read" | "semantic.ann_checksum" | "semantic.ann_deserialize" + ))); +} + +#[test] +#[ignore = "Requires verified Google pair, normal worker and CPU FFmpeg helpers"] +fn retained_query_reader_follows_early_text_and_full_media_without_reencoding_documents( +) -> Result<()> { + assert_eq!(std::env::var("SEARCHI_FORCE_CPU")?, "1"); + let models = std::env::var_os("SEARCHI_GEMMA_MODEL_DIR").expect("Verified models"); + let worker = std::env::var_os("SEARCHI_GEMMA_WORKER").expect("Normal worker"); + let tools = FfmpegTools { + ffmpeg: PathBuf::from(std::env::var_os("SEARCHI_TEST_FFMPEG").expect("CPU ffmpeg")), + ffprobe: PathBuf::from(std::env::var_os("SEARCHI_TEST_FFPROBE").expect("CPU ffprobe")), + }; + let temp = tempfile::tempdir()?; + let root = temp.path().join("corpus"); + fs::create_dir(&root)?; + let supplier = root.join("supplier.txt"); + fs::write( + &supplier, + "The supplier agreement may be terminated with sixty days written notice.", + )?; + fs::write( + root.join("garden.txt"), + "A vegetable bed and rainwater barrel are planned for the garden.", + )?; + fs::copy( + Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../searchi-semantic/tests/fixtures/digup-photo.jpg"), + root.join("photo.jpg"), + )?; + let db = temp.path().join("index.db"); + let mut lexical = SearchIndex::open(&db)?; + lexical.index_folder(&root)?; + let graphs = PreparedGraphs::default(); + let calls = Arc::new(AtomicUsize::new(0)); + let google = GemmaEmbedder::load_query_with_device(&models, &worker, 1, Device::Cpu)?; + let mut reader = HybridSearch::with_provider( + &db, + Box::new(CountedQuery { + google, + calls: calls.clone(), + }), + )?; + reader.share_prepared_graphs(&graphs)?; + reader.warmup()?; // No active generation yet. + let query = "Wie beenden wir die Zusammenarbeit mit dem Lieferanten?"; + assert!(reader + .search(query, 5, &SearchOptions::default())? + .is_empty()); + assert_eq!(calls.load(Ordering::Relaxed), 2); + let mut publisher = HybridSearch::open(&db, &models, &worker, 1)?; + publisher.share_prepared_graphs(&graphs)?; + let text = publisher + .index_text_embeddings_preserving_media()? + .expect("Initial text build"); + assert_eq!(text.document_inference_calls, 2); + assert_eq!(text.media_files, 0); + let text_status = HybridSearch::embedding_status_for_coverage(&db, SemanticCoverage::TextOnly)?; + assert!(!text_status.stale); + assert_eq!(text_status.pending, 0); + assert_eq!(text_status.coverage, Some(SemanticCoverage::TextOnly)); + assert!(HybridSearch::embedding_status(&db)?.stale); + let early = reader.search_measured(query, 5, &SearchOptions::default(), true)?; + assert_eq!(early.results[0].filename, "supplier.txt"); + assert!(early.results[0].snippet.contains("sixty days")); + assert_eq!(early.results[0].source.as_ref().unwrap().kind, "line"); + assert_retained_trace(&early); + assert_eq!(calls.load(Ordering::Relaxed), 2); + let before = text_vectors(&db)?; + let full = publisher.index_embeddings_with_media(&tools, &MediaLimits::default())?; + assert_eq!( + full.document_inference_calls, 0, + "Promotion reuses text's identical encoder/window policy" + ); + assert_eq!(full.image_inference_calls, 1); + assert_eq!( + text_vectors(&db)?, + before, + "Document vector bytes/citations survive promotion" + ); + let full_status = HybridSearch::embedding_status(&db)?; + assert_eq!(full_status.coverage, Some(SemanticCoverage::FullMedia)); + assert!(!full_status.stale); + assert!(!full_status.text_indexing_allowed); + let promoted = reader.search_measured(query, 5, &SearchOptions::default(), true)?; + assert_eq!(promoted.results[0].filename, "supplier.txt"); + assert_retained_trace(&promoted); + assert_eq!( + calls.load(Ordering::Relaxed), + 2, + "No repeated warmup, restart or cached query inference" + ); + assert_eq!(reader.worker_info().unwrap().role, "query"); + assert!(publisher + .index_text_embeddings_preserving_media()? + .is_none()); + assert_eq!( + HybridSearch::embedding_status(&db)?.generation, + Some(full.generation) + ); + fs::write( + &supplier, + "An updated supplier agreement has ninety days written notice.", + )?; + lexical.index_folder(&root)?; + let stale = reader.search_measured(query, 5, &SearchOptions::default(), true)?; + assert!( + stale + .results + .iter() + .all(|result| result.filename != "supplier.txt"), + "Old extraction/semantic chunks stay suppressed" + ); + assert_eq!(calls.load(Ordering::Relaxed), 2); + assert!( + publisher + .index_text_embeddings_preserving_media()? + .is_none(), + "Projector repair must not downgrade stale full generation" + ); + Ok(()) +} diff --git a/docs/EARLY-TEXT-SETUP.md b/docs/EARLY-TEXT-SETUP.md new file mode 100644 index 0000000..c89dc64 --- /dev/null +++ b/docs/EARLY-TEXT-SETUP.md @@ -0,0 +1,70 @@ +# Text readiness during media setup + +The desktop can prepare the real Google query session as soon as the pinned +309,855,456-byte backbone is verified. The 554,821,024-byte media projector +continues downloading independently. Verification events follow exact SHA checks +and atomic installation; byte-count progress never substitutes for verification. +The complete model bundle still requires both artifacts. + +The UI distinguishes warmed search, text/model readiness, media/model readiness +and indexing admission. A media-download error remains visible with a retry +action while the retained text query session stays usable. Worker capability +alone does not establish that the media projector is installed. + +## Index coverage and promotion + +Initial text indexing can run during projector download after query preparation. +Text-only status counts text passages, rather than repeatedly treating every +unindexed media file as a reason to rebuild. The automatic attempt key includes +requested coverage, so completed media setup can trigger full promotion. + +A retained reader selects a compatible active publication inside the same SQL +read snapshot. Prepared-graph lookup uses that publication's actual fingerprint. +The query model and task-specific vector LRU remain loaded through promotion. +Model, dimensions and text-window identity remain strict. + +Text tasks use a validated component policy shared by new text-only/full-media +generations. Cross-coverage donors are admitted only from the current active +generation after bounded metadata checks; the existing exact-policy donor +lookup across completed generations remains available. Media donors still +require the exact media generation policy. Completed chunks, exact extraction +bytes and persisted ranges must pass the existing provenance checks. +Historical text-only or custom-media text tasks using the previous +whole-generation policy can need a one-time re-encode; their lexical and +semantic readability is preserved. + +Early setup never downgrades an existing compatible full media generation, +including a stale one during projector repair. The final decision is made under +the index build lock; a stale UI status snapshot cannot bypass it. Updates resume +after media setup is ready. Explicit CLI text-only builds retain their behavior. + +## Actual validation + +Windows VM tests force CPU. The genuine Google `early_text_google_cpu` regression +passed in **42.94 seconds** with the verified normal worker, two text sources and +one real image. It opened and warmed the reader before the first publication, +then verified early remembered-text retrieval and citations, prepared ANN handoff, +full image promotion with **zero document re-encodes**, identical text-vector +bytes and retained query-cache calls. A changed source was suppressed, and the +guarded text builder preserved the full generation. This is a correctness fixture, +not a cold-download, first-input, p95 or representative-library performance result. + +Seven additional service control tests cover coverage, mutation, strict identity, +donor reuse and publication admission; they do not produce semantic-quality +evidence. Desktop state tests cover early text admission, media failure and full +index preservation. Frontend tests cover coverage promotion, retry visibility, +legacy status and incompatible-generation readiness. + +The final Windows CPU workspace run passed **137 tests**, with **24 explicitly +ignored** integration/resource checks. Desktop Rust passed **32 tests**, frontend +passed **51**, and both strict all-target Clippy checks, formatting and the +frontend production build passed. The actual Google integration above was run +separately; ignored entries in the default suite are not execution evidence. + +Native UI interaction during a genuine delayed HTTPS media download, failure, +retry and promotion remains to be executed. The prepared Linux QA plan uses only +an owned loopback CONNECT proxy and isolated app environment; it does not change +the product's pinned download URL or verification. Native Windows/macOS, GPU, +large libraries and actual first-input measurements remain open. Document indexing +still needs to finish a publishable generation; earlier model readiness does not +make a large folder's content instantaneously available. diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index f7d281e..bf5278b 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -193,8 +193,12 @@ Audit and measured priorities: [EMBEDDINGGEMMA2-AUDIT.md](EMBEDDINGGEMMA2-AUDIT. downloading; preserve verified assets and honest media setup status. - [x] Typed backbone/projector verification callbacks with full-pair setup semantics preserved; 16 asset tests passed, including gated media download. - - [ ] Integrate explicit text/media coverage, retained query readiness and - media retry status across engine/desktop; verify actual first use. + - [x] Integrate explicit text/media coverage, retained query readiness and + media retry status across engine/desktop; genuine CPU promotion reuses + identical document vectors with zero document re-encodes. + See [early text setup](EARLY-TEXT-SETUP.md). + - [ ] Verify native first use during genuine delayed media download, + failure/retry and promotion; record actual first-input timing separately. - [ ] Buffered bounded media IPC and linear UTF-8 source-window offset processing. - [ ] Evaluate 256d MRL and content-specific vision budgets on independent data. - [ ] Validate GPU precision, synchronization-aware timings, parity and fallback