diff --git a/apps/desktop/src-tauri/Cargo.lock b/apps/desktop/src-tauri/Cargo.lock index 9b0e7c5..f438f29 100644 --- a/apps/desktop/src-tauri/Cargo.lock +++ b/apps/desktop/src-tauri/Cargo.lock @@ -3784,13 +3784,16 @@ name = "searchi-core" version = "0.1.0" dependencies = [ "anyhow", + "libc", "lopdf", "quick-xml", "rusqlite", "searchi-media", "serde", + "serde_json", "sha2 0.10.9", "walkdir", + "windows-sys 0.61.2", "zip", ] diff --git a/apps/desktop/src-tauri/src/main.rs b/apps/desktop/src-tauri/src/main.rs index 1ad99b8..0e066e7 100644 --- a/apps/desktop/src-tauri/src/main.rs +++ b/apps/desktop/src-tauri/src/main.rs @@ -45,6 +45,7 @@ struct AppState { semantic: Arc, watching: Arc, index_options: searchi_core::IndexOptions, + document_worker: PathBuf, previews: Arc, } @@ -446,6 +447,7 @@ async fn index_folder(root: String, state: State<'_, AppState>) -> Result) -> Result, roots: Vec, options: searchi_core::IndexOptions, + document_worker: PathBuf, ) -> Result { watching::WatchService::start(roots, options.excluded_paths.clone(), move |request| { if indexing @@ -488,6 +492,7 @@ fn start_folder_watching( let _ = app.emit("folder-indexing-changed", ()); let result = (|| -> Result { let mut index = SearchIndex::open(&database).map_err(|error| error.to_string())?; + index.set_document_worker(document_worker.clone()); let report = match request.paths { Some(paths) => index.index_paths_cancellable( &request.root, @@ -566,6 +571,14 @@ fn main() { )?; std::fs::create_dir_all(&directory)?; let database = directory.join("searchi.sqlite"); + // Bundle resources live outside the executable directory on macOS. + let document_worker = std::env::var_os("SEARCHI_EXTRACT_WORKER") + .map(PathBuf::from) + .unwrap_or(app.path().resource_dir()?.join(if cfg!(windows) { + "searchi-extract-worker.exe" + } else { + "searchi-extract-worker" + })); let lexical = Arc::new(Mutex::new(SearchIndex::open(&database)?)); let warm_database = database.clone(); let warm_target = lexical.clone(); @@ -632,7 +645,7 @@ fn main() { let indexing = Arc::new(AtomicBool::new(false)); let watching = Arc::new(start_folder_watching( app.handle().clone(), database.clone(), lexical.clone(), indexing.clone(), - settings.value.roots.clone(), index_options.clone(), + settings.value.roots.clone(), index_options.clone(), document_worker.clone(), ).map_err(std::io::Error::other)?); let semantic = Arc::new( SemanticState::new(model_directory, worker, media_tools) @@ -666,6 +679,7 @@ fn main() { semantic, watching, index_options, + document_worker, previews: previews.clone(), }); Ok(()) diff --git a/crates/searchi-core/src/extraction.rs b/crates/searchi-core/src/extraction.rs index 9e53a74..b19d8b1 100644 --- a/crates/searchi-core/src/extraction.rs +++ b/crates/searchi-core/src/extraction.rs @@ -1,4 +1,4 @@ -//! Bounded inputs and outputs; parser allocations and CPU are not process isolated. +//! Bounded snapshots; documents use the configured isolated process adapter. use crate::SourceLocation; use anyhow::{bail, Context, Result}; use quick_xml::{events::Event, name::ResolveResult, NsReader, XmlVersion}; @@ -7,8 +7,36 @@ use std::{ fs::File, io::{Cursor, Read}, path::Path, + sync::atomic::AtomicBool, + time::Duration, }; +pub(crate) enum DocumentExtraction { + Worker(crate::extraction_process::ProcessConfig), + InternalInProcess, +} + +impl DocumentExtraction { + pub fn normal() -> Result { + let executable = match std::env::var_os("SEARCHI_EXTRACT_WORKER") { + Some(path) => std::path::PathBuf::from(path), + None => std::env::current_exe()?.with_file_name(if cfg!(windows) { + "searchi-extract-worker.exe" + } else { + "searchi-extract-worker" + }), + }; + Ok(Self::worker(executable)) + } + + pub fn worker(executable: std::path::PathBuf) -> Self { + Self::Worker(crate::extraction_process::ProcessConfig { + executable, + timeout: Duration::from_secs(30), + }) + } +} + const MAX_TEXT: usize = 2 * 1024 * 1024; const MAX_DOCUMENT: usize = 32 * 1024 * 1024; const MAX_XML: usize = 8 * 1024 * 1024; @@ -75,17 +103,65 @@ pub(crate) fn read_source(path: &Path) -> Result> { Ok(Some(SourceInput { bytes, hash })) } -pub(crate) fn extract(path: &Path, input: SourceInput) -> Result>> { +pub(crate) fn verify_document_snapshot( + path: &Path, + expected_hash: &str, + initial: &std::fs::Metadata, +) -> Result<()> { + let verified = read_source(path)?.context("Document source no longer readable")?; + let after = std::fs::metadata(path)?; + if verified.hash != expected_hash + || after.len() != initial.len() + || after.modified()? != initial.modified()? + { + bail!("Document changed during extraction; retry indexing; reconciliation rolled back"); + } + Ok(()) +} + +pub(crate) fn extract( + path: &Path, + input: SourceInput, + documents: &DocumentExtraction, + cancelled: Option<&AtomicBool>, +) -> Result>> { let extension = path .extension() .and_then(|value| value.to_str()) .unwrap_or("") .to_ascii_lowercase(); - if extension == "pdf" { - return pdf(&input.bytes).map(Some); - } - if extension == "docx" { - return docx(&input.bytes).map(Some); + let format = match extension.as_str() { + "pdf" => Some(crate::extraction_protocol::DocumentFormat::Pdf), + "docx" => Some(crate::extraction_protocol::DocumentFormat::Docx), + _ => None, + }; + if let Some(format) = format { + if matches!(documents, DocumentExtraction::InternalInProcess) { + return document_bytes(format, &input.bytes).map(Some); + } + let DocumentExtraction::Worker(config) = documents else { + unreachable!() + }; + let request = crate::extraction_protocol::RequestHeader::from_bytes(format, &input.bytes)?; + let never_cancelled = AtomicBool::new(false); + let response = crate::extraction_process::run( + config, + request, + input.bytes, + cancelled.unwrap_or(&never_cancelled), + )?; + return match response.outcome { + crate::extraction_protocol::ExtractionOutcome::Extracted { passages } => Ok(Some( + passages + .into_iter() + .map(|p| Passage { + text: p.text, + source: p.source, + }) + .collect(), + )), + crate::extraction_protocol::ExtractionOutcome::Rejected { reason } => bail!("{reason}"), + }; } let code = crate::is_code_path(path); if input.bytes.contains(&0) { @@ -327,8 +403,28 @@ mod provenance_tests { format!("{:x}", Sha256::digest(b"original parser input")) ); std::fs::write(&path, "replacement on disk")?; - let passages = extract(&path, input)?.unwrap(); + let passages = + extract(&path, input, &DocumentExtraction::InternalInProcess, None)?.unwrap(); assert_eq!(passages[0].text, "original parser input"); Ok(()) } + + #[test] + fn document_publication_rejects_same_stat_byte_changes() -> Result<()> { + let directory = tempfile::tempdir()?; + let path = directory.path().join("snapshot.pdf"); + std::fs::write(&path, b"original")?; + let metadata = std::fs::metadata(&path)?; + let input = read_source(&path)?.unwrap(); + verify_document_snapshot(&path, &input.hash, &metadata)?; + std::fs::write(&path, b"modified")?; + File::options() + .write(true) + .open(&path)? + .set_times(std::fs::FileTimes::new().set_modified(metadata.modified()?))?; + assert_eq!(std::fs::metadata(&path)?.len(), metadata.len()); + assert_eq!(std::fs::metadata(&path)?.modified()?, metadata.modified()?); + assert!(verify_document_snapshot(&path, &input.hash, &metadata).is_err()); + Ok(()) + } } diff --git a/crates/searchi-core/src/lib.rs b/crates/searchi-core/src/lib.rs index 639d6e3..d6d3e35 100644 --- a/crates/searchi-core/src/lib.rs +++ b/crates/searchi-core/src/lib.rs @@ -1,8 +1,8 @@ //! Offline lexical search engine shared by the CLI and desktop shell. //! //! Reconciliation is atomic. Incomplete directory walks never delete records. -//! Text and document extraction have byte limits, but still run in this process. -//! Worker isolation remains a production milestone. +//! Document extraction uses a bounded helper process; plain text/code snapshots +//! remain in-process. Native platform release validation remains a milestone. mod code; mod extraction; pub mod extraction_process; @@ -28,6 +28,8 @@ use walkdir::WalkDir; const SCHEMA_VERSION: i64 = 13; const MAX_RESULTS: usize = 1000; +type PreviousExtraction = (i64, i64, String, Option, Option); + #[derive(Debug, Default, Serialize)] pub struct IndexReport { pub discovered: usize, @@ -113,6 +115,7 @@ pub struct SearchIndex { connection: Connection, database_path: Option, gram_frequencies: RefCell>, + document_extraction: extraction::DocumentExtraction, } impl SearchIndex { @@ -287,9 +290,22 @@ impl SearchIndex { connection, database_path, gram_frequencies: RefCell::new(HashMap::new()), + document_extraction: extraction::DocumentExtraction::normal()?, }) } + /// Select the trusted packaged document helper (for example desktop resources). + /// Failure to launch never falls back to parsing documents in this process. + pub fn set_document_worker(&mut self, executable: impl Into) { + self.document_extraction = extraction::DocumentExtraction::worker(executable.into()); + } + + /// Internal parser-fixture adapter. Never enable in a distributed application. + #[doc(hidden)] + pub fn use_in_process_document_parser_for_tests(&mut self) { + self.document_extraction = extraction::DocumentExtraction::InternalInProcess; + } + /// SHA-256 of the exact raw bytes used for the indexed extraction. Legacy /// records return None until reindexed; callers must not infer provenance /// from size/mtime or hash a later file read on their behalf. @@ -653,11 +669,11 @@ impl SearchIndex { .and_then(|value| value.to_str()) .unwrap_or("") .to_ascii_lowercase(); - let previous: Option<(i64, i64, String, Option)> = transaction + let previous: Option = transaction .query_row( - "SELECT size,mtime,root,extracted_sha256 FROM files WHERE path=?1", + "SELECT size,mtime,root,extracted_sha256,extraction_error FROM files WHERE path=?1", [&path], - |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?, row.get(4)?)), ) .optional()?; // Supported inputs are bounded and fingerprinted even when file @@ -685,25 +701,41 @@ impl SearchIndex { .and_then(|input| input.as_ref()) .map(|input| input.hash.clone()); if !media_hint - && previous - .as_ref() - .is_some_and(|(old_size, old_mtime, old_root, old_hash)| { + && previous.as_ref().is_some_and( + |(old_size, old_mtime, old_root, old_hash, old_error)| { *old_size == size && *old_mtime == mtime && old_root == &root_key && input.is_ok() && old_hash == &source_hash - }) + && old_error.is_none() + }, + ) { report.unchanged += 1; continue; } let extracted = input.and_then(|input| match input { - Some(input) => extraction::extract(entry.path(), input), + Some(input) => extraction::extract( + entry.path(), + input, + &self.document_extraction, + cancelled, + ), None => Ok(None), }); let (extracted, extraction_error) = match extracted { Ok(extracted) => (extracted, None), + Err(error) + if error + .downcast_ref::() + .is_some_and(|error| { + matches!(error, extraction_process::ProcessError::Cancelled) + }) => + { + return Err(error) + .context("Index reconciliation cancelled; transaction rolled back"); + } Err(error) if matches!(extension.as_str(), "pdf" | "docx") => { report.failures += 1; let reason: String = format!("{error:#}").chars().take(2048).collect(); @@ -714,6 +746,12 @@ impl SearchIndex { .with_context(|| format!("read {path}; reconciliation rolled back")) } }; + check_index_cancelled(cancelled)?; + if matches!(extension.as_str(), "pdf" | "docx") { + if let Some(hash) = &source_hash { + extraction::verify_document_snapshot(entry.path(), hash, &metadata)?; + } + } if extracted.is_none() { report.skipped += 1; } diff --git a/crates/searchi-core/tests/document_publication.rs b/crates/searchi-core/tests/document_publication.rs new file mode 100644 index 0000000..fa86d73 --- /dev/null +++ b/crates/searchi-core/tests/document_publication.rs @@ -0,0 +1,107 @@ +//! Deterministic parent/index publication tests using an owned protocol fixture. +//! Genuine parser execution is covered separately in documents.rs. +use anyhow::Result; +use searchi_core::{ + extraction_protocol::{DocumentFormat, RequestHeader}, + IndexOptions, SearchIndex, +}; +use std::{ + fs, + sync::atomic::{AtomicBool, Ordering}, + time::{Duration, Instant}, +}; + +#[test] +fn mid_worker_same_stat_changes_and_cancellation_roll_back_index() -> Result<()> { + for cancel in [false, true] { + let directory = tempfile::tempdir()?; + let root = directory.path().join("files"); + fs::create_dir(&root)?; + let source = root.join("contract.pdf"); + fs::write(root.join("notes.txt"), "preserved evidence")?; + let mut index = SearchIndex::open(":memory:")?; + index.index_folder(&root)?; + fs::write(&source, b"original")?; + let initial = fs::metadata(&source)?; + let digest = RequestHeader::from_bytes(DocumentFormat::Pdf, b"original")?.input_sha256; + let marker = directory.path().join("admitted"); + let release = directory.path().join("release"); + let fixture_source = directory.path().join("worker.rs"); + let executable = directory + .path() + .join(format!("fixture{}", std::env::consts::EXE_SUFFIX)); + let response = serde_json::json!({ + "version":1,"input_sha256":digest, + "outcome":{"status":"extracted","passages":[{"text":"stale document text","source":{"kind":"page","number":1}}]} + }).to_string(); + fs::write( + &fixture_source, + format!( + r#" +use std::io::{{Read,Write}}; +fn main() {{ + let mut bytes=Vec::new(); std::io::stdin().read_to_end(&mut bytes).unwrap(); + std::fs::write({marker:?}, b"admitted").unwrap(); + while !std::path::Path::new({release:?}).exists() {{ std::thread::sleep(std::time::Duration::from_millis(5)); }} + let response={response:?}.as_bytes(); let mut out=std::io::stdout(); + out.write_all(&(response.len() as u32).to_le_bytes()).unwrap(); out.write_all(response).unwrap(); +}} +"# + ), + )?; + let status = std::process::Command::new("rustc") + .arg(&fixture_source) + .arg("-o") + .arg(&executable) + .status()?; + anyhow::ensure!(status.success(), "Compile owned protocol fixture"); + index.set_document_worker(executable); + let cancelled = AtomicBool::new(false); + std::thread::scope(|scope| { + let changer = scope.spawn(|| -> Result<()> { + let deadline = Instant::now() + Duration::from_secs(10); + while !marker.exists() && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(5)); + } + if !marker.exists() { + cancelled.store(true, Ordering::Release); + anyhow::bail!("Worker did not admit fixture input"); + } + if cancel { + cancelled.store(true, Ordering::Release); + } else { + fs::write(&source, b"modified")?; + fs::File::options() + .write(true) + .open(&source)? + .set_times(fs::FileTimes::new().set_modified(initial.modified()?))?; + assert_eq!(fs::metadata(&source)?.len(), initial.len()); + assert_eq!(fs::metadata(&source)?.modified()?, initial.modified()?); + fs::write(&release, b"release")?; + } + Ok(()) + }); + let result = + index.index_folder_cancellable(&root, &IndexOptions::default(), &cancelled); + assert!( + result.is_err(), + "Mutation or cancellation must roll back publication" + ); + let message = format!("{:#}", result.unwrap_err()); + assert!( + message.contains(if cancel { + "cancelled" + } else { + "changed during extraction" + }), + "{message}" + ); + changer.join().unwrap().unwrap(); + }); + assert_eq!(index.stats()?.files, 1); + assert_eq!(index.search("evidence", 10)?.len(), 1); + assert!(index.search("stale", 10)?.is_empty()); + assert!(!index.contains_path(&source.to_string_lossy())?); + } + Ok(()) +} diff --git a/crates/searchi-core/tests/documents.rs b/crates/searchi-core/tests/documents.rs index 6b983a6..3c71191 100644 --- a/crates/searchi-core/tests/documents.rs +++ b/crates/searchi-core/tests/documents.rs @@ -57,6 +57,7 @@ fn compressed_document_limits_are_enforced() -> Result<()> { ); docx(&root.path().join("over-limit.docx"), &xml)?; let mut index = SearchIndex::open(":memory:")?; + index.use_in_process_document_parser_for_tests(); assert_eq!(index.index_folder(root.path())?.failures, 2); assert_eq!(index.stats()?.content_files, 0); assert!(index.search("bomb.pdf", 10)?[0].extraction_error.is_some()); @@ -90,6 +91,7 @@ fn encrypted_pdf_is_explicitly_unsupported() -> Result<()> { document.encrypt(&state)?; document.save(&path)?; let mut index = SearchIndex::open(":memory:")?; + index.use_in_process_document_parser_for_tests(); assert_eq!(index.index_folder(root.path())?.failures, 1); assert!(index.search("protected.pdf", 10)?[0] .extraction_error @@ -122,6 +124,72 @@ fn docx(path: &Path, xml: &str) -> Result<()> { Ok(()) } +#[test] +fn missing_worker_preserves_filename_clears_passages_and_retries() -> Result<()> { + let root = tempfile::tempdir()?; + pdf(&root.path().join("agreement.pdf"))?; + let mut index = SearchIndex::open(":memory:")?; + index.use_in_process_document_parser_for_tests(); + index.index_folder(root.path())?; + assert!(!index.search("cancellation", 10)?.is_empty()); + index.set_document_worker(root.path().join("missing-helper.exe")); + // Force a changed snapshot so a previous successful extraction is replaced. + pdf_with_text( + &root.path().join("agreement.pdf"), + &["different cancellation"], + )?; + assert_eq!(index.index_folder(root.path())?.failures, 1); + assert!(index.search("cancellation", 10)?.is_empty()); + assert!(index.search("agreement.pdf", 10)?[0] + .extraction_error + .is_some()); + // A same-stat, same-byte failed source must not be skipped on retry. + index.use_in_process_document_parser_for_tests(); + assert_eq!(index.index_folder(root.path())?.indexed, 1); + assert!(!index.search("cancellation", 10)?.is_empty()); + Ok(()) +} + +#[test] +#[ignore = "requires built helper in SEARCHI_EXTRACT_WORKER"] +fn normal_adapter_uses_real_document_worker_and_cancellation_rolls_back() -> Result<()> { + anyhow::ensure!( + std::env::var_os("SEARCHI_EXTRACT_WORKER").is_some(), + "Set actual document helper path" + ); + let root = tempfile::tempdir()?; + pdf(&root.path().join("agreement.pdf"))?; + let mut index = SearchIndex::open(":memory:")?; + assert_eq!(index.index_folder(root.path())?.failures, 0); + assert_eq!( + index.search("cancellation", 10)?[0] + .source + .as_ref() + .unwrap() + .number, + 2 + ); + let before = index.stats()?.files; + let cancelled = std::sync::atomic::AtomicBool::new(true); + assert!(index + .index_folder_cancellable( + root.path(), + &searchi_core::IndexOptions::default(), + &cancelled + ) + .is_err()); + assert_eq!(index.stats()?.files, before); + assert_eq!( + index.search("cancellation", 10)?[0] + .source + .as_ref() + .unwrap() + .number, + 2 + ); + Ok(()) +} + #[test] fn real_pdf_and_docx_return_matching_locations_and_persist() -> Result<()> { let root = tempfile::tempdir()?; @@ -132,6 +200,7 @@ fn real_pdf_and_docx_return_matching_locations_and_persist() -> Result<()> { )?; let db = root.path().join("index.db"); let mut index = SearchIndex::open(&db)?; + index.use_in_process_document_parser_for_tests(); assert_eq!(index.index_folder(root.path())?.indexed, 2); let result = index.search("cancellation", 10)?; assert_eq!(result[0].source.as_ref().unwrap().kind, "page"); @@ -180,6 +249,7 @@ fn corrupt_document_reports_failure_and_preserves_filename_search() -> Result<() let root = tempfile::tempdir()?; fs::write(root.path().join("notes.txt"), "persistent evidence")?; let mut index = SearchIndex::open(":memory:")?; + index.use_in_process_document_parser_for_tests(); index.index_folder(root.path())?; fs::write(root.path().join("broken.pdf"), b"not a PDF")?; assert_eq!(index.index_folder(root.path())?.failures, 1); @@ -199,7 +269,8 @@ fn corrupt_document_reports_failure_and_preserves_filename_search() -> Result<() index.index_folder(root.path())?; assert_eq!(index.search("cancellation", 10)?.len(), 1); fs::write(root.path().join("agreement.pdf"), "corrupted replacement")?; - assert_eq!(index.index_folder(root.path())?.failures, 1); + // Previously rejected DOCX is retried too; failures are per current scan. + assert_eq!(index.index_folder(root.path())?.failures, 2); assert!(index.search("cancellation", 10)?.is_empty()); assert!(index.search("agreement.pdf", 10)?[0] .extraction_error diff --git a/docs/EXTRACTION-ISOLATION.md b/docs/EXTRACTION-ISOLATION.md index 2d9c9ec..bc31020 100644 --- a/docs/EXTRACTION-ISOLATION.md +++ b/docs/EXTRACTION-ISOLATION.md @@ -126,6 +126,27 @@ the initially ignored actual-worker core test then passed explicitly, giving 60 executed checks. Strict all-target Clippy passed for both crates. No GPU or model inference was used. Windows Job configuration/verification is exercised, but adversarial memory-exhaustion behavior and native Unix limits remain open. -Default indexing still uses the earlier in-process parser; worker discovery, -post-parser source validation, production adapter activation and required -release staging are the next steps. No published installer contains this helper. +## Shared indexing adapter + +Current source uses the helper for PDF/DOCX by default. Headless clients discover +it beside their executable; desktop indexing explicitly selects the bundle +resource directory, including watcher reconciliation. An internal environment +override supports validation. The in-process adapter is an explicit test API, +never a failure fallback. Missing or failed helpers leave filenames searchable, +record an extraction error, clear obsolete passages and retry on later rescans. + +After parsing, bounded rereading checks the original byte digest and metadata +before publication. Cancellation rolls back reconciliation. Six document tests +passed, including genuine PDF extraction through the normal worker adapter. +A controlled subprocess test also passed mutation with identical size/restored +mtime during parsing and cancellation during parsing; both preserve the old +index through rollback. Strict core Clippy and formatting passed. An earlier +full-core run hit a two-second parent fixture timeout; that broader run is not +claimed green by these focused results. + +Required helper build/staging and release packaging remain open. No published +installer contains this helper. Do not publish current source without it. + +Desktop integration validation: 46 native tests passed on Windows with CPU-only +settings. The desktop lockfile now records the shared process adapter dependencies. +This checks compiled integration, not installed helper discovery or native UI interaction. diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 4d5b658..8eba803 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -314,3 +314,8 @@ Completion requires recorded evidence; targets are not measured results. The partial features above do not close broader production milestones such as crash-resumable jobs, fuzzy retrieval, watcher reliability or performance targets. + +## Document worker follow-up + +- [x] Shared PDF/DOCX process adapter, exact-byte post-parser validation, cancellation rollback and desktop resource discovery. See [evidence](EXTRACTION-ISOLATION.md). +- [ ] Build and require the extraction helper in desktop/headless release packages; test relocated package indexing before publishing.