karl/searchi
1
0 Fork 0
Code Issues Pull Requests Packages Projects Releases 10 Wiki Activity

Reuse completed Google embeddings and resume semantic staging

Browse Source
This commit is contained in:
karl committed 2026-10-13 00:58:57 +02:00
1 parent 0b56ab553c
commit 5c18e7b49c
11 files changed
+1699 -170
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

SKILLS.md
+11
View File
@@ -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
+15 -78
View File
@@ -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
/// storage errors leave the previous generation active. A crashed build is
/// reclaimed on the next explicit rebuild under an OS advisory lock.
/// Build or resume a replacement generation in bounded batches. Completed
/// groups can be reused; failures retain staging work and the ready reader.
pub fn index_embeddings(&mut self) -> Result<EmbeddingReport> {
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<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)
self.rebuild_generation(fingerprint, |service, generation, work| {
service.build(generation, work)
})
}
/// Filter each retrieval lane before ranking, preserve filename priority,
crates/searchi-service/src/media.rs
+151 -65
View File
@@ -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| {
let mut report = service.build(generation)?;
self.rebuild_generation(fingerprint, |service, generation, work| {
service.build(generation, work)?;
let mut after = 0;
let mut media_count = 0usize;
loop {
@@ -60,53 +62,7 @@ impl HybridSearch {
media_count <= 10000,
"Media generation exceeds 10000-file resource budget"
);
let before_chunks = report.chunks;
match service.build_media_file(
generation,
id,
path,
kind,
tools,
limits,
&mut report,
) {
Ok(()) => report.media_files += 1,
Err(error) => {
if error.downcast_ref::<InferenceFailure>().is_some() {
return Err(error);
}
service.store.media_failure(
generation,
&path.to_string_lossy(),
&format!("{error:#}"),
)?;
report.chunks = before_chunks;
if report.failures.len() < 1000 {
report.failures.push(MediaFailure {
path: path.to_string_lossy().into_owned(),
message: format!("{error:#}").chars().take(2048).collect(),
});
}
report.media_failures += 1;
}
}
}
}
Ok(report)
})
}
#[allow(clippy::too_many_arguments)]
fn build_media_file(
&mut self,
generation: i64,
id: i64,
path: &Path,
kind: MediaKind,
tools: &FfmpegTools,
limits: &MediaLimits,
report: &mut EmbeddingReport,
) -> Result<()> {
let before_chunks = work.report.chunks;
let identity = MediaIdentity::read(
path,
if kind == MediaKind::Image {
@@ -114,24 +70,119 @@ impl HybridSearch {
} else {
limits.max_file_bytes
},
);
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,
)?;
let mut ordinal = 0;
if kind == MediaKind::Image {
validate_image(path)?;
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,
&task.path,
&format!("{error:#}"),
)?;
work.report.chunks = before_chunks;
if work.report.failures.len() < 1000 {
work.report.failures.push(MediaFailure {
path: task.path,
message: format!("{error:#}").chars().take(2048).collect(),
});
}
work.report.media_failures += 1;
}
}
}
}
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!(
report.chunks < MAX_SCAN_CHUNKS,
work.report.chunks + count <= MAX_SCAN_CHUNKS,
"Media exceeds generation chunk budget"
);
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])?;
report.chunks += 1;
} else {
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()
@@ -144,7 +195,40 @@ impl HybridSearch {
metadata.duration_ms,
identity.bytes,
identity.modified_ns,
)?;
)
}
#[allow(clippy::too_many_arguments)]
fn build_media_file(
&mut self,
generation: i64,
id: i64,
path: &Path,
kind: MediaKind,
tools: &FfmpegTools,
limits: &MediaLimits,
identity: &MediaIdentity,
report: &mut EmbeddingReport,
) -> Result<()> {
let mut ordinal = 0;
if kind == MediaKind::Image {
validate_image(path)?;
ensure!(
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])?;
report.chunks += 1;
} else {
let metadata = probe(tools, path, limits)?;
self.refresh_media_metadata(path, identity, &metadata)?;
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
+272
View File
@@ -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
+28 -4
View File
@@ -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
+671
View File
@@ -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
+21 -10
View File
@@ -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");
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;")?;
assert_eq!(
generations, 2,
"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!(message.contains("rejected generation activation"));
assert!(message.contains("rejected generation cleanup"));
assert!(format!("{failure:#}").contains("rejected generation activation"));
assert_eq!(active(&connection)?, original);
assert_reader(&mut service)?;
connection.execute_batch(
"DROP TRIGGER reject_generation_activation; DROP TRIGGER reject_generation_cleanup;",
)?;
assert_eq!(
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
+417
View File
@@ -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
+63
View File
@@ -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
+42 -11
View File
@@ -1,8 +1,13 @@
# Incremental semantic indexing and recovery
Status: implementation design and acceptance gates; not implemented. Current
semantic rebuilds regenerate vectors and the ANN graph. Interrupted staging is
discarded by the next rebuild. The previous ready generation remains searchable.
Status: completed-source reuse and compatible staging recovery implemented in
schema 7. Unchanged vectors are copied with current citations; the ANN graph
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.
- [ ] Rebuild mixed text/code/image/audio/video fixtures unchanged with zero
- [x] Schema-6 storage migration preserves active-generation search; legacy
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.
- [ ] Re-encode text after a basename rename; reuse a folder-only move with
current citations. Verify media bytes before reusing a renamed media file.
- [x] Change one document and one image; re-encode only affected groups.
Image regression preserves file size and exact mtime to exercise SHA identity.
- [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.
- [ ] Use genuine Google CPU inference for these integration checks. Storage
- [x] Inject activation failure, preserve existing results and resume completed
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
+8 -2
View File
@@ -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.