feat: add desktop document search and verified native CPU embeddings

crates/searchi-core/src/extraction.rs

@@ -0,0 +1,267 @@
+//! Bounded inputs and outputs; parser allocations and CPU are not process isolated.
+use crate::SourceLocation;
+use anyhow::{bail, Context, Result};
+use quick_xml::{events::Event, name::ResolveResult, NsReader, XmlVersion};
+use std::{
+    fs::File,
+    io::{Cursor, Read},
+    path::Path,
+};
+
+const MAX_TEXT: usize = 2 * 1024 * 1024;
+const MAX_DOCUMENT: usize = 32 * 1024 * 1024;
+const MAX_XML: usize = 8 * 1024 * 1024;
+const MAX_PDF_STREAM: usize = 4 * 1024 * 1024;
+const MAX_PASSAGES: usize = 10_000;
+
+pub(crate) struct Passage {
+    pub text: String,
+    pub source: SourceLocation,
+}
+
+pub(crate) fn extract(path: &Path) -> Result<Option<Vec<Passage>>> {
+    let extension = path
+        .extension()
+        .and_then(|value| value.to_str())
+        .unwrap_or("")
+        .to_ascii_lowercase();
+    if extension == "pdf" || extension == "docx" {
+        let bytes = read_bounded(path, MAX_DOCUMENT)?;
+        return if extension == "pdf" {
+            pdf(&bytes).map(Some)
+        } else {
+            docx(&bytes).map(Some)
+        };
+    }
+    if !matches!(
+        extension.as_str(),
+        "txt"
+            | "md"
+            | "rs"
+            | "py"
+            | "js"
+            | "ts"
+            | "tsx"
+            | "jsx"
+            | "json"
+            | "toml"
+            | "yaml"
+            | "yml"
+            | "csv"
+            | "html"
+            | "css"
+            | "xml"
+            | "log"
+            | "sh"
+    ) {
+        return Ok(None);
+    }
+    let file = File::open(path)?;
+    if file.metadata()?.len() > MAX_TEXT as u64 {
+        return Ok(None);
+    }
+    let mut bytes = Vec::new();
+    file.take(MAX_TEXT as u64 + 1).read_to_end(&mut bytes)?;
+    if bytes.len() > MAX_TEXT || bytes.contains(&0) {
+        return Ok(None);
+    }
+    let Ok(text) = String::from_utf8(bytes) else {
+        return Ok(None);
+    };
+    let lines = text.split_inclusive('\n').collect::<Vec<_>>();
+    Ok(Some(
+        lines
+            .chunks(64)
+            .enumerate()
+            .map(|(index, lines)| Passage {
+                text: lines.concat(),
+                source: SourceLocation {
+                    kind: "line".into(),
+                    number: (index * 64 + 1) as u32,
+                },
+            })
+            .collect(),
+    ))
+}
+
+fn read_bounded(path: &Path, maximum: usize) -> Result<Vec<u8>> {
+    let file = File::open(path)?;
+    if file.metadata()?.len() > maximum as u64 {
+        bail!("document input exceeds {maximum} bytes");
+    }
+    let mut bytes = Vec::new();
+    file.take(maximum as u64 + 1).read_to_end(&mut bytes)?;
+    if bytes.len() > maximum {
+        bail!("document input exceeds {maximum} bytes");
+    }
+    Ok(bytes)
+}
+
+fn pdf(bytes: &[u8]) -> Result<Vec<Passage>> {
+    let document = lopdf::Document::load_mem_with_options(
+        bytes,
+        lopdf::LoadOptions {
+            strict: true,
+            max_decompressed_size: Some(MAX_PDF_STREAM),
+            ..Default::default()
+        },
+    )
+    .context("invalid, encrypted, or over-limit PDF")?;
+    if document.was_encrypted() || document.is_encrypted() {
+        bail!("encrypted PDF is unsupported");
+    }
+    let pages = document.get_pages();
+    if pages.is_empty() {
+        bail!("PDF has no pages");
+    }
+    if pages.len() > 1000 {
+        bail!("PDF exceeds 1000 pages");
+    }
+    let mut passages = Vec::new();
+    let mut total = 0;
+    for number in pages.keys() {
+        let text = document
+            .extract_text_with_limit(&[*number], MAX_PDF_STREAM)
+            .with_context(|| format!("PDF text extraction failed on page {number}"))?;
+        total += text.len();
+        if total > MAX_TEXT {
+            bail!("PDF text exceeds {MAX_TEXT} bytes");
+        }
+        if !text.trim().is_empty() {
+            passages.push(Passage {
+                text,
+                source: SourceLocation {
+                    kind: "page".into(),
+                    number: *number,
+                },
+            });
+        }
+    }
+    Ok(passages)
+}
+
+fn docx(bytes: &[u8]) -> Result<Vec<Passage>> {
+    let mut archive =
+        zip::ZipArchive::new(Cursor::new(bytes)).context("invalid DOCX ZIP container")?;
+    if archive.len() > 10_000 {
+        bail!("DOCX exceeds 10000 archive entries");
+    }
+    let mut xml = String::new();
+    let part = archive
+        .by_name("word/document.xml")
+        .context("DOCX lacks word/document.xml")?;
+    if part.size() > MAX_XML as u64 {
+        bail!("DOCX document XML exceeds {MAX_XML} bytes");
+    }
+    part.take(MAX_XML as u64 + 1)
+        .read_to_string(&mut xml)
+        .context("DOCX XML must be valid UTF-8")?;
+    if xml.len() > MAX_XML {
+        bail!("DOCX document XML exceeds {MAX_XML} bytes");
+    }
+    let mut reader = NsReader::from_str(&xml);
+    let mut passages = Vec::new();
+    let mut paragraph = String::new();
+    let mut number = 0;
+    let mut in_paragraph = false;
+    let mut in_text = false;
+    let mut total = 0;
+    let mut depth = 0usize;
+    let mut saw_document = false;
+    loop {
+        let (namespace, event) = reader.read_resolved_event().context("invalid DOCX XML")?;
+        let word_namespace = matches!(namespace, ResolveResult::Bound(value) if matches!(value.as_ref(),
+            "http://schemas.openxmlformats.org/wordprocessingml/2006/main" | "http://purl.oclc.org/ooxml/wordprocessingml/main"));
+        match &event {
+            Event::Start(start) => {
+                if depth == 0 {
+                    if saw_document || !word_namespace || start.local_name().as_ref() != "document"
+                    {
+                        bail!("invalid DOCX document root");
+                    }
+                    saw_document = true;
+                }
+                depth += 1;
+                if depth > 128 {
+                    bail!("DOCX XML exceeds 128 nesting levels");
+                }
+            }
+            Event::End(_) => depth = depth.checked_sub(1).context("invalid DOCX XML depth")?,
+            Event::Empty(_) if depth == 0 => bail!("invalid DOCX document root"),
+            _ => {}
+        }
+        match event {
+            Event::Start(event) if word_namespace => match event.local_name().as_ref() {
+                "p" => {
+                    if in_paragraph {
+                        bail!("nested DOCX paragraphs are unsupported");
+                    }
+                    number += 1;
+                    if number > MAX_PASSAGES {
+                        bail!("DOCX exceeds 10000 paragraphs");
+                    }
+                    in_paragraph = true;
+                    paragraph.clear();
+                }
+                "t" => in_text = in_paragraph,
+                _ => {}
+            },
+            Event::End(event) if word_namespace => match event.local_name().as_ref() {
+                "t" => in_text = false,
+                "p" => {
+                    in_paragraph = false;
+                    in_text = false;
+                    total += paragraph.len();
+                    if total > MAX_TEXT {
+                        bail!("DOCX text exceeds {MAX_TEXT} bytes");
+                    }
+                    if !paragraph.trim().is_empty() {
+                        passages.push(Passage {
+                            text: std::mem::take(&mut paragraph),
+                            source: SourceLocation {
+                                kind: "paragraph".into(),
+                                number: number as u32,
+                            },
+                        });
+                    }
+                }
+                _ => {}
+            },
+            Event::Empty(event) if word_namespace => match event.local_name().as_ref() {
+                "tab" if in_paragraph => paragraph.push('\t'),
+                "br" | "cr" if in_paragraph => paragraph.push('\n'),
+                "p" => {
+                    number += 1;
+                    if number > MAX_PASSAGES {
+                        bail!("DOCX exceeds 10000 paragraphs");
+                    }
+                }
+                _ => {}
+            },
+            Event::Text(event) if in_text => {
+                paragraph.push_str(&event.xml_content(XmlVersion::Implicit1_0))
+            }
+            Event::CData(event) if in_text => {
+                paragraph.push_str(&event.xml_content(XmlVersion::Implicit1_0))
+            }
+            Event::GeneralRef(event) if in_text => {
+                let reference = format!("&{};", event.as_ref());
+                paragraph.push_str(
+                    &quick_xml::escape::unescape(&reference).context("unsupported DOCX entity")?,
+                );
+            }
+            Event::DocType(_) => bail!("DOCX XML document types are unsupported"),
+            Event::Eof => {
+                if in_paragraph || depth != 0 || !saw_document {
+                    bail!("incomplete DOCX document");
+                }
+                break;
+            }
+            _ => {}
+        }
+        if total + paragraph.len() > MAX_TEXT {
+            bail!("DOCX text exceeds {MAX_TEXT} bytes");
+        }
+    }
+    Ok(passages)
+}

crates/searchi-core/src/lib.rs

@@ -1,21 +1,20 @@
 //! Offline lexical search engine shared by the CLI and desktop shell.
 //!
 //! Reconciliation is atomic. Incomplete directory walks never delete records.
-//! Text extraction is deliberately bounded; binary document extractors belong in
-//! isolated workers in the next milestone, not in this indexing transaction.
+//! Text and document extraction have byte limits, but still run in this process.
+//! Worker isolation remains a production milestone.
+mod extraction;
 use anyhow::{bail, Context, Result};
 use rusqlite::{params, Connection, OptionalExtension};
 use serde::Serialize;
 use std::{
     fs,
-    io::Read,
     path::{Path, PathBuf},
     time::UNIX_EPOCH,
 };
 use walkdir::WalkDir;
 
-const SCHEMA_VERSION: i64 = 2;
-const MAX_TEXT_BYTES: u64 = 2 * 1024 * 1024;
+const SCHEMA_VERSION: i64 = 4;
 const MAX_RESULTS: usize = 1000;
 
 #[derive(Debug, Default, Serialize)]
@@ -25,6 +24,7 @@ pub struct IndexReport {
     pub removed: usize,
     pub skipped: usize,
     pub unchanged: usize,
+    pub failures: usize,
 }
 
 /// Names of directories to prune anywhere below an explicitly chosen root.
