From b3ebd7a894ae71386fe809f335573ffd3e312d18 Mon Sep 17 00:00:00 2001 From: karl Date: Mon, 12 Oct 2026 23:46:01 +0200 Subject: [PATCH] Harden worker IPC and recoverable runtime staging --- .gitignore | 4 + apps/desktop/src-tauri/src/data_directory.rs | 44 +++ apps/desktop/src-tauri/src/main.rs | 27 +- crates/searchi-assets/src/lib.rs | 47 +++ crates/searchi-semantic/src/gemma.rs | 270 ++++++++++++-- design-qa.md | 11 + docs/BUILDING.md | 49 ++- docs/ROADMAP.md | 22 +- docs/SEMANTIC.md | 6 + docs/VALIDATION.md | 33 ++ tools/embeddinggemma2/stage_runtime.py | 348 +++++++++++++++++- tools/embeddinggemma2/test_stage_runtime.py | 366 +++++++++++++++++++ 12 files changed, 1152 insertions(+), 75 deletions(-) create mode 100644 apps/desktop/src-tauri/src/data_directory.rs diff --git a/.gitignore b/.gitignore index 0d8734f..10ec936 100644 --- a/.gitignore +++ b/.gitignore @@ -7,6 +7,10 @@ dist/ .searchi/ __pycache__/ /apps/desktop/src-tauri/gen/ +/apps/desktop/src-tauri/.searchi-stage.lock +/apps/desktop/src-tauri/.searchi-stage.json +/apps/desktop/src-tauri/.resources.stage-*/ +/apps/desktop/src-tauri/.resources.backup-*/ /apps/desktop/src-tauri/resources/searchi-gemma-worker* /apps/desktop/src-tauri/resources/vulkan-1.dll /apps/desktop/src-tauri/resources/*.dll diff --git a/apps/desktop/src-tauri/src/data_directory.rs b/apps/desktop/src-tauri/src/data_directory.rs new file mode 100644 index 0000000..916ea1c --- /dev/null +++ b/apps/desktop/src-tauri/src/data_directory.rs @@ -0,0 +1,44 @@ +//! Explicit data-directory isolation for portable use and native release QA. +//! Ordinary launches continue to use the operating system's app-data directory. +use std::{ffi::OsString, io, path::PathBuf}; + +pub fn resolve(default: PathBuf, override_path: Option) -> io::Result { + let Some(value) = override_path else { + return Ok(default); + }; + let path = PathBuf::from(value); + if !path.is_absolute() || path.as_os_str().len() > 32768 { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "SEARCHI_DATA_DIR must be an absolute bounded directory path", + )); + } + Ok(path) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn explicit_directory_isolates_settings_and_index_without_changing_default() { + let default = std::env::temp_dir().join("searchi-default"); + let isolated = std::env::temp_dir().join("searchi-isolated"); + assert_eq!(resolve(default.clone(), None).unwrap(), default); + assert_eq!( + resolve(default, Some(isolated.clone().into_os_string())).unwrap(), + isolated + ); + } + + #[test] + fn invalid_override_is_rejected_without_falling_back_to_user_data() { + let default = std::env::temp_dir().join("searchi-default"); + for value in [OsString::new(), OsString::from("relative/data")] { + assert_eq!( + resolve(default.clone(), Some(value)).unwrap_err().kind(), + io::ErrorKind::InvalidInput + ); + } + } +} diff --git a/apps/desktop/src-tauri/src/main.rs b/apps/desktop/src-tauri/src/main.rs index e5f7274..7e1b451 100644 --- a/apps/desktop/src-tauri/src/main.rs +++ b/apps/desktop/src-tauri/src/main.rs @@ -1,6 +1,10 @@ #![cfg_attr(not(debug_assertions), windows_subsystem = "windows")] +#[cfg(all(not(debug_assertions), not(feature = "custom-protocol")))] +compile_error!("Release desktop builds require --features custom-protocol to embed the frontend"); + mod ai_commands; +mod data_directory; mod model_download; mod semantic_commands; mod settings; @@ -394,7 +398,10 @@ fn main() { } }).build()) .setup(|app| { - let directory = app.path().app_data_dir()?; + let directory = data_directory::resolve( + app.path().app_data_dir()?, + std::env::var_os("SEARCHI_DATA_DIR"), + )?; std::fs::create_dir_all(&directory)?; let database = directory.join("searchi.sqlite"); let lexical = Arc::new(Mutex::new(SearchIndex::open(&database)?)); @@ -413,15 +420,15 @@ fn main() { let development_root = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../.."); let development_models = development_root.join(".searchi/models/embeddinggemma2"); let bundled_models = app.path().resource_dir()?.join("embeddinggemma2"); - let model_directory = settings.value.model_directory.clone().unwrap_or_else(|| { - if cfg!(debug_assertions) && development_models.is_dir() { - development_models.canonicalize().unwrap_or(development_models) - } else if bundled_models.is_dir() { - bundled_models - } else { - searchi_assets::default_model_directory().unwrap_or_else(|_| directory.join("Models")) - } - }); + let model_directory = if let Some(configured) = settings.value.model_directory.clone() { + configured + } else if cfg!(debug_assertions) && development_models.is_dir() { + development_models.canonicalize().unwrap_or(development_models) + } else if searchi_assets::has_model_files(&bundled_models)? { + bundled_models + } else { + searchi_assets::default_model_directory().unwrap_or_else(|_| directory.join("Models")) + }; let worker_name = if cfg!(windows) { "searchi-gemma-worker.exe" } else { "searchi-gemma-worker" }; let development_worker = development_root.join(".searchi/runtime/cpu") .join(if cfg!(windows) { "Release" } else { "" }).join(worker_name); diff --git a/crates/searchi-assets/src/lib.rs b/crates/searchi-assets/src/lib.rs index 0971089..56f1c15 100644 --- a/crates/searchi-assets/src/lib.rs +++ b/crates/searchi-assets/src/lib.rs @@ -94,6 +94,27 @@ pub fn default_model_directory() -> Result { .join("embeddinggemma2")) } +/// Cheap location selection for optional bundled assets. This checks regular +/// files and expected lengths only; `verify_cached` still verifies SHA-256 +/// before setup considers the model usable. Partial bundles use managed cache. +pub fn has_model_files(directory: &Path) -> Result { + model_files_present(directory, &manifest()?.files) +} + +fn model_files_present(directory: &Path, artifacts: &[Artifact]) -> Result { + for artifact in artifacts { + let metadata = match fs::metadata(directory.join(&artifact.path)) { + Ok(metadata) => metadata, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false), + Err(error) => return Err(error).context("inspect bundled model artifact"), + }; + if !metadata.is_file() || metadata.len() != artifact.bytes { + return Ok(false); + } + } + Ok(true) +} + /// Offline validation includes file length and full SHA-256, not mere existence. pub fn verify_cached(directory: &Path) -> Result { for artifact in manifest()?.files { @@ -383,6 +404,32 @@ mod tests { ); } + #[test] + fn optional_bundle_requires_both_regular_files_at_expected_sizes() { + let directory = tempfile::tempdir().unwrap(); + let artifacts = [artifact("one.gguf", b"abc"), artifact("two.gguf", b"xyz")]; + assert!(!model_files_present(directory.path(), &artifacts).unwrap()); + fs::write(directory.path().join("one.gguf"), b"abc").unwrap(); + assert!(!model_files_present(directory.path(), &artifacts).unwrap()); + fs::create_dir(directory.path().join("two.gguf")).unwrap(); + assert!(!model_files_present(directory.path(), &artifacts).unwrap()); + fs::remove_dir(directory.path().join("two.gguf")).unwrap(); + fs::write(directory.path().join("two.gguf"), b"xyz").unwrap(); + assert!(model_files_present(directory.path(), &artifacts).unwrap()); + fs::write(directory.path().join("one.gguf"), b"ab").unwrap(); + assert!(!model_files_present(directory.path(), &artifacts).unwrap()); + } + + #[test] + fn bundled_file_presence_never_substitutes_for_sha_verification() { + let directory = tempfile::tempdir().unwrap(); + let artifact = artifact("one.gguf", b"abc"); + let path = directory.path().join(&artifact.path); + fs::write(&path, b"xyz").unwrap(); + assert!(model_files_present(directory.path(), std::slice::from_ref(&artifact)).unwrap()); + assert!(!verified(&path, &artifact).unwrap()); + } + #[test] fn same_size_corruption_is_detected_and_missing_cache_is_offline() { let directory = tempfile::tempdir().unwrap(); diff --git a/crates/searchi-semantic/src/gemma.rs b/crates/searchi-semantic/src/gemma.rs index c4f92af..7b710f6 100644 --- a/crates/searchi-semantic/src/gemma.rs +++ b/crates/searchi-semantic/src/gemma.rs @@ -60,7 +60,10 @@ fn query_operation(info: &WorkerInfo, code: bool) -> Result<&'static str> { /// Persistent isolated native worker. No HTTP or remote model fallback. pub struct GemmaEmbedder { child: Child, - stdin: ChildStdin, + requests: Option>>, + writes: Receiver>, + writer_thread: Option>, + reader_thread: Option>, responses: Receiver>, pub info: WorkerInfo, healthy: bool, @@ -73,24 +76,48 @@ pub struct GemmaEmbedder { impl Drop for GemmaEmbedder { fn drop(&mut self) { - let _ = self.child.kill(); - let _ = self.child.wait(); + self.stop_worker(); } } +type WriterTransport = ( + mpsc::SyncSender>, + Receiver>, + thread::JoinHandle<()>, +); + +fn spawn_writer(mut stdin: ChildStdin) -> WriterTransport { + let (requests, pending) = mpsc::sync_channel::>(1); + let (completed, writes) = mpsc::sync_channel(1); + let writer = thread::spawn(move || { + while let Ok(encoded) = pending.recv() { + let result = stdin + .write_all(&encoded) + .and_then(|_| stdin.write_all(b"\n")) + .and_then(|_| stdin.flush()) + .map_err(|error| format!("Cannot write local Gemma worker request: {error}")); + let failed = result.is_err(); + if completed.try_send(result).is_err() || failed { + break; + } + } + }); + (requests, writes, writer) +} + #[derive(Debug)] -struct StartupTransportFailure(String); -impl std::fmt::Display for StartupTransportFailure { +struct WorkerTransportFailure(String); +impl std::fmt::Display for WorkerTransportFailure { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { formatter.write_str(&self.0) } } -impl std::error::Error for StartupTransportFailure {} +impl std::error::Error for WorkerTransportFailure {} /// Retry only process/transport failure, never an incompatible protocol or model. fn start_with_cpu_fallback(cpu: bool, mut start: impl FnMut(bool) -> Result) -> Result { match start(cpu) { - Err(error) if !cpu && error.downcast_ref::().is_some() => { + Err(error) if !cpu && error.downcast_ref::().is_some() => { start(true).with_context(|| { format!("CPU initialization failed after automatic worker startup failure: {error}") }) @@ -163,7 +190,7 @@ impl GemmaEmbedder { command.creation_flags(0x08000000); // CREATE_NO_WINDOW } let mut child = command.spawn().map_err(|error| { - StartupTransportFailure(format!("Cannot start local Gemma worker: {error}")) + WorkerTransportFailure(format!("Cannot start local Gemma worker: {error}")) })?; let (Some(stdin), Some(stdout)) = (child.stdin.take(), child.stdout.take()) else { let _ = child.kill(); @@ -171,12 +198,12 @@ impl GemmaEmbedder { anyhow::bail!("Gemma worker protocol pipes missing"); }; let (sender, responses) = mpsc::sync_channel(1); - thread::spawn(move || { + let reader_thread = thread::spawn(move || { let mut reader = BufReader::new(stdout); loop { let response = read_response(&mut reader); let failed = response.is_err(); - if sender.send(response).is_err() || failed { + if sender.try_send(response).is_err() || failed { break; } } @@ -186,13 +213,14 @@ impl GemmaEmbedder { other => { let _ = child.kill(); let _ = child.wait(); + let _ = reader_thread.join(); // Malformed/oversized JSON is a protocol violation, not evidence // that hardware acceleration failed. Do not mask it with retry. if matches!(&other, Ok(Err(error)) if error.starts_with("Invalid Gemma worker JSON:") || error == "Gemma worker response exceeds protocol byte budget") { anyhow::bail!("Invalid Gemma worker startup protocol: {other:?}"); } - return Err(StartupTransportFailure(format!( + return Err(WorkerTransportFailure(format!( "Gemma worker did not initialize: {other:?}" )) .into()); @@ -203,6 +231,7 @@ impl GemmaEmbedder { Err(error) => { let _ = child.kill(); let _ = child.wait(); + let _ = reader_thread.join(); return Err(error).context("Invalid Gemma worker handshake"); } }; @@ -216,11 +245,16 @@ impl GemmaEmbedder { { let _ = child.kill(); let _ = child.wait(); + let _ = reader_thread.join(); anyhow::bail!("Gemma worker handshake violates configured model/device contract"); } + let (requests, writes, writer_thread) = spawn_writer(stdin); Ok(Self { child, - stdin, + requests: Some(requests), + writes, + writer_thread: Some(writer_thread), + reader_thread: Some(reader_thread), responses, info, healthy: true, @@ -246,7 +280,10 @@ impl GemmaEmbedder { ); let response = match self.exchange(&encoded) { Ok(response) => response, - Err(error) if can_retry_cpu(&self.info.device, self.force_cpu) => { + Err(error) + if can_retry_cpu(&self.info.device, self.force_cpu) + && error.downcast_ref::().is_some() => + { // GPU runtimes can abort rather than return an encoder error. // Restart once as CPU and retry this same bounded request. let _ = self.child.wait(); @@ -296,33 +333,76 @@ impl GemmaEmbedder { Ok(response) } + fn stop_worker(&mut self) { + self.requests.take(); + // The shipped native worker does not spawn descendants. Killing/reaping + // it closes its pipe ends and releases any blocked writer/read thread. + let _ = self.child.kill(); + let _ = self.child.wait(); + if let Some(writer) = self.writer_thread.take() { + let _ = writer.join(); + } + if let Some(reader) = self.reader_thread.take() { + let _ = reader.join(); + } + } + fn exchange(&mut self, encoded: &[u8]) -> Result { + self.exchange_with_timeout(encoded, Duration::from_secs(180)) + } + + fn exchange_with_timeout(&mut self, encoded: &[u8], timeout: Duration) -> Result { + let deadline = Instant::now() + timeout; let write_started = self.query_profile.as_ref().map(|_| Instant::now()); - let sent = (|| { - self.stdin.write_all(encoded)?; - self.stdin.write_all(b"\n")?; - self.stdin.flush() - })() - .context("Cannot write local Gemma worker request"); - self.record_query_duration("request_write_us", write_started); - if let Err(error) = sent { + let result = (|| { + self.requests + .as_ref() + .ok_or_else(|| WorkerTransportFailure("Gemma worker writer stopped".into()))? + .try_send(encoded.to_vec()) + .map_err(|_| { + WorkerTransportFailure("Gemma worker request queue unavailable".into()) + })?; + match self + .writes + .recv_timeout(deadline.saturating_duration_since(Instant::now())) + { + Ok(Ok(())) => (), + other => { + return Err(WorkerTransportFailure(format!( + "Gemma worker request write failed or timed out: {other:?}" + )) + .into()) + } + } + self.record_query_duration("request_write_us", write_started); + let wait_started = self.query_profile.as_ref().map(|_| Instant::now()); + let response = match self + .responses + .recv_timeout(deadline.saturating_duration_since(Instant::now())) + { + Ok(Ok(value)) => value, + Ok(Err(error)) + if error.starts_with("Invalid Gemma worker JSON:") + || error == "Gemma worker response exceeds protocol byte budget" => + { + anyhow::bail!("Invalid Gemma worker response protocol: {error}") + } + other => { + return Err(WorkerTransportFailure(format!( + "Gemma worker request failed or timed out: {other:?}" + )) + .into()) + } + }; + self.record_query_duration("worker_wait_and_response_decode_us", wait_started); + Ok(response) + })(); + if result.is_err() { self.healthy = false; self.info.ready = false; - let _ = self.child.kill(); - return Err(error); + self.stop_worker(); } - let wait_started = self.query_profile.as_ref().map(|_| Instant::now()); - let response = match self.responses.recv_timeout(Duration::from_secs(180)) { - Ok(Ok(value)) => value, - other => { - self.healthy = false; - self.info.ready = false; - let _ = self.child.kill(); - anyhow::bail!("Gemma worker request failed: {other:?}"); - } - }; - self.record_query_duration("worker_wait_and_response_decode_us", wait_started); - Ok(response) + result } pub fn split_windows(&mut self, text: &str, title: &str) -> Result> { @@ -710,6 +790,122 @@ fn verify_artifact(path: &Path, bytes: u64, sha: &str) -> Result<()> { #[cfg(test)] mod tests { use super::*; + // This helper is an inert pipe-protocol fixture, never an embedding provider. + fn stalled_protocol_helper(read_request: bool) -> GemmaEmbedder { + #[cfg(windows)] + let mut command = { + use std::os::windows::process::CommandExt; + let mut command = Command::new("powershell.exe"); + command.args([ + "-NoProfile", + "-NonInteractive", + "-Command", + if read_request { + "[Console]::In.ReadLine() | Out-Null; Start-Sleep -Seconds 20" + } else { + "Start-Sleep -Seconds 20" + }, + ]); + command.creation_flags(0x08000000); + command + }; + #[cfg(not(windows))] + let mut command = { + let mut command = Command::new("sh"); + command.args([ + "-c", + if read_request { + "read line; exec sleep 20" + } else { + "exec sleep 20" + }, + ]); + command + }; + let mut child = command + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + let (requests, writes, writer_thread) = spawn_writer(child.stdin.take().unwrap()); + let stdout = child.stdout.take().unwrap(); + let (sender, responses) = mpsc::sync_channel(1); + let reader_thread = thread::spawn(move || { + let response = read_response(&mut BufReader::new(stdout)); + let _ = sender.try_send(response); + }); + GemmaEmbedder { + child, + requests: Some(requests), + writes, + writer_thread: Some(writer_thread), + reader_thread: Some(reader_thread), + responses, + info: WorkerInfo { + ready: true, + model: "pipe protocol fixture".into(), + dimensions: 0, + device: "cpu".into(), + threads: 1, + runtime: "fixture".into(), + media_supported: false, + code_query_supported: false, + }, + healthy: true, + force_cpu: true, + model_dir: PathBuf::new(), + worker_path: PathBuf::new(), + projector_verified: false, + query_profile: None, + } + } + + #[test] + fn disconnected_writer_queue_is_transport_failure_and_reaps_helper() { + let mut fixture = stalled_protocol_helper(false); + let (sender, receiver) = mpsc::sync_channel(1); + drop(receiver); + fixture.requests = Some(sender); + let error = fixture + .exchange_with_timeout(b"{}", Duration::from_millis(300)) + .unwrap_err(); + assert!(error.downcast_ref::().is_some()); + assert!(fixture.child.try_wait().unwrap().is_some()); + assert!(!fixture.healthy && fixture.requests.is_none()); + assert!(fixture.writer_thread.is_none() && fixture.reader_thread.is_none()); + } + + #[test] + fn entire_request_deadline_kills_and_joins_worker_that_never_reads_stdin() { + let mut fixture = stalled_protocol_helper(false); + let started = Instant::now(); + let error = fixture + .exchange_with_timeout(&vec![b'x'; MAX_REQUEST - 1], Duration::from_millis(300)) + .unwrap_err(); + assert!(error.downcast_ref::().is_some()); + assert!(error.to_string().contains("write failed or timed out")); + assert!(started.elapsed() < Duration::from_secs(5)); + assert!(fixture.child.try_wait().unwrap().is_some()); + assert!(!fixture.healthy && !fixture.info.ready); + assert!(fixture.writer_thread.is_none() && fixture.reader_thread.is_none()); + } + + #[test] + fn response_wait_uses_same_deadline_and_reaps_silent_reader() { + let mut fixture = stalled_protocol_helper(true); + let started = Instant::now(); + let error = fixture + .exchange_with_timeout(b"{}", Duration::from_millis(300)) + .unwrap_err(); + assert!(error.downcast_ref::().is_some()); + assert!(error.to_string().contains("request failed or timed out")); + assert!(started.elapsed() < Duration::from_secs(5)); + assert!(fixture.child.try_wait().unwrap().is_some()); + assert!(fixture.requests.is_none()); + assert!(fixture.writer_thread.is_none() && fixture.reader_thread.is_none()); + } + #[test] fn automatic_startup_transport_failure_retries_cpu_once() { let mut attempts = Vec::new(); @@ -718,7 +914,7 @@ mod tests { if cpu { Ok("cpu ready") } else { - Err(StartupTransportFailure("worker closed its output before ready".into()).into()) + Err(WorkerTransportFailure("worker closed its output before ready".into()).into()) } }) .unwrap(); @@ -732,7 +928,7 @@ mod tests { let mut attempts = Vec::new(); let result: Result<()> = start_with_cpu_fallback(cpu, |selected_cpu| { attempts.push(selected_cpu); - Err(StartupTransportFailure("worker exited".into()).into()) + Err(WorkerTransportFailure("worker exited".into()).into()) }); let error = format!("{:#}", result.unwrap_err()); assert_eq!(attempts, if cpu { vec![true] } else { vec![false, true] }); diff --git a/design-qa.md b/design-qa.md index 4a837e9..af143d1 100644 --- a/design-qa.md +++ b/design-qa.md @@ -2,6 +2,17 @@ final result: browser captures checked; native desktop visual gate remains open +## Native release startup evidence (2026-10-12) + +The extracted embedded-frontend Windows Release archive launches outside the +checkout with isolated validation data and CPU-only inference/rendering. +Accessibility confirms the `http://tauri.localhost/` frontend, one indexed file, +one extracted content segment and ready Google semantic search using CPU. +This verifies startup and engine status, not search interaction or appearance. +Native input fails with `GetCursorPos 0x80070005`; capture fails with +`CreateForMonitor 0x80070057`. No native screenshot was obtained. Native focus, +keyboard navigation, transitions and visual inspection remain open. + ## Browser-rendered implementation evidence (2026-10-12) Four 1440×1000 captures now show the production frontend against the actual diff --git a/docs/BUILDING.md b/docs/BUILDING.md index b1b9060..a970447 100644 --- a/docs/BUILDING.md +++ b/docs/BUILDING.md @@ -35,8 +35,13 @@ cargo test -p searchi-service --test google_cpu -- --ignored --nocapture cargo test -p searchi-service --lib google_cpu_preserves -- --ignored --nocapture ``` -The worker is statically linked with the MSVC CRT on Windows. Staging copies -it and the llama.cpp license into the desktop resources directory. Add +The explicit internal CPU worker uses the static MSVC CRT on Windows; normal +auto builds use the shared runtime described below. Staging verifies a complete +candidate resource directory, then swaps it into place under a process lock. +Its recovery journal restores an interrupted swap on the next invocation; +failed restoration retains the previous backup. Existing unrelated fonts and +media resources are preserved. Fault-injected process interruption is tested; +power-loss durability is not certified. Add `--bundle-model` to staging for an offline model bundle after downloading; otherwise desktop setup provisions verified models on first launch. Search itself performs no network inference or downloads. The worker now uses the verified media projector for genuine image/audio embeddings. The FFmpeg script stages verified standalone decoders and their notices. @@ -59,25 +64,55 @@ supports `--backend auto|metal|vulkan|cuda` builds with CPU fallback; do not exercise those acceleration backends in this VM. Their execution and failure recovery require validation on the corresponding release hardware. +### Isolated native desktop validation + +Set `SEARCHI_DATA_DIR` to an absolute directory to isolate the desktop's +settings and index for release QA or portable use. Normal launches use the OS +app-data directory. Invalid overrides fail startup instead of falling back to +the user's regular data. Model downloads still use the configured model +directory or managed cache, so isolated tests should also configure a verified +fixture model directory in their own settings. + +```powershell +$env:SEARCHI_DATA_DIR = 'C:\searchi-qa-data' +$env:SEARCHI_FORCE_CPU = '1' +$env:WEBVIEW2_ADDITIONAL_BROWSER_ARGUMENTS = '--disable-gpu --disable-accelerated-video-decode' +``` + +Extract and launch the entire Release archive outside the source checkout. +Direct Cargo release builds must embed the frontend: + +```powershell +cargo build --manifest-path apps/desktop/src-tauri/Cargo.toml --release --features custom-protocol --locked +``` + +The crate rejects Release compilation without this feature; the Tauri build +CLI enables it automatically. A development-server URL is not a release gate. +Record native window interaction separately from process startup, browser +screenshots and worker inference; those are different release gates. + ## Headless Linux Google search These are the current script and CLI interfaces, not evidence of an executed -Linux release. On a native Linux build host, run from the repository root: +Linux release. On a native Linux build host with Vulkan development libraries +and `glslc`, run from the repository root: ```sh -python3 tools/embeddinggemma2/build_runtime.py --backend cpu --jobs 2 +python3 tools/embeddinggemma2/build_runtime.py --backend auto --jobs 2 python3 tools/embeddinggemma2/download.py python3 tools/media-runtime/provision.py -export SEARCHI_FORCE_CPU=1 cargo build --release -p searchi-cli --locked ./target/release/searchi index --db searchi.sqlite --root /path/to/documents -./target/release/searchi embed --db searchi.sqlite --model-dir .searchi/models/embeddinggemma2 --embedding-worker .searchi/runtime/cpu/searchi-gemma-worker --ffmpeg .searchi/media-runtime/bin/ffmpeg --ffprobe .searchi/media-runtime/bin/ffprobe --cpu-threads 1 -./target/release/searchi search --db searchi.sqlite 'supplier cancellation terms' --model-dir .searchi/models/embeddinggemma2 --embedding-worker .searchi/runtime/cpu/searchi-gemma-worker --cpu-threads 1 +./target/release/searchi embed --db searchi.sqlite --model-dir .searchi/models/embeddinggemma2 --embedding-worker .searchi/runtime/auto/searchi-gemma-worker --ffmpeg .searchi/media-runtime/bin/ffmpeg --ffprobe .searchi/media-runtime/bin/ffprobe --cpu-threads 1 +./target/release/searchi search --db searchi.sqlite 'supplier cancellation terms' --model-dir .searchi/models/embeddinggemma2 --embedding-worker .searchi/runtime/auto/searchi-gemma-worker --cpu-threads 1 ``` `embed` builds document, image, video and audio/music vectors by default. Use `--text-only` explicitly to omit media. Default media indexing needs paired FFmpeg/FFprobe paths or packaged helpers beside the installed CLI. +This normal worker detects usable acceleration and falls back to CPU. Internal +CPU-only model tests may use `--backend cpu` and `SEARCHI_FORCE_CPU=1`; those +test artifacts are not a separate distribution edition. `searchi setup --model-dir PATH` is an alternative verified model-provisioning command. It provisions model assets, not the native worker. Package the worker diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 2867897..be90adb 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -67,11 +67,16 @@ Status: implementation started; no production release yet. Windows VM tests use - [x] Windows desktop archive packager includes explicit runtime/helper files and notices, checks architecture and records per-file checksums; excludes model caches and user data. Controlled packaging regression tests pass. -- [ ] Verify embedded-frontend Release archive launch outside the source tree, - clean-machine prerequisites, GPU dependencies, upgrades and signing. +- [x] Embedded-frontend Windows Release archive launches outside the source tree + with isolated data and a ready real Google CPU semantic generation. +- [ ] Verify native interactions, clean-machine prerequisites, GPU dependencies, + upgrades and signing. - [x] Preflight native license and the complete offline Google model pair before modifying staged resources; corrupted/missing assets preserve existing worker. -- [ ] Atomic resource staging and interrupted-copy recovery for release bundles. +- [x] Transactional resource staging, verified copied bytes, process locking, + interrupted-copy/swap recovery and backup preservation on failed rollback. +- [ ] Verify power-loss durability and release-tool interaction with resource + directory swaps on each OS; fault-injected process interruption is narrower. - [x] Shared monotonic Basic/Debug search metrics, granular bounded engine traces, actual worker query diagnostics, and headless JSON performance output. - [ ] Verify detailed performance UI interaction on native desktop platforms; @@ -84,7 +89,9 @@ 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. -- [ ] Execute Linux/macOS platform tests; create Windows/macOS/Linux installers. +- [x] Native Linux headless workspace tests: 93 passed, 11 explicitly ignored. +- [ ] Execute Linux native Google worker and desktop/macOS platform tests; + create Windows/macOS/Linux installers. - [ ] Signing/notarization configuration, upgrade/rollback validation, release instructions. ## 6. Local agents and optional answers @@ -115,8 +122,11 @@ Audit and measured priorities: [EMBEDDINGGEMMA2-AUDIT.md](EMBEDDINGGEMMA2-AUDIT. preserve filtering, source freshness, failures and actual warmup probes. - [x] Reduce semantic settle delay to 120 ms and retain old ready generation during rebuilding; test initial-generation and activation readiness boundaries. -- [ ] Bound whole IPC write/read deadlines and recover from a nonreading helper. -- [ ] Declare/test x86 CPU minimum or baseline/AVX2 runtime selection. +- [x] Bound whole IPC write/read deadlines and recover from a nonreading helper; + CPU-only real helper-process regressions cover blocked writes and responses. +- [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. - [ ] Benchmark genuine token-budgeted multi-sequence batching and isolation. - [ ] Implement and measure shared foreground/indexer resource budgets/priority. diff --git a/docs/SEMANTIC.md b/docs/SEMANTIC.md index 88e3152..20e3d4b 100644 --- a/docs/SEMANTIC.md +++ b/docs/SEMANTIC.md @@ -20,6 +20,12 @@ under the opt-in minilm-proof feature and tools/semantic-proof/HISTORICAL.md. The persistent native helper uses bounded NDJSON over stdin/stdout. It has no HTTP server or download code. Rust verifies the text model's length and SHA256 before starting it. Setup provisions assets; search never downloads them. +Each request has one 180-second IPC deadline across pipe writing and response +reading. Single-slot queues bound pending work. A stalled write or response +kills/reaps the helper and joins both IPC threads; only transport failures +qualify for a bounded CPU restart. Malformed/oversized responses remain +protocol errors. Cleanup assumes the shipped helper spawns no descendants +holding its pipe handles, matching the native implementation. Production device mode is auto, with available compiled acceleration and CPU fallback. Model/context initialization and returned inference/nonfinite-output diff --git a/docs/VALIDATION.md b/docs/VALIDATION.md index 4cd4be6..f58d129 100644 --- a/docs/VALIDATION.md +++ b/docs/VALIDATION.md @@ -7,6 +7,39 @@ and `SEARCHI_FORCE_CPU=1`. ## Current integration checkpoint (2026-10-12) +The reliability checkpoint passes 93 workspace tests on Windows and 93 on +native Linux (Ubuntu 24.04 under WSL); each explicitly ignores 11 optional +integration tests. Linux uses its native Rust toolchain and an isolated ext4 +build cache. This verifies headless core portability, not a Linux desktop, +Linux native Google worker, GPU or installer. Windows desktop passes 17 tests; +strict workspace and desktop Clippy checks pass. Five genuine Google model +integration tests pass in forced-CPU mode. + +Worker IPC now applies one deadline across sending and receiving, with bounded +queues and owned writer/reader threads. Transport failures terminate and reap +the worker; eligible GPU transport failures retry on CPU. Invalid protocol +responses remain errors. Fifteen semantic unit tests pass, including a blocked +pipe and disconnected writer; one optional integration test is ignored. + +Resource staging passes 24 tests on Windows (one symlink-permission skip) and +23 on Linux (two Windows-only skips). Locking, recovery, rollback preservation, +link rejection and concurrent edits before the final swap are covered. Actual +automatic-backend staging also succeeds. These are process-failure checks; +power-loss durability and concurrent arbitrary writes after the final check +remain release gates. + +The latest unsigned Windows archive launches outside the checkout with an +embedded frontend at `http://tauri.localhost/`, isolated data, one real indexed +document and a ready Google CPU semantic generation. The archive is +145,217,175 bytes, SHA-256 +`fb454a9dbf258f51044796ab24a5942a8f90804d0824ca0e2dc5f59b7847ec3b`. +Native accessibility confirms startup and status only. Input and screenshot +capture remain denied by the environment; native interaction, appearance, +clean-machine installation and GPU behavior are unverified. Release builds +without the embedded-frontend feature now fail explicitly. + +The following package-dispatch evidence predates this reliability checkpoint: + The normal Windows x64 package now includes baseline and optimized CPU DLLs with CPU/OS feature checks, plus the Vulkan backend. Shared DLLs use a common MSVC CRT: a first `/MT` prototype crashed because model file handles crossed diff --git a/tools/embeddinggemma2/stage_runtime.py b/tools/embeddinggemma2/stage_runtime.py index 1ce52a1..7bba1c9 100644 --- a/tools/embeddinggemma2/stage_runtime.py +++ b/tools/embeddinggemma2/stage_runtime.py @@ -1,16 +1,327 @@ """Stage the native worker for installers; optional offline model bundle.""" import argparse +from contextlib import contextmanager import hashlib import json import os import re from pathlib import Path import shutil +import stat +import time +import uuid ROOT = Path(__file__).resolve().parents[2] 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"} +def _is_link(path): + """Reject Windows reparse points, including junctions, on Python 3.11+.""" + path = Path(path) + if path.is_symlink(): + return True + try: + attributes = getattr(path.lstat(), "st_file_attributes", 0) + except FileNotFoundError: + return False + return bool(attributes & getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400)) + + +def _checked_path(path, parent): + """Validate absolute workspace boundaries before any move or recursive delete.""" + path, parent = Path(path).absolute(), Path(parent).resolve() + if _is_link(path) or path.parent.resolve() != parent: + raise RuntimeError("Staging path escapes its checked parent or is a link") + if path.exists() and path.resolve() != parent / path.name: + raise RuntimeError("Staging path resolves outside its checked parent") + return path + + +def _check_tree(directory): + if _is_link(directory): + raise RuntimeError("Resource staging refuses symlinks or junctions") + count = 0 + for parent, directories, files in os.walk(directory, followlinks=False): + for name in [*directories, *files]: + count += 1 + if count > 100000: + raise RuntimeError("Resource staging exceeds its file-count limit") + path = Path(parent) / name + if _is_link(path) or not path.resolve().is_relative_to(directory.resolve()): + raise RuntimeError("Resource staging refuses symlinks or junctions") + + +def _remove_tree(path, parent): + checked = _checked_path(path, parent) + if checked.exists(): + _check_tree(checked) + shutil.rmtree(checked) + + +def _move_directory(source, destination, parent): + """Bounded retry for transient Windows sharing locks during directory swaps.""" + for attempt in range(4): + source = _checked_path(source, parent) + destination = _checked_path(destination, parent) + try: + os.replace(source, destination) + return + except PermissionError: + if attempt == 3: + raise + time.sleep(0.05 * (2 ** attempt)) + + +def _record(path): + path = Path(path) + if _is_link(path) or path.absolute() != path.resolve() or not path.is_file(): + raise RuntimeError(f"Staging source is missing or is a link: {path.name}") + with path.open("rb") as stream: + return path.stat().st_size, hashlib.file_digest(stream, "sha256").hexdigest() + + +def _verified_copy(source, destination, expected=None): + source, destination = Path(source), Path(destination) + expected = expected or _record(source) + if destination.absolute() != destination.resolve(): + raise RuntimeError("Staging copy target resolves through a link") + shutil.copy2(source, destination) + if _record(destination) != expected: + raise RuntimeError(f"Staging source changed or copy was incomplete: {source.name}") + return str(destination) + + +@contextmanager +def staging_lock(parent): + """Kernel-held lock releases after process crashes; PID is informational.""" + path = _checked_path(parent / ".searchi-stage.lock", parent) + with path.open("a+b") as stream: + stream.seek(0, os.SEEK_END) + if stream.tell() == 0: + stream.write(b"\0") + stream.flush() + stream.seek(0) + try: + if os.name == "nt": + import msvcrt + msvcrt.locking(stream.fileno(), msvcrt.LK_NBLCK, 1) + else: + import fcntl + fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError as error: + raise RuntimeError("Another resource staging process holds the lock") from error + try: + stream.seek(1) + stream.truncate() + stream.write(f"pid={os.getpid()}".encode()) + stream.flush() + yield + finally: + stream.seek(0) + if os.name == "nt": + msvcrt.locking(stream.fileno(), msvcrt.LK_UNLCK, 1) + else: + fcntl.flock(stream.fileno(), fcntl.LOCK_UN) + + +def _journal_paths(destination, state): + if (not isinstance(state, dict) or state.get("schema") != 1 + or state.get("owner") != "searchi-runtime-staging" + or not isinstance(state.get("transaction"), str) + or not re.fullmatch(r"[0-9a-f]{32}", state["transaction"]) + or type(state.get("had_destination")) is not bool): + raise RuntimeError("Invalid resource staging recovery journal") + parent = destination.parent.resolve() + transaction = state["transaction"] + return (_checked_path(parent / f".resources.stage-{transaction}", parent), + _checked_path(parent / f".resources.backup-{transaction}", parent)) + + +def recover_staging(destination): + """Recover interrupted copy/swap while caller holds the staging lock. + + This handles process interruption, not certified power-loss durability. + An unsuccessful restoration retains the journal and original backup. + """ + parent = destination.parent.resolve() + destination = _checked_path(destination, parent) + journal = _checked_path(parent / ".searchi-stage.json", parent) + if not journal.exists(): + return + if journal.stat().st_size > 4096: + raise RuntimeError("Oversized resource staging recovery journal") + state = json.loads(journal.read_text(encoding="utf-8")) + candidate, backup = _journal_paths(destination, state) + for directory in [destination, candidate, backup]: + if directory.exists(): + _check_tree(directory) + marker = ".searchi-stage-complete" + + def complete(directory): + path = directory / marker + return path.is_file() and path.stat().st_size == 32 and path.read_text() == state["transaction"] + + if not destination.exists(): + if backup.exists(): + # If restoration fails, do not discard either recoverable directory. + _move_directory(backup, destination, parent) + elif candidate.exists() and complete(candidate): + _move_directory(candidate, destination, parent) + elif state["had_destination"]: + raise RuntimeError("Original staging directory and backup are both missing; recovery retained") + if destination.exists() and backup.exists(): + if not complete(destination): + raise RuntimeError("Ambiguous resource staging recovery; original backup retained") + _remove_tree(backup, parent) + _remove_tree(candidate, parent) + if destination.exists(): + completion = destination / marker + if completion.exists(): + if not complete(destination): + raise RuntimeError("Invalid resource staging completion marker") + completion.unlink() + journal.unlink() + + +def _previous_owned(destination): + names = {"searchi-gemma-worker.exe", "searchi-gemma-worker", "runtime-libraries.json", + "LLAMA-LICENSE.txt", "VULKAN-LICENSE.txt", "MSVC-REDISTRIBUTION.txt", "vulkan-1.dll"} + manifest_path = Path(__file__).with_name("model-manifest.json") + if manifest_path.stat().st_size > 65536: + raise RuntimeError("Pinned model manifest is oversized") + model_manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + files = model_manifest.get("files") if isinstance(model_manifest, dict) else None + if not isinstance(files, list) or not 1 <= len(files) <= 4: + raise RuntimeError("Invalid pinned model manifest") + for record in files: + name = record.get("path") if isinstance(record, dict) else None + if not isinstance(name, str) or not re.fullmatch(r"[A-Za-z0-9_-]+(?:\.[A-Za-z0-9_-]+)*\.gguf", name): + raise RuntimeError("Invalid pinned model resource path") + names.add(f"embeddinggemma2/{name}") + inventory = destination / "runtime-libraries.json" + if inventory.exists(): + if inventory.stat().st_size > 65536: + raise RuntimeError("Previous runtime inventory is oversized") + previous = json.loads(inventory.read_text(encoding="utf-8")) + records = previous.get("libraries") if isinstance(previous, dict) else None + if not isinstance(records, list) or len(records) > 64: + raise RuntimeError("Invalid previous runtime inventory") + for record in records: + name = record.get("path") if isinstance(record, dict) else None + if not isinstance(name, str) or not re.fullmatch(r"[A-Za-z0-9_-]+\.dll", name): + raise RuntimeError("Invalid previous runtime library path") + names.add(name) + return names + + +def _resource_key(path): + value = Path(path).as_posix() + return value.casefold() if os.name == "nt" else value + + +def _preserved_snapshot(destination, owned): + """Bounded content/tree snapshot detects edits to resources being retained.""" + if not destination.exists(): + return None + _check_tree(destination) + snapshot = {} + count = 0 + for parent, directories, files in os.walk(destination, followlinks=False): + for name in [*directories, *files]: + count += 1 + if count > 100000: + raise RuntimeError("Resource staging exceeds its snapshot file-count limit") + path = Path(parent) / name + key = _resource_key(path.relative_to(destination)) + if key in owned: + if not path.is_file(): + raise RuntimeError("Owned runtime resource is not a file") + continue + if path.is_dir(): + snapshot[key] = ("directory",) + else: + record = _record(path) + metadata = path.stat() + snapshot[key] = ("file", *record, metadata.st_mtime_ns, metadata.st_ino, metadata.st_dev) + return snapshot + + +def stage_resources(destination, sources, validate=None): + """Replace a verified complete resource directory and preserve other assets.""" + destination = Path(destination).absolute() + if destination.parent != destination.parent.resolve(): + raise RuntimeError("Staging parent resolves through a link outside its expected location") + destination.parent.mkdir(parents=True, exist_ok=True) + parent = destination.parent.resolve() + destination = _checked_path(destination, parent) + with staging_lock(parent): + recover_staging(destination) + if destination.exists(): + _check_tree(destination) + if (destination / ".searchi-stage-complete").exists(): + raise RuntimeError("Reserved staging completion marker exists without its journal") + owned = _previous_owned(destination) + owned_keys = {_resource_key(path) for path in owned} + preserved = _preserved_snapshot(destination, owned_keys) + records = [] + targets = set() + for source, relative in sources: + relative = Path(relative) + if relative.is_absolute() or relative.anchor or relative.drive or ".." in relative.parts or not relative.parts: + raise RuntimeError("Invalid staging resource target") + key = relative.as_posix().lower() + if key in targets: + raise RuntimeError("Duplicate staging resource target") + targets.add(key) + records.append((Path(source), relative, _record(source))) + state = {"schema": 1, "owner": "searchi-runtime-staging", "transaction": uuid.uuid4().hex, + "had_destination": destination.exists()} + candidate, backup = _journal_paths(destination, state) + journal = _checked_path(parent / ".searchi-stage.json", parent) + pending = _checked_path(parent / f".searchi-stage-journal-{state['transaction']}.pending", parent) + try: + with pending.open("x", encoding="utf-8") as stream: + json.dump(state, stream) + stream.flush() + os.fsync(stream.fileno()) + os.replace(pending, journal) + finally: + pending.unlink(missing_ok=True) + try: + if destination.exists(): + def ignore_owned(directory, names): + relative = Path(directory).relative_to(destination) + return [name for name in names if _resource_key(relative / name) in owned_keys] + + shutil.copytree(destination, candidate, copy_function=_verified_copy, ignore=ignore_owned) + else: + candidate.mkdir() + model_directory = candidate / "embeddinggemma2" + if model_directory.is_dir() and not any(model_directory.iterdir()): + if not model_directory.resolve().is_relative_to(candidate.resolve()): + raise RuntimeError("Owned model directory escapes candidate") + model_directory.rmdir() + for source, relative, expected in records: + target = candidate / relative + if not target.resolve().is_relative_to(candidate.resolve()): + raise RuntimeError("Staging resource target escapes candidate directory") + target.parent.mkdir(parents=True, exist_ok=True) + _verified_copy(source, target, expected) + _check_tree(candidate) + if validate: + validate(candidate) + if _preserved_snapshot(destination, owned_keys) != preserved: + raise RuntimeError("Preserved resources changed during staging; original retained") + (candidate / ".searchi-stage-complete").write_text(state["transaction"]) + if destination.exists(): + _move_directory(destination, backup, parent) + _move_directory(candidate, destination, parent) + recover_staging(destination) + except Exception: + recover_staging(destination) + raise + + def verify_runtime(source, required=False): """Preflight an explicit, bounded dependency inventory, never a DLL glob.""" inventory = source.parent / "runtime-libraries.json" @@ -74,6 +385,12 @@ def main(): parser.add_argument("--backend", choices=["cpu", "auto", "metal", "vulkan", "cuda"], default="auto") parser.add_argument("--bundle-model", action="store_true") args = parser.parse_args() + destination = ROOT / "apps/desktop/src-tauri/resources" + if destination.parent.absolute() != destination.parent.resolve(): + raise RuntimeError("Staging parent resolves through a link outside its expected location") + destination.parent.mkdir(parents=True, exist_ok=True) + with staging_lock(destination.parent): + recover_staging(destination) name = "searchi-gemma-worker.exe" if os.name == "nt" else "searchi-gemma-worker" source = ROOT / ".searchi/runtime" / args.backend if os.name == "nt": @@ -91,29 +408,30 @@ def main(): raise RuntimeError("Microsoft CRT redistribution notice is missing") dependencies.append(crt_notice) artifacts = [] + model_manifest = None if args.bundle_model: - manifest = json.loads(Path(__file__).with_name("model-manifest.json").read_text()) - artifacts = verify_bundle(ROOT, manifest) + model_manifest = json.loads(Path(__file__).with_name("model-manifest.json").read_text()) + artifacts = verify_bundle(ROOT, model_manifest) loader = None if os.name == "nt" and args.backend in {"auto", "vulkan"}: from provision_vulkan import provision _, loader = provision() - destination = ROOT / "apps/desktop/src-tauri/resources" - destination.mkdir(parents=True, exist_ok=True) - shutil.copy2(source, destination / name) - for dependency in dependencies: - shutil.copy2(dependency, destination / dependency.name) - # Include the upstream runtime license with the statically linked helper. - shutil.copy2(license_path, destination / "LLAMA-LICENSE.txt") + sources = [(source, source.name), (license_path, "LLAMA-LICENSE.txt")] + sources.extend((path, path.name) for path in dependencies) if loader: for name in ["vulkan-1.dll", "VULKAN-LICENSE.txt"]: if name not in {path.name for path in dependencies}: - shutil.copy2(loader / name, destination / name) - if args.bundle_model: - model_destination = destination / "embeddinggemma2" - model_destination.mkdir(exist_ok=True) - for artifact in artifacts: - shutil.copy2(artifact, model_destination / artifact.name) + sources.append((loader / name, name)) + sources.extend((path, f"embeddinggemma2/{path.name}") for path in artifacts) + + def validate(candidate): + verify_runtime(candidate / source.name, os.name == "nt" and args.backend in {"auto", "vulkan"}) + if model_manifest: + for artifact, spec in zip(artifacts, model_manifest["files"], strict=True): + if _record(candidate / "embeddinggemma2" / artifact.name) != (spec["bytes"], spec["sha256"]): + raise RuntimeError("Copied model checksum mismatch") + + stage_resources(destination, sources, validate) print(f"Staged native worker: {destination / source.name}") diff --git a/tools/embeddinggemma2/test_stage_runtime.py b/tools/embeddinggemma2/test_stage_runtime.py index bd01a78..a4c444d 100644 --- a/tools/embeddinggemma2/test_stage_runtime.py +++ b/tools/embeddinggemma2/test_stage_runtime.py @@ -2,6 +2,7 @@ import hashlib import json import os +import subprocess from pathlib import Path import tempfile import unittest @@ -11,6 +12,371 @@ import stage_runtime class StagingTests(unittest.TestCase): + def resources(self, directory): + root = Path(directory) + destination = root / "src-tauri/resources" + destination.mkdir(parents=True) + (destination / "searchi-gemma-worker.exe").write_bytes(b"previous worker") + (destination / "fonts").mkdir() + (destination / "fonts/Inter.woff2").write_bytes(b"font") + (destination / "media-runtime").mkdir() + (destination / "media-runtime/ffmpeg.exe").write_bytes(b"media") + (destination / "user-note.txt").write_bytes(b"preserve user file") + source = root / "worker.exe" + source.write_bytes(b"next worker") + return destination, source + + def test_failed_copy_preserves_previous_complete_resources(self): + with tempfile.TemporaryDirectory() as directory: + destination, source = self.resources(directory) + original = stage_runtime.shutil.copy2 + + def fail_worker(path, target, *args, **kwargs): + if Path(path) == source: + raise OSError("injected copy failure") + return original(path, target, *args, **kwargs) + + with patch.object(stage_runtime.shutil, "copy2", fail_worker): + with self.assertRaisesRegex(OSError, "copy failure"): + stage_runtime.stage_resources(destination, [(source, "searchi-gemma-worker.exe")]) + self.assertEqual((destination / "searchi-gemma-worker.exe").read_bytes(), b"previous worker") + self.assertEqual((destination / "user-note.txt").read_bytes(), b"preserve user file") + self.assertFalse((destination.parent / ".searchi-stage.json").exists()) + self.assertEqual(list(destination.parent.glob(".resources.stage-*")), []) + + def test_changed_copy_bytes_reject_before_swap(self): + with tempfile.TemporaryDirectory() as directory: + destination, source = self.resources(directory) + original = stage_runtime.shutil.copy2 + + def corrupt_worker(path, target, *args, **kwargs): + result = original(path, target, *args, **kwargs) + if Path(path) == source: + Path(target).write_bytes(b"bad worker!") + return result + + with patch.object(stage_runtime.shutil, "copy2", corrupt_worker): + with self.assertRaisesRegex(RuntimeError, "changed or copy was incomplete"): + stage_runtime.stage_resources(destination, [(source, "searchi-gemma-worker.exe")]) + self.assertEqual((destination / "searchi-gemma-worker.exe").read_bytes(), b"previous worker") + + def test_source_change_after_preflight_rejects_before_swap(self): + with tempfile.TemporaryDirectory() as directory: + destination, source = self.resources(directory) + original = stage_runtime.shutil.copy2 + + def mutate_source(path, target, *args, **kwargs): + if Path(path) == source: + source.write_bytes(b"evil worker") + return original(path, target, *args, **kwargs) + + with patch.object(stage_runtime.shutil, "copy2", mutate_source): + with self.assertRaisesRegex(RuntimeError, "changed or copy was incomplete"): + stage_runtime.stage_resources(destination, [(source, "searchi-gemma-worker.exe")]) + self.assertEqual((destination / "searchi-gemma-worker.exe").read_bytes(), b"previous worker") + + def test_resource_targets_reject_traversal_and_windows_drive_roots(self): + with tempfile.TemporaryDirectory() as directory: + destination, source = self.resources(directory) + targets = ["../outside"] + if os.name == "nt": + targets.extend(["C:outside", "\\outside"]) + for target in targets: + with self.subTest(target=target), self.assertRaisesRegex(RuntimeError, "resource target"): + stage_runtime.stage_resources(destination, [(source, target)]) + self.assertFalse((destination.parent / ".searchi-stage.json").exists()) + self.assertEqual((destination / "searchi-gemma-worker.exe").read_bytes(), b"previous worker") + + def test_locked_destination_rename_preserves_old_resources(self): + with tempfile.TemporaryDirectory() as directory: + destination, source = self.resources(directory) + original = stage_runtime.os.replace + + def fail_old_rename(path, target): + if Path(path) == destination: + raise PermissionError("destination is locked") + return original(path, target) + + with patch.object(stage_runtime.os, "replace", fail_old_rename): + with self.assertRaisesRegex(PermissionError, "locked"): + stage_runtime.stage_resources(destination, [(source, "searchi-gemma-worker.exe")]) + self.assertEqual((destination / "searchi-gemma-worker.exe").read_bytes(), b"previous worker") + self.assertFalse((destination.parent / ".searchi-stage.json").exists()) + + def test_interrupted_between_renames_restores_old_before_source_preflight(self): + class Interrupt(BaseException): + pass + + with tempfile.TemporaryDirectory() as directory: + destination, source = self.resources(directory) + original = stage_runtime.os.replace + + def interrupt_candidate(path, target): + if Path(path).name.startswith(".resources.stage-"): + raise Interrupt() + return original(path, target) + + with patch.object(stage_runtime.os, "replace", interrupt_candidate): + with self.assertRaises(Interrupt): + stage_runtime.stage_resources(destination, [(source, "searchi-gemma-worker.exe")]) + self.assertFalse(destination.exists()) + self.assertEqual(len(list(destination.parent.glob(".resources.backup-*"))), 1) + with self.assertRaisesRegex(RuntimeError, "source is missing"): + stage_runtime.stage_resources(destination, [(source.with_name("missing"), "searchi-gemma-worker.exe")]) + self.assertEqual((destination / "searchi-gemma-worker.exe").read_bytes(), b"previous worker") + self.assertFalse((destination.parent / ".searchi-stage.json").exists()) + + def test_interrupted_after_install_commits_verified_new_resources(self): + class Interrupt(BaseException): + pass + + with tempfile.TemporaryDirectory() as directory: + destination, source = self.resources(directory) + original = stage_runtime.os.replace + + def interrupt_after_candidate(path, target): + original(path, target) + if Path(path).name.startswith(".resources.stage-"): + raise Interrupt() + + with patch.object(stage_runtime.os, "replace", interrupt_after_candidate): + with self.assertRaises(Interrupt): + stage_runtime.stage_resources(destination, [(source, "searchi-gemma-worker.exe")]) + with self.assertRaisesRegex(RuntimeError, "source is missing"): + stage_runtime.stage_resources(destination, [(source.with_name("missing"), "searchi-gemma-worker.exe")]) + self.assertEqual((destination / "searchi-gemma-worker.exe").read_bytes(), b"next worker") + self.assertEqual((destination / "fonts/Inter.woff2").read_bytes(), b"font") + self.assertEqual((destination / "media-runtime/ffmpeg.exe").read_bytes(), b"media") + self.assertEqual((destination / "user-note.txt").read_bytes(), b"preserve user file") + self.assertEqual(list(destination.parent.glob(".resources.backup-*")), []) + self.assertFalse((destination / ".searchi-stage-complete").exists()) + + def test_failed_rollback_retains_backup_and_journal_for_retry(self): + with tempfile.TemporaryDirectory() as directory: + destination, source = self.resources(directory) + original = stage_runtime.os.replace + + def fail_candidate_and_restore(path, target): + if Path(path).name.startswith((".resources.stage-", ".resources.backup-")): + raise PermissionError("injected rename failure") + return original(path, target) + + with patch.object(stage_runtime.os, "replace", fail_candidate_and_restore): + with self.assertRaisesRegex(PermissionError, "rename failure"): + stage_runtime.stage_resources(destination, [(source, "searchi-gemma-worker.exe")]) + backup = next(destination.parent.glob(".resources.backup-*")) + self.assertEqual((backup / "searchi-gemma-worker.exe").read_bytes(), b"previous worker") + self.assertTrue((destination.parent / ".searchi-stage.json").exists()) + with self.assertRaisesRegex(RuntimeError, "source is missing"): + stage_runtime.stage_resources(destination, [(source.with_name("missing"), "searchi-gemma-worker.exe")]) + self.assertEqual((destination / "searchi-gemma-worker.exe").read_bytes(), b"previous worker") + + def test_profile_switch_removes_only_previous_inventory_libraries(self): + with tempfile.TemporaryDirectory() as directory: + destination, source = self.resources(directory) + old = destination / "ggml-cpu-old.dll" + old.write_bytes(b"old backend") + (destination / "user-plugin.dll").write_bytes(b"user plugin") + (destination / "runtime-libraries.json").write_text(json.dumps({"libraries": [{"path": old.name}]})) + stage_runtime.stage_resources(destination, [(source, "searchi-gemma-worker.exe")]) + self.assertFalse((destination / old.name).exists()) + self.assertFalse((destination / "runtime-libraries.json").exists()) + self.assertEqual((destination / "user-plugin.dll").read_bytes(), b"user plugin") + self.assertEqual((destination / "fonts/Inter.woff2").read_bytes(), b"font") + + def test_offline_to_online_removes_only_exact_pinned_model_pair(self): + with tempfile.TemporaryDirectory() as directory: + destination, source = self.resources(directory) + model_directory = destination / "embeddinggemma2" + model_directory.mkdir() + manifest = json.loads(Path(stage_runtime.__file__).with_name("model-manifest.json").read_text()) + for spec in manifest["files"]: + (model_directory / spec["path"]).write_bytes(b"old unverified model") + (model_directory / "user-note.txt").write_bytes(b"preserve unrelated asset") + stage_runtime.stage_resources(destination, [(source, "searchi-gemma-worker.exe")]) + for spec in manifest["files"]: + self.assertFalse((destination / "embeddinggemma2" / spec["path"]).exists()) + self.assertEqual((destination / "embeddinggemma2/user-note.txt").read_bytes(), b"preserve unrelated asset") + self.assertEqual((destination / "fonts/Inter.woff2").read_bytes(), b"font") + + def test_online_staging_removes_empty_owned_model_directory(self): + with tempfile.TemporaryDirectory() as directory: + destination, source = self.resources(directory) + model_directory = destination / "embeddinggemma2" + model_directory.mkdir() + manifest = json.loads(Path(stage_runtime.__file__).with_name("model-manifest.json").read_text()) + for spec in manifest["files"]: + (model_directory / spec["path"]).write_bytes(b"old model") + stage_runtime.stage_resources(destination, [(source, "searchi-gemma-worker.exe")]) + self.assertFalse((destination / "embeddinggemma2").exists()) + + def test_old_owned_worker_and_model_pair_are_never_copied(self): + with tempfile.TemporaryDirectory() as directory: + destination, source = self.resources(directory) + model_directory = destination / "embeddinggemma2" + model_directory.mkdir() + manifest = json.loads(Path(stage_runtime.__file__).with_name("model-manifest.json").read_text()) + old_paths = {destination / "searchi-gemma-worker.exe"} + for spec in manifest["files"]: + path = model_directory / spec["path"] + path.write_bytes(b"old owned model") + old_paths.add(path) + original = stage_runtime.shutil.copy2 + + original_record = stage_runtime._record + + def reject_owned_copy(path, target, *args, **kwargs): + if Path(path) in old_paths: + self.fail("Previously owned runtime/model asset was unnecessarily copied") + return original(path, target, *args, **kwargs) + + def reject_owned_hash(path): + if Path(path) in old_paths: + self.fail("Previously owned runtime/model asset was unnecessarily hashed") + return original_record(path) + + with patch.object(stage_runtime.shutil, "copy2", reject_owned_copy), patch.object(stage_runtime, "_record", reject_owned_hash): + stage_runtime.stage_resources(destination, [(source, "searchi-gemma-worker.exe")]) + self.assertEqual((destination / "searchi-gemma-worker.exe").read_bytes(), b"next worker") + self.assertFalse((destination / "embeddinggemma2").exists()) + + def test_concurrent_preserved_edit_aborts_and_keeps_latest_original(self): + with tempfile.TemporaryDirectory() as directory: + destination, source = self.resources(directory) + + def edit_original(_candidate): + (destination / "user-note.txt").write_bytes(b"latest user edit") + + with self.assertRaisesRegex(RuntimeError, "Preserved resources changed"): + stage_runtime.stage_resources(destination, [(source, "searchi-gemma-worker.exe")], edit_original) + self.assertEqual((destination / "user-note.txt").read_bytes(), b"latest user edit") + self.assertEqual((destination / "searchi-gemma-worker.exe").read_bytes(), b"previous worker") + self.assertFalse((destination.parent / ".searchi-stage.json").exists()) + + def test_concurrent_preserved_addition_aborts_and_keeps_new_file(self): + with tempfile.TemporaryDirectory() as directory: + destination, source = self.resources(directory) + + def add_original(_candidate): + (destination / "fonts/new-font.woff2").write_bytes(b"new user asset") + + with self.assertRaisesRegex(RuntimeError, "Preserved resources changed"): + stage_runtime.stage_resources(destination, [(source, "searchi-gemma-worker.exe")], add_original) + self.assertEqual((destination / "fonts/new-font.woff2").read_bytes(), b"new user asset") + self.assertEqual((destination / "searchi-gemma-worker.exe").read_bytes(), b"previous worker") + self.assertFalse((destination.parent / ".searchi-stage.json").exists()) + + def test_concurrent_staging_lock_fails_without_mutation(self): + with tempfile.TemporaryDirectory() as directory: + destination, source = self.resources(directory) + with stage_runtime.staging_lock(destination.parent): + with self.assertRaisesRegex(RuntimeError, "holds the lock"): + stage_runtime.stage_resources(destination, [(source, "searchi-gemma-worker.exe")]) + self.assertEqual((destination / "searchi-gemma-worker.exe").read_bytes(), b"previous worker") + + def test_malicious_recovery_journal_cannot_escape_sibling_paths(self): + with tempfile.TemporaryDirectory() as directory: + destination, source = self.resources(directory) + outside = Path(directory) / "outside" + outside.mkdir() + (outside / "user-data.txt").write_bytes(b"never remove") + journal = destination.parent / ".searchi-stage.json" + journal.write_text(json.dumps({"schema": 1, "owner": "searchi-runtime-staging", "transaction": "../outside", "had_destination": True})) + with self.assertRaisesRegex(RuntimeError, "recovery journal"): + stage_runtime.stage_resources(destination, [(source, "searchi-gemma-worker.exe")]) + self.assertEqual((outside / "user-data.txt").read_bytes(), b"never remove") + self.assertEqual((destination / "searchi-gemma-worker.exe").read_bytes(), b"previous worker") + + def test_existing_unrelated_symlink_rejects_before_stage_changes(self): + with tempfile.TemporaryDirectory() as directory: + destination, source = self.resources(directory) + outside = Path(directory) / "outside.txt" + outside.write_bytes(b"private data") + link = destination / "user-link.txt" + try: + link.symlink_to(outside) + except OSError: + self.skipTest("Host does not permit creating test symlinks") + try: + with self.assertRaisesRegex(RuntimeError, "symlinks or junctions"): + stage_runtime.stage_resources(destination, [(source, "searchi-gemma-worker.exe")]) + self.assertEqual(outside.read_bytes(), b"private data") + self.assertTrue(link.is_symlink()) + self.assertEqual((destination / "searchi-gemma-worker.exe").read_bytes(), b"previous worker") + finally: + link.unlink() + + def test_first_install_interrupt_after_verified_candidate_recovers(self): + class Interrupt(BaseException): + pass + + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + destination = root / "src-tauri/resources" + source = root / "worker.exe" + source.write_bytes(b"first worker") + original = stage_runtime.os.replace + + def interrupt_candidate(path, target): + if Path(path).name.startswith(".resources.stage-"): + raise Interrupt() + return original(path, target) + + with patch.object(stage_runtime.os, "replace", interrupt_candidate): + with self.assertRaises(Interrupt): + stage_runtime.stage_resources(destination, [(source, "searchi-gemma-worker.exe")]) + self.assertFalse(destination.exists()) + with stage_runtime.staging_lock(destination.parent): + stage_runtime.recover_staging(destination) + self.assertEqual((destination / "searchi-gemma-worker.exe").read_bytes(), b"first worker") + self.assertFalse((destination.parent / ".searchi-stage.json").exists()) + + @unittest.skipUnless(os.name == "nt", "Windows junction boundary regression") + def test_existing_junction_rejects_without_following_outside_directory(self): + with tempfile.TemporaryDirectory() as directory: + destination, source = self.resources(directory) + outside = Path(directory) / "outside" + outside.mkdir() + (outside / "private.txt").write_bytes(b"untouched") + link = destination / "user-junction" + result = subprocess.run(["cmd.exe", "/c", "mklink", "/J", str(link), str(outside)], capture_output=True) + if result.returncode: + self.skipTest("Host does not permit creating a test junction") + try: + self.assertTrue(stage_runtime._is_link(link)) + with self.assertRaisesRegex(RuntimeError, "symlinks or junctions"): + stage_runtime.stage_resources(destination, [(source, "searchi-gemma-worker.exe")]) + self.assertEqual((outside / "private.txt").read_bytes(), b"untouched") + self.assertEqual((destination / "searchi-gemma-worker.exe").read_bytes(), b"previous worker") + finally: + # Remove only the verified test junction, never its target tree. + self.assertEqual(link.absolute().parent, destination.absolute()) + self.assertTrue(stage_runtime._is_link(link)) + link.rmdir() + + def test_interrupted_partial_candidate_keeps_old_resources_and_recovers(self): + class Interrupt(BaseException): + pass + + with tempfile.TemporaryDirectory() as directory: + destination, source = self.resources(directory) + original = stage_runtime.shutil.copy2 + + def interrupt_copy(path, target, *args, **kwargs): + if Path(path) == source: + Path(target).write_bytes(b"partial") + raise Interrupt() + return original(path, target, *args, **kwargs) + + with patch.object(stage_runtime.shutil, "copy2", interrupt_copy): + with self.assertRaises(Interrupt): + stage_runtime.stage_resources(destination, [(source, "searchi-gemma-worker.exe")]) + self.assertEqual((destination / "searchi-gemma-worker.exe").read_bytes(), b"previous worker") + with stage_runtime.staging_lock(destination.parent): + stage_runtime.recover_staging(destination) + self.assertEqual(list(destination.parent.glob(".resources.stage-*")), []) + self.assertFalse((destination.parent / ".searchi-stage.json").exists()) + @unittest.skipUnless(os.name == "nt", "Windows app-local CRT staging") def test_missing_crt_notice_preserves_existing_staged_worker(self): with tempfile.TemporaryDirectory() as directory: