From 6b78c8bb7f1e481edf5605d13e1ebee1569b3047 Mon Sep 17 00:00:00 2001 From: karl Date: Tue, 13 Oct 2026 01:36:34 +0200 Subject: [PATCH] Keep prepared search readers warm and verify abrupt indexing recovery --- AGENTS.md | 4 +- Cargo.lock | 1 + SKILLS.md | 11 + .../resources/media-runtime/manifest.json | 24 +- .../src-tauri/src/semantic_commands.rs | 32 +- crates/searchi-cli/Cargo.toml | 1 + crates/searchi-cli/tests/crash_recovery.rs | 297 ++++++++++++++++++ crates/searchi-service/src/lib.rs | 7 + crates/searchi-service/src/prepared.rs | 161 ++++++++++ crates/searchi-service/src/store.rs | 135 +++++++- .../tests/prepared_google_cpu.rs | 87 +++++ docs/BUILDING.md | 12 +- docs/PERFORMANCE.md | 16 + docs/RECOVERY-AND-HANDOFF.md | 100 ++++++ docs/ROADMAP.md | 12 +- tools/package_windows.py | 81 ++++- tools/test_package_windows.py | 71 +++++ 17 files changed, 1011 insertions(+), 41 deletions(-) create mode 100644 crates/searchi-cli/tests/crash_recovery.rs create mode 100644 crates/searchi-service/src/prepared.rs create mode 100644 crates/searchi-service/tests/prepared_google_cpu.rs create mode 100644 docs/RECOVERY-AND-HANDOFF.md diff --git a/AGENTS.md b/AGENTS.md index b10a789..a4a00f9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -42,7 +42,9 @@ readiness without completed release gates and recorded evidence. Keep running appropriate local checks and recording actual platform evidence. - Agents own assigned paths and must coordinate shared interface changes. Do not commit other agents' unfinished work. - Prepare indexes and the real Google query session in the background. Retain - warmed readers and keep keyword search usable during setup. Measure startup, + warmed readers and keep keyword search usable during setup. Register prepared + graphs before SQL activation; retain the query model/LRU across publication + instead of repeating warmup under the foreground mutex. Measure startup, actual first input, and warm p95 separately. - UI should belong to each OS: native typography/chrome/dialogs, themes, keyboard access, reduced motion and high contrast. Brief transitions must not diff --git a/Cargo.lock b/Cargo.lock index 65247f6..bd504df 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2247,6 +2247,7 @@ dependencies = [ "clap", "futures-util", "rmcp", + "rusqlite", "searchi-assets", "searchi-core", "searchi-service", diff --git a/SKILLS.md b/SKILLS.md index adba8a6..0570e97 100644 --- a/SKILLS.md +++ b/SKILLS.md @@ -80,6 +80,17 @@ SEARCHI_TEST_FFMPEG and SEARCHI_TEST_FFPROBE. Run the actual Google integration: cargo test -p searchi-service --test publication_google_cpu --test incremental_google_cpu --test google_cpu --locked -- --ignored --nocapture --test-threads=1 ``` +Abrupt process recovery uses the same verified asset/helper variables: + +```sh +cargo test -p searchi-cli --test crash_recovery --locked -- --ignored --nocapture --test-threads=1 +cargo test -p searchi-service --test prepared_google_cpu --locked -- --ignored --nocapture --test-threads=1 +``` + +These tests use actual Google CPU inference. Crash tests terminate only their +owned CLI process tree. Do not equate process termination with power-loss proof, +or tiny prepared-reader timings with large-corpus/native UI performance. + Completed-group reuse must preserve actual vector bytes and remap current source citations. Verify exact parser-input provenance and reject stale extraction before encoding; legacy records require a normal rescan. Do not substitute simulated vectors or treat zero encoder calls as a diff --git a/apps/desktop/src-tauri/resources/media-runtime/manifest.json b/apps/desktop/src-tauri/resources/media-runtime/manifest.json index 74366a0..082dd42 100644 --- a/apps/desktop/src-tauri/resources/media-runtime/manifest.json +++ b/apps/desktop/src-tauri/resources/media-runtime/manifest.json @@ -3,9 +3,25 @@ "release": "autobuild-2026-10-12-13-12", "version": "9.0.2-27-g7e93014c5b", "platforms": { - "windows-x86_64": {"archive": "ffmpeg-n9.0.2-27-g7e93014c5b-win64-lgpl-9.0.zip", "bytes": 171476229, "sha256": "104a019fe34c0747b208bde4842a0bac8b70a307cc2bb33aa4648fc15ddb2e39"}, - "windows-aarch64": {"archive": "ffmpeg-n9.0.2-27-g7e93014c5b-winarm64-lgpl-9.0.zip", "bytes": 118006941, "sha256": "68d3bac585b80b6f17e459eb4a203ed1f01b7635d4c611f1685de1ae65ca0987"}, - "linux-x86_64": {"archive": "ffmpeg-n9.0.2-27-g7e93014c5b-linux64-lgpl-9.0.tar.xz", "bytes": 137901812, "sha256": "12445b38a4700ba1ae7c9c7e57062f602c927dc4187555bfadc9f3e2f75f05b0"}, - "linux-aarch64": {"archive": "ffmpeg-n9.0.2-27-g7e93014c5b-linuxarm64-lgpl-9.0.tar.xz", "bytes": 117008524, "sha256": "c9160ae8440c17c016fb2f6b3fcce4bb9d633a15845e908b2303948a7724f227"} + "windows-x86_64": { + "archive": "ffmpeg-n9.0.2-27-g7e93014c5b-win64-lgpl-9.0.zip", + "bytes": 171476229, + "sha256": "104a019fe34c0747b208bde4842a0bac8b70a307cc2bb33aa4648fc15ddb2e39" + }, + "windows-aarch64": { + "archive": "ffmpeg-n9.0.2-27-g7e93014c5b-winarm64-lgpl-9.0.zip", + "bytes": 118006941, + "sha256": "68d3bac585b80b6f17e459eb4a203ed1f01b7635d4c611f1685de1ae65ca0987" + }, + "linux-x86_64": { + "archive": "ffmpeg-n9.0.2-27-g7e93014c5b-linux64-lgpl-9.0.tar.xz", + "bytes": 137901812, + "sha256": "12445b38a4700ba1ae7c9c7e57062f602c927dc4187555bfadc9f3e2f75f05b0" + }, + "linux-aarch64": { + "archive": "ffmpeg-n9.0.2-27-g7e93014c5b-linuxarm64-lgpl-9.0.tar.xz", + "bytes": 117008524, + "sha256": "c9160ae8440c17c016fb2f6b3fcce4bb9d633a15845e908b2303948a7724f227" + } } } diff --git a/apps/desktop/src-tauri/src/semantic_commands.rs b/apps/desktop/src-tauri/src/semantic_commands.rs index 77cba1d..63d2e5d 100644 --- a/apps/desktop/src-tauri/src/semantic_commands.rs +++ b/apps/desktop/src-tauri/src/semantic_commands.rs @@ -25,6 +25,7 @@ pub struct SemanticState { downloaded_bytes: AtomicU64, error: Mutex>, service: Mutex>, + prepared_graphs: searchi_service::PreparedGraphs, media_tools: Option, media_setup_error: Option, } @@ -51,6 +52,7 @@ impl SemanticState { downloaded_bytes: AtomicU64::new(0), error: Mutex::new(None), service: Mutex::new(None), + prepared_graphs: searchi_service::PreparedGraphs::default(), media_tools, media_setup_error, } @@ -76,6 +78,9 @@ impl SemanticState { let mut service = HybridSearch::open(database, &self.model_directory, &self.worker, threads) .map_err(|error| error.to_string())?; + service + .share_prepared_graphs(&self.prepared_graphs) + .map_err(|error| error.to_string())?; service.warmup().map_err(|error| error.to_string())?; *self .service @@ -242,36 +247,15 @@ pub async fn index_embeddings(state: State<'_, AppState>) -> Result Self { + #[cfg(unix)] + { + use std::os::unix::process::CommandExt; + // Every worker/decoder inherits this exclusive process group. + command.process_group(0); + } + command + .env("SEARCHI_FORCE_CPU", "1") + .stdin(Stdio::null()) + .stdout(File::create(stdout).unwrap()) + .stderr(File::create(stderr).unwrap()); + Self { + child: command.spawn().unwrap(), + } + } + fn terminate(&mut self) -> std::io::Result<()> { + if self.child.try_wait()?.is_some() { + return Ok(()); + } + #[cfg(windows)] + let status = Command::new("taskkill.exe") + .args(["/PID", &self.child.id().to_string(), "/T", "/F"]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status()?; + #[cfg(unix)] + let status = Command::new("/bin/kill") + .args(["-KILL", "--", &format!("-{}", self.child.id())]) + .status()?; + if !status.success() { + return Err(std::io::Error::other( + "Could not terminate owned CLI process tree", + )); + } + self.child.wait()?; + Ok(()) + } + fn finish(&mut self, timeout: Duration, stderr: &Path) { + let deadline = Instant::now() + timeout; + loop { + if let Some(status) = self.child.try_wait().unwrap() { + assert!( + status.success(), + "CLI failed: {}", + fs::read_to_string(stderr).unwrap() + ); + return; + } + assert!(Instant::now() < deadline, "Owned CLI exceeded deadline"); + thread::sleep(Duration::from_millis(20)); + } + } +} +impl Drop for OwnedCli { + fn drop(&mut self) { + let _ = self.terminate(); + } +} + +fn cli(db: &Path, command: &str) -> Command { + let mut cmd = Command::new(env!("CARGO_BIN_EXE_searchi")); + cmd.arg(command).arg("--db").arg(db); + cmd +} +fn run(command: Command, directory: &Path, label: &str) -> Value { + let stdout = directory.join(format!("{label}.stdout")); + let stderr = directory.join(format!("{label}.stderr")); + let mut child = OwnedCli::spawn(command, &stdout, &stderr); + child.finish(Duration::from_secs(900), &stderr); + serde_json::from_slice(&fs::read(stdout).unwrap()).unwrap() +} +fn embed(db: &Path, model: &Path, worker: &Path, ffmpeg: &Path, ffprobe: &Path) -> Command { + let mut command = cli(db, "embed"); + command + .arg("--model-dir") + .arg(model) + .arg("--embedding-worker") + .arg(worker) + .arg("--cpu-threads") + .arg("1") + .arg("--ffmpeg") + .arg(ffmpeg) + .arg("--ffprobe") + .arg(ffprobe); + command +} +fn index(db: &Path, root: &Path, directory: &Path) { + let stdout = directory.join("index.stdout"); + let stderr = directory.join("index.stderr"); + let mut command = cli(db, "index"); + command.arg("--root").arg(root); + let mut child = OwnedCli::spawn(command, &stdout, &stderr); + child.finish(Duration::from_secs(60), &stderr); +} +fn required(name: &str) -> PathBuf { + PathBuf::from(std::env::var_os(name).unwrap_or_else(|| panic!("{name} required"))) +} +fn active(db: &Connection) -> i64 { + db.query_row( + "SELECT active_generation FROM semantic_meta WHERE singleton=1", + [], + |r| r.get(0), + ) + .unwrap() +} +type CommittedChunk = (String, i64, String, Vec); +fn completed(db: &Connection, generation: i64) -> Vec { + let mut statement = db.prepare("SELECT c.task_key,c.ordinal,c.path,c.vector FROM semantic_chunks c JOIN semantic_tasks t ON t.generation=c.generation AND t.task_key=c.task_key WHERE c.generation=?1 AND t.state='done' ORDER BY c.task_key,c.ordinal").unwrap(); + statement + .query_map([generation], |r| { + Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)) + }) + .unwrap() + .collect::>() + .unwrap() +} +fn write_audio(path: &Path) { + // Standard PCM WAV: three overlapping audio windows under normal limits. + let samples = 16_000_u32 * 65; + let mut wav = File::create(path).unwrap(); + for bytes in [ + b"RIFF".as_slice(), + &(36 + samples * 2).to_le_bytes(), + b"WAVEfmt ", + &16_u32.to_le_bytes(), + &1_u16.to_le_bytes(), + &1_u16.to_le_bytes(), + &16_000_u32.to_le_bytes(), + &32_000_u32.to_le_bytes(), + &2_u16.to_le_bytes(), + &16_u16.to_le_bytes(), + b"data", + &(samples * 2).to_le_bytes(), + ] { + wav.write_all(bytes).unwrap(); + } + let mut pcm = Vec::with_capacity(samples as usize * 2); + for n in 0..samples { + let value = ((n as f64 * 440.0 * std::f64::consts::TAU / 16_000.0).sin() * 4096.0) as i16; + pcm.extend(value.to_le_bytes()); + } + wav.write_all(&pcm).unwrap(); +} + +#[test] +#[ignore = "Requires verified Google assets/CPU auto worker and FFmpeg; kills only owned CLI tree"] +fn real_google_cli_kill_preserves_active_and_resumes_only_complete_groups() { + assert_eq!(std::env::var("SEARCHI_FORCE_CPU").unwrap(), "1"); + let model = required("SEARCHI_GEMMA_MODEL_DIR"); + let worker = required("SEARCHI_GEMMA_WORKER"); + let ffmpeg = required("SEARCHI_TEST_FFMPEG"); + let ffprobe = required("SEARCHI_TEST_FFPROBE"); + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("sources"); + fs::create_dir(&root).unwrap(); + fs::write( + root.join("agreement.txt"), + "The supplier partnership can end with sixty days written notice.", + ) + .unwrap(); + let path = temp.path().join("index.db"); + index(&path, &root, temp.path()); + run( + embed(&path, &model, &worker, &ffmpeg, &ffprobe), + temp.path(), + "initial", + ); + let db = Connection::open(&path).unwrap(); + db.busy_timeout(Duration::from_secs(2)).unwrap(); + let original = active(&db); + let old_vectors = completed(&db, original); + fs::write( + root.join("retry.rs"), + "pub fn exponential_backoff(attempt: u32) -> u64 { 100 * 2_u64.pow(attempt.min(8)) }\n", + ) + .unwrap(); + write_audio(&root.join("tone.wav")); + index(&path, &root, temp.path()); + let stdout = temp.path().join("interrupted.stdout"); + let stderr = temp.path().join("interrupted.stderr"); + let mut child = OwnedCli::spawn( + embed(&path, &model, &worker, &ffmpeg, &ffprobe), + &stdout, + &stderr, + ); + let deadline = Instant::now() + Duration::from_secs(300); + let generation = loop { + assert!( + child.child.try_wait().unwrap().is_none(), + "CLI finished before interruption boundary: {}", + fs::read_to_string(&stderr).unwrap() + ); + let boundary: Option = db.query_row("SELECT g.id FROM semantic_generations g WHERE g.id!=?1 AND (SELECT count(*) FROM semantic_tasks t WHERE t.generation=g.id AND t.state='done')>=2 AND EXISTS(SELECT 1 FROM semantic_tasks t JOIN semantic_chunks c ON c.generation=t.generation AND c.path=t.path AND c.passage_id=t.passage_id WHERE t.generation=g.id AND t.state='running' AND t.source_type='media' AND c.task_key IS NULL AND c.source_type!='text') LIMIT 1", [original], |r| r.get(0)).ok(); + if let Some(generation) = boundary { + break generation; + } + assert!( + Instant::now() < deadline, + "No committed partial media boundary within deadline" + ); + thread::sleep(Duration::from_millis(20)); + }; + child.terminate().unwrap(); + assert_eq!( + active(&db), + original, + "Abrupt kill must not publish staging" + ); + assert_eq!(completed(&db, original), old_vectors); + let saved = completed(&db, generation); + assert_eq!(saved.len(), 2); + let partial: usize = db.query_row("SELECT count(*) FROM semantic_chunks c JOIN semantic_tasks t ON t.generation=c.generation AND t.path=c.path AND t.passage_id=c.passage_id WHERE c.generation=?1 AND t.state='running' AND t.source_type='media' AND c.task_key IS NULL AND c.source_type!='text'",[generation],|r|r.get(0)).unwrap(); + assert!( + partial > 0, + "Need actual committed incomplete Google vectors" + ); + let mut old_search = cli(&path, "search"); + old_search + .arg("How can we stop working with the vendor?") + .arg("--json") + .arg("--model-dir") + .arg(&model) + .arg("--embedding-worker") + .arg(&worker) + .arg("--cpu-threads") + .arg("1"); + let previous_results = run(old_search, temp.path(), "old-reader"); + assert_eq!(previous_results[0]["filename"], "agreement.txt"); + assert_eq!( + active(&db), + original, + "Searching must not activate unfinished work" + ); + let report = run( + embed(&path, &model, &worker, &ffmpeg, &ffprobe), + temp.path(), + "resumed", + ); + assert_eq!(report["resumed"], true); + assert_eq!(report["generation"], generation); + assert_eq!(report["document_inference_calls"], 0); + assert_eq!(report["reused_sources"], 2); + assert_eq!(report["encoded_sources"], 1); + assert_eq!( + report["audio_inference_calls"], 3, + "All incomplete media windows must be encoded afresh" + ); + assert_eq!(report["media_failures"], 0); + assert_eq!(active(&db), generation); + let after = completed(&db, generation); + for chunk in saved { + assert!( + after.contains(&chunk), + "Committed real vector bytes/citations changed" + ); + } + let (total,unique):(usize,usize)=db.query_row("SELECT count(*),count(DISTINCT task_key||':'||ordinal) FROM semantic_chunks WHERE generation=?1",[generation],|r|Ok((r.get(0)?,r.get(1)?))).unwrap(); + assert_eq!( + (total, unique), + (5, 5), + "Partial media chunks must not leak or duplicate" + ); + let mut search = cli(&path, "search"); + search + .arg("retry failed requests with exponential delays") + .arg("--code-only") + .arg("--json") + .arg("--model-dir") + .arg(&model) + .arg("--embedding-worker") + .arg(&worker) + .arg("--cpu-threads") + .arg("1"); + let results = run(search, temp.path(), "citation"); + assert_eq!(results[0]["filename"], "retry.rs"); + assert_eq!(results[0]["source"]["kind"], "line"); + assert_eq!(results[0]["source"]["number"], 1); + eprintln!("Real CPU CLI crash recovery: generation={generation}, committed_sources=2, discarded_partial_chunks={partial}, resumed_audio_calls=3"); +} diff --git a/crates/searchi-service/src/lib.rs b/crates/searchi-service/src/lib.rs index ab19193..1b2bc1b 100644 --- a/crates/searchi-service/src/lib.rs +++ b/crates/searchi-service/src/lib.rs @@ -1,6 +1,8 @@ //! Shared offline hybrid search using real Google embeddings and persisted, //! filtered HNSW retrieval. Production-scale validation remains a release gate. mod media; +mod prepared; +pub use prepared::PreparedGraphs; mod pipeline; mod routing; pub use routing::SemanticOrigin; @@ -172,6 +174,11 @@ pub struct HybridSearch { } impl HybridSearch { + /// Connect background publication to an existing warmed reader. + pub fn share_prepared_graphs(&mut self, graphs: &PreparedGraphs) -> Result<()> { + self.store.share_prepared_graphs(graphs) + } + /// Actual model/runtime/device reported by the persistent native worker. pub fn worker_info(&self) -> Option { self.embedder.worker_info() diff --git a/crates/searchi-service/src/prepared.rs b/crates/searchi-service/src/prepared.rs new file mode 100644 index 0000000..de166b6 --- /dev/null +++ b/crates/searchi-service/src/prepared.rs @@ -0,0 +1,161 @@ +//! Process-local handoff of immutable graphs prepared by background indexing. +use anyhow::{ensure, Result}; +use std::{ + collections::VecDeque, + path::{Path, PathBuf}, + sync::{Arc, Mutex}, +}; +use usearch::Index; + +/// Share between services for the same database. No model or query data is shared. +#[derive(Clone, Default)] +pub struct PreparedGraphs(Arc>); +#[derive(Default)] +struct State { + identity: Option<(PathBuf, usize)>, + entries: VecDeque, +} +#[derive(Clone)] +struct Entry { + generation: i64, + fingerprint: String, + checksum: String, + graph: Arc, +} +impl PreparedGraphs { + pub(crate) fn bind(&self, database: &Path, dimensions: usize) -> Result<()> { + let identity = (database.canonicalize()?, dimensions); + let mut state = self + .0 + .lock() + .map_err(|_| anyhow::anyhow!("Prepared graph registry unavailable"))?; + if let Some(existing) = &state.identity { + ensure!( + *existing == identity, + "Prepared graphs belong to another database or dimension" + ); + } else { + state.identity = Some(identity); + } + Ok(()) + } + pub(crate) fn get( + &self, + generation: i64, + fingerprint: &str, + checksum: &str, + ) -> Result>> { + let state = self + .0 + .lock() + .map_err(|_| anyhow::anyhow!("Prepared graph registry unavailable"))?; + Ok(state + .entries + .iter() + .find(|entry| { + entry.generation == generation + && entry.fingerprint == fingerprint + && entry.checksum == checksum + }) + .map(|entry| entry.graph.clone())) + } + pub(crate) fn publish( + &self, + generation: i64, + fingerprint: String, + checksum: String, + graph: Arc, + ) -> Result { + let mut state = self + .0 + .lock() + .map_err(|_| anyhow::anyhow!("Prepared graph registry unavailable"))?; + state.entries.push_back(Entry { + generation, + fingerprint, + checksum, + graph: graph.clone(), + }); + while state.entries.len() > 2 { + state.entries.pop_front(); + } + Ok(Publication { + registry: self.clone(), + graph, + committed: false, + }) + } +} +/// Failed SQL publication removes only this publisher's entry. +pub(crate) struct Publication { + registry: PreparedGraphs, + graph: Arc, + committed: bool, +} +impl Publication { + pub(crate) fn commit(mut self) { + self.committed = true; + } +} +impl Drop for Publication { + fn drop(&mut self) { + if !self.committed { + if let Ok(mut state) = self.registry.0.lock() { + state + .entries + .retain(|entry| !Arc::ptr_eq(&entry.graph, &self.graph)); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + fn graph() -> Arc { + Arc::new( + Index::new(&usearch::IndexOptions { + dimensions: 768, + ..Default::default() + }) + .unwrap(), + ) + } + #[test] + fn identity_checksum_and_rollback_are_enforced() -> Result<()> { + let directory = tempfile::tempdir()?; + let database = directory.path().join("index.db"); + std::fs::write(&database, [])?; + let other = directory.path().join("other.db"); + std::fs::write(&other, [])?; + let registry = PreparedGraphs::default(); + registry.bind(&database, 768)?; + assert!(registry.bind(&other, 768).is_err()); + assert!(registry.bind(&database, 256).is_err()); + let active = graph(); + registry + .publish(1, "policy".into(), "one".into(), active.clone())? + .commit(); + assert!(registry.get(1, "policy", "wrong")?.is_none()); + assert!(registry.get(1, "wrong", "one")?.is_none()); + { + let _pending = registry.publish(2, "policy".into(), "two".into(), graph())?; + assert!(registry.get(2, "policy", "two")?.is_some()); + } + assert!(registry.get(2, "policy", "two")?.is_none()); + assert!(Arc::ptr_eq( + &active, + ®istry.get(1, "policy", "one")?.unwrap() + )); + registry + .publish(2, "policy".into(), "two".into(), graph())? + .commit(); + registry + .publish(3, "policy".into(), "three".into(), graph())? + .commit(); + assert_eq!(registry.0.lock().unwrap().entries.len(), 2); + assert!(registry.get(1, "policy", "one")?.is_none()); + assert_eq!(active.dimensions(), 768); // Eviction never invalidates a retained reader. + Ok(()) + } +} diff --git a/crates/searchi-service/src/store.rs b/crates/searchi-service/src/store.rs index dc3c3ff..00ad1a2 100644 --- a/crates/searchi-service/src/store.rs +++ b/crates/searchi-service/src/store.rs @@ -45,7 +45,8 @@ pub(crate) struct Store { connection: Connection, lock_path: PathBuf, dimensions: usize, - ann_cache: Option<(i64, Index)>, + ann_cache: Option<(i64, Arc)>, + prepared_graphs: crate::PreparedGraphs, eligibility_cache: Option, } @@ -68,6 +69,19 @@ thread_local! { static HASH_CALLS:std::cell::Cell=const {std::cell::Cell: thread_local! { static MEDIA_STAT_CALLS:std::cell::Cell=const {std::cell::Cell::new(0)}; } impl Store { + pub(crate) fn share_prepared_graphs(&mut self, graphs: &crate::PreparedGraphs) -> Result<()> { + graphs.bind( + Path::new( + self.connection + .path() + .context("Database path unavailable")?, + ), + self.dimensions, + )?; + self.prepared_graphs = graphs.clone(); + Ok(()) + } + 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()?) } @@ -170,6 +184,7 @@ impl Store { lock_path, dimensions, ann_cache: None, + prepared_graphs: crate::PreparedGraphs::default(), eligibility_cache: None, }) } @@ -328,6 +343,16 @@ impl Store { // before acquiring the write transaction. The revision check below // still closes races with indexed changes after this validation. validate_sources()?; + let fingerprint: String = self.connection.query_row( + "SELECT fingerprint FROM semantic_generations WHERE id=?1", + [generation], + |row| row.get(0), + )?; + let ann = Arc::new(ann); + // Readers observing SQL publication already have a prepared immutable graph. + let publication = + self.prepared_graphs + .publish(generation, fingerprint, checksum.clone(), ann.clone())?; let transaction = self .connection .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; @@ -366,6 +391,7 @@ impl Store { transaction.execute_batch("UPDATE semantic_meta SET dirty_floor=max(dirty_floor,corpus_revision-100000) WHERE singleton=1; DELETE FROM semantic_dirty_paths WHERE revision<(SELECT dirty_floor FROM semantic_meta WHERE singleton=1);")?; transaction.commit()?; + publication.commit(); self.ann_cache = Some((generation, ann)); self.eligibility_cache = None; Ok(()) @@ -505,6 +531,23 @@ impl Store { return Ok(Vec::new()); }; trace.end(generation_span, Default::default()); + if !self + .ann_cache + .as_ref() + .is_some_and(|(generation, _)| *generation == active) + { + let prepared = trace.timed("semantic.prepared_graph_lookup", parent, |_, _| { + let checksum: String = snapshot.query_row( + "SELECT ann_sha256 FROM semantic_generations WHERE id=?1", + [active], + |row| row.get(0), + )?; + self.prepared_graphs.get(active, fingerprint, &checksum) + })?; + if let Some(graph) = prepared { + self.ann_cache = Some((active, graph)); + } + } let ann_load = trace.begin("semantic.ann_load", parent); trace.counter( ann_load, @@ -553,7 +596,7 @@ impl Store { ann.reserve_capacity_and_threads(ann.size().max(1), 1)?; trace.counter(deserialize_span, "chunks", ann.size() as u64); trace.end(deserialize_span, Default::default()); - self.ann_cache = Some((active, ann)); + self.ann_cache = Some((active, Arc::new(ann))); } trace.end(ann_load, Default::default()); let filter_span = trace.begin("semantic.filter_normalization", parent); @@ -1417,6 +1460,94 @@ mod tests { Ok(()) } + #[test] + fn failed_sql_activation_removes_prepared_pending_and_preserves_live_reader() -> Result<()> { + // Storage publication invariant only; real Google coverage is a separate integration. + let directory = tempfile::tempdir()?; + let root = directory.path().join("docs"); + std::fs::create_dir(&root)?; + std::fs::write(root.join("notes.txt"), "original relevant content")?; + let database = directory.path().join("index.db"); + let mut lexical = SearchIndex::open(&database)?; + lexical.index_folder(&root)?; + let passage = lexical.passage_batch(0, 1)?.remove(0); + let windows = [TextWindow { + text: passage.text.clone(), + start_byte: 0, + end_byte: passage.text.len(), + }]; + let mut vector = vec![0.0; DIMENSIONS]; + vector[0] = 1.0; + let policy = "test-storage-publication"; + let graphs = crate::PreparedGraphs::default(); + let mut writer = Store::open(&database, DIMENSIONS)?; + writer.share_prepared_graphs(&graphs)?; + let active = writer.begin(policy)?; + writer.insert(active, &passage, 0, &windows, &[vector.clone()])?; + writer.activate(active)?; + let checksum: String = writer.connection.query_row( + "SELECT ann_sha256 FROM semantic_generations WHERE id=?1", + [active], + |r| r.get(0), + )?; + let mut reader = Store::open(&database, DIMENSIONS)?; + reader.share_prepared_graphs(&graphs)?; + assert_eq!( + reader + .search(&vector, policy, 10, &SearchOptions::default())? + .len(), + 1 + ); + let retained = reader.ann_cache.as_ref().unwrap().1.clone(); + let pending = writer.begin(policy)?; + writer.insert(pending, &passage, 0, &windows, &[vector.clone()])?; + let prepared = writer.build_ann(pending)?; + let mut serialized = vec![0; prepared.serialized_length()]; + prepared.save_to_buffer(&mut serialized)?; + let pending_checksum = format!("{:x}", Sha256::digest(&serialized)); + writer.connection.execute_batch("CREATE TRIGGER reject_activation BEFORE UPDATE OF active_generation ON semantic_meta BEGIN SELECT RAISE(ABORT,'activation rejected'); END")?; + assert!(writer + .activate(pending) + .unwrap_err() + .to_string() + .contains("activation rejected")); + assert_eq!( + writer.connection.query_row( + "SELECT active_generation FROM semantic_meta WHERE singleton=1", + [], + |r| r.get::<_, i64>(0) + )?, + active + ); + assert!(graphs.get(pending, policy, &pending_checksum)?.is_none()); + assert!(Arc::ptr_eq( + &retained, + &graphs.get(active, policy, &checksum)?.unwrap() + )); + assert_eq!( + reader + .search(&vector, policy, 10, &SearchOptions::default())? + .len(), + 1 + ); + assert!(Arc::ptr_eq( + &retained, + &reader.ann_cache.as_ref().unwrap().1 + )); + writer + .connection + .execute_batch("DROP TRIGGER reject_activation")?; + writer.activate(pending)?; + assert_eq!( + reader + .search(&vector, policy, 10, &SearchOptions::default())? + .len(), + 1 + ); + assert_eq!(reader.ann_cache.as_ref().unwrap().0, pending); + Ok(()) + } + #[test] fn generation_persists_failed_staging_preserves_active_and_changes_hide_stale_vectors( ) -> Result<()> { diff --git a/crates/searchi-service/tests/prepared_google_cpu.rs b/crates/searchi-service/tests/prepared_google_cpu.rs new file mode 100644 index 0000000..f2870d6 --- /dev/null +++ b/crates/searchi-service/tests/prepared_google_cpu.rs @@ -0,0 +1,87 @@ +//! Genuine Google CPU proof for background prepared-graph publication. +use anyhow::Result; +use searchi_core::SearchIndex; +use searchi_service::{HybridSearch, PreparedGraphs, SearchOptions}; +use std::fs; + +#[test] +#[ignore = "Requires verified Google models and normal worker; CPU only"] +fn prepared_publication_keeps_warmed_query_reader_hot() -> Result<()> { + assert_eq!(std::env::var("SEARCHI_FORCE_CPU")?, "1"); + let model = std::env::var_os("SEARCHI_GEMMA_MODEL_DIR").expect("Verified models required"); + let worker = std::env::var_os("SEARCHI_GEMMA_WORKER").expect("Normal worker required"); + let temp = tempfile::tempdir()?; + let root = temp.path().join("documents"); + fs::create_dir(&root)?; + let file = root.join("supplier.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 graphs = PreparedGraphs::default(); + let mut indexer = HybridSearch::open(&db, &model, &worker, 1)?; + indexer.share_prepared_graphs(&graphs)?; + indexer.index_embeddings()?; + let mut reader = HybridSearch::open(&db, &model, &worker, 1)?; + reader.share_prepared_graphs(&graphs)?; + reader.warmup()?; + let query = "How can we end our supplier partnership?"; + reader.search(query, 10, &SearchOptions::default())?; + fs::write( + &file, + "The supplier agreement ends with ninety days written notice.", + )?; + lexical.index_folder(&root)?; + indexer.index_embeddings()?; + drop(indexer); // The shared graph outlives the background model worker. + let response = reader.search_measured(query, 10, &SearchOptions::default(), true)?; + assert_eq!(response.results[0].filename, "supplier.txt"); + assert!(response.results[0].snippet.contains("ninety days")); + let trace = response.metrics.trace.expect("Detailed trace required"); + assert!( + trace + .iter() + .any(|span| span.counters.get("query_cache_hit") == Some(&1)), + "The retained query model must preserve its real-vector LRU across publication" + ); + let loads: Vec<_> = trace + .iter() + .filter(|span| span.name == "semantic.ann_load") + .collect(); + assert!(!loads.is_empty()); + assert!(loads + .iter() + .all(|span| 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" + ))); + let unseen = reader.search_measured( + "What notice is required for the contract to end?", + 10, + &SearchOptions::default(), + true, + )?; + assert!(unseen.results[0].snippet.contains("ninety days")); + let unseen_trace = unseen.metrics.trace.expect("Unseen query trace required"); + assert!( + unseen_trace + .iter() + .any(|span| span.counters.get("query_cache_hit") == Some(&0)), + "A distinct foreground input must execute genuine query inference" + ); + assert!(unseen_trace + .iter() + .filter(|span| span.name == "semantic.ann_load") + .all(|span| span.counters.get("cache_hit") == Some(&1))); + assert!(!unseen_trace.iter().any(|span| matches!( + span.name.as_str(), + "semantic.ann_blob_read" | "semantic.ann_checksum" | "semantic.ann_deserialize" + ))); + eprintln!("Prepared reader CPU fixture: retained_query_ms={:.3}, unseen_query_ms={:.3}; one document, not a scale benchmark", + response.metrics.total_ms, unseen.metrics.total_ms); + Ok(()) +} diff --git a/docs/BUILDING.md b/docs/BUILDING.md index bf7a1ff..b9c8f2c 100644 --- a/docs/BUILDING.md +++ b/docs/BUILDING.md @@ -201,10 +201,13 @@ signing/notarization requires Apple tooling and credentials. See the ### Windows archive for source installations -After staging the CPU worker and media helpers, build an embedded-frontend +Stage the normal hardware-detecting worker and media helpers, then build an embedded-frontend desktop binary and collect its runtime files: ```powershell +python tools/embeddinggemma2/build_runtime.py --backend auto --jobs 2 +python tools/embeddinggemma2/stage_runtime.py --backend auto +python tools/media-runtime/provision.py --stage Set-Location apps/desktop npm run tauri -- build --no-bundle Set-Location ../.. @@ -223,6 +226,10 @@ using CPU and OS capabilities, and the verified Vulkan loader DLL and its license; GPU availability is detected at runtime, with CPU fallback. The SDK is used only to compile the app. All runtime DLLs are listed by size and SHA-256 in `runtime-libraries.json`. +Packaging also verifies the original pinned decoder ZIP and compares staged +FFmpeg/FFprobe, notices and manifest against it. Keep that verified archive in +`.searchi/media-runtime` while preparing the package; changed staged helpers +or notices are rejected before replacing a previous archive. Staging and packaging reject missing or changed listed dependencies before replacing output. CPU-only VM smoke tests can also omit optimized modules to exercise the packaged baseline; this does not certify older physical hardware. @@ -250,6 +257,7 @@ need both verified model files. See [semantic bounds and evidence](SEMANTIC.md) for current corpus, localization, and performance limitations. Keep signing keys, repository tokens, and model credentials outside source control. FFmpeg provisioning supports manifest-pinned Windows/Linux x86-64 and ARM64 artifacts; -only Windows x86-64 has executed here. macOS decoder packaging remains a gate. +Windows and Linux x86-64 startup and CPU decoding have executed here; +other architectures and macOS decoder packaging remain gates. See [media runtime provenance](../tools/media-runtime/README.md) and [measured media limitations](MEDIA-QUALITY.md). diff --git a/docs/PERFORMANCE.md b/docs/PERFORMANCE.md index 950aaf5..3217ee0 100644 --- a/docs/PERFORMANCE.md +++ b/docs/PERFORMANCE.md @@ -166,3 +166,19 @@ include the active VM-contention caveat: native photo inference and other work may overlap. This is an absolute small-corpus comparison, not an old-baseline regression claim or proof of model/media tracing overhead. The extra cost supports keeping detailed mode optional rather than claiming it is free. + +## Prepared-reader publication + +The desktop shares a bounded process-local prepared ANN between indexing and +its retained query service. The graph is registered before SQL activation and +matched by canonical database, dimensions, generation, policy and checksum. +`semantic.prepared_graph_lookup` measures that handoff; a successful handoff +sets `semantic.ann_load` cache_hit to one, without blob read/checksum/deserialization. +The query model and real-vector LRU remain warm. Eligibility refresh and query +inference still have costs and are measured independently. + +The actual CPU regression checks both a repeated cached query and an unseen +query after publication; the unseen input must record query_cache_hit zero. +See [RECOVERY-AND-HANDOFF.md](RECOVERY-AND-HANDOFF.md). Its single-document +request times do not establish representative-corpus p95, native first input +or foreground priority during indexing. Shared CPU admission remains open. diff --git a/docs/RECOVERY-AND-HANDOFF.md b/docs/RECOVERY-AND-HANDOFF.md new file mode 100644 index 0000000..58b7572 --- /dev/null +++ b/docs/RECOVERY-AND-HANDOFF.md @@ -0,0 +1,100 @@ +# Process recovery and prepared-reader handoff + +2026-10-13 development validation. Production release gates remain open. + +## Abrupt process termination + +`crates/searchi-cli/tests/crash_recovery.rs` invokes the real installed-layout +CLI adapter with verified Google EmbeddingGemma 2 assets, one CPU thread and +actual FFmpeg helpers. It creates two document/code source groups and a +65-second PCM WAV with three normal audio windows. It polls SQLite for both +completed document groups plus a committed, unowned chunk belonging to the +running media task, then terminates only its owned process tree. + +Windows uses PID-scoped `taskkill /T /F`; Unix assigns a dedicated process group +and kills that group. Cleanup does not kill processes by name and does not panic +during test unwinding. No user app/settings/index is used. + +Assertions verify that the previous active generation and completed vector +bytes remain intact. Restart resumes the same generation, makes zero document +encoder calls, re-encodes all three windows of the unfinished audio source, +and publishes five unique chunks without leaked partial vectors. A real +Google CodeRetrieval query checks the current source line citation. + +Windows and native Linux actual runs passed in 157.86 s and 233.65 s. These are suite elapsed times with compiler +contention, not a search benchmark. Power loss, arbitrary batch boundaries and +all crash scenarios are not established by one process-termination case. + +## Prepared graphs and warmed query model + +The desktop already used separate background and query model workers, but +publication could send the next foreground request through blob reading, +SHA verification and ANN deserialization. It also withdrew readiness and +repeated query warmup while holding the retained session mutex. + +The publisher now shares its immutable prepared `Arc` before +SQL activation through a process-local `PreparedGraphs` handle. That handle +binds to the canonical database and vector dimension; entries match generation, +full policy fingerprint and persisted ANN checksum. At most two registry entries +are retained. Failed SQL activation removes only its unpublished entry; readers +retain their own Arc when an old registry entry is evicted. + +The desktop query worker keeps its model and query-vector LRU ready throughout +publication. Detailed traces include `semantic.prepared_graph_lookup`; the new +active graph is a cache hit instead of foreground blob loading. Existing source +eligibility, strict filtering and generation snapshot checks still apply. + +Storage regressions cover identity/dimension/checksum mismatches, bounded +eviction, retained-reader lifetime and SQL activation rejection/retry. The real +Google CPU test drops the background worker, searches the updated document and +asserts no blob-read/checksum/deserialization spans. It verifies both a retained +query-vector hit and an unseen query with actual neural inference. + +Windows one-document fixture: retained query 4.389 ms, unseen query 111.086 ms, +29.05 s suite elapsed. These measure the shared engine request in debug mode, +not native UI first input, application startup, a representative corpus or p95. +The same native Linux fixture passed: retained query 2.107 ms, unseen query +79.614 ms, suite 21.98 s. Different cache/contention conditions prevent treating +these numbers as Windows-versus-Linux comparative performance. + +## Windows package media provenance + +The packager now verifies the pinned original decoder ZIP size and checksum, +then compares the staged manifest, helpers and notices with bounded archive +members. Valid PE headers and a freshly recorded package checksum are not +accepted as decoder provenance. Fifteen packaging tests cover changed helpers, +notices, manifests, archive members and changes during ZIP writing; failure +preserves the previous output. Media staging was refreshed from the verified +cached archive before packaging. + +Current embedded-frontend Windows release built successfully. Its 145,265,448-byte unsigned ZIP +contains 39 verified files; optimized and baseline-only packaged CPU inference +and decoder startup passed. Archive SHA-256: +`cf7953827be90255f182932fe53c16faf76701098d2636c8860b6ed4e0eb6ec4`. +Both worker smoke tests request auto with the GPU provider disabled only in the +child process. Actual GPU, native UI interaction and clean-machine installation +are not certified by those smoke tests. + +## Remaining performance/release work + +Background and query workers still receive independent thread budgets; shared +CPU admission and foreground priority are open. Publication no longer charges +foreground graph deserialization, but eligibility refresh, indexing I/O and +CPU contention need representative measurements. Native compositor/input timing, +large-corpus p95, older CPUs, actual GPU/device-loss, power-loss durability, +installers/upgrades/signatures and macOS packaging remain open. + +## Executed checks + +- Windows and native Linux workspace tests: 106 passed each, 18 ignored. +- Windows native desktop unit suite: 17 passed. +- Strict Windows/Linux workspace and Windows desktop Clippy plus both formatting checks passed. +- Real Google CPU process-kill and prepared-reader regressions passed on both OSes. +- Windows packager tests: 15 passed. Current embedded-frontend release build, + archive integrity, packaged optimized/baseline CPU inference and decoder startup + passed; no accelerator was used. + +The prepared graph handle is process-local: foreign CLI/MCP index writers cannot +share its native Arc. Cold CLI initialization and cross-process reader refresh +remain separate costs. Representative concurrent-indexing/foreground measurements +and shared CPU thread admission remain the next performance milestone. diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 8f8bd4b..41a1eda 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -30,7 +30,7 @@ Status: implementation started; no production release yet. Windows VM tests use - [ ] Versioned embeddings, ANN index, exact-match priority and hybrid fusion. - [ ] English/German relevance corpus and measured top-10 recall. - [ ] Atomic model/index replacement and resumable persistent work queues. -- [ ] Completed-source manifests with title/model/window policy keys and +- [x] Completed-source manifests with title/model/window policy keys and citation remapping; reuse only verified complete document/media groups. - [ ] Durable generation tasks, crash resume and final corpus reconciliation; see [incremental semantic design](INCREMENTAL-SEMANTIC.md). @@ -100,7 +100,7 @@ Status: implementation started; no production release yet. Windows VM tests use Warm p95 alone does not satisfy the first-use requirement. - [ ] Test migrations, interrupted work, invalid documents, inaccessible/offline roots and cloud placeholders. - [ ] Automated native build matrix and documented supported cross-compilation paths. -- [x] Native Linux headless workspace tests: 104 passed, 16 explicitly ignored. +- [x] Native Linux headless workspace tests: 106 passed, 18 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, @@ -136,6 +136,9 @@ Audit and measured priorities: [EMBEDDINGGEMMA2-AUDIT.md](EMBEDDINGGEMMA2-AUDIT. paired repeated-query/filter-change results with unchanged retrieval evidence. - [x] Bounded provider-scoped actual-query vector LRU with accurate cache traces; preserve filtering, source freshness, failures and actual warmup probes. +- [x] Bounded shared prepared-ANN handoff before SQL activation; retain query + model/LRU and readiness across publication. Actual CPU first new query has + no foreground graph load; [recovery/handoff evidence](RECOVERY-AND-HANDOFF.md). - [x] Reduce semantic settle delay to 120 ms and retain old ready generation during rebuilding; test initial-generation and activation readiness boundaries. - [x] Bound whole IPC write/read deadlines and recover from a nonreading helper; @@ -153,7 +156,10 @@ Audit and measured priorities: [EMBEDDINGGEMMA2-AUDIT.md](EMBEDDINGGEMMA2-AUDIT. on normal rescans; stale/missing agent excerpts fail closed. - [x] Actual Google CPU same-metadata text mutation before/during inference: reject publication, rescan without metadata workarounds and reuse unaffected vectors. -- [ ] Process-kill/concurrent deletion/media mutation, component-policy reuse and +- [x] Abrupt owned CLI termination after a committed partial audio chunk; + resume completed documents and reencode unfinished media without duplicates. + Process termination does not prove power-loss durability. +- [ ] Concurrent deletion/media mutation, component-policy reuse and indexing/foreground measurements. See [INCREMENTAL-SEMANTIC.md](INCREMENTAL-SEMANTIC.md). - [ ] Benchmark genuine token-budgeted multi-sequence batching and isolation. diff --git a/tools/package_windows.py b/tools/package_windows.py index 303c113..4d7eef8 100644 --- a/tools/package_windows.py +++ b/tools/package_windows.py @@ -10,11 +10,80 @@ import os import re from pathlib import Path import struct +import stat import tempfile import zipfile ROOT = Path(__file__).resolve().parents[1] CRT_LIBRARIES = {"concrt140.dll", "msvcp140.dll", "msvcp140_1.dll", "msvcp140_2.dll", "msvcp140_atomic_wait.dll", "msvcp140_codecvt_ids.dll", "vccorlib140.dll", "vcruntime140.dll", "vcruntime140_1.dll", "vcruntime140_threads.dll"} +MAX_MEMBER = 256 * 1024 * 1024 +MAX_MEDIA_TOTAL = 1024 * 1024 * 1024 + + +def digest(path): + with path.open("rb") as stream: + return hashlib.file_digest(stream, "sha256").hexdigest() + + +def media_provenance(root, resources, architecture): + """Compare staged helpers/notices with the checksum-pinned upstream ZIP.""" + manifest_path = root / "tools/media-runtime/manifest.json" + if not manifest_path.is_file() or manifest_path.stat().st_size > 65536: + raise ValueError("Required pinned media manifest is missing or oversized") + manifest_bytes = manifest_path.read_bytes() + manifest = json.loads(manifest_bytes) + artifact = manifest["platforms"][f"windows-{architecture}"] + name = artifact.get("archive") + if not isinstance(name, str) or not re.fullmatch(r"[A-Za-z0-9_.-]+\.zip", name): + raise ValueError("Invalid media archive name") + archive = root / ".searchi/media-runtime" / name + if (not archive.is_file() or archive.is_symlink() or archive.absolute() != archive.resolve() + or archive.stat().st_size != artifact["bytes"] or digest(archive) != artifact["sha256"]): + raise ValueError("Media archive provenance mismatch") + staged_manifest = resources / "media-runtime/manifest.json" + if (not staged_manifest.is_file() or staged_manifest.stat().st_size != len(manifest_bytes) + or staged_manifest.read_bytes() != manifest_bytes): + raise ValueError("Staged media manifest differs from pinned manifest") + selected = {"media-runtime/manifest.json": {"bytes": len(manifest_bytes), "sha256": hashlib.sha256(manifest_bytes).hexdigest()}} + selected_keys = {name.casefold() for name in selected} + with zipfile.ZipFile(archive) as package: + members = package.infolist() + if len(members) > 10000: + raise ValueError("Media archive exceeds member ceiling") + total = 0 + for member in members: + total += member.file_size + if member.file_size < 0 or member.file_size > MAX_MEMBER or total > MAX_MEDIA_TOTAL: + raise ValueError("Media archive exceeds size ceiling") + parts = member.filename.replace("\\", "/").split("/") + basename = parts[-1] + if basename in {"ffmpeg.exe", "ffprobe.exe"}: + target = "media-runtime/bin/" + basename + elif basename.lower().startswith(("license", "copying")) or basename.lower() in {"readme.txt", "readme.md"}: + if any(part in {"", ".", ".."} or ":" in part for part in parts): + raise ValueError("Unsafe media notice path") + target = "media-runtime/licenses/" + "/".join(parts) + else: + continue + mode = member.external_attr >> 16 + if (any(part in {"", ".", ".."} or ":" in part for part in parts) + or member.is_dir() or stat.S_IFMT(mode) not in {0, stat.S_IFREG} or member.flag_bits & 1 + or member.file_size <= 0 or target.casefold() in selected_keys): + raise ValueError("Invalid or duplicate media member") + path = resources / target + if (not path.is_file() or path.is_symlink() or path.absolute() != path.resolve() + or not path.resolve().is_relative_to(resources.resolve())): + raise ValueError(f"Staged media helper/notice missing or outside resources: {target}") + with package.open(member) as stream: + checksum = hashlib.file_digest(stream, "sha256").hexdigest() + if path.stat().st_size != member.file_size or digest(path) != checksum: + raise ValueError(f"Media helper/notice differs from pinned archive: {target}") + selected[target] = {"bytes": member.file_size, "sha256": checksum} + selected_keys.add(target.casefold()) + if (not {"media-runtime/bin/ffmpeg.exe", "media-runtime/bin/ffprobe.exe"}.issubset(selected) + or not any(name.startswith("media-runtime/licenses/") for name in selected)): + raise ValueError("Media archive lacks required helpers or notices") + return selected def pe_machine(path): @@ -34,6 +103,10 @@ def pe_machine(path): def package(root, executable, output): resources = root / "apps/desktop/src-tauri/resources" + machine = pe_machine(executable) + architecture = {0x8664: "x86_64", 0xAA64: "aarch64"}.get(machine) + if architecture is None: + raise ValueError("Unsupported Windows architecture") inventory = resources / "runtime-libraries.json" if not inventory.is_file() or inventory.stat().st_size > 65536: raise ValueError("Required runtime dependency inventory is missing or oversized") @@ -69,6 +142,8 @@ def package(root, executable, output): raise ValueError(f"Runtime checksum mismatch: {name}") if "bytes" in record and path.stat().st_size != record["bytes"]: raise ValueError(f"Runtime size mismatch: {name}") + media = media_provenance(root, resources, architecture) + expected.update(media) selected = [(executable, "searchi.exe"), (root / "LICENSE", "LICENSE.txt")] for name in ["searchi-gemma-worker.exe", "LLAMA-LICENSE.txt", "MSVC-REDISTRIBUTION.txt", "vulkan-1.dll", "VULKAN-LICENSE.txt", "media-runtime/manifest.json", "media-runtime/bin/ffmpeg.exe", "media-runtime/bin/ffprobe.exe", @@ -77,17 +152,13 @@ def package(root, executable, output): selected.append((inventory, inventory.name)) selected_names = {name for _, name in selected} selected.extend((resources / name, name) for name in expected if name not in selected_names) - for directory in ["licenses", "media-runtime/licenses"]: + for directory in ["licenses"]: notices = list((resources / directory).rglob("*.txt")) if not notices: raise ValueError(f"Required notices are missing: {directory}") selected_names = {name for _, name in selected} selected.extend((path, path.relative_to(resources).as_posix()) for path in notices if path.relative_to(resources).as_posix() not in selected_names) - machine = pe_machine(executable) - architecture = {0x8664: "x86_64", 0xAA64: "aarch64"}.get(machine) - if architecture is None: - raise ValueError("Unsupported Windows architecture") for path, name in selected: if not path.is_file(): raise ValueError(f"Required package file is missing: {name}") diff --git a/tools/test_package_windows.py b/tools/test_package_windows.py index 18895eb..0f57f19 100644 --- a/tools/test_package_windows.py +++ b/tools/test_package_windows.py @@ -45,9 +45,80 @@ class PackageTests(unittest.TestCase): path.parent.mkdir(parents=True, exist_ok=True) path.write_text("notice") (resources / "settings.json").write_text("private settings") + media = root / ".searchi/media-runtime" + media.mkdir(parents=True) + upstream = media / "pinned-windows.zip" + with zipfile.ZipFile(upstream, "w") as archive: + for staged, member in [("bin/ffmpeg.exe", "bin/ffmpeg.exe"), ("bin/ffprobe.exe", "bin/ffprobe.exe"), ("licenses/LICENSE.txt", "LICENSE.txt")]: + archive.writestr(member, (resources / "media-runtime" / staged).read_bytes()) + manifest = root / "tools/media-runtime/manifest.json" + manifest.parent.mkdir(parents=True) + manifest.write_text(json.dumps({"platforms": {"windows-x86_64": { + "archive": upstream.name, "bytes": upstream.stat().st_size, + "sha256": hashlib.sha256(upstream.read_bytes()).hexdigest()}}})) + (resources / "media-runtime/manifest.json").write_bytes(manifest.read_bytes()) inventory(resources) return root, resources, app, root / "archive.zip" + def test_valid_pe_media_mutation_and_notice_corruption_preserve_previous_archive(self): + for name in ["bin/ffmpeg.exe", "bin/ffprobe.exe", "licenses/LICENSE.txt", "manifest.json"]: + with self.subTest(name=name), tempfile.TemporaryDirectory() as directory: + root, resources, app, output = self.fixture(directory) + output.write_bytes(b"previous package") + path = resources / "media-runtime" / name + path.write_bytes(path.read_bytes() + b"tampered") + with self.assertRaisesRegex(ValueError, "differs from pinned"): + package(root, app, output) + self.assertEqual(output.read_bytes(), b"previous package") + self.assertEqual(list(output.parent.glob("*.partial")), []) + + def test_pinned_archive_corruption_preserves_previous_package(self): + with tempfile.TemporaryDirectory() as directory: + root, _, app, output = self.fixture(directory) + output.write_bytes(b"previous package") + source = root / ".searchi/media-runtime/pinned-windows.zip" + source.write_bytes(source.read_bytes() + b"changed") + with self.assertRaisesRegex(ValueError, "provenance mismatch"): + package(root, app, output) + self.assertEqual(output.read_bytes(), b"previous package") + + def test_media_change_during_archive_write_preserves_previous_package(self): + with tempfile.TemporaryDirectory() as directory: + root, resources, app, output = self.fixture(directory) + output.write_bytes(b"previous package") + original = zipfile.ZipFile.open + def mutate(archive, name, mode="r", *args, **kwargs): + if name == "media-runtime/bin/ffmpeg.exe" and mode == "w": + path = resources / name + path.write_bytes(path.read_bytes() + b"changed") + return original(archive, name, mode, *args, **kwargs) + with mock.patch.object(zipfile.ZipFile, "open", mutate), self.assertRaisesRegex(ValueError, "changed while packaging"): + package(root, app, output) + self.assertEqual(output.read_bytes(), b"previous package") + + def test_duplicate_helper_and_link_notice_are_rejected_before_output(self): + for kind in ["duplicate", "link"]: + with self.subTest(kind=kind), tempfile.TemporaryDirectory() as directory: + root, resources, app, output = self.fixture(directory) + output.write_bytes(b"previous package") + upstream = root / ".searchi/media-runtime/pinned-windows.zip" + with zipfile.ZipFile(upstream, "a") as archive: + if kind == "duplicate": + archive.writestr("other/ffmpeg.exe", (resources / "media-runtime/bin/ffmpeg.exe").read_bytes()) + else: + entry = zipfile.ZipInfo("COPYING.txt") + entry.create_system = 3 + entry.external_attr = 0o120777 << 16 + archive.writestr(entry, "somewhere") + manifest = root / "tools/media-runtime/manifest.json" + data = json.loads(manifest.read_text()) + data["platforms"]["windows-x86_64"].update(bytes=upstream.stat().st_size, sha256=hashlib.sha256(upstream.read_bytes()).hexdigest()) + manifest.write_text(json.dumps(data)) + (resources / "media-runtime/manifest.json").write_bytes(manifest.read_bytes()) + with self.assertRaisesRegex(ValueError, "Invalid or duplicate"): + package(root, app, output) + self.assertEqual(output.read_bytes(), b"previous package") + def test_complete_archive_excludes_settings_and_verifies_recorded_bytes(self): with tempfile.TemporaryDirectory() as directory: root, _, app, output = self.fixture(directory)