@@ -60,6 +60,27 @@ pub struct SearchResult {
     pub snippet: String,
     pub score: f64,
     pub match_kind: String,
+    pub source: Option<SourceLocation>,
+    pub extraction_error: Option<String>,
+}
+
+/// A verified matching passage location; numbers are one based.
+#[derive(Debug, Clone, Serialize)]
+pub struct SourceLocation {
+    pub kind: String,
+    pub number: u32,
+}
+
+/// A bounded corpus page for downstream real embedding runtimes. IDs identify
+/// current index rows, not immutable content; consumers must validate content
+/// fingerprints across reconciliations before reusing persisted embeddings.
+#[derive(Debug, Serialize)]
+pub struct IndexedPassage {
+    pub id: i64,
+    pub path: String,
+    pub filename: String,
+    pub text: String,
+    pub source: SourceLocation,
 }
 
 #[derive(Debug, Serialize)]
@@ -103,7 +124,7 @@ impl SearchIndex {
                 id INTEGER PRIMARY KEY, path TEXT NOT NULL UNIQUE,
                 root TEXT NOT NULL, filename TEXT NOT NULL, content TEXT NOT NULL,
                 size INTEGER NOT NULL DEFAULT -1, mtime INTEGER NOT NULL DEFAULT -1,
-                extension TEXT NOT NULL DEFAULT '');
+                extension TEXT NOT NULL DEFAULT '', extraction_error TEXT);
             CREATE INDEX IF NOT EXISTS files_root ON files(root);
             CREATE VIRTUAL TABLE IF NOT EXISTS search USING fts5(filename, path, content,
                 content='files', content_rowid='id', tokenize='unicode61');
@@ -128,9 +149,21 @@ impl SearchIndex {
                 ALTER TABLE files ADD COLUMN extension TEXT NOT NULL DEFAULT '';
                 PRAGMA user_version=2;",
                 )?;
-            } else {
-                initialization.execute_batch("PRAGMA user_version=2;")?;
             }
+            if version > 0 && version < 4 {
+                initialization.execute("ALTER TABLE files ADD COLUMN extraction_error TEXT", [])?;
+            }
+            initialization.execute_batch(
+                "CREATE VIRTUAL TABLE IF NOT EXISTS passages USING fts5(
+                text, file_path UNINDEXED, kind UNINDEXED, number UNINDEXED, tokenize='unicode61');
+                CREATE TRIGGER IF NOT EXISTS files_passages_delete AFTER DELETE ON files BEGIN
+                    DELETE FROM passages WHERE file_path=old.path; END;",
+            )?;
+            if version != 4 {
+                // Re-extract records from older schemas to populate genuine locations.
+                initialization.execute("UPDATE files SET mtime=-1", [])?;
+            }
+            initialization.execute_batch("PRAGMA user_version=4;")?;
             initialization.commit()?;
         }
         Ok(Self {
@@ -218,17 +251,46 @@ impl SearchIndex {
                 report.unchanged += 1;
                 continue;
             }
-            let content = extract_text(entry.path()).with_context(|| format!("read {path}"))?;
-            if content.is_none() {
+            let (extracted, extraction_error) = match extraction::extract(entry.path()) {
+                Ok(extracted) => (extracted, None),
+                Err(error) if matches!(extension.as_str(), "pdf" | "docx") => {
+                    report.failures += 1;
+                    let reason: String = format!("{error:#}").chars().take(2048).collect();
+                    (None, Some(reason))
+                }
+                Err(error) => {
+                    return Err(error)
+                        .with_context(|| format!("read {path}; reconciliation rolled back"))
+                }
+            };
+            if extracted.is_none() {
                 report.skipped += 1;
             }
+            let extracted = extracted.unwrap_or_default();
+            let content = extracted
+                .iter()
+                .map(|passage| passage.text.as_str())
+                .collect::<Vec<_>>()
+                .join("\n");
             transaction.execute(
-                "INSERT INTO files(path,root,filename,content,size,mtime,extension) VALUES (?1,?2,?3,?4,?5,?6,?7)
+                "INSERT INTO files(path,root,filename,content,size,mtime,extension,extraction_error) VALUES (?1,?2,?3,?4,?5,?6,?7,?8)
                 ON CONFLICT(path) DO UPDATE SET root=excluded.root, filename=excluded.filename,
                 content=excluded.content, size=excluded.size, mtime=excluded.mtime,
-                extension=excluded.extension",
-                params![path, root_key, filename, content.unwrap_or_default(), size, mtime, extension],
+                extension=excluded.extension, extraction_error=excluded.extraction_error",
+                params![path, root_key, filename, content, size, mtime, extension, extraction_error],
             )?;
+            transaction.execute("DELETE FROM passages WHERE file_path=?1", [&path])?;
+            for passage in extracted {
+                transaction.execute(
+                    "INSERT INTO passages(text,file_path,kind,number) VALUES (?1,?2,?3,?4)",
+                    params![
+                        passage.text,
+                        path,
+                        passage.source.kind,
+                        passage.source.number
+                    ],
+                )?;
+            }
             report.indexed += 1;
         }
         report.removed = transaction.execute(
@@ -315,7 +377,7 @@ impl SearchIndex {
             FROM search WHERE search MATCH ?1)
             SELECT f.path, f.filename, COALESCE(h.excerpt, ''), COALESCE(h.rank, 0),
                 CASE WHEN lower(f.filename)=lower(?2) THEN 0
-                WHEN instr(lower(f.filename), lower(?2)) > 0 THEN 1 ELSE 2 END AS tier
+                WHEN instr(lower(f.filename), lower(?2)) > 0 THEN 1 ELSE 2 END AS tier, f.extraction_error
             FROM files f LEFT JOIN hits h ON h.id=f.id
             WHERE (h.id IS NOT NULL OR instr(lower(f.filename), lower(?2)) > 0)
             AND (?4 IS NULL OR substr(f.path,1,length(?4))=?4)
@@ -324,7 +386,7 @@ impl SearchIndex {
             AND (?8 IS NULL OR f.mtime/1000000>=?8) AND (?9 IS NULL OR f.mtime/1000000<=?9)
             ORDER BY tier, h.rank, f.path LIMIT ?3",
         )?;
-        let results = statement
+        let mut results = statement
             .query_map(
                 params![
                     fts,
@@ -350,13 +412,56 @@ impl SearchIndex {
                             _ => "content",
                         }
                         .to_owned(),
+                        source: None,
+                        extraction_error: row.get(5)?,
                     })
                 },
             )?
             .collect::<rusqlite::Result<Vec<_>>>()?;
+        let mut passages = self.connection.prepare("SELECT kind,number,snippet(passages,0,'','',' … ',24)
+            FROM passages WHERE passages MATCH ?1 AND file_path=?2 ORDER BY bm25(passages),rowid LIMIT 1")?;
+        for result in &mut results {
+            let passage: Option<(String, u32, String)> = passages
+                .query_row(params![fts, result.path], |row| {
+                    Ok((row.get(0)?, row.get(1)?, row.get(2)?))
+                })
+                .optional()?;
+            if let Some((kind, number, snippet)) = passage {
+                result.source = Some(SourceLocation { kind, number });
+                result.snippet = snippet;
+            }
+        }
         Ok(results)
     }
 
+    pub fn contains_path(&self, path: &str) -> Result<bool> {
+        Ok(self.connection.query_row(
+            "SELECT EXISTS(SELECT 1 FROM files WHERE path=?1)",
+            [path],
+            |row| row.get(0),
+        )?)
+    }
+
+    pub fn passage_batch(&self, after_id: i64, limit: usize) -> Result<Vec<IndexedPassage>> {
+        let mut statement = self.connection.prepare("SELECT p.rowid,f.path,f.filename,p.text,p.kind,p.number
+            FROM passages p JOIN files f ON f.path=p.file_path WHERE p.rowid>?1 ORDER BY p.rowid LIMIT ?2")?;
+        let passages = statement
+            .query_map(params![after_id, limit.min(256) as i64], |row| {
+                Ok(IndexedPassage {
+                    id: row.get(0)?,
+                    path: row.get(1)?,
+                    filename: row.get(2)?,
+                    text: row.get(3)?,
+                    source: SourceLocation {
+                        kind: row.get(4)?,
+                        number: row.get(5)?,
+                    },
+                })
+            })?
+            .collect::<rusqlite::Result<Vec<_>>>()?;
+        Ok(passages)
+    }
+
     pub fn stats(&self) -> Result<IndexStats> {
         Ok(self.connection.query_row(
             "SELECT count(*), count(NULLIF(content, '')) FROM files",
@@ -371,47 +476,6 @@ impl SearchIndex {
     }
 }
 
-fn extract_text(path: &Path) -> Result<Option<String>> {
-    let extension = path
-        .extension()
-        .and_then(|value| value.to_str())
-        .unwrap_or("")
-        .to_ascii_lowercase();
-    if !matches!(
-        extension.as_str(),
-        "txt"
-            | "md"
-            | "rs"
-            | "py"
-            | "js"
-            | "ts"
-            | "tsx"
-            | "jsx"
-            | "json"
-            | "toml"
-            | "yaml"
-            | "yml"
-            | "csv"
-            | "html"
-            | "css"
-            | "xml"
-            | "log"
-            | "sh"
-    ) {
-        return Ok(None);
-    }
-    let file = fs::File::open(path)?;
-    if file.metadata()?.len() > MAX_TEXT_BYTES {
-        return Ok(None);
-    }
-    let mut bytes = Vec::new();
-    file.take(MAX_TEXT_BYTES + 1).read_to_end(&mut bytes)?;
-    if bytes.len() as u64 > MAX_TEXT_BYTES || bytes.contains(&0) {
-        return Ok(None);
-    }
-    Ok(String::from_utf8(bytes).ok())
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